Add database schema and repository methods for auto-abort previous builds feature

Co-authored-by: prakarsh-dt <[email protected]>

Author: Copilot

internal/sql/repository/pipelineConfig/CiPipelineRepository.go

@@ -50,6 +50,7 @@ type CiPipeline struct {
 	ScanEnabled              bool   `sql:"scan_enabled,notnull"`
 	IsDockerConfigOverridden bool   `sql:"is_docker_config_overridden, notnull"`
 	PipelineType             string `sql:"ci_pipeline_type"`
+	AutoAbortPreviousBuilds  bool   `sql:"auto_abort_previous_builds,notnull"`
 	sql.AuditLog
 	CiPipelineMaterials []*CiPipelineMaterial
 	CiTemplate          *CiTemplate

internal/sql/repository/pipelineConfig/CiWorkflowRepository.go

@@ -38,6 +38,7 @@ type CiWorkflowRepository interface {
 	FindCiWorkflowGitTriggersById(id int) (workflow *CiWorkflow, err error)
 	FindCiWorkflowGitTriggersByIds(ids []int) ([]*CiWorkflow, error)
 	FindByName(name string) (*CiWorkflow, error)
+	FindRunningWorkflowsForPipeline(pipelineId int) ([]*CiWorkflow, error)
 
 	FindLastTriggeredWorkflowByCiIds(pipelineId []int) (ciWorkflow []*CiWorkflow, err error)
 	FindWorkflowsByCiWorkflowIds(ciWorkflowIds []int) (ciWorkflow []*CiWorkflow, err error)
@@ -187,6 +188,20 @@ func (impl *CiWorkflowRepositoryImpl) FindByStatusesIn(activeStatuses []string)
 	return ciWorkFlows, err
 }
 
+func (impl *CiWorkflowRepositoryImpl) FindRunningWorkflowsForPipeline(pipelineId int) ([]*CiWorkflow, error) {
+	var ciWorkFlows []*CiWorkflow
+	// Status values for running/pending workflows that can be aborted
+	runningStatuses := []string{"Running", "Starting", "Pending"}
+	
+	err := impl.dbConnection.Model(&ciWorkFlows).
+		Column("ci_workflow.*", "CiPipeline").
+		Where("ci_workflow.ci_pipeline_id = ?", pipelineId).
+		Where("ci_workflow.status in (?)", pg.In(runningStatuses)).
+		Order("ci_workflow.started_on DESC").
+		Select()
+	return ciWorkFlows, err
+}
+
 // FindByPipelineId gets only those workflowWithArtifact whose parent_ci_workflow_id is null, this is done to accommodate multiple ci_artifacts through a single workflow(parent), making child workflows for other ci_artifacts (this has been done due to design understanding and db constraint) single workflow single ci-artifact
 func (impl *CiWorkflowRepositoryImpl) FindByPipelineId(pipelineId int, offset int, limit int) ([]WorkflowWithArtifact, error) {
 	var wfs []WorkflowWithArtifact

pkg/build/trigger/HandlerService.go

@@ -84,6 +84,7 @@ type HandlerService interface {
 	GetRunningWorkflowLogs(workflowId int, followLogs bool) (*bufio.Reader, func() error, error)
 	GetHistoricBuildLogs(workflowId int, ciWorkflow *pipelineConfig.CiWorkflow) (map[string]string, error)
 	DownloadCiWorkflowArtifacts(pipelineId int, buildId int) (*os.File, error)
+	abortPreviousRunningBuilds(pipelineId int, triggeredBy int32) error
 }
 
 // CATEGORY=CI_BUILDX
@@ -707,6 +708,13 @@ func (impl *HandlerServiceImpl) triggerCiPipeline(trigger *types.CiTriggerReques
 		return 0, err
 	}
 
+	// Check if auto-abort is enabled for this pipeline and abort previous builds if needed
+	err = impl.abortPreviousRunningBuilds(trigger.PipelineId, trigger.TriggeredBy)
+	if err != nil {
+		impl.Logger.Errorw("error in aborting previous running builds", "pipelineId", trigger.PipelineId, "err", err)
+		// Log error but don't fail the trigger - previous builds aborting is a best-effort operation
+	}
+
 	err = impl.executeCiPipeline(workflowRequest)
 	if err != nil {
 		impl.Logger.Errorw("error in executing ci pipeline", "err", err)
@@ -2065,3 +2073,84 @@ func (impl *HandlerServiceImpl) DownloadCiWorkflowArtifacts(pipelineId int, buil
 	impl.Logger.Infow("Downloaded ", "filename", file.Name(), "bytes", numBytes)
 	return file, nil
 }
+
+// abortPreviousRunningBuilds checks if auto-abort is enabled for the pipeline and aborts previous running builds
+func (impl *HandlerServiceImpl) abortPreviousRunningBuilds(pipelineId int, triggeredBy int32) error {
+	// Get pipeline configuration to check if auto-abort is enabled
+	ciPipeline, err := impl.ciPipelineRepository.FindById(pipelineId)
+	if err != nil {
+		impl.Logger.Errorw("error in finding ci pipeline", "pipelineId", pipelineId, "err", err)
+		return err
+	}
+
+	// Check if auto-abort is enabled for this pipeline
+	if !ciPipeline.AutoAbortPreviousBuilds {
+		impl.Logger.Debugw("auto-abort not enabled for pipeline", "pipelineId", pipelineId)
+		return nil
+	}
+
+	// Find all running/pending workflows for this pipeline
+	runningWorkflows, err := impl.ciWorkflowRepository.FindRunningWorkflowsForPipeline(pipelineId)
+	if err != nil {
+		impl.Logger.Errorw("error in finding running workflows for pipeline", "pipelineId", pipelineId, "err", err)
+		return err
+	}
+
+	if len(runningWorkflows) == 0 {
+		impl.Logger.Debugw("no running workflows found to abort for pipeline", "pipelineId", pipelineId)
+		return nil
+	}
+
+	impl.Logger.Infow("found running workflows to abort due to auto-abort configuration", 
+		"pipelineId", pipelineId, "workflowCount", len(runningWorkflows), "triggeredBy", triggeredBy)
+
+	// Abort each running workflow
+	for _, workflow := range runningWorkflows {
+		// Check if the workflow is in a critical phase that should not be aborted
+		if impl.isWorkflowInCriticalPhase(workflow) {
+			impl.Logger.Infow("skipping abort of workflow in critical phase", 
+				"workflowId", workflow.Id, "status", workflow.Status, "pipelineId", pipelineId)
+			continue
+		}
+
+		// Attempt to cancel the build
+		_, err := impl.CancelBuild(workflow.Id, false)
+		if err != nil {
+			impl.Logger.Errorw("error aborting previous running build", 
+				"workflowId", workflow.Id, "pipelineId", pipelineId, "err", err)
+			// Continue with other workflows even if one fails
+			continue
+		}
+
+		impl.Logger.Infow("successfully aborted previous running build due to auto-abort", 
+			"workflowId", workflow.Id, "pipelineId", pipelineId, "abortedBy", triggeredBy)
+	}
+
+	return nil
+}
+
+// isWorkflowInCriticalPhase determines if a workflow is in a critical phase and should not be aborted
+// This protects builds that are in the final stages like pushing cache or artifacts
+func (impl *HandlerServiceImpl) isWorkflowInCriticalPhase(workflow *pipelineConfig.CiWorkflow) bool {
+	// For now, we consider "Starting" as safe to abort, but "Running" needs more careful consideration
+	// In the future, this could be extended to check actual workflow steps/stages
+	
+	// If workflow has been running for less than 2 minutes, it's likely still in setup phase
+	if workflow.Status == "Running" && workflow.StartedOn.IsZero() == false {
+		runningDuration := time.Since(workflow.StartedOn)
+		if runningDuration < 2*time.Minute {
+			impl.Logger.Debugw("workflow is in early running phase, safe to abort", 
+				"workflowId", workflow.Id, "runningDuration", runningDuration.String())
+			return false
+		}
+		
+		// For workflows running longer, we should be more cautious
+		// This could be extended to check actual workflow phases using workflow service APIs
+		impl.Logger.Debugw("workflow has been running for a while, considering as critical phase", 
+			"workflowId", workflow.Id, "runningDuration", runningDuration.String())
+		return true
+	}
+	
+	// "Starting" and "Pending" are generally safe to abort
+	return false
+}

pkg/build/trigger/HandlerService_test.go

@@ -0,0 +1,102 @@
+/*
+ * Copyright (c) 2024. Devtron Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package trigger
+
+import (
+	"testing"
+	"time"
+
+	"github.com/devtron-labs/devtron/internal/sql/repository/pipelineConfig"
+	"github.com/devtron-labs/devtron/pkg/pipeline/types"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestHandlerServiceImpl_isWorkflowInCriticalPhase(t *testing.T) {
+	// Create a handler service instance for testing
+	handlerService := &HandlerServiceImpl{}
+
+	t.Run("Starting workflow should not be in critical phase", func(t *testing.T) {
+		workflow := &pipelineConfig.CiWorkflow{
+			Id:        1,
+			Status:    "Starting",
+			StartedOn: time.Now(),
+		}
+		result := handlerService.isWorkflowInCriticalPhase(workflow)
+		assert.False(t, result, "Starting workflow should not be in critical phase")
+	})
+
+	t.Run("Pending workflow should not be in critical phase", func(t *testing.T) {
+		workflow := &pipelineConfig.CiWorkflow{
+			Id:        2,
+			Status:    "Pending",
+			StartedOn: time.Now(),
+		}
+		result := handlerService.isWorkflowInCriticalPhase(workflow)
+		assert.False(t, result, "Pending workflow should not be in critical phase")
+	})
+
+	t.Run("Recently started Running workflow should not be in critical phase", func(t *testing.T) {
+		workflow := &pipelineConfig.CiWorkflow{
+			Id:        3,
+			Status:    "Running",
+			StartedOn: time.Now().Add(-1 * time.Minute), // Started 1 minute ago
+		}
+		result := handlerService.isWorkflowInCriticalPhase(workflow)
+		assert.False(t, result, "Recently started Running workflow should not be in critical phase")
+	})
+
+	t.Run("Long running workflow should be in critical phase", func(t *testing.T) {
+		workflow := &pipelineConfig.CiWorkflow{
+			Id:        4,
+			Status:    "Running",
+			StartedOn: time.Now().Add(-5 * time.Minute), // Started 5 minutes ago
+		}
+		result := handlerService.isWorkflowInCriticalPhase(workflow)
+		assert.True(t, result, "Long running workflow should be in critical phase")
+	})
+
+	t.Run("Running workflow with zero StartedOn should not be in critical phase", func(t *testing.T) {
+		workflow := &pipelineConfig.CiWorkflow{
+			Id:        5,
+			Status:    "Running",
+			StartedOn: time.Time{}, // Zero time
+		}
+		result := handlerService.isWorkflowInCriticalPhase(workflow)
+		assert.False(t, result, "Running workflow with zero StartedOn should not be in critical phase")
+	})
+}
+
+func TestCiTriggerRequest_HasPipelineId(t *testing.T) {
+	t.Run("CiTriggerRequest should contain PipelineId", func(t *testing.T) {
+		triggerRequest := &types.CiTriggerRequest{
+			PipelineId:  123,
+			TriggeredBy: 1,
+		}
+		assert.Equal(t, 123, triggerRequest.PipelineId, "CiTriggerRequest should have PipelineId field")
+		assert.Equal(t, int32(1), triggerRequest.TriggeredBy, "CiTriggerRequest should have TriggeredBy field")
+	})
+}
+
+func TestCiPipeline_HasAutoAbortField(t *testing.T) {
+	t.Run("CiPipeline should have AutoAbortPreviousBuilds field", func(t *testing.T) {
+		pipeline := &pipelineConfig.CiPipeline{
+			Id:                      1,
+			AutoAbortPreviousBuilds: true,
+		}
+		assert.True(t, pipeline.AutoAbortPreviousBuilds, "CiPipeline should have AutoAbortPreviousBuilds field")
+	})
+}

scripts/sql/34504200_ci_pipeline_auto_abort_config.down.sql

@@ -0,0 +1,3 @@
+-- Remove auto-abort configuration from ci_pipeline
+DROP INDEX IF EXISTS idx_ci_pipeline_auto_abort;
+ALTER TABLE ci_pipeline DROP COLUMN IF EXISTS auto_abort_previous_builds;

scripts/sql/34504200_ci_pipeline_auto_abort_config.up.sql

@@ -0,0 +1,8 @@
+-- Add configuration for auto-abort previous builds feature
+ALTER TABLE ci_pipeline 
+ADD COLUMN IF NOT EXISTS auto_abort_previous_builds BOOLEAN DEFAULT FALSE;
+
+-- Add index for performance when querying by pipeline id and auto abort setting
+CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_ci_pipeline_auto_abort 
+ON ci_pipeline (id, auto_abort_previous_builds) 
+WHERE auto_abort_previous_builds = TRUE;