updated: review comments

internal/util/CollectionUtil.go

@@ -27,3 +27,10 @@ func CompareUnOrdered(a, b []int) bool {
 	sort.Ints(b)
 	return cmp.Equal(a, b)
 }
+
+func GetMaxInteger(a, b int) int {
+	if a > b {
+		return a
+	}
+	return b
+}

pkg/app/AppService.go

@@ -105,7 +105,8 @@ type AppServiceConfig struct {
 	HelmPipelineStatusCheckEligibleTime string `env:"HELM_PIPELINE_STATUS_CHECK_ELIGIBLE_TIME" envDefault:"120"`             //in seconds
 	ExposeCDMetrics                     bool   `env:"EXPOSE_CD_METRICS" envDefault:"false"`
 	EnableAsyncInstallDevtronChart      bool   `env:"ENABLE_ASYNC_INSTALL_DEVTRON_CHART" envDefault:"true"`
-	DevtronChartInstallTimeout          int    `env:"DEVTRON_CHART_INSTALL_TIMEOUT" envDefault:"6"` //in minutes
+	DevtronChartInstallRequestTimeout   int    `env:"DEVTRON_CHART_INSTALL_REQUEST_TIMEOUT" envDefault:"6"` //in minutes
+	HelmInstallationTimeout             int    `env:"HELM_INSTALLATION_TIMEOUT" envDefault:"5"`             //in minutes
 }
 
 func GetAppServiceConfig() (*AppServiceConfig, error) {
@@ -1892,35 +1893,41 @@ func (impl *AppServiceImpl) GetTriggerEvent(deploymentAppType string, triggeredA
 
 func (impl *AppServiceImpl) HandleCDTriggerRelease(overrideRequest *bean.ValuesOverrideRequest, ctx context.Context, triggeredAt time.Time, deployedBy int32) (releaseNo int, manifest []byte, err error) {
 	if impl.isDevtronAsyncInstallModeEnabled(overrideRequest.DeploymentAppType) {
-		event := &bean.AsyncCdDeployEvent{
-			ValuesOverrideRequest: overrideRequest,
-			TriggeredAt:           triggeredAt,
-			TriggeredBy:           deployedBy,
-		}
-		payload, err := json.Marshal(event)
-		if err != nil {
-			impl.logger.Errorw("failed to marshal helm async CD deploy event request", "request", event, "err", err)
-			return 0, manifest, err
-		}
-
-		// publish nats event for async installation
-		err = impl.pubsubClient.Publish(pubsub.DEVTRON_CHART_INSTALL_TOPIC, string(payload))
-		if err != nil {
-			impl.logger.Errorw("failed to publish trigger request event", "topic", pubsub.DEVTRON_CHART_INSTALL_TOPIC, "payload", payload, "err", err)
-		}
-
-		//update workflow runner status, used in app workflow view
-		err = impl.UpdateCDWorkflowRunnerStatus(ctx, overrideRequest, triggeredAt, pipelineConfig.WorkflowInQueue)
-		if err != nil {
-			return 0, manifest, err
-		}
-		return 0, manifest, nil
+		// asynchronous mode of installation
+		return impl.TriggerHelmAsyncRelease(overrideRequest, ctx, triggeredAt, deployedBy)
 	}
 	// synchronous mode of installation
 	return impl.TriggerRelease(overrideRequest, ctx, triggeredAt, deployedBy)
 }
 
-// TriggerRelease used to trigger Install/Upgrade Devtron App releases synchronously
+// TriggerHelmAsyncRelease will publish async helm Install/Upgrade request event for Devtron App releases
+func (impl *AppServiceImpl) TriggerHelmAsyncRelease(overrideRequest *bean.ValuesOverrideRequest, ctx context.Context, triggeredAt time.Time, triggeredBy int32) (releaseNo int, manifest []byte, err error) {
+	event := &bean.AsyncCdDeployEvent{
+		ValuesOverrideRequest: overrideRequest,
+		TriggeredAt:           triggeredAt,
+		TriggeredBy:           triggeredBy,
+	}
+	payload, err := json.Marshal(event)
+	if err != nil {
+		impl.logger.Errorw("failed to marshal helm async CD deploy event request", "request", event, "err", err)
+		return 0, manifest, err
+	}
+
+	// publish nats event for async installation
+	err = impl.pubsubClient.Publish(pubsub.DEVTRON_CHART_INSTALL_TOPIC, string(payload))
+	if err != nil {
+		impl.logger.Errorw("failed to publish trigger request event", "topic", pubsub.DEVTRON_CHART_INSTALL_TOPIC, "payload", payload, "err", err)
+	}
+
+	//update workflow runner status, used in app workflow view
+	err = impl.UpdateCDWorkflowRunnerStatus(ctx, overrideRequest, triggeredAt, pipelineConfig.WorkflowInQueue)
+	if err != nil {
+		return 0, manifest, err
+	}
+	return 0, manifest, nil
+}
+
+// TriggerRelease will trigger Install/Upgrade request for Devtron App releases synchronously
 func (impl *AppServiceImpl) TriggerRelease(overrideRequest *bean.ValuesOverrideRequest, ctx context.Context, triggeredAt time.Time, triggeredBy int32) (releaseNo int, manifest []byte, err error) {
 	// Handling for auto trigger
 	if overrideRequest.UserId == 0 {
@@ -3149,7 +3156,7 @@ func (impl *AppServiceImpl) createHelmAppForCdPipeline(overrideRequest *bean.Val
 				HistoryMax:        impl.helmAppService.GetRevisionHistoryMaxValue(client2.SOURCE_DEVTRON_APP),
 				ChartContent:      &client2.ChartContent{Content: referenceChartByte},
 			}
-
+			// For cases where helm release was not found, kubelink will install the same configuration
 			updateApplicationResponse, err := impl.helmAppClient.UpdateApplication(ctx, req)
 			if err != nil {
 				impl.logger.Errorw("error in updating helm application for cd pipeline", "err", err)
@@ -3244,7 +3251,7 @@ func (impl *AppServiceImpl) UpdateCDWorkflowRunnerStatus(ctx context.Context, ov
 		// if the current cdWfr status is already a terminal status and then don't update the status
 		// e.g: Status : Failed --> Progressing (not allowed)
 		if slices.Contains(pipelineConfig.WfrTerminalStatusList, cdWfr.Status) {
-			impl.logger.Errorw("deployment has already been terminated for workflow runner", "workflowRunnerId", cdWfr.Id, "err", err)
+			impl.logger.Warnw("deployment has already been terminated for workflow runner", "workflowRunnerId", cdWfr.Id, "err", err)
 			return nil
 		}
 		cdWfr.Status = status

pkg/pipeline/CdHandler.go

@@ -479,7 +479,12 @@ func (impl *CdHandlerImpl) CheckHelmAppStatusPeriodicallyAndUpdateInDb(helmPipel
 			impl.Logger.Warnw("release mismatched, skipping helm apps status update for this trigger", "appIdentifier", appIdentifier, "err", err)
 			continue
 		} else if helmInstalledDevtronApp.GetReleaseStatus() == serverBean.HelmReleaseStatusPendingInstall {
-			if time.Now().After(helmInstalledDevtronApp.GetLastDeployed().AsTime().Add(5 * time.Minute)) {
+			appServiceConfig, err := app.GetAppServiceConfig()
+			if err != nil {
+				impl.Logger.Errorw("error in parsing app status config variables", "err", err)
+				continue
+			}
+			if time.Now().After(helmInstalledDevtronApp.GetLastDeployed().AsTime().Add(time.Duration(appServiceConfig.HelmInstallationTimeout) * time.Minute)) {
 				// If release status is in pending-install for more than 5 mins, then mark the deployment as failure
 				wfr.Status = pipelineConfig.WorkflowFailed
 				wfr.FinishedOn = time.Now()

pkg/pipeline/WorkflowDagExecutor.go

@@ -265,7 +265,7 @@ func NewWorkflowDagExecutorImpl(Logger *zap.SugaredLogger, pipelineRepository pi
 	if err != nil {
 		return nil
 	}
-	err = wde.SubscribeDevtronAsyncInstallRequest()
+	err = wde.SubscribeDevtronAsyncHelmInstallRequest()
 	if err != nil {
 		return nil
 	}
@@ -312,55 +312,52 @@ func (impl *WorkflowDagExecutorImpl) Subscribe() error {
 	return nil
 }
 
-func (impl *WorkflowDagExecutorImpl) SubscribeDevtronAsyncInstallRequest() error {
+func (impl *WorkflowDagExecutorImpl) extractOverrideRequestFromCDAsyncInstallEvent(msg *pubsub.PubSubMsg) *bean.AsyncCdDeployEvent {
+	CDAsyncInstallNatsMessage := &bean.AsyncCdDeployEvent{}
+	err := json.Unmarshal([]byte(msg.Data), CDAsyncInstallNatsMessage)
+	if err != nil {
+		impl.logger.Errorw("error in unmarshalling CD async install request nats message", "err", err)
+		return nil
+	}
+	overrideRequest := CDAsyncInstallNatsMessage.ValuesOverrideRequest
+	pipeline, err := impl.pipelineRepository.FindById(overrideRequest.PipelineId)
+	if err != nil {
+		impl.logger.Errorw("error in fetching pipeline by pipelineId", "err", err)
+		return nil
+	}
+	impl.appService.SetPipelineFieldsInOverrideRequest(overrideRequest, pipeline)
+	return CDAsyncInstallNatsMessage
+}
+
+func (impl *WorkflowDagExecutorImpl) SubscribeDevtronAsyncHelmInstallRequest() error {
 	callback := func(msg *pubsub.PubSubMsg) {
-		impl.logger.Debug("received Devtron App helm async install request event, SubscribeDevtronAsyncInstallRequest", "data", msg.Data)
-		CDAsyncInstallNatsMessage := &bean.AsyncCdDeployEvent{}
-		err := json.Unmarshal([]byte(msg.Data), CDAsyncInstallNatsMessage)
-		if err != nil {
-			impl.logger.Errorw("error in unmarshalling CD async install request nats message, SubscribeDevtronAsyncInstallRequest", "err", err)
-			return
-		}
+		impl.logger.Debug("received Devtron App helm async install request event, SubscribeDevtronAsyncHelmInstallRequest", "data", msg.Data)
+		CDAsyncInstallNatsMessage := impl.extractOverrideRequestFromCDAsyncInstallEvent(msg)
 		overrideRequest := CDAsyncInstallNatsMessage.ValuesOverrideRequest
-		appServiceConfig, err := app.GetAppServiceConfig()
-		if err != nil {
-			impl.logger.Errorw("error in parsing app status config variables, SubscribeDevtronAsyncInstallRequest", "err", err)
-			return
-		}
-		pipeline, err := impl.pipelineRepository.FindById(overrideRequest.PipelineId)
-		if err != nil {
-			impl.logger.Errorw("error in fetching pipeline by pipelineId, SubscribeDevtronAsyncInstallRequest", "err", err)
-			return
-		}
-		impl.appService.SetPipelineFieldsInOverrideRequest(overrideRequest, pipeline)
-		if overrideRequest.DeploymentAppType != bean2.Helm {
-			impl.logger.Errorw("invalid deployment type for CD async install event, SubscribeDevtronAsyncInstallRequest", "overrideRequest", overrideRequest)
-			return
-		}
-		if overrideRequest.WfrId < 1 {
-			impl.logger.Errorw("invalid overrideRequest WfrId for CD async install event, SubscribeDevtronAsyncInstallRequest", "overrideRequest", overrideRequest)
-			return
-		}
 		cdWfr, err := impl.cdWorkflowRepository.FindWorkflowRunnerById(overrideRequest.WfrId)
 		if err != nil && err != pg.ErrNoRows {
-			impl.logger.Errorw("err on fetching cd workflow, SubscribeDevtronAsyncInstallRequest", "err", err)
+			impl.logger.Errorw("err on fetching cd workflow, SubscribeDevtronAsyncHelmInstallRequest", "err", err)
 			return
 		}
-
 		// skip if the cdWfr.Status is already in a terminal state
 		if cdWfr != nil && slices.Contains(pipelineConfig.WfrTerminalStatusList, cdWfr.Status) {
-			impl.logger.Errorw("err on fetching cd workflow, SubscribeDevtronAsyncInstallRequest", "err", err)
+			impl.logger.Warnw("skipped deployment as the runner status is already in terminal state", "cdWfrId", cdWfr, "status", cdWfr.Status)
 			return
 		}
-
-		ctx, cancel := context.WithTimeout(context.Background(), time.Duration(appServiceConfig.DevtronChartInstallTimeout)*time.Minute)
+		appServiceConfig, err := app.GetAppServiceConfig()
+		if err != nil {
+			impl.logger.Errorw("error in parsing app status config variables, SubscribeDevtronAsyncHelmInstallRequest", "err", err)
+			return
+		}
+		timeout := util.GetMaxInteger(appServiceConfig.HelmInstallationTimeout, appServiceConfig.DevtronChartInstallRequestTimeout)
+		ctx, cancel := context.WithTimeout(context.Background(), time.Duration(timeout)*time.Minute)
 		defer cancel()
 		_, span := otel.Tracer("orchestrator").Start(ctx, "appService.TriggerRelease")
 		releaseId, _, releaseErr := impl.appService.TriggerRelease(overrideRequest, ctx, CDAsyncInstallNatsMessage.TriggeredAt, CDAsyncInstallNatsMessage.TriggeredBy)
 		span.End()
 		if releaseErr != nil {
 			if err = impl.MarkCurrentDeploymentFailed(cdWfr, releaseErr, overrideRequest.UserId); err != nil {
-				impl.logger.Errorw("error while updating current runner status to failed, SubscribeDevtronAsyncInstallRequest", "cdWfr", cdWfr.Id, "err", err)
+				impl.logger.Errorw("error while updating current runner status to failed, SubscribeDevtronAsyncHelmInstallRequest", "cdWfr", cdWfr.Id, "err", err)
 			}
 			return
 		} else {