Merge remote-tracking branch 'origin/optimize-ci-cd-workflow' into optimize-ci-cd-workflow

Author: prakash100198

internal/sql/repository/pipelineConfig/CdWorfkflowRepository.go

@@ -19,7 +19,6 @@ package pipelineConfig
 import (
 	"context"
 	"errors"
-	"fmt"
 	apiBean "github.com/devtron-labs/devtron/api/bean"
 	"github.com/devtron-labs/devtron/internal/sql/repository"
 	"github.com/devtron-labs/devtron/internal/sql/repository/pipelineConfig/bean/workflow"
@@ -29,7 +28,6 @@ import (
 	"github.com/go-pg/pg"
 	"go.opentelemetry.io/otel"
 	"go.uber.org/zap"
-	"strings"
 	"time"
 )
 
@@ -71,7 +69,7 @@ type CdWorkflowRepository interface {
 	IsLatestWf(pipelineId int, wfId int) (bool, error)
 	FindLatestCdWorkflowByPipelineId(pipelineIds []int) (*CdWorkflow, error)
 	FindLatestCdWorkflowByPipelineIdV2(pipelineIds []int) ([]*CdWorkflow, error)
-	FetchAllCdStagesLatestEntity(pipelineWorkflowPairs map[int][]apiBean.WorkflowType) ([]*CdWorkflowStatus, error)
+	FetchAllCdStagesLatestEntity(pipelineIds []int) ([]*CdWorkflowStatus, error)
 	FetchAllCdStagesLatestEntityStatus(wfrIds []int) ([]*CdWorkflowRunner, error)
 	ExistsByStatus(status string) (bool, error)
 	FetchEnvAllCdStagesLatestEntityStatus(wfrIds []int, envID int) ([]*CdWorkflowRunner, error)
@@ -580,15 +578,15 @@ func (impl *CdWorkflowRepositoryImpl) IsLatestWf(pipelineId int, wfId int) (bool
 	return !exists, err
 }
 
-func (impl *CdWorkflowRepositoryImpl) FetchAllCdStagesLatestEntity(pipelineWorkflowPairs map[int][]apiBean.WorkflowType) ([]*CdWorkflowStatus, error) {
+func (impl *CdWorkflowRepositoryImpl) FetchAllCdStagesLatestEntity(pipelineIds []int) ([]*CdWorkflowStatus, error) {
 	var cdWorkflowStatus []*CdWorkflowStatus
-	if len(pipelineWorkflowPairs) == 0 {
+	if len(pipelineIds) == 0 {
 		return cdWorkflowStatus, nil
 	}
 	query := "select p.ci_pipeline_id, wf.pipeline_id, wfr.workflow_type, max(wfr.id) as wfr_id from cd_workflow_runner wfr" +
 		" inner join cd_workflow wf on wf.id=wfr.cd_workflow_id" +
 		" inner join pipeline p on p.id = wf.pipeline_id" +
-		" where (wf.pipeline_id, wfr.workflow_type) in (" + buildPipelineTypeValuesList(pipelineWorkflowPairs) + ")" +
+		" where wf.pipeline_id in (" + sqlIntSeq(pipelineIds) + ")" +
 		" group by p.ci_pipeline_id, wf.pipeline_id, wfr.workflow_type order by wfr_id desc;"
 	_, err := impl.dbConnection.Query(&cdWorkflowStatus, query)
 	if err != nil {
@@ -598,15 +596,6 @@ func (impl *CdWorkflowRepositoryImpl) FetchAllCdStagesLatestEntity(pipelineWorkf
 	return cdWorkflowStatus, nil
 }
 
-func buildPipelineTypeValuesList(pairs map[int][]apiBean.WorkflowType) string {
-	var values []string
-	for pipelineId, workflowTypes := range pairs {
-		for _, workflowType := range workflowTypes {
-			values = append(values, fmt.Sprintf("(%d,'%s')", pipelineId, workflowType))
-		}
-	}
-	return strings.Join(values, ",")
-}
 func (impl *CdWorkflowRepositoryImpl) FetchAllCdStagesLatestEntityStatus(wfrIds []int) ([]*CdWorkflowRunner, error) {
 	var wfrList []*CdWorkflowRunner
 	err := impl.dbConnection.Model(&wfrList).

pkg/workflow/cd/read/CdWorkflowRunnerReadService.go

@@ -104,9 +104,8 @@ func (impl *CdWorkflowRunnerReadServiceImpl) GetWfrStatusForLatestRunners(pipeli
 	}
 
 	// calculating all the pipelines not present in the index table cdWorkflowLatest
-	pipelinesAbsentInCache := make(map[int][]bean2.WorkflowType)
+	absentPipelineIds := make([]int, 0)
 	for _, item := range pipelines {
-
 		var isPreCDConfigured, isPostCDConfigured bool
 		if configuredStages, ok := pipelineStageMap[item.Id]; ok {
 			for _, stage := range configuredStages {
@@ -119,13 +118,7 @@ func (impl *CdWorkflowRunnerReadServiceImpl) GetWfrStatusForLatestRunners(pipeli
 		}
 
 		if _, ok := cdWorfklowLatestMap[item.Id]; !ok {
-			pipelinesAbsentInCache[item.Id] = append(pipelinesAbsentInCache[item.Id], bean2.CD_WORKFLOW_TYPE_DEPLOY)
-			if isPreCDConfigured {
-				pipelinesAbsentInCache[item.Id] = append(pipelinesAbsentInCache[item.Id], bean2.CD_WORKFLOW_TYPE_PRE)
-			}
-			if isPostCDConfigured {
-				pipelinesAbsentInCache[item.Id] = append(pipelinesAbsentInCache[item.Id], bean2.CD_WORKFLOW_TYPE_POST)
-			}
+			absentPipelineIds = append(absentPipelineIds, item.Id)
 		} else {
 			isPreCDStageAbsent, isPostCdStageAbsent, isDeployStageAbsent := true, true, true
 			for _, stage := range cdWorfklowLatestMap[item.Id] {
@@ -138,21 +131,15 @@ func (impl *CdWorkflowRunnerReadServiceImpl) GetWfrStatusForLatestRunners(pipeli
 					isDeployStageAbsent = false
 				}
 			}
-			if isDeployStageAbsent {
-				pipelinesAbsentInCache[item.Id] = append(pipelinesAbsentInCache[item.Id], bean2.CD_WORKFLOW_TYPE_DEPLOY)
-			}
-			if isPreCDConfigured && isPreCDStageAbsent {
-				pipelinesAbsentInCache[item.Id] = append(pipelinesAbsentInCache[item.Id], bean2.CD_WORKFLOW_TYPE_PRE)
-			}
-			if isPostCDConfigured && isPostCdStageAbsent {
-				pipelinesAbsentInCache[item.Id] = append(pipelinesAbsentInCache[item.Id], bean2.CD_WORKFLOW_TYPE_POST)
+			if isDeployStageAbsent || (isPreCDConfigured && isPreCDStageAbsent) || (isPostCDConfigured && isPostCdStageAbsent) {
+				absentPipelineIds = append(absentPipelineIds, item.Id)
 			}
 		}
 	}
-	if len(pipelinesAbsentInCache) > 0 {
-		remainingRunners, err := impl.cdWorkflowRepository.FetchAllCdStagesLatestEntity(pipelinesAbsentInCache)
+	if len(absentPipelineIds) > 0 {
+		remainingRunners, err := impl.cdWorkflowRepository.FetchAllCdStagesLatestEntity(absentPipelineIds)
 		if err != nil {
-			impl.logger.Errorw("error in fetching all cd stages latest entity", "pipelinesAbsentInCache", pipelinesAbsentInCache, "err", err)
+			impl.logger.Errorw("error in fetching all cd stages latest entity", "pipelineIds", absentPipelineIds, "err", err)
 			return nil, err
 		}
 		result = append(result, remainingRunners...)