feat: multiple images handling for single workflow for ECR Plugin Poll Images (#4027)

* container registry handling for ci_job ci pipeline

* Handling multiple images from Ci Complete event

* parent ci workflow id

* getting workflows without parent_ci_workflow_id

* Setting pod status as successful

* getting all workflows

* adding new api for getting all artifacts

* parent ci workflow fetching null

* remving parent_ci_workflow_id from workflow_response

* Excluding parent workflow ci artifact

* query change

* query

* self review

* self review

* self review comments

* review comments

* IT case for fetching ci artifact for ci job type

* handling nil pointer for docker registry id

* reverting pipeline builder for now

* reverting pipeline builder for now

* pushing changes for docker config update

* checking len before querying

* review comments

* script number change

* review comments logging errors

---------

Co-authored-by: ayushmaheshwari <[email protected]>

Author: Shivam-nagar23

api/restHandler/app/BuildPipelineRestHandler.go

@@ -52,6 +52,7 @@ type DevtronAppBuildRestHandler interface {
 	HandleWorkflowWebhook(w http.ResponseWriter, r *http.Request)
 	GetBuildLogs(w http.ResponseWriter, r *http.Request)
 	FetchWorkflowDetails(w http.ResponseWriter, r *http.Request)
+	GetArtifactsForCiJob(w http.ResponseWriter, r *http.Request)
 	// CancelWorkflow CancelBuild
 	CancelWorkflow(w http.ResponseWriter, r *http.Request)
 
@@ -1567,6 +1568,51 @@ func (handler PipelineConfigRestHandlerImpl) FetchWorkflowDetails(w http.Respons
 	common.WriteJsonResp(w, err, resp, http.StatusOK)
 }
 
+func (handler PipelineConfigRestHandlerImpl) GetArtifactsForCiJob(w http.ResponseWriter, r *http.Request) {
+	userId, err := handler.userAuthService.GetLoggedInUser(r)
+	if userId == 0 || err != nil {
+		common.WriteJsonResp(w, err, "Unauthorized User", http.StatusUnauthorized)
+		return
+	}
+	vars := mux.Vars(r)
+	pipelineId, err := strconv.Atoi(vars["pipelineId"])
+	if err != nil {
+		common.WriteJsonResp(w, err, nil, http.StatusBadRequest)
+		return
+	}
+	buildId, err := strconv.Atoi(vars["workflowId"])
+	if err != nil || buildId == 0 {
+		common.WriteJsonResp(w, err, nil, http.StatusBadRequest)
+		return
+	}
+	handler.Logger.Infow("request payload, GetArtifactsForCiJob", "pipelineId", pipelineId, "buildId", buildId, "buildId", buildId)
+	ciPipeline, err := handler.ciPipelineRepository.FindById(pipelineId)
+	if err != nil {
+		common.WriteJsonResp(w, err, nil, http.StatusInternalServerError)
+		return
+	}
+	//RBAC
+	token := r.Header.Get("token")
+	object := handler.enforcerUtil.GetAppRBACNameByAppId(ciPipeline.AppId)
+	if ok := handler.enforcer.Enforce(token, casbin.ResourceApplications, casbin.ActionGet, object); !ok {
+		common.WriteJsonResp(w, err, "Unauthorized User", http.StatusForbidden)
+		return
+	}
+	//RBAC
+	resp, err := handler.ciHandler.FetchArtifactsForCiJob(buildId)
+	if err != nil {
+		handler.Logger.Errorw("service err, FetchArtifactsForCiJob", "err", err, "pipelineId", pipelineId, "buildId", buildId, "buildId", buildId)
+		if util.IsErrNoRows(err) {
+			err = &util.ApiError{Code: "404", HttpStatusCode: http.StatusNotFound, UserMessage: "no artifact found"}
+			common.WriteJsonResp(w, err, nil, http.StatusOK)
+		} else {
+			common.WriteJsonResp(w, err, nil, http.StatusInternalServerError)
+		}
+		return
+	}
+	common.WriteJsonResp(w, err, resp, http.StatusOK)
+}
+
 func (handler PipelineConfigRestHandlerImpl) GetCiPipelineByEnvironment(w http.ResponseWriter, r *http.Request) {
 	vars := mux.Vars(r)
 	userId, err := handler.userAuthService.GetLoggedInUser(r)

api/router/PipelineConfigRouter.go

@@ -109,6 +109,7 @@ func (router PipelineConfigRouterImpl) initPipelineConfigRouter(configRouter *mu
 	configRouter.Path("/ci-pipeline/refresh-material/{gitMaterialId}").HandlerFunc(router.restHandler.RefreshMaterials).Methods("GET")
 
 	configRouter.Path("/{appId}/ci-pipeline/{pipelineId}/workflow/{workflowId}").HandlerFunc(router.restHandler.FetchWorkflowDetails).Methods("GET")
+	configRouter.Path("/ci-pipeline/{pipelineId}/workflow/{workflowId}/ci-job/artifacts").HandlerFunc(router.restHandler.GetArtifactsForCiJob).Methods("GET")
 	configRouter.Path("/ci-pipeline/{pipelineId}/artifacts/{workflowId}").HandlerFunc(router.restHandler.DownloadCiWorkflowArtifacts).Methods("GET")
 
 	configRouter.Path("/ci-pipeline/{pipelineId}/git-changes/{ciMaterialId}").HandlerFunc(router.restHandler.FetchChanges).Methods("GET")

api/router/pubsub/CiEventHandler.go

@@ -118,19 +118,26 @@ func (impl *CiEventHandlerImpl) Subscribe() error {
 			}
 		} else if ciCompleteEvent.ImageDetailsFromCR != nil {
 			if len(ciCompleteEvent.ImageDetailsFromCR.ImageDetails) > 0 {
-				detail := util.GetLatestImageAccToImagePushedAt(ciCompleteEvent.ImageDetailsFromCR.ImageDetails)
-				request, err := impl.BuildCIArtifactRequestForImageFromCR(detail, ciCompleteEvent.ImageDetailsFromCR.Region, ciCompleteEvent)
+				imageDetails := util.GetReverseSortedImageDetails(ciCompleteEvent.ImageDetailsFromCR.ImageDetails)
+				digestWorkflowMap, err := impl.webhookService.HandleMultipleImagesFromEvent(imageDetails, *ciCompleteEvent.WorkflowId)
 				if err != nil {
-					impl.logger.Error("Error while creating request for pipelineID", "pipelineId", ciCompleteEvent.PipelineId, "err", err)
+					impl.logger.Errorw("error in getting digest workflow map", "err", err, "workflowId", ciCompleteEvent.WorkflowId)
 					return
 				}
-				resp, err := impl.webhookService.HandleCiSuccessEvent(ciCompleteEvent.PipelineId, request, detail.ImagePushedAt)
-				if err != nil {
-					impl.logger.Error("Error while sending event for CI success for pipelineID", "pipelineId",
-						ciCompleteEvent.PipelineId, "request", request, "err", err)
-					return
+				for _, detail := range imageDetails {
+					request, err := impl.BuildCIArtifactRequestForImageFromCR(detail, ciCompleteEvent.ImageDetailsFromCR.Region, ciCompleteEvent, digestWorkflowMap[*detail.ImageDigest].Id)
+					if err != nil {
+						impl.logger.Error("Error while creating request for pipelineID", "pipelineId", ciCompleteEvent.PipelineId, "err", err)
+						return
+					}
+					resp, err := impl.webhookService.HandleCiSuccessEvent(ciCompleteEvent.PipelineId, request, detail.ImagePushedAt)
+					if err != nil {
+						impl.logger.Error("Error while sending event for CI success for pipelineID", "pipelineId",
+							ciCompleteEvent.PipelineId, "request", request, "err", err)
+						return
+					}
+					impl.logger.Debug("response of handle ci success event for multiple images from plugin", "resp", resp)
 				}
-				impl.logger.Debug(resp)
 			}
 
 		} else {
@@ -219,7 +226,7 @@ func (impl *CiEventHandlerImpl) BuildCiArtifactRequest(event CiCompleteEvent) (*
 	return request, nil
 }
 
-func (impl *CiEventHandlerImpl) BuildCIArtifactRequestForImageFromCR(imageDetails types.ImageDetail, region string, event CiCompleteEvent) (*pipeline.CiArtifactWebhookRequest, error) {
+func (impl *CiEventHandlerImpl) BuildCIArtifactRequestForImageFromCR(imageDetails types.ImageDetail, region string, event CiCompleteEvent, workflowId int) (*pipeline.CiArtifactWebhookRequest, error) {
 	if event.TriggeredBy == 0 {
 		event.TriggeredBy = 1 // system triggered event
 	}
@@ -229,7 +236,7 @@ func (impl *CiEventHandlerImpl) BuildCIArtifactRequestForImageFromCR(imageDetail
 		DataSource:         event.DataSource,
 		PipelineName:       event.PipelineName,
 		UserId:             event.TriggeredBy,
-		WorkflowId:         event.WorkflowId,
+		WorkflowId:         &workflowId,
 		IsArtifactUploaded: event.IsArtifactUploaded,
 	}
 	return request, nil

internal/sql/repository/CiArtifactRepository.go

@@ -71,6 +71,7 @@ type CiArtifactRepository interface {
 	GetByImageDigest(imageDigest string) (artifact *CiArtifact, err error)
 	GetByIds(ids []int) ([]*CiArtifact, error)
 	GetArtifactByCdWorkflowId(cdWorkflowId int) (artifact *CiArtifact, err error)
+	GetArtifactsByParentCiWorkflowId(parentCiWorkflowId int) ([]string, error)
 }
 
 type CiArtifactRepositoryImpl struct {
@@ -569,3 +570,15 @@ func (impl CiArtifactRepositoryImpl) GetArtifactByCdWorkflowId(cdWorkflowId int)
 		Select()
 	return artifact, err
 }
+
+// GetArtifactsByParentCiWorkflowId will get all artifacts of child workflow sorted by descending order to fetch latest at top, child workflow required for handling container image polling plugin as there can be multiple images from a single parent workflow, which are accommodated in child workflows
+func (impl CiArtifactRepositoryImpl) GetArtifactsByParentCiWorkflowId(parentCiWorkflowId int) ([]string, error) {
+	var artifacts []string
+	query := "SELECT cia.image FROM ci_artifact cia where cia.ci_workflow_id in (SELECT wf.id from ci_workflow wf where wf.parent_ci_workflow_id = ? ) ORDER BY cia.created_on DESC ;"
+	_, err := impl.dbConnection.Query(&artifacts, query, parentCiWorkflowId)
+	if err != nil {
+		impl.logger.Errorw("error occurred while fetching artifacts for parent ci workflow id", "err", err)
+		return nil, err
+	}
+	return artifacts, err
+}

internal/sql/repository/pipelineConfig/CiWorkflowRepository.go

@@ -72,6 +72,7 @@ type CiWorkflow struct {
 	CiBuildType           string            `sql:"ci_build_type"`
 	EnvironmentId         int               `sql:"environment_id"`
 	ReferenceCiWorkflowId int               `sql:"ref_ci_workflow_id"`
+	ParentCiWorkFlowId int               `sql:"parent_ci_workflow_id"`
 	CiPipeline            *CiPipeline
 }
 
@@ -99,6 +100,7 @@ type WorkflowWithArtifact struct {
 	EnvironmentId      int               `json:"environmentId"`
 	EnvironmentName    string            `json:"environmentName"`
 	RefCiWorkflowId    int               `json:"referenceCiWorkflowId"`
+	ParentCiWorkflowId int               `json:"parent_ci_workflow_id"`
 }
 
 type GitCommit struct {
@@ -168,9 +170,10 @@ func (impl *CiWorkflowRepositoryImpl) FindByStatusesIn(activeStatuses []string)
 	return ciWorkFlows, err
 }
 
+// FindByPipelineId gets only those workflowWithArtifact whose parent_ci_workflow_id is null, this is done to accommodate multiple ci_artifacts through a single workflow(parent), making child workflows for other ci_artifacts (this has been done due to design understanding and db constraint) single workflow single ci-artifact
 func (impl *CiWorkflowRepositoryImpl) FindByPipelineId(pipelineId int, offset int, limit int) ([]WorkflowWithArtifact, error) {
 	var wfs []WorkflowWithArtifact
-	queryTemp := "select cia.id as ci_artifact_id, env.environment_name, cia.image, cia.is_artifact_uploaded, wf.*, u.email_id from ci_workflow wf left join users u on u.id = wf.triggered_by left join ci_artifact cia on wf.id = cia.ci_workflow_id left join environment env on env.id = wf.environment_id where wf.ci_pipeline_id = ? order by wf.started_on desc offset ? limit ?;"
+	queryTemp := "select cia.id as ci_artifact_id, env.environment_name, cia.image, cia.is_artifact_uploaded, wf.*, u.email_id from ci_workflow wf left join users u on u.id = wf.triggered_by left join ci_artifact cia on wf.id = cia.ci_workflow_id left join environment env on env.id = wf.environment_id where wf.ci_pipeline_id = ? and parent_ci_workflow_id is null order by wf.started_on desc offset ? limit ?;"
 	_, err := impl.dbConnection.Query(&wfs, queryTemp, pipelineId, offset, limit)
 	if err != nil {
 		return nil, err

pkg/pipeline/BuildPipelineConfigService.go

@@ -802,6 +802,39 @@ func (impl PipelineBuilderImpl) UpdateCiTemplate(updateRequest *bean.CiConfigReq
 	}
 
 	originalCiConf.CiBuildConfig = ciBuildConfig
+	//TODO: below update code is a hack for ci_job and should be reviewed
+
+	// updating ci_template_override for ci_pipeline type = CI_JOB because for this pipeling ci_template and ci_template_override are kept same as
+	pipelines, err := impl.ciPipelineRepository.FindByAppId(originalCiConf.AppId)
+	if err != nil && err != pg.ErrNoRows {
+		impl.logger.Errorw("error in finding pipeline for app")
+	}
+	ciPipelineIds := make([]int, 0)
+	ciPipelineIdsMap := make(map[int]*pipelineConfig.CiPipeline)
+	for ind, p := range pipelines {
+		ciPipelineIds[ind] = p.Id
+		ciPipelineIdsMap[p.Id] = p
+	}
+	var ciTemplateOverrides []*pipelineConfig.CiTemplateOverride
+	if len(ciPipelineIds) > 0 {
+		ciTemplateOverrides, err = impl.ciTemplateOverrideRepository.FindByCiPipelineIds(ciPipelineIds)
+		if err != nil && err != pg.ErrNoRows {
+			impl.logger.Errorw("error in fetching ci tempalate by pipeline ids", "err", err, "ciPipelineIds", ciPipelineIds)
+		}
+	}
+	for _, ciTemplateOverride := range ciTemplateOverrides {
+		if _, ok := ciPipelineIdsMap[ciTemplateOverride.CiPipelineId]; ok {
+			if ciPipelineIdsMap[ciTemplateOverride.CiPipelineId].PipelineType == string(bean.CI_JOB) {
+				ciTemplateOverride.DockerRepository = updateRequest.DockerRepository
+				ciTemplateOverride.DockerRegistryId = updateRequest.DockerRegistry
+				_, err = impl.ciTemplateOverrideRepository.Update(ciTemplateOverride)
+				if err != nil {
+					impl.logger.Errorw("error in updating ci template for ci_job", "err", err)
+				}
+			}
+		}
+	}
+	// update completed for ci_pipeline_type = ci_job
 
 	err = impl.CiTemplateHistoryService.SaveHistory(ciTemplateBean, "update")
 

pkg/pipeline/CiHandler.go

@@ -64,7 +64,7 @@ type CiHandler interface {
 	FetchMaterialsByPipelineId(pipelineId int, showAll bool) ([]pipelineConfig.CiPipelineMaterialResponse, error)
 	FetchMaterialsByPipelineIdAndGitMaterialId(pipelineId int, gitMaterialId int, showAll bool) ([]pipelineConfig.CiPipelineMaterialResponse, error)
 	FetchWorkflowDetails(appId int, pipelineId int, buildId int) (WorkflowResponse, error)
-
+	FetchArtifactsForCiJob(buildId int) (*ArtifactsForCiJob, error)
 	//FetchBuildById(appId int, pipelineId int) (WorkflowResponse, error)
 	CancelBuild(workflowId int) (int, error)
 
@@ -171,6 +171,10 @@ type WorkflowResponse struct {
 	ReferenceWorkflowId  int                                         `json:"referenceWorkflowId"`
 }
 
+type ArtifactsForCiJob struct {
+	Artifacts []string `json:"artifacts"`
+}
+
 type GitTriggerInfoResponse struct {
 	CiMaterials      []pipelineConfig.CiPipelineMaterialResponse `json:"ciMaterials"`
 	TriggeredByEmail string                                      `json:"triggeredByEmail"`
@@ -740,6 +744,17 @@ func (impl *CiHandlerImpl) FetchWorkflowDetails(appId int, pipelineId int, build
 	return workflowResponse, nil
 }
 
+func (impl *CiHandlerImpl) FetchArtifactsForCiJob(buildId int) (*ArtifactsForCiJob, error) {
+	artifacts, err := impl.ciArtifactRepository.GetArtifactsByParentCiWorkflowId(buildId)
+	if err != nil {
+		impl.Logger.Errorw("error in fetching artifacts by parent ci workflow id", "err", err, "buildId", buildId)
+		return nil, err
+	}
+	artifactsResponse := &ArtifactsForCiJob{
+		Artifacts: artifacts,
+	}
+	return artifactsResponse, nil
+}
 func (impl *CiHandlerImpl) GetRunningWorkflowLogs(pipelineId int, workflowId int) (*bufio.Reader, func() error, error) {
 	ciWorkflow, err := impl.ciWorkflowRepository.FindById(workflowId)
 	if err != nil {

pkg/pipeline/CiHandlerIT_test.go

@@ -0,0 +1,32 @@
+package pipeline
+
+import (
+	"github.com/devtron-labs/devtron/internal/sql/repository"
+	"github.com/devtron-labs/devtron/internal/util"
+	"github.com/devtron-labs/devtron/pkg/sql"
+	"github.com/stretchr/testify/assert"
+	"testing"
+	"time"
+)
+
+func TestCiHandlerImpl_FetchArtifactsForCiJob(t *testing.T) {
+	t.SkipNow()
+	ciHandler := initCiHandler()
+
+	t.Run("Fetch Ci Artifacts For Ci Job type", func(tt *testing.T) {
+		buildId := 304 // Mocked because child workflows are only created dynamic based on number of images which are available after polling
+		time.Sleep(5 * time.Second)
+		_, err := ciHandler.FetchArtifactsForCiJob(buildId)
+		assert.Nil(t, err)
+
+	})
+}
+
+func initCiHandler() *CiHandlerImpl {
+	sugaredLogger, _ := util.InitLogger()
+	config, _ := sql.GetConfig()
+	db, _ := sql.NewDbConnection(config, sugaredLogger)
+	ciArtifactRepositoryImpl := repository.NewCiArtifactRepositoryImpl(db, sugaredLogger)
+	ciHandlerImpl := NewCiHandlerImpl(sugaredLogger, nil, nil, nil, nil, nil, nil, ciArtifactRepositoryImpl, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil)
+	return ciHandlerImpl
+}

pkg/pipeline/PipelineBuilder.go

@@ -576,13 +576,15 @@ func (impl *PipelineBuilderImpl) getCiTemplateVariables(appId int) (ciConfig *be
 	}
 
 	var regHost string
+	var templateDockerRegistryId string
 	dockerRegistry := template.DockerRegistry
 	if dockerRegistry != nil {
 		regHost, err = dockerRegistry.GetRegistryLocation()
 		if err != nil {
 			impl.logger.Errorw("invalid reg url", "err", err)
 			return nil, err
 		}
+		templateDockerRegistryId = dockerRegistry.Id
 	}
 	ciConfig = &bean.CiConfigRequest{
 		Id:                template.Id,
@@ -599,6 +601,7 @@ func (impl *PipelineBuilderImpl) getCiTemplateVariables(appId int) (ciConfig *be
 		CreatedBy:         template.CreatedBy,
 		CreatedOn:         template.CreatedOn,
 		CiGitMaterialId:   template.GitMaterialId,
+		DockerRegistry:    templateDockerRegistryId,
 	}
 	if dockerRegistry != nil {
 		ciConfig.DockerRegistry = dockerRegistry.Id

pkg/pipeline/WebhookService.go

@@ -23,6 +23,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"github.com/argoproj/argo-workflows/v3/pkg/apis/workflow/v1alpha1"
+	"github.com/aws/aws-sdk-go-v2/service/ecr/types"
 	"github.com/devtron-labs/devtron/client/events"
 	"github.com/devtron-labs/devtron/internal/sql/repository"
 	"github.com/devtron-labs/devtron/internal/sql/repository/pipelineConfig"
@@ -55,6 +56,7 @@ type WebhookService interface {
 	HandleCiSuccessEvent(ciPipelineId int, request *CiArtifactWebhookRequest, imagePushedAt *time.Time) (id int, err error)
 	HandleExternalCiWebhook(externalCiId int, request *CiArtifactWebhookRequest, auth func(token string, projectObject string, envObject string) bool) (id int, err error)
 	HandleCiStepFailedEvent(ciPipelineId int, request *CiArtifactWebhookRequest) (err error)
+	HandleMultipleImagesFromEvent(imageDetails []types.ImageDetail, ciWorkflowId int) (map[string]*pipelineConfig.CiWorkflow, error)
 }
 
 type WebhookServiceImpl struct {
@@ -399,3 +401,44 @@ func (impl *WebhookServiceImpl) BuildPayload(request *CiArtifactWebhookRequest,
 	payload.DockerImageUrl = request.Image
 	return payload
 }
+
+// HandleMultipleImagesFromEvent handles multiple images from plugin and creates ci workflow for n-1 images for mapping in ci_artifact
+func (impl *WebhookServiceImpl) HandleMultipleImagesFromEvent(imageDetails []types.ImageDetail, ciWorkflowId int) (map[string]*pipelineConfig.CiWorkflow, error) {
+	ciWorkflow, err := impl.ciWorkflowRepository.FindById(ciWorkflowId)
+	if err != nil {
+		impl.logger.Errorw("error in finding ci workflow by id ", "err", err, "ciWorkFlowId", ciWorkflowId)
+		return nil, err
+	}
+
+	//creating n-1 workflows for rest images, oldest will be mapped to original workflow id.
+	digestWorkflowMap := make(map[string]*pipelineConfig.CiWorkflow)
+	// mapping oldest to original ciworkflowId
+	digestWorkflowMap[*imageDetails[0].ImageDigest] = ciWorkflow
+	for i := 1; i < len(imageDetails); i++ {
+		workflow := &pipelineConfig.CiWorkflow{
+			Name:               ciWorkflow.Name + fmt.Sprintf("-child-%d", i),
+			Status:             ciWorkflow.Status,
+			PodStatus:          string(v1alpha1.NodeSucceeded),
+			StartedOn:          time.Now(),
+			Namespace:          ciWorkflow.Namespace,
+			LogLocation:        ciWorkflow.LogLocation,
+			TriggeredBy:        ciWorkflow.TriggeredBy,
+			CiPipelineId:       ciWorkflow.CiPipelineId,
+			CiArtifactLocation: ciWorkflow.CiArtifactLocation,
+			BlobStorageEnabled: ciWorkflow.BlobStorageEnabled,
+			PodName:            ciWorkflow.PodName,
+			CiBuildType:        ciWorkflow.CiBuildType,
+			ParentCiWorkFlowId: ciWorkflow.Id,
+			GitTriggers:        ciWorkflow.GitTriggers,
+			Message:            ciWorkflow.Message,
+		}
+		err = impl.ciWorkflowRepository.SaveWorkFlow(workflow)
+		if err != nil {
+			impl.logger.Errorw("error in saving workflow for child workflow", "err", err, "parentCiWorkflowId", ciWorkflowId)
+			return nil, err
+		}
+		digestWorkflowMap[*imageDetails[i].ImageDigest] = workflow
+
+	}
+	return digestWorkflowMap, nil
+}