Merge pull request #6320 from devtron-labs/build-worker-status

feat: separate build worker status timeline

Author: ayu-devtron

Wire.go

@@ -155,6 +155,8 @@ import (
 	repository3 "github.com/devtron-labs/devtron/pkg/pipeline/history/repository"
 	repository5 "github.com/devtron-labs/devtron/pkg/pipeline/repository"
 	"github.com/devtron-labs/devtron/pkg/pipeline/types"
+	"github.com/devtron-labs/devtron/pkg/pipeline/workflowStatus"
+	repository6 "github.com/devtron-labs/devtron/pkg/pipeline/workflowStatus/repository"
 	"github.com/devtron-labs/devtron/pkg/plugin"
 	"github.com/devtron-labs/devtron/pkg/policyGovernance"
 	resourceGroup2 "github.com/devtron-labs/devtron/pkg/resourceGroup"
@@ -471,6 +473,11 @@ func InitializeApp() (*App, error) {
 		pipeline.NewCiServiceImpl,
 		wire.Bind(new(pipeline.CiService), new(*pipeline.CiServiceImpl)),
 
+		workflowStatus.NewWorkflowStageFlowStatusServiceImpl,
+		wire.Bind(new(workflowStatus.WorkFlowStageStatusService), new(*workflowStatus.WorkFlowStageStatusServiceImpl)),
+		repository6.NewWorkflowStageRepositoryImpl,
+		wire.Bind(new(repository6.WorkflowStageRepository), new(*repository6.WorkflowStageRepositoryImpl)),
+
 		pipelineConfig.NewCiWorkflowRepositoryImpl,
 		wire.Bind(new(pipelineConfig.CiWorkflowRepository), new(*pipelineConfig.CiWorkflowRepositoryImpl)),
 

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

@@ -25,6 +25,7 @@ import (
 	"github.com/devtron-labs/devtron/internal/sql/constants"
 	"github.com/devtron-labs/devtron/pkg/build/artifacts/imageTagging"
 	bean2 "github.com/devtron-labs/devtron/pkg/build/pipeline/bean"
+	constants2 "github.com/devtron-labs/devtron/pkg/pipeline/constants"
 	"github.com/devtron-labs/devtron/util/stringsUtil"
 	"golang.org/x/exp/maps"
 	"io"
@@ -45,7 +46,6 @@ import (
 	"github.com/devtron-labs/devtron/internal/util"
 	"github.com/devtron-labs/devtron/pkg/auth/authorisation/casbin"
 	"github.com/devtron-labs/devtron/pkg/bean"
-	"github.com/devtron-labs/devtron/pkg/pipeline"
 	bean1 "github.com/devtron-labs/devtron/pkg/pipeline/bean"
 	"github.com/devtron-labs/devtron/pkg/pipeline/types"
 	resourceGroup "github.com/devtron-labs/devtron/pkg/resourceGroup"
@@ -663,7 +663,7 @@ func (handler *PipelineConfigRestHandlerImpl) validateCiTriggerRBAC(token string
 	// This is being done for jobs, jobs execute in default-env (devtron-ci) namespace by default. so considering DefaultCiNamespace as env for rbac enforcement
 	envName := ""
 	if triggerEnvironmentId == 0 {
-		envName = pipeline.DefaultCiWorkflowNamespace
+		envName = constants2.DefaultCiWorkflowNamespace
 	}
 	appObject := handler.enforcerUtil.GetAppRBACNameByAppId(ciPipeline.AppId)
 	workflowObject := handler.enforcerUtil.GetWorkflowRBACByCiPipelineId(ciPipelineId, workflowName)

env_gen.json

@@ -76,6 +76,7 @@
  | DEFAULT_TARGET_PLATFORM | string | |  |  | false |
  | DOCKER_BUILD_CACHE_PATH | string |/var/lib/docker |  |  | false |
  | ENABLE_BUILD_CONTEXT | bool |false |  |  | false |
+ | ENABLE_WORKFLOW_EXECUTION_STAGE | bool |true | if enabled then we will display build stages separately for CI/Job/Pre-Post CD | true | false |
  | EXTERNAL_BLOB_STORAGE_CM_NAME | string |blob-storage-cm |  |  | false |
  | EXTERNAL_BLOB_STORAGE_SECRET_NAME | string |blob-storage-secret |  |  | false |
  | EXTERNAL_CD_NODE_LABEL_SELECTOR |  | |  |  | false |

env_gen.md

@@ -40,8 +40,8 @@ type CdWorkflowRepository interface {
 	FindCdWorkflowMetaByEnvironmentId(appId int, environmentId int, offset int, size int) ([]CdWorkflowRunner, error)
 	FindCdWorkflowMetaByPipelineId(pipelineId int, offset int, size int) ([]CdWorkflowRunner, error)
 	FindArtifactByPipelineIdAndRunnerType(pipelineId int, runnerType apiBean.WorkflowType, limit int, runnerStatuses []string) ([]CdWorkflowRunner, error)
-	SaveWorkFlowRunner(wfr *CdWorkflowRunner) (*CdWorkflowRunner, error)
-	UpdateWorkFlowRunner(wfr *CdWorkflowRunner) error
+	SaveWorkFlowRunnerWithTx(wfr *CdWorkflowRunner, tx *pg.Tx) (*CdWorkflowRunner, error)
+	UpdateWorkFlowRunnerWithTx(wfr *CdWorkflowRunner, tx *pg.Tx) error
 	UpdateIsArtifactUploaded(wfrId int, isArtifactUploaded workflow.ArtifactUploadedType) error
 	GetPreviousQueuedRunners(cdWfrId, pipelineId int) ([]*CdWorkflowRunner, error)
 	UpdateRunnerStatusToFailedForIds(errMsg string, triggeredBy int32, cdWfrIds ...int) error
@@ -451,14 +451,14 @@ func (impl *CdWorkflowRepositoryImpl) FindLastPreOrPostTriggeredByEnvironmentId(
 	return wfr, err
 }
 
-func (impl *CdWorkflowRepositoryImpl) SaveWorkFlowRunner(wfr *CdWorkflowRunner) (*CdWorkflowRunner, error) {
-	err := impl.dbConnection.Insert(wfr)
+func (impl *CdWorkflowRepositoryImpl) SaveWorkFlowRunnerWithTx(wfr *CdWorkflowRunner, tx *pg.Tx) (*CdWorkflowRunner, error) {
+	err := tx.Insert(wfr)
 	return wfr, err
 }
 
-func (impl *CdWorkflowRepositoryImpl) UpdateWorkFlowRunner(wfr *CdWorkflowRunner) error {
+func (impl *CdWorkflowRepositoryImpl) UpdateWorkFlowRunnerWithTx(wfr *CdWorkflowRunner, tx *pg.Tx) error {
 	wfr.Message = util.GetTruncatedMessage(wfr.Message, 1000)
-	err := impl.dbConnection.Update(wfr)
+	err := tx.Update(wfr)
 	return err
 }
 

internal/sql/repository/pipelineConfig/CdWorfkflowRepository.go

@@ -26,9 +26,9 @@ import (
 )
 
 type CiWorkflowRepository interface {
-	SaveWorkFlow(wf *CiWorkflow) error
+	SaveWorkFlowWithTx(wf *CiWorkflow, tx *pg.Tx) error
 	FindLastTriggeredWorkflow(pipelineId int) (*CiWorkflow, error)
-	UpdateWorkFlow(wf *CiWorkflow) error
+	UpdateWorkFlowWithTx(wf *CiWorkflow, tx *pg.Tx) error
 	UpdateArtifactUploaded(id int, isUploaded workflow.ArtifactUploadedType) error
 	FindByStatusesIn(activeStatuses []string) ([]*CiWorkflow, error)
 	FindByPipelineId(pipelineId int, offset int, size int) ([]WorkflowWithArtifact, error)
@@ -263,13 +263,13 @@ func (impl *CiWorkflowRepositoryImpl) FindCiWorkflowGitTriggersByIds(ids []int)
 	return workflows, err
 }
 
-func (impl *CiWorkflowRepositoryImpl) SaveWorkFlow(wf *CiWorkflow) error {
-	err := impl.dbConnection.Insert(wf)
+func (impl *CiWorkflowRepositoryImpl) SaveWorkFlowWithTx(wf *CiWorkflow, tx *pg.Tx) error {
+	err := tx.Insert(wf)
 	return err
 }
 
-func (impl *CiWorkflowRepositoryImpl) UpdateWorkFlow(wf *CiWorkflow) error {
-	err := impl.dbConnection.Update(wf)
+func (impl *CiWorkflowRepositoryImpl) UpdateWorkFlowWithTx(wf *CiWorkflow, tx *pg.Tx) error {
+	err := tx.Update(wf)
 	return err
 }
 

internal/sql/repository/pipelineConfig/CiWorkflowRepository.go

@@ -6,7 +6,7 @@ import (
 	"github.com/devtron-labs/devtron/client/argocdServer/bean"
 )
 
-var WfrTerminalStatusList = []string{WorkflowAborted, WorkflowFailed, WorkflowSucceeded, bean.HIBERNATING, string(health.HealthStatusHealthy), string(health.HealthStatusDegraded)}
+var WfrTerminalStatusList = []string{WorkflowAborted, WorkflowFailed, WorkflowSucceeded, bean.HIBERNATING, string(health.HealthStatusHealthy), string(health.HealthStatusDegraded), WorkflowTimedOut, WorkflowCancel}
 
 type WorkflowStatus int
 
@@ -34,6 +34,7 @@ const (
 	WorkflowTypeDeploy         = "DEPLOY"
 	WorkflowTypePre            = "PRE"
 	WorkflowTypePost           = "POST"
+	WorkflowWaitingToStart     = "WaitingToStart"
 )
 
 func (a WorkflowStatus) String() string {
@@ -60,3 +61,6 @@ type CdWorkflowRunnerArtifactMetadata struct {
 	ParentCiArtifact int  `pg:"parent_ci_artifact"`
 	Scanned          bool `pg:"scanned"`
 }
+
+const WorkflowCancel = "CANCELLED"
+const POD_DELETED_MESSAGE = "pod deleted"

internal/sql/repository/pipelineConfig/bean/workflow/cdWorkflow/CdWorkflowBean.go

@@ -38,6 +38,7 @@ import (
 	bean6 "github.com/devtron-labs/devtron/pkg/deployment/manifest/deploymentTemplate/bean"
 	"github.com/devtron-labs/devtron/pkg/deployment/manifest/deploymentTemplate/read"
 	bean4 "github.com/devtron-labs/devtron/pkg/deployment/trigger/devtronApps/bean"
+	"github.com/devtron-labs/devtron/pkg/workflow/cd"
 	"io/ioutil"
 	"net/url"
 	"path"
@@ -125,6 +126,7 @@ type AppServiceImpl struct {
 	appListingService                      AppListingService
 	deploymentConfigService                common2.DeploymentConfigService
 	envConfigOverrideReadService           read.EnvConfigOverrideService
+	cdWorkflowRunnerService                cd.CdWorkflowRunnerService
 }
 
 type AppService interface {
@@ -164,7 +166,8 @@ func NewAppService(
 	deploymentTemplateService deploymentTemplate.DeploymentTemplateService,
 	appListingService AppListingService,
 	deploymentConfigService common2.DeploymentConfigService,
-	envConfigOverrideReadService read.EnvConfigOverrideService) *AppServiceImpl {
+	envConfigOverrideReadService read.EnvConfigOverrideService,
+	cdWorkflowRunnerService cd.CdWorkflowRunnerService) *AppServiceImpl {
 	appServiceImpl := &AppServiceImpl{
 		mergeUtil:                              mergeUtil,
 		pipelineOverrideRepository:             pipelineOverrideRepository,
@@ -194,6 +197,7 @@ func NewAppService(
 		appListingService:                      appListingService,
 		deploymentConfigService:                deploymentConfigService,
 		envConfigOverrideReadService:           envConfigOverrideReadService,
+		cdWorkflowRunnerService:                cdWorkflowRunnerService,
 	}
 	return appServiceImpl
 }
@@ -1005,7 +1009,7 @@ func (impl *AppServiceImpl) UpdateCdWorkflowRunnerByACDObject(app *v1alpha1.Appl
 	}
 	wfr.UpdatedBy = 1
 	wfr.UpdatedOn = time.Now()
-	err = impl.cdWorkflowRepository.UpdateWorkFlowRunner(wfr)
+	err = impl.cdWorkflowRunnerService.UpdateCdWorkflowRunnerWithStage(wfr)
 	if err != nil {
 		impl.logger.Errorw("error on update cd workflow runner", "wfr", wfr, "app", app, "err", err)
 		return err

pkg/app/AppService.go

@@ -119,7 +119,7 @@ func (impl *TriggerServiceImpl) TriggerPostStage(request bean.TriggerRequest) (*
 		runner.Status = cdWorkflow.WorkflowFailed
 		runner.Message = err.Error()
 		runner.FinishedOn = time.Now()
-		_ = impl.cdWorkflowRepository.UpdateWorkFlowRunner(runner)
+		_ = impl.cdWorkflowRunnerService.UpdateCdWorkflowRunnerWithStage(runner)
 		return nil, err
 	}
 
@@ -129,7 +129,7 @@ func (impl *TriggerServiceImpl) TriggerPostStage(request bean.TriggerRequest) (*
 		runner.Status = cdWorkflow.WorkflowFailed
 		runner.Message = err.Error()
 		runner.FinishedOn = time.Now()
-		_ = impl.cdWorkflowRepository.UpdateWorkFlowRunner(runner)
+		_ = impl.cdWorkflowRunnerService.UpdateCdWorkflowRunnerWithStage(runner)
 		return nil, err
 	}
 	manifestPushTempate, err := impl.getManifestPushTemplateForPostStage(request, envDevploymentConfig, jobHelmPackagePath, cdStageWorkflowRequest, cdWf, runner, pipeline, triggeredBy, triggeredAt)
@@ -143,7 +143,7 @@ func (impl *TriggerServiceImpl) TriggerPostStage(request bean.TriggerRequest) (*
 		return nil, err
 	}
 	wfr.ImagePathReservationIds = pluginImagePathReservationIds
-	err = impl.cdWorkflowRepository.UpdateWorkFlowRunner(&wfr)
+	err = impl.cdWorkflowRunnerService.UpdateCdWorkflowRunnerWithStage(&wfr)
 	if err != nil {
 		impl.logger.Error("error in updating image path reservation ids in cd workflow runner", "err", "err")
 	}

pkg/deployment/trigger/devtronApps/PostStageTriggerService.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"github.com/argoproj/argo-workflows/v3/pkg/apis/workflow/v1alpha1"
 	blob_storage "github.com/devtron-labs/common-lib/blob-storage"
 	commonBean "github.com/devtron-labs/common-lib/workflow"
 	bean2 "github.com/devtron-labs/devtron/api/bean"
@@ -135,11 +136,11 @@ func (impl *TriggerServiceImpl) TriggerPreStage(request bean.TriggerRequest) (*b
 		runner.Status = cdWorkflow.WorkflowFailed
 		runner.Message = err.Error()
 		runner.FinishedOn = time.Now()
-		_ = impl.cdWorkflowRepository.UpdateWorkFlowRunner(runner)
+		_ = impl.cdWorkflowRunnerService.UpdateCdWorkflowRunnerWithStage(runner)
 		return nil, err
 	} else {
 		runner.ImagePathReservationIds = imagePathReservationIds
-		_ = impl.cdWorkflowRepository.UpdateWorkFlowRunner(runner)
+		_ = impl.cdWorkflowRunnerService.UpdateCdWorkflowRunnerWithStage(runner)
 	}
 
 	_, span = otel.Tracer("orchestrator").Start(ctx, "cdWorkflowService.SubmitWorkflow")
@@ -152,7 +153,7 @@ func (impl *TriggerServiceImpl) TriggerPreStage(request bean.TriggerRequest) (*b
 		runner.Status = cdWorkflow.WorkflowFailed
 		runner.Message = err.Error()
 		runner.FinishedOn = time.Now()
-		_ = impl.cdWorkflowRepository.UpdateWorkFlowRunner(runner)
+		_ = impl.cdWorkflowRunnerService.UpdateCdWorkflowRunnerWithStage(runner)
 		return nil, err
 	}
 	manifestPushTemplate, err := impl.getManifestPushTemplateForPreStage(ctx, envDeploymentConfig, pipeline, artifact, jobHelmPackagePath, cdWf, runner, triggeredBy, triggeredAt, request)
@@ -251,6 +252,7 @@ func (impl *TriggerServiceImpl) createStartingWfAndRunner(request bean.TriggerRe
 		WorkflowType:          request.WorkflowType,
 		ExecutorType:          impl.config.GetWorkflowExecutorType(),
 		Status:                cdWorkflow.WorkflowStarting, // starting PreStage
+		PodStatus:             string(v1alpha1.NodePending),
 		TriggeredBy:           triggeredBy,
 		StartedOn:             triggeredAt,
 		Namespace:             request.RunStageInEnvNamespace,
@@ -262,7 +264,7 @@ func (impl *TriggerServiceImpl) createStartingWfAndRunner(request bean.TriggerRe
 		ReferenceId:           request.TriggerContext.ReferenceId,
 	}
 	_, span := otel.Tracer("orchestrator").Start(ctx, "cdWorkflowRepository.SaveWorkFlowRunner")
-	_, err = impl.cdWorkflowRepository.SaveWorkFlowRunner(runner)
+	_, err = impl.cdWorkflowRunnerService.SaveCDWorkflowRunnerWithStage(runner)
 	span.End()
 	if err != nil {
 		return nil, nil, err
@@ -311,7 +313,7 @@ func (impl *TriggerServiceImpl) checkVulnerabilityStatusAndFailWfIfNeeded(ctx co
 		runner.FinishedOn = time.Now()
 		runner.UpdatedOn = time.Now()
 		runner.UpdatedBy = triggeredBy
-		err = impl.cdWorkflowRepository.UpdateWorkFlowRunner(runner)
+		err = impl.cdWorkflowRunnerService.UpdateCdWorkflowRunnerWithStage(runner)
 		if err != nil {
 			impl.logger.Errorw("error in updating wfr status due to vulnerable image", "err", err)
 			return err

pkg/deployment/trigger/devtronApps/PreStageTriggerService.go

@@ -172,6 +172,7 @@ type TriggerServiceImpl struct {
 	gitOperationService                 git.GitOperationService
 	attributeService                    attributes.AttributesService
 	clusterRepository                   repository5.ClusterRepository
+	cdWorkflowRunnerService             cd.CdWorkflowRunnerService
 }
 
 func NewTriggerServiceImpl(logger *zap.SugaredLogger,
@@ -230,6 +231,7 @@ func NewTriggerServiceImpl(logger *zap.SugaredLogger,
 	gitOperationService git.GitOperationService,
 	attributeService attributes.AttributesService,
 	clusterRepository repository5.ClusterRepository,
+	cdWorkflowRunnerService cd.CdWorkflowRunnerService,
 ) (*TriggerServiceImpl, error) {
 	impl := &TriggerServiceImpl{
 		logger:                              logger,
@@ -292,6 +294,7 @@ func NewTriggerServiceImpl(logger *zap.SugaredLogger,
 		ciCdPipelineOrchestrator:    ciCdPipelineOrchestrator,
 		gitOperationService:         gitOperationService,
 		attributeService:            attributeService,
+		cdWorkflowRunnerService:     cdWorkflowRunnerService,
 
 		clusterRepository: clusterRepository,
 	}
@@ -518,7 +521,7 @@ func (impl *TriggerServiceImpl) ManualCdTrigger(triggerContext bean.TriggerConte
 			AuditLog:     sql.AuditLog{CreatedOn: triggeredAt, CreatedBy: overrideRequest.UserId, UpdatedOn: triggeredAt, UpdatedBy: overrideRequest.UserId},
 			ReferenceId:  triggerContext.ReferenceId,
 		}
-		savedWfr, err := impl.cdWorkflowRepository.SaveWorkFlowRunner(runner)
+		savedWfr, err := impl.cdWorkflowRunnerService.SaveCDWorkflowRunnerWithStage(runner)
 		if err != nil {
 			impl.logger.Errorw("err in creating cdWorkflowRunner, ManualCdTrigger", "cdWorkflowId", cdWorkflowId, "err", err)
 			return 0, "", nil, err
@@ -649,7 +652,7 @@ func (impl *TriggerServiceImpl) TriggerAutomaticDeployment(request bean.TriggerR
 		AuditLog:     sql.AuditLog{CreatedOn: triggeredAt, CreatedBy: triggeredBy, UpdatedOn: triggeredAt, UpdatedBy: triggeredBy},
 		ReferenceId:  request.TriggerContext.ReferenceId,
 	}
-	savedWfr, err := impl.cdWorkflowRepository.SaveWorkFlowRunner(runner)
+	savedWfr, err := impl.cdWorkflowRunnerService.SaveCDWorkflowRunnerWithStage(runner)
 	if err != nil {
 		return err
 	}