Refactor WorkflowStatusLatestService to derive status from workflow runner tables and remove redundant status parameters. Integrate CI status optimization fallback in CiHandlerImpl.

Author: prakash100198

pkg/pipeline/CiHandler.go

@@ -49,6 +49,7 @@ import (
 	"github.com/devtron-labs/devtron/pkg/pipeline/executors"
 	"github.com/devtron-labs/devtron/pkg/pipeline/types"
 	"github.com/devtron-labs/devtron/pkg/resourceGroup"
+	workflowStatusLatest "github.com/devtron-labs/devtron/pkg/workflow/status"
 	"github.com/devtron-labs/devtron/util/rbac"
 	"github.com/go-pg/pg"
 	"go.uber.org/zap"
@@ -97,13 +98,15 @@ type CiHandlerImpl struct {
 	config                       *types.CiConfig
 	k8sCommonService             k8sPkg.K8sCommonService
 	workFlowStageStatusService   workflowStatus.WorkFlowStageStatusService
+	workflowStatusUpdateService  workflowStatusLatest.WorkflowStatusUpdateService
 }
 
 func NewCiHandlerImpl(Logger *zap.SugaredLogger, ciService CiService, ciPipelineMaterialRepository pipelineConfig.CiPipelineMaterialRepository, gitSensorClient gitSensor.Client, ciWorkflowRepository pipelineConfig.CiWorkflowRepository,
 	ciArtifactRepository repository.CiArtifactRepository, userService user.UserService, eventClient client.EventClient, eventFactory client.EventFactory, ciPipelineRepository pipelineConfig.CiPipelineRepository,
 	appListingRepository repository.AppListingRepository, cdPipelineRepository pipelineConfig.PipelineRepository, enforcerUtil rbac.EnforcerUtil, resourceGroupService resourceGroup.ResourceGroupService, envRepository repository2.EnvironmentRepository,
 	imageTaggingService imageTagging.ImageTaggingService, k8sCommonService k8sPkg.K8sCommonService, appWorkflowRepository appWorkflow.AppWorkflowRepository, customTagService CustomTagService,
 	workFlowStageStatusService workflowStatus.WorkFlowStageStatusService,
+	workflowStatusUpdateService workflowStatusLatest.WorkflowStatusUpdateService,
 ) *CiHandlerImpl {
 	cih := &CiHandlerImpl{
 		Logger:                       Logger,
@@ -126,6 +129,7 @@ func NewCiHandlerImpl(Logger *zap.SugaredLogger, ciService CiService, ciPipeline
 		appWorkflowRepository:        appWorkflowRepository,
 		k8sCommonService:             k8sCommonService,
 		workFlowStageStatusService:   workFlowStageStatusService,
+		workflowStatusUpdateService:  workflowStatusUpdateService,
 	}
 	config, err := types.GetCiConfig()
 	if err != nil {
@@ -644,10 +648,16 @@ func (impl *CiHandlerImpl) stateChanged(status string, podStatus string, msg str
 }
 
 func (impl *CiHandlerImpl) FetchCiStatusForTriggerViewV1(appId int) ([]*pipelineConfig.CiWorkflowStatus, error) {
-	ciWorkflowStatuses, err := impl.ciWorkflowRepository.FIndCiWorkflowStatusesByAppId(appId)
-	if err != nil && !util.IsErrNoRows(err) {
-		impl.Logger.Errorw("err in fetching ciWorkflowStatuses from ciWorkflowRepository", "appId", appId, "err", err)
-		return ciWorkflowStatuses, err
+	// Try to use the optimized workflow status latest service first
+	ciWorkflowStatuses, err := impl.workflowStatusUpdateService.FetchCiStatusForTriggerViewOptimized(appId)
+	if err != nil {
+		impl.Logger.Errorw("error in fetching ci status from optimized service, falling back to old method", "appId", appId, "err", err)
+		// Fallback to old method if optimized service fails
+		ciWorkflowStatuses, err = impl.ciWorkflowRepository.FIndCiWorkflowStatusesByAppId(appId)
+		if err != nil && !util.IsErrNoRows(err) {
+			impl.Logger.Errorw("err in fetching ciWorkflowStatuses from ciWorkflowRepository", "appId", appId, "err", err)
+			return ciWorkflowStatuses, err
+		}
 	}
 
 	return ciWorkflowStatuses, err

pkg/workflow/status/WorkflowStatusLatestService.go

@@ -164,7 +164,7 @@ func (impl *WorkflowStatusLatestServiceImpl) GetCiWorkflowStatusLatestByAppId(ap
 }
 
 // CD Workflow Status Latest methods implementation
-func (impl *WorkflowStatusLatestServiceImpl) SaveOrUpdateCdWorkflowStatusLatest(pipelineId, appId, environmentId, workflowRunnerId int, workflowType, status string, userId int32) error {
+func (impl *WorkflowStatusLatestServiceImpl) SaveOrUpdateCdWorkflowStatusLatest(pipelineId, appId, environmentId, workflowRunnerId int, workflowType string, userId int32) error {
 	// Check if entry exists
 	existingEntry, err := impl.workflowStatusLatestRepository.GetCdWorkflowStatusLatestByPipelineIdAndWorkflowType(pipelineId, workflowType)
 	if err != nil && err != pg.ErrNoRows {
@@ -181,7 +181,6 @@ func (impl *WorkflowStatusLatestServiceImpl) SaveOrUpdateCdWorkflowStatusLatest(
 			EnvironmentId:    environmentId,
 			WorkflowType:     workflowType,
 			WorkflowRunnerId: workflowRunnerId,
-			Status:           status,
 		}
 		model.CreatedBy = userId
 		model.CreatedOn = now
@@ -192,7 +191,6 @@ func (impl *WorkflowStatusLatestServiceImpl) SaveOrUpdateCdWorkflowStatusLatest(
 	} else {
 		// Update existing entry
 		existingEntry.WorkflowRunnerId = workflowRunnerId
-		existingEntry.Status = status
 		existingEntry.UpdatedBy = userId
 		existingEntry.UpdatedOn = now
 
@@ -211,13 +209,20 @@ func (impl *WorkflowStatusLatestServiceImpl) GetCdWorkflowStatusLatestByPipeline
 		return nil, err
 	}
 
+	// Get status from cd_workflow_runner table
+	cdWorkflowRunner, err := impl.cdWorkflowRepository.FindBasicWorkflowRunnerById(model.WorkflowRunnerId)
+	if err != nil {
+		impl.logger.Errorw("error in getting cd workflow runner", "err", err, "workflowRunnerId", model.WorkflowRunnerId)
+		return nil, err
+	}
+
 	return &CdWorkflowStatusLatest{
 		PipelineId:       model.PipelineId,
 		AppId:            model.AppId,
 		EnvironmentId:    model.EnvironmentId,
 		WorkflowType:     model.WorkflowType,
 		WorkflowRunnerId: model.WorkflowRunnerId,
-		Status:           model.Status,
+		Status:           cdWorkflowRunner.Status,
 	}, nil
 }
 
@@ -230,13 +235,20 @@ func (impl *WorkflowStatusLatestServiceImpl) GetCdWorkflowStatusLatestByAppId(ap
 
 	var result []*CdWorkflowStatusLatest
 	for _, model := range models {
+		// Get status from cd_workflow_runner table
+		cdWorkflowRunner, err := impl.cdWorkflowRepository.FindBasicWorkflowRunnerById(model.WorkflowRunnerId)
+		if err != nil {
+			impl.logger.Errorw("error in getting cd workflow runner", "err", err, "workflowRunnerId", model.WorkflowRunnerId)
+			continue // Skip this entry if we can't get the workflow runner
+		}
+
 		result = append(result, &CdWorkflowStatusLatest{
 			PipelineId:       model.PipelineId,
 			AppId:            model.AppId,
 			EnvironmentId:    model.EnvironmentId,
 			WorkflowType:     model.WorkflowType,
 			WorkflowRunnerId: model.WorkflowRunnerId,
-			Status:           model.Status,
+			Status:           cdWorkflowRunner.Status,
 		})
 	}
 
@@ -252,13 +264,20 @@ func (impl *WorkflowStatusLatestServiceImpl) GetCdWorkflowStatusLatestByPipeline
 
 	var result []*CdWorkflowStatusLatest
 	for _, model := range models {
+		// Get status from cd_workflow_runner table
+		cdWorkflowRunner, err := impl.cdWorkflowRepository.FindBasicWorkflowRunnerById(model.WorkflowRunnerId)
+		if err != nil {
+			impl.logger.Errorw("error in getting cd workflow runner", "err", err, "workflowRunnerId", model.WorkflowRunnerId)
+			continue // Skip this entry if we can't get the workflow runner
+		}
+
 		result = append(result, &CdWorkflowStatusLatest{
 			PipelineId:       model.PipelineId,
 			AppId:            model.AppId,
 			EnvironmentId:    model.EnvironmentId,
 			WorkflowType:     model.WorkflowType,
 			WorkflowRunnerId: model.WorkflowRunnerId,
-			Status:           model.Status,
+			Status:           cdWorkflowRunner.Status,
 		})
 	}
 

pkg/workflow/status/WorkflowStatusUpdateService.go

@@ -23,8 +23,8 @@ import (
 
 type WorkflowStatusUpdateService interface {
 	// Methods to update latest status tables when workflow status changes
-	UpdateCiWorkflowStatusLatest(pipelineId, appId, ciWorkflowId int, status string, userId int32) error
-	UpdateCdWorkflowStatusLatest(pipelineId, appId, environmentId, workflowRunnerId int, workflowType, status string, userId int32) error
+	UpdateCiWorkflowStatusLatest(pipelineId, appId, ciWorkflowId int, userId int32) error
+	UpdateCdWorkflowStatusLatest(pipelineId, appId, environmentId, workflowRunnerId int, workflowType string, userId int32) error
 
 	// Methods to fetch optimized status for trigger view
 	FetchCiStatusForTriggerViewOptimized(appId int) ([]*pipelineConfig.CiWorkflowStatus, error)
@@ -58,12 +58,12 @@ func NewWorkflowStatusUpdateServiceImpl(
 	}
 }
 
-func (impl *WorkflowStatusUpdateServiceImpl) UpdateCiWorkflowStatusLatest(pipelineId, appId, ciWorkflowId int, status string, userId int32) error {
-	return impl.workflowStatusLatestService.SaveOrUpdateCiWorkflowStatusLatest(pipelineId, appId, ciWorkflowId, status, userId)
+func (impl *WorkflowStatusUpdateServiceImpl) UpdateCiWorkflowStatusLatest(pipelineId, appId, ciWorkflowId int, userId int32) error {
+	return impl.workflowStatusLatestService.SaveOrUpdateCiWorkflowStatusLatest(pipelineId, appId, ciWorkflowId, userId)
 }
 
-func (impl *WorkflowStatusUpdateServiceImpl) UpdateCdWorkflowStatusLatest(pipelineId, appId, environmentId, workflowRunnerId int, workflowType, status string, userId int32) error {
-	return impl.workflowStatusLatestService.SaveOrUpdateCdWorkflowStatusLatest(pipelineId, appId, environmentId, workflowRunnerId, workflowType, status, userId)
+func (impl *WorkflowStatusUpdateServiceImpl) UpdateCdWorkflowStatusLatest(pipelineId, appId, environmentId, workflowRunnerId int, workflowType string, userId int32) error {
+	return impl.workflowStatusLatestService.SaveOrUpdateCdWorkflowStatusLatest(pipelineId, appId, environmentId, workflowRunnerId, workflowType, userId)
 }
 
 func (impl *WorkflowStatusUpdateServiceImpl) FetchCiStatusForTriggerViewOptimized(appId int) ([]*pipelineConfig.CiWorkflowStatus, error) {