feat: add or delete CD pipelines from workflow (#4398)

* first commit

* fixes

* PR comments

* removed enum because of cyclic dependency

* adding deploymentappcreated in cd-pipeline list api

* fix

* dfs fix in env filter

* level wise sorting

* sorting wfm

* cleaning up dead code

* pr comments

* aritfacts list fix

Author: subhashish-devtron

internal/sql/repository/appWorkflow/AppWorkflowRepository.go

@@ -61,7 +61,7 @@ type AppWorkflowRepository interface {
 	FindByCDPipelineIds(cdPipelineIds []int) ([]*AppWorkflowMapping, error)
 	FindByWorkflowIds(workflowIds []int) ([]*AppWorkflowMapping, error)
 	FindMappingByAppIds(appIds []int) ([]*AppWorkflowMapping, error)
-	UpdateParentComponentDetails(tx *pg.Tx, oldComponentId int, oldComponentType string, newComponentId int, newComponentType string) error
+	UpdateParentComponentDetails(tx *pg.Tx, oldComponentId int, oldComponentType string, newComponentId int, newComponentType string, componentIdsFilter []int) error
 	FindWFMappingByComponent(componentType string, componentId int) (*AppWorkflowMapping, error)
 	FindByComponentId(componentId int) ([]*AppWorkflowMapping, error)
 }
@@ -474,19 +474,23 @@ func (impl AppWorkflowRepositoryImpl) FindMappingByAppIds(appIds []int) ([]*AppW
 	return appWorkflowsMapping, err
 }
 
-func (impl AppWorkflowRepositoryImpl) UpdateParentComponentDetails(tx *pg.Tx, oldParentId int, oldParentType string, newParentId int, newParentType string) error {
+func (impl AppWorkflowRepositoryImpl) UpdateParentComponentDetails(tx *pg.Tx, oldParentId int, oldParentType string, newParentId int, newParentType string, componentIdFilter []int) error {
 
 	/*updateQuery := fmt.Sprintf(" UPDATE app_workflow_mapping "+
 		" SET parent_type = (select type from new_app_workflow_mapping),parent_id = (select id from new_app_workflow_mapping) where parent_id = %v and parent_type='%v' and active = true", oldComponentId, oldComponentType)
 
 	finalQuery := withQuery + updateQuery*/
-	_, err := tx.Model((*AppWorkflowMapping)(nil)).
+	query := tx.Model((*AppWorkflowMapping)(nil)).
 		Set("parent_type = ?", newParentType).
 		Set("parent_id = ?", newParentId).
 		Where("parent_type = ?", oldParentType).
 		Where("parent_id = ?", oldParentId).
-		Where("active = true").
-		Update()
+		Where("active = true")
+
+	if len(componentIdFilter) > 0 {
+		query = query.Where("component_id in (?)", pg.In(componentIdFilter)).Where("type = ?", "CD_PIPELINE")
+	}
+	_, err := query.Update()
 	return err
 }
 

pkg/appClone/AppCloneService.go

@@ -594,6 +594,7 @@ func (impl *AppCloneServiceImpl) CreateWf(oldAppId, newAppId int, userId int32,
 	if err != nil {
 		return nil, err
 	}
+
 	impl.logger.Debugw("workflow found", "wf", refAppWFs)
 
 	createWorkflowMappingDtoResp := CreateWorkflowMappingDto{
@@ -673,6 +674,9 @@ func (impl *AppCloneServiceImpl) createWfInstances(refWfMappings []appWorkflow.A
 	var ciMapping []appWorkflow.AppWorkflowMappingDto
 	var cdMappings []appWorkflow.AppWorkflowMappingDto
 	var webhookMappings []appWorkflow.AppWorkflowMappingDto
+
+	refWfMappings = appWorkflow.LevelWiseSort(refWfMappings)
+
 	for _, appWf := range refWfMappings {
 		if appWf.Type == appWorkflow2.CIPIPELINE {
 			ciMapping = append(ciMapping, appWf)

pkg/appWorkflow/AppWorkflowService.go

@@ -32,6 +32,7 @@ import (
 	"github.com/devtron-labs/devtron/pkg/sql"
 	"github.com/devtron-labs/devtron/pkg/user"
 	bean3 "github.com/devtron-labs/devtron/pkg/user/bean"
+	"github.com/devtron-labs/devtron/pkg/variables/utils"
 	"github.com/devtron-labs/devtron/util/rbac"
 	"github.com/go-pg/pg"
 	"go.uber.org/zap"
@@ -741,7 +742,7 @@ func (impl AppWorkflowServiceImpl) FilterWorkflows(triggerViewConfig *TriggerVie
 			continue
 		}
 
-		identifierToFilteredWorkflowMapping, leafPipelines := fetchLeafPipelinesAndPopulateChildrenIdsInWorkflowMapping(workflow.AppWorkflowMappingDto)
+		identifierToFilteredWorkflowMapping, leafPipelines, _ := processWorkflowMappingTree(workflow.AppWorkflowMappingDto)
 
 		identifierToFilteredWorkflowMapping = filterMappingOnFilteredCdPipelineIds(identifierToFilteredWorkflowMapping, leafPipelines, cdPipelineIdsFiltered)
 
@@ -766,24 +767,34 @@ func extractOutFilteredWorkflowMappings(appWorkflowMappings []AppWorkflowMapping
 	return newAppWorkflowMappingDto
 }
 
-// fetchLeafPipelinesAndPopulateChildrenIdsInWorkflowMapping function fetches all the leaf cd pipelines and append
-// the children pipelineIds into ChildPipelinesIds object in AppWorkflowMappingDto and returns both object.
-func fetchLeafPipelinesAndPopulateChildrenIdsInWorkflowMapping(appWorkflowMappings []AppWorkflowMappingDto) (map[PipelineIdentifier]*AppWorkflowMappingDto, []*AppWorkflowMappingDto) {
+// processWorkflowMappingTree function processed the wf mapping array into a tree structure
+// returns a map of identifier to mapping, leaf nodes and the root node
+func processWorkflowMappingTree(appWorkflowMappings []AppWorkflowMappingDto) (map[PipelineIdentifier]*AppWorkflowMappingDto, []AppWorkflowMappingDto, *AppWorkflowMappingDto) {
 	identifierToFilteredWorkflowMapping := make(map[PipelineIdentifier]*AppWorkflowMappingDto)
-	leafPipelines := make([]*AppWorkflowMappingDto, 0)
+	leafPipelines := make([]AppWorkflowMappingDto, 0)
+	var rootPipeline *AppWorkflowMappingDto
+	//initializing the nodes with empty children and collecting leaf
 	for i, appWorkflowMapping := range appWorkflowMappings {
-		if appWorkflowMapping.IsLast {
-			leafPipelines = append(leafPipelines, &appWorkflowMappings[i])
-		}
+		appWorkflowMappings[i].ChildPipelinesIds = mapset.NewSet()
 		identifierToFilteredWorkflowMapping[appWorkflowMapping.getPipelineIdentifier()] = &appWorkflowMappings[i]
-		if appWorkflowMappings[i].ChildPipelinesIds == nil {
-			appWorkflowMappings[i].ChildPipelinesIds = mapset.NewSet()
+
+		//collecting leaf pipelines
+		if appWorkflowMapping.IsLast {
+			leafPipelines = append(leafPipelines, appWorkflowMapping)
 		}
-		if appWorkflow, ok := identifierToFilteredWorkflowMapping[appWorkflowMapping.getParentPipelineIdentifier()]; ok {
-			appWorkflow.ChildPipelinesIds.Add(appWorkflowMapping.ComponentId)
+	}
+
+	for _, appWorkflowMapping := range identifierToFilteredWorkflowMapping {
+		// populating children in parent nodes
+		parentId := appWorkflowMapping.getParentPipelineIdentifier()
+		componentId := appWorkflowMapping.ComponentId
+		if parentMapping, hasParent := identifierToFilteredWorkflowMapping[parentId]; hasParent && !parentMapping.ChildPipelinesIds.Contains(componentId) {
+			parentMapping.ChildPipelinesIds.Add(componentId)
+		} else if !hasParent {
+			rootPipeline = appWorkflowMapping
 		}
 	}
-	return identifierToFilteredWorkflowMapping, leafPipelines
+	return identifierToFilteredWorkflowMapping, leafPipelines, rootPipeline
 }
 
 // filterMappingOnFilteredCdPipelineIds iterates over all leaf cd-pipelines, if that leaf cd-pipeline is present in the
@@ -792,7 +803,7 @@ func fetchLeafPipelinesAndPopulateChildrenIdsInWorkflowMapping(appWorkflowMappin
 // cd-pipeline from componentIdWorkflowMapping's list of AppWorkflowMappingDto and also truncate the child
 // cd-pipeline id present in the parent's ChildPipelinesIds object inside AppWorkflowMappingDto.
 func filterMappingOnFilteredCdPipelineIds(identifierToFilteredWorkflowMapping map[PipelineIdentifier]*AppWorkflowMappingDto,
-	leafPipelines []*AppWorkflowMappingDto, cdPipelineIdsFiltered mapset.Set) map[PipelineIdentifier]*AppWorkflowMappingDto {
+	leafPipelines []AppWorkflowMappingDto, cdPipelineIdsFiltered mapset.Set) map[PipelineIdentifier]*AppWorkflowMappingDto {
 
 	leafPipelineSize := len(leafPipelines)
 	for i := 0; i < leafPipelineSize; i++ {
@@ -803,9 +814,11 @@ func filterMappingOnFilteredCdPipelineIds(identifierToFilteredWorkflowMapping ma
 			parent := leafPipelines[i].getParentPipelineIdentifier()
 			identifierToFilteredWorkflowMapping[parent].ChildPipelinesIds.Remove(leafPipelines[i].ComponentId)
 		}
-		if identifierToFilteredWorkflowMapping[leafPipelines[i].getParentPipelineIdentifier()].ChildPipelinesIds.Cardinality() == 0 {
+		parentPipelineIdentifier := leafPipelines[i].getParentPipelineIdentifier()
+		childPipelineIds := identifierToFilteredWorkflowMapping[parentPipelineIdentifier].ChildPipelinesIds
+		if childPipelineIds.Cardinality() == 0 {
 			//this means this pipeline has become leaf, so append this pipelineId in leafPipelines for further processing
-			leafPipelines = append(leafPipelines, identifierToFilteredWorkflowMapping[leafPipelines[i].getParentPipelineIdentifier()])
+			leafPipelines = append(leafPipelines, *identifierToFilteredWorkflowMapping[leafPipelines[i].getParentPipelineIdentifier()])
 			leafPipelineSize += 1
 		}
 
@@ -849,3 +862,40 @@ func (impl AppWorkflowServiceImpl) FindAppWorkflowByCiPipelineId(ciPipelineId in
 	return appWorkflowMapping, nil
 
 }
+
+// LevelWiseSort performs level wise sort for workflow mappings starting from leaves
+// This will break if ever the workflow mappings array break the assumption of being a DAG with one root node
+func LevelWiseSort(appWorkflowMappings []AppWorkflowMappingDto) []AppWorkflowMappingDto {
+
+	if len(appWorkflowMappings) < 2 {
+		return appWorkflowMappings
+	}
+
+	identifierToNodeMapping, _, root := processWorkflowMappingTree(appWorkflowMappings)
+
+	result := make([]AppWorkflowMappingDto, 0)
+	nodesInCurrentLevel := append(make([]AppWorkflowMappingDto, 0), *root)
+	for len(result) != len(appWorkflowMappings) {
+		result = append(result, nodesInCurrentLevel...)
+		childrenOfCurrentLevel := make([]AppWorkflowMappingDto, 0)
+		for _, node := range nodesInCurrentLevel {
+			childrenOfCurrentLevel = append(childrenOfCurrentLevel, getMappingsFromIds(identifierToNodeMapping, utils.ToIntArray(node.ChildPipelinesIds.ToSlice()))...)
+		}
+		// cloning slice elements
+		nodesInCurrentLevel = append(childrenOfCurrentLevel, []AppWorkflowMappingDto{}...)
+	}
+
+	return result
+}
+
+func getMappingsFromIds(identifierToNodeMapping map[PipelineIdentifier]*AppWorkflowMappingDto, ids []int) []AppWorkflowMappingDto {
+	result := make([]AppWorkflowMappingDto, 0)
+	for _, id := range ids {
+		identifier := PipelineIdentifier{
+			PipelineType: CD_PIPELINE_TYPE,
+			PipelineId:   id,
+		}
+		result = append(result, *identifierToNodeMapping[identifier])
+	}
+	return result
+}

pkg/bean/app.go

@@ -588,8 +588,17 @@ type CDPipelineConfigObject struct {
 	CustomTagStage                *repository.PipelineStageType          `json:"customTagStage"`
 	EnableCustomTag               bool                                   `json:"enableCustomTag"`
 	SwitchFromCiPipelineId        int                                    `json:"switchFromCiPipelineId"`
+	CDPipelineAddType             CDPipelineAddType                      `json:"addType"`
+	ChildPipelineId               int                                    `json:"childPipelineId"`
 }
 
+type CDPipelineAddType string
+
+const (
+	SEQUENTIAL CDPipelineAddType = "SEQUENTIAL"
+	PARALLEL   CDPipelineAddType = "PARALLEL"
+)
+
 func (cdpipelineConfig *CDPipelineConfigObject) IsSwitchCiPipelineRequest() bool {
 	return cdpipelineConfig.SwitchFromCiPipelineId > 0 && cdpipelineConfig.AppWorkflowId > 0
 }

pkg/pipeline/AppArtifactManager.go

@@ -864,11 +864,14 @@ func (impl *AppArtifactManagerImpl) BuildArtifactsForCdStageV2(listingFilterOpts
 	if listingFilterOpts.ParentCdId > 0 {
 		//TODO: check if we can fetch LastSuccessfulTriggerOnParent wfr along with last running wf
 		parentCdWfrList, err := impl.cdWorkflowRepository.FindArtifactByPipelineIdAndRunnerType(listingFilterOpts.ParentCdId, bean.CD_WORKFLOW_TYPE_DEPLOY, 1, []string{application.Healthy, application.SUCCEEDED, application.Progressing})
-		if err != nil || len(parentCdWfrList) == 0 {
+		if err != nil {
 			impl.logger.Errorw("error in getting artifact for parent cd", "parentCdPipelineId", listingFilterOpts.ParentCdId)
 			return ciArtifacts, totalCount, err
 		}
-		artifactRunningOnParentCd = parentCdWfrList[0].CdWorkflow.CiArtifact.Id
+
+		if len(parentCdWfrList) != 0 {
+			artifactRunningOnParentCd = parentCdWfrList[0].CdWorkflow.CiArtifact.Id
+		}
 	}
 
 	for _, artifact := range cdArtifacts {

pkg/pipeline/BuildPipelineSwitchService.go

@@ -329,5 +329,5 @@ func (impl *BuildPipelineSwitchServiceImpl) saveHistoryOfOverriddenTemplate(ciPi
 }
 
 func (impl *BuildPipelineSwitchServiceImpl) updateLinkedAppWorkflowMappings(tx *pg.Tx, oldAppWorkflowMapping *appWorkflow.AppWorkflowMapping, newAppWorkflowMapping *appWorkflow.AppWorkflowMapping) error {
-	return impl.appWorkflowRepository.UpdateParentComponentDetails(tx, oldAppWorkflowMapping.ComponentId, oldAppWorkflowMapping.Type, newAppWorkflowMapping.ComponentId, newAppWorkflowMapping.Type)
+	return impl.appWorkflowRepository.UpdateParentComponentDetails(tx, oldAppWorkflowMapping.ComponentId, oldAppWorkflowMapping.Type, newAppWorkflowMapping.ComponentId, newAppWorkflowMapping.Type, nil)
 }

pkg/pipeline/CiCdPipelineOrchestrator.go

@@ -1778,6 +1778,7 @@ func (impl CiCdPipelineOrchestratorImpl) GetCdPipelinesForApp(appId int) (cdPipe
 			PreStageConfigMapSecretNames:  preStageConfigmapSecrets,
 			PostStageConfigMapSecretNames: postStageConfigmapSecrets,
 			DeploymentAppType:             dbPipeline.DeploymentAppType,
+			DeploymentAppCreated:          dbPipeline.DeploymentAppCreated,
 			DeploymentAppDeleteRequest:    dbPipeline.DeploymentAppDeleteRequest,
 			IsVirtualEnvironment:          dbPipeline.Environment.IsVirtualEnvironment,
 		}

pkg/pipeline/DeploymentPipelineConfigService.go

@@ -639,10 +639,8 @@ func (impl *CdPipelineConfigServiceImpl) DeleteCdPipeline(pipeline *pipelineConf
 	if err != nil && err != pg.ErrNoRows {
 		impl.logger.Errorw("error in getting children cd details", "err", err)
 		return deleteResponse, err
-	} else if len(childNodes) > 0 {
-		impl.logger.Debugw("cannot delete cd pipeline, contains children cd")
-		return deleteResponse, fmt.Errorf("Please delete children CD pipelines before deleting this pipeline.")
 	}
+
 	//getting deployment group for this pipeline
 	deploymentGroupNames, err := impl.deploymentGroupRepository.GetNamesByAppIdAndEnvId(pipeline.EnvironmentId, pipeline.AppId)
 	if err != nil && err != pg.ErrNoRows {
@@ -679,7 +677,7 @@ func (impl *CdPipelineConfigServiceImpl) DeleteCdPipeline(pipeline *pipelineConf
 		impl.logger.Errorw("error in deleting workflow mapping", "err", err)
 		return deleteResponse, err
 	}
-	if appWorkflowMapping.ParentType == appWorkflow.WEBHOOK {
+	if appWorkflowMapping.ParentType == appWorkflow.WEBHOOK && len(childNodes) == 0 {
 		childNodes, err := impl.appWorkflowRepository.FindWFCDMappingByExternalCiId(appWorkflowMapping.ParentId)
 		if err != nil && !util.IsErrNoRows(err) {
 			impl.logger.Errorw("error in fetching external ci", "err", err)
@@ -726,6 +724,14 @@ func (impl *CdPipelineConfigServiceImpl) DeleteCdPipeline(pipeline *pipelineConf
 		return deleteResponse, err
 	}
 
+	if len(childNodes) > 0 {
+		err = impl.appWorkflowRepository.UpdateParentComponentDetails(tx, appWorkflowMapping.ComponentId, appWorkflowMapping.Type, appWorkflowMapping.ParentId, appWorkflowMapping.ParentType, nil)
+		if err != nil {
+			impl.logger.Errorw("error updating wfm for children pipelines of pipeline", "err", err, "id", appWorkflowMapping.Id)
+			return deleteResponse, err
+		}
+	}
+
 	if pipeline.PreStageConfig != "" {
 		err = impl.prePostCdScriptHistoryService.CreatePrePostCdScriptHistory(pipeline, tx, repository4.PRE_CD_TYPE, false, 0, time.Time{})
 		if err != nil {
@@ -1027,6 +1033,7 @@ func (impl *CdPipelineConfigServiceImpl) GetCdPipelinesForApp(appId int) (cdPipe
 			RunPreStageInEnv:              dbPipeline.RunPreStageInEnv,
 			RunPostStageInEnv:             dbPipeline.RunPostStageInEnv,
 			DeploymentAppType:             dbPipeline.DeploymentAppType,
+			DeploymentAppCreated:          dbPipeline.DeploymentAppCreated,
 			ParentPipelineType:            appToWorkflowMapping.ParentType,
 			ParentPipelineId:              appToWorkflowMapping.ParentId,
 			DeploymentAppDeleteRequest:    dbPipeline.DeploymentAppDeleteRequest,
@@ -1714,6 +1721,18 @@ func (impl *CdPipelineConfigServiceImpl) createCdPipeline(ctx context.Context, a
 					parentPipelineId = pipeline.SourceToNewPipelineId[pipeline.ParentPipelineId]
 				}
 			}
+
+			if pipeline.CDPipelineAddType == bean.SEQUENTIAL {
+				childPipelineIds := make([]int, 0)
+				if pipeline.ChildPipelineId > 0 {
+					childPipelineIds = append(childPipelineIds, pipeline.ChildPipelineId)
+				}
+				err = impl.appWorkflowRepository.UpdateParentComponentDetails(tx, parentPipelineId, parentPipelineType, pipelineId, "CD_PIPELINE", childPipelineIds)
+				if err != nil {
+					return 0, err
+				}
+			}
+
 			appWorkflowMap := &appWorkflow.AppWorkflowMapping{
 				AppWorkflowId: pipeline.AppWorkflowId,
 				ParentId:      parentPipelineId,
@@ -1956,14 +1975,11 @@ func (impl *CdPipelineConfigServiceImpl) DeleteCdPipelinePartial(pipeline *pipel
 		deleteResponse.ClusterReachable = false
 	}
 	//getting children CD pipeline details
-	childNodes, err := impl.appWorkflowRepository.FindWFCDMappingByParentCDPipelineId(pipeline.Id)
 	if err != nil && err != pg.ErrNoRows {
 		impl.logger.Errorw("error in getting children cd details", "err", err)
 		return deleteResponse, err
-	} else if len(childNodes) > 0 {
-		impl.logger.Debugw("cannot delete cd pipeline, contains children cd")
-		return deleteResponse, fmt.Errorf("Please delete children CD pipelines before deleting this pipeline.")
 	}
+
 	//getting deployment group for this pipeline
 	deploymentGroupNames, err := impl.deploymentGroupRepository.GetNamesByAppIdAndEnvId(pipeline.EnvironmentId, pipeline.AppId)
 	if err != nil && err != pg.ErrNoRows {

pkg/variables/utils/conversion-utils.go

@@ -20,6 +20,15 @@ func ToStringArray(interfaceArr []interface{}) []string {
 	return stringArr
 }
 
+// ToIntArray converts an array of interface{} back to an array of int
+func ToIntArray(interfaceArr []interface{}) []int {
+	intArr := make([]int, len(interfaceArr))
+	for i, v := range interfaceArr {
+		intArr[i] = v.(int)
+	}
+	return intArr
+}
+
 func FilterDuplicatesInStringArray(items []string) []string {
 	itemsSet := mapset.NewSetFromSlice(ToInterfaceArray(items))
 	uniqueItems := ToStringArray(itemsSet.ToSlice())