Merge pull request #5990 from devtron-labs/force-abort-fix

fix: Force abort fix

Author: prakash100198

api/restHandler/app/pipeline/configure/BuildPipelineRestHandler.go

@@ -1562,12 +1562,16 @@ func (handler *PipelineConfigRestHandlerImpl) CancelWorkflow(w http.ResponseWrit
 		return
 	}
 	var forceAbort bool
-	forceAbort, err = strconv.ParseBool(queryVars.Get("forceAbort"))
-	if err != nil {
-		handler.Logger.Errorw("request err, CancelWorkflow", "err", err)
-		common.WriteJsonResp(w, err, nil, http.StatusBadRequest)
-		return
+	forceAbortQueryParam := queryVars.Get("forceAbort")
+	if len(forceAbortQueryParam) > 0 {
+		forceAbort, err = strconv.ParseBool(forceAbortQueryParam)
+		if err != nil {
+			handler.Logger.Errorw("request err, CancelWorkflow", "err", err)
+			common.WriteJsonResp(w, err, nil, http.StatusBadRequest)
+			return
+		}
 	}
+
 	handler.Logger.Infow("request payload, CancelWorkflow", "workflowId", workflowId, "pipelineId", pipelineId)
 
 	ciPipeline, err := handler.ciPipelineRepository.FindById(pipelineId)

api/restHandler/app/pipeline/configure/DeploymentPipelineRestHandler.go

@@ -2071,6 +2071,16 @@ func (handler *PipelineConfigRestHandlerImpl) CancelStage(w http.ResponseWriter,
 		common.WriteJsonResp(w, err, nil, http.StatusInternalServerError)
 		return
 	}
+	var forceAbort bool
+	forceAbortQueryParam := r.URL.Query().Get("forceAbort")
+	if len(forceAbortQueryParam) > 0 {
+		forceAbort, err = strconv.ParseBool(forceAbortQueryParam)
+		if err != nil {
+			handler.Logger.Errorw("request err, CancelWorkflow", "err", err)
+			common.WriteJsonResp(w, err, nil, http.StatusBadRequest)
+			return
+		}
+	}
 	handler.Logger.Infow("request payload, CancelStage", "pipelineId", pipelineId, "workflowRunnerId", workflowRunnerId)
 
 	//RBAC
@@ -2082,7 +2092,7 @@ func (handler *PipelineConfigRestHandlerImpl) CancelStage(w http.ResponseWriter,
 	}
 	//RBAC
 
-	resp, err := handler.cdHandler.CancelStage(workflowRunnerId, userId)
+	resp, err := handler.cdHandler.CancelStage(workflowRunnerId, forceAbort, userId)
 	if err != nil {
 		handler.Logger.Errorw("service err, CancelStage", "err", err, "pipelineId", pipelineId, "workflowRunnerId", workflowRunnerId)
 		if util.IsErrNoRows(err) {

pkg/pipeline/CdHandler.go

@@ -21,6 +21,7 @@ import (
 	"errors"
 	"fmt"
 	"github.com/devtron-labs/devtron/internal/sql/repository/pipelineConfig/adapter/cdWorkflow"
+	bean2 "github.com/devtron-labs/devtron/pkg/bean"
 	common2 "github.com/devtron-labs/devtron/pkg/deployment/common"
 	util2 "github.com/devtron-labs/devtron/pkg/pipeline/util"
 	"os"
@@ -64,7 +65,7 @@ type CdHandler interface {
 	FetchCdWorkflowDetails(appId int, environmentId int, pipelineId int, buildId int) (types.WorkflowResponse, error)
 	DownloadCdWorkflowArtifacts(buildId int) (*os.File, error)
 	FetchCdPrePostStageStatus(pipelineId int) ([]pipelineBean.CdWorkflowWithArtifact, error)
-	CancelStage(workflowRunnerId int, userId int32) (int, error)
+	CancelStage(workflowRunnerId int, forceAbort bool, userId int32) (int, error)
 	FetchAppWorkflowStatusForTriggerView(appId int) ([]*pipelineConfig.CdWorkflowStatus, error)
 	FetchAppWorkflowStatusForTriggerViewForEnvironment(request resourceGroup2.ResourceGroupingRequest, token string) ([]*pipelineConfig.CdWorkflowStatus, error)
 	FetchAppDeploymentStatusForEnvironments(request resourceGroup2.ResourceGroupingRequest, token string) ([]*pipelineConfig.AppDeploymentStatus, error)
@@ -133,16 +134,12 @@ func NewCdHandlerImpl(Logger *zap.SugaredLogger, userService user.UserService,
 	return cdh
 }
 
-func (impl *CdHandlerImpl) CancelStage(workflowRunnerId int, userId int32) (int, error) {
+func (impl *CdHandlerImpl) CancelStage(workflowRunnerId int, forceAbort bool, userId int32) (int, error) {
 	workflowRunner, err := impl.cdWorkflowRepository.FindWorkflowRunnerById(workflowRunnerId)
 	if err != nil {
 		impl.Logger.Errorw("err", "err", err)
 		return 0, err
 	}
-	if !(string(v1alpha1.NodePending) == workflowRunner.Status || string(v1alpha1.NodeRunning) == workflowRunner.Status) {
-		impl.Logger.Info("cannot cancel stage, stage not in progress")
-		return 0, errors.New("cannot cancel stage, stage not in progress")
-	}
 	pipeline, err := impl.pipelineRepository.FindById(workflowRunner.CdWorkflow.PipelineId)
 	if err != nil {
 		impl.Logger.Errorw("error while fetching cd pipeline", "err", err)
@@ -175,11 +172,35 @@ func (impl *CdHandlerImpl) CancelStage(workflowRunnerId int, userId int32) (int,
 		}
 	}
 	// Terminate workflow
-	err = impl.workflowService.TerminateWorkflow(workflowRunner.ExecutorType, workflowRunner.Name, workflowRunner.Namespace, restConfig, isExtCluster, nil)
-	if err != nil {
+	cancelWfDtoRequest := &types.CancelWfRequestDto{
+		ExecutorType: workflowRunner.ExecutorType,
+		WorkflowName: workflowRunner.Name,
+		Namespace:    workflowRunner.Namespace,
+		RestConfig:   restConfig,
+		IsExt:        isExtCluster,
+		Environment:  nil,
+	}
+	err = impl.workflowService.TerminateWorkflow(cancelWfDtoRequest)
+	if err != nil && forceAbort {
+		impl.Logger.Errorw("error in terminating workflow, with force abort flag as true", "workflowName", workflowRunner.Name, "err", err)
+		cancelWfDtoRequest.WorkflowGenerateName = fmt.Sprintf("%d-%s", workflowRunnerId, workflowRunner.Name)
+		err1 := impl.workflowService.TerminateDanglingWorkflows(cancelWfDtoRequest)
+		if err1 != nil {
+			impl.Logger.Errorw("error in terminating dangling workflows", "cancelWfDtoRequest", cancelWfDtoRequest, "err", err)
+			// ignoring error here in case of force abort, confirmed from product
+		}
+	} else if err != nil {
 		impl.Logger.Error("cannot terminate wf runner", "err", err)
 		return 0, err
 	}
+	if forceAbort {
+		err = impl.handleForceAbortCaseForCdStage(workflowRunner, forceAbort)
+		if err != nil {
+			impl.Logger.Errorw("error in handleForceAbortCaseForCdStage", "forceAbortFlag", forceAbort, "workflowRunner", workflowRunner, "err", err)
+			return 0, err
+		}
+		return workflowRunner.Id, nil
+	}
 	if len(workflowRunner.ImagePathReservationIds) > 0 {
 		err := impl.customTagService.DeactivateImagePathReservationByImageIds(workflowRunner.ImagePathReservationIds)
 		if err != nil {
@@ -198,6 +219,34 @@ func (impl *CdHandlerImpl) CancelStage(workflowRunnerId int, userId int32) (int,
 	return workflowRunner.Id, nil
 }
 
+func (impl *CdHandlerImpl) updateWorkflowRunnerForForceAbort(workflowRunner *pipelineConfig.CdWorkflowRunner) error {
+	workflowRunner.Status = executors.WorkflowCancel
+	workflowRunner.PodStatus = string(bean2.Failed)
+	workflowRunner.Message = FORCE_ABORT_MESSAGE_AFTER_STARTING_STAGE
+	err := impl.cdWorkflowRepository.UpdateWorkFlowRunner(workflowRunner)
+	if err != nil {
+		impl.Logger.Errorw("error in updating workflow status in cd workflow runner in force abort case", "err", err)
+		return err
+	}
+	return nil
+}
+
+func (impl *CdHandlerImpl) handleForceAbortCaseForCdStage(workflowRunner *pipelineConfig.CdWorkflowRunner, forceAbort bool) error {
+	isWorkflowInNonTerminalStage := workflowRunner.Status == string(v1alpha1.NodePending) || workflowRunner.Status == string(v1alpha1.NodeRunning)
+	if !isWorkflowInNonTerminalStage {
+		if forceAbort {
+			return impl.updateWorkflowRunnerForForceAbort(workflowRunner)
+		} else {
+			return &util.ApiError{Code: "200", HttpStatusCode: 400, UserMessage: "cannot cancel stage, stage not in progress"}
+		}
+	}
+	//this arises when someone deletes the workflow in resource browser and wants to force abort a cd stage(pre/post)
+	if workflowRunner.Status == string(v1alpha1.NodeRunning) && forceAbort {
+		return impl.updateWorkflowRunnerForForceAbort(workflowRunner)
+	}
+	return nil
+}
+
 func (impl *CdHandlerImpl) UpdateWorkflow(workflowStatus v1alpha1.WorkflowStatus) (int, string, error) {
 	wfStatusRs := impl.extractWorkfowStatus(workflowStatus)
 	workflowName, status, podStatus, message, podName := wfStatusRs.WorkflowName, wfStatusRs.Status, wfStatusRs.PodStatus, wfStatusRs.Message, wfStatusRs.PodName

pkg/pipeline/CiHandler.go

@@ -163,7 +163,7 @@ const Running = "Running"
 const Starting = "Starting"
 const POD_DELETED_MESSAGE = "pod deleted"
 const TERMINATE_MESSAGE = "workflow shutdown with strategy: Terminate"
-const ABORT_MESSAGE_AFTER_STARTING_STAGE = "workflow shutdown with strategy: Force Abort"
+const FORCE_ABORT_MESSAGE_AFTER_STARTING_STAGE = "workflow shutdown with strategy: Force Abort"
 
 func (impl *CiHandlerImpl) CheckAndReTriggerCI(workflowStatus v1alpha1.WorkflowStatus) error {
 
@@ -592,20 +592,9 @@ func (impl *CiHandlerImpl) GetBuildHistory(pipelineId int, appId int, offset int
 func (impl *CiHandlerImpl) CancelBuild(workflowId int, forceAbort bool) (int, error) {
 	workflow, err := impl.ciWorkflowRepository.FindById(workflowId)
 	if err != nil {
-		impl.Logger.Errorw("err", "err", err)
+		impl.Logger.Errorw("error in finding ci-workflow by workflow id", "ciWorkflowId", workflowId, "err", err)
 		return 0, err
 	}
-	if !(string(v1alpha1.NodePending) == workflow.Status || string(v1alpha1.NodeRunning) == workflow.Status) {
-		if forceAbort {
-			return impl.cancelBuildAfterStartWorkflowStage(workflow)
-		} else {
-			return 0, &util.ApiError{Code: "200", HttpStatusCode: 400, UserMessage: "cannot cancel build, build not in progress"}
-		}
-	}
-	//this arises when someone deletes the workflow in resource browser and wants to force abort a ci
-	if workflow.Status == string(v1alpha1.NodeRunning) && forceAbort {
-		return impl.cancelBuildAfterStartWorkflowStage(workflow)
-	}
 	isExt := workflow.Namespace != DefaultCiWorkflowNamespace
 	var env *repository3.Environment
 	var restConfig *rest.Config
@@ -615,15 +604,40 @@ func (impl *CiHandlerImpl) CancelBuild(workflowId int, forceAbort bool) (int, er
 			return 0, err
 		}
 	}
-
 	// Terminate workflow
-	err = impl.workflowService.TerminateWorkflow(workflow.ExecutorType, workflow.Name, workflow.Namespace, restConfig, isExt, env)
-	if err != nil && strings.Contains(err.Error(), "cannot find workflow") {
+	cancelWfDtoRequest := &types.CancelWfRequestDto{
+		ExecutorType: workflow.ExecutorType,
+		WorkflowName: workflow.Name,
+		Namespace:    workflow.Namespace,
+		RestConfig:   restConfig,
+		IsExt:        isExt,
+		Environment:  env,
+	}
+	// Terminate workflow
+	err = impl.workflowService.TerminateWorkflow(cancelWfDtoRequest)
+	if err != nil && forceAbort {
+		impl.Logger.Errorw("error in terminating workflow, with force abort flag flag as true", "workflowName", workflow.Name, "err", err)
+
+		cancelWfDtoRequest.WorkflowGenerateName = fmt.Sprintf("%d-%s", workflowId, workflow.Name)
+		err1 := impl.workflowService.TerminateDanglingWorkflows(cancelWfDtoRequest)
+		if err1 != nil {
+			impl.Logger.Errorw("error in terminating dangling workflows", "cancelWfDtoRequest", cancelWfDtoRequest, "err", err)
+			// ignoring error here in case of force abort, confirmed from product
+		}
+	} else if err != nil && strings.Contains(err.Error(), "cannot find workflow") {
 		return 0, &util.ApiError{Code: "200", HttpStatusCode: http.StatusBadRequest, UserMessage: err.Error()}
 	} else if err != nil {
 		impl.Logger.Errorw("cannot terminate wf", "err", err)
 		return 0, err
 	}
+	if forceAbort {
+		err = impl.handleForceAbortCaseForCi(workflow, forceAbort)
+		if err != nil {
+			impl.Logger.Errorw("error in handleForceAbortCaseForCi", "forceAbortFlag", forceAbort, "workflow", workflow, "err", err)
+			return 0, err
+		}
+		return workflow.Id, nil
+	}
 
 	workflow.Status = executors.WorkflowCancel
 	if workflow.ExecutorType == cdWorkflow.WORKFLOW_EXECUTOR_TYPE_SYSTEM {
@@ -652,16 +666,32 @@ func (impl *CiHandlerImpl) CancelBuild(workflowId int, forceAbort bool) (int, er
 	return workflow.Id, nil
 }
 
-func (impl *CiHandlerImpl) cancelBuildAfterStartWorkflowStage(workflow *pipelineConfig.CiWorkflow) (int, error) {
+func (impl *CiHandlerImpl) handleForceAbortCaseForCi(workflow *pipelineConfig.CiWorkflow, forceAbort bool) error {
+	isWorkflowInNonTerminalStage := workflow.Status == string(v1alpha1.NodePending) || workflow.Status == string(v1alpha1.NodeRunning)
+	if !isWorkflowInNonTerminalStage {
+		if forceAbort {
+			return impl.updateWorkflowForForceAbort(workflow)
+		} else {
+			return &util.ApiError{Code: "200", HttpStatusCode: 400, UserMessage: "cannot cancel build, build not in progress"}
+		}
+	}
+	//this arises when someone deletes the workflow in resource browser and wants to force abort a ci
+	if workflow.Status == string(v1alpha1.NodeRunning) && forceAbort {
+		return impl.updateWorkflowForForceAbort(workflow)
+	}
+	return nil
+}
+
+func (impl *CiHandlerImpl) updateWorkflowForForceAbort(workflow *pipelineConfig.CiWorkflow) error {
 	workflow.Status = executors.WorkflowCancel
 	workflow.PodStatus = string(bean.Failed)
-	workflow.Message = ABORT_MESSAGE_AFTER_STARTING_STAGE
+	workflow.Message = FORCE_ABORT_MESSAGE_AFTER_STARTING_STAGE
 	err := impl.ciWorkflowRepository.UpdateWorkFlow(workflow)
 	if err != nil {
 		impl.Logger.Errorw("error in updating workflow status", "err", err)
-		return 0, err
+		return err
 	}
-	return workflow.Id, nil
+	return nil
 }
 
 func (impl *CiHandlerImpl) getRestConfig(workflow *pipelineConfig.CiWorkflow) (*rest.Config, error) {

pkg/pipeline/WorkflowService.go

@@ -22,6 +22,7 @@ import (
 	"errors"
 	v1alpha12 "github.com/argoproj/argo-workflows/v3/pkg/client/clientset/versioned/typed/workflow/v1alpha1"
 	"github.com/argoproj/argo-workflows/v3/workflow/util"
+	"github.com/devtron-labs/common-lib/utils"
 	"github.com/devtron-labs/common-lib/utils/k8s"
 	"github.com/devtron-labs/common-lib/utils/k8s/commonBean"
 	"github.com/devtron-labs/devtron/api/bean"
@@ -39,6 +40,8 @@ import (
 	v12 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/client-go/rest"
+	"net/http"
+	"strconv"
 	"strings"
 )
 
@@ -51,7 +54,8 @@ type WorkflowService interface {
 	GetWorkflowStatus(executorType cdWorkflow.WorkflowExecutorType, name string, namespace string, restConfig *rest.Config) (*types.WorkflowStatus, error)
 	// ListAllWorkflows(namespace string) (*v1alpha1.WorkflowList, error)
 	// UpdateWorkflow(wf *v1alpha1.Workflow) (*v1alpha1.Workflow, error)
-	TerminateWorkflow(executorType cdWorkflow.WorkflowExecutorType, name string, namespace string, restConfig *rest.Config, isExt bool, environment *repository.Environment) error
+	TerminateWorkflow(cancelWfDtoRequest *types.CancelWfRequestDto) error
+	TerminateDanglingWorkflows(cancelWfDtoRequest *types.CancelWfRequestDto) error
 }
 
 type WorkflowServiceImpl struct {
@@ -157,12 +161,10 @@ func (impl *WorkflowServiceImpl) createWorkflowTemplate(workflowRequest *types.W
 	}
 
 	workflowMainContainer, err := workflowRequest.GetWorkflowMainContainer(impl.ciCdConfig, infraConfiguration, workflowJson, &workflowTemplate, workflowConfigMaps, workflowSecrets)
-
 	if err != nil {
 		impl.Logger.Errorw("error occurred while getting workflow main container", "err", err)
 		return bean3.WorkflowTemplate{}, err
 	}
-
 	workflowTemplate.Containers = []v12.Container{workflowMainContainer}
 	impl.updateBlobStorageConfig(workflowRequest, &workflowTemplate)
 	if workflowRequest.Type == bean3.CI_WORKFLOW_PIPELINE_TYPE || workflowRequest.Type == bean3.JOB_WORKFLOW_PIPELINE_TYPE {
@@ -352,27 +354,42 @@ func (impl *WorkflowServiceImpl) GetWorkflowStatus(executorType cdWorkflow.Workf
 	return wfStatus, err
 }
 
-func (impl *WorkflowServiceImpl) TerminateWorkflow(executorType cdWorkflow.WorkflowExecutorType, name string, namespace string, restConfig *rest.Config, isExt bool, environment *repository.Environment) error {
-	impl.Logger.Debugw("terminating wf", "name", name)
+func (impl *WorkflowServiceImpl) TerminateWorkflow(cancelWfDtoRequest *types.CancelWfRequestDto) error {
+	impl.Logger.Debugw("terminating wf", "name", cancelWfDtoRequest.WorkflowName)
 	var err error
-	if executorType != "" {
-		workflowExecutor := impl.getWorkflowExecutor(executorType)
+	if cancelWfDtoRequest.ExecutorType != "" {
+		workflowExecutor := impl.getWorkflowExecutor(cancelWfDtoRequest.ExecutorType)
 		if workflowExecutor == nil {
 			return errors.New("workflow executor not found")
 		}
-		if restConfig == nil {
-			restConfig = impl.config
+		if cancelWfDtoRequest.RestConfig == nil {
+			cancelWfDtoRequest.RestConfig = impl.config
 		}
-		err = workflowExecutor.TerminateWorkflow(name, namespace, restConfig)
+		err = workflowExecutor.TerminateWorkflow(cancelWfDtoRequest.WorkflowName, cancelWfDtoRequest.Namespace, cancelWfDtoRequest.RestConfig)
 	} else {
-		wfClient, err := impl.getWfClient(environment, namespace, isExt)
+		wfClient, err := impl.getWfClient(cancelWfDtoRequest.Environment, cancelWfDtoRequest.Namespace, cancelWfDtoRequest.IsExt)
 		if err != nil {
 			return err
 		}
-		err = util.TerminateWorkflow(context.Background(), wfClient, name)
+		err = util.TerminateWorkflow(context.Background(), wfClient, cancelWfDtoRequest.WorkflowName)
+	}
+	return err
+}
+
+func (impl *WorkflowServiceImpl) TerminateDanglingWorkflows(cancelWfDtoRequest *types.CancelWfRequestDto) error {
+	impl.Logger.Debugw("terminating dangling wf", "name", cancelWfDtoRequest.WorkflowName)
+	var err error
+	workflowExecutor := impl.getWorkflowExecutor(cancelWfDtoRequest.ExecutorType)
+	if workflowExecutor == nil {
+		return &utils.ApiError{HttpStatusCode: http.StatusNotFound, Code: strconv.Itoa(http.StatusNotFound), InternalMessage: "workflow executor not found", UserMessage: "workflow executor not found"}
 	}
+	if cancelWfDtoRequest.RestConfig == nil {
+		cancelWfDtoRequest.RestConfig = impl.config
+	}
+	err = workflowExecutor.TerminateDanglingWorkflow(cancelWfDtoRequest.WorkflowGenerateName, cancelWfDtoRequest.Namespace, cancelWfDtoRequest.RestConfig)
 	return err
 }
+
 func (impl *WorkflowServiceImpl) getRuntimeEnvClientInstance(environment *repository.Environment) (v1alpha12.WorkflowInterface, error) {
 	restConfig, err, _ := impl.k8sCommonService.GetRestConfigByClusterId(context.Background(), environment.ClusterId)
 	if err != nil {