misc: notification for failed Deployment (#6611)

* notification for failed Deployment

* notification for failed Deployment IN ARGO

* review comments

* async func

* fix

* wire

Author: prkhrkat

Wire.go

@@ -303,6 +303,9 @@ func InitializeApp() (*App, error) {
 		pipeline.NewCiMaterialConfigServiceImpl,
 		wire.Bind(new(pipeline.CiMaterialConfigService), new(*pipeline.CiMaterialConfigServiceImpl)),
 
+		app.NewDeploymentEventHandlerImpl,
+		wire.Bind(new(app.DeploymentEventHandler), new(*app.DeploymentEventHandlerImpl)),
+
 		pipeline.NewAppArtifactManagerImpl,
 		wire.Bind(new(pipeline.AppArtifactManager), new(*pipeline.AppArtifactManagerImpl)),
 		pipeline.NewDevtronAppCMCSServiceImpl,

env_gen.json

@@ -20,7 +20,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"net/url"
 	"strconv"
 	"time"
 
@@ -47,7 +46,6 @@ import (
 
 	"github.com/argoproj/argo-cd/v2/pkg/apis/application/v1alpha1"
 	"github.com/caarlos0/env"
-	"github.com/devtron-labs/common-lib/async"
 	k8sCommonBean "github.com/devtron-labs/common-lib/utils/k8s/commonBean"
 	"github.com/devtron-labs/common-lib/utils/k8s/health"
 	"github.com/devtron-labs/devtron/api/bean"
@@ -126,7 +124,7 @@ type AppServiceImpl struct {
 	deploymentConfigService                common2.DeploymentConfigService
 	envConfigOverrideReadService           read.EnvConfigOverrideService
 	cdWorkflowRunnerService                cd.CdWorkflowRunnerService
-	asyncRunnable                          *async.Runnable
+	deploymentEventHandler                 DeploymentEventHandler
 }
 
 type AppService interface {
@@ -135,7 +133,6 @@ type AppService interface {
 	GetConfigMapAndSecretJson(appId int, envId int) ([]byte, error)
 	UpdateCdWorkflowRunnerByACDObject(app *v1alpha1.Application, cdWfrId int, updateTimedOutStatus bool) error
 	UpdateDeploymentStatusForGitOpsPipelines(app *v1alpha1.Application, applicationClusterId int, statusTime time.Time, isAppStore bool) (bool, bool, *chartConfig.PipelineOverride, error)
-	WriteCDSuccessEvent(appId int, envId int, override *chartConfig.PipelineOverride)
 	CreateGitOpsRepo(app *app.App, targetRevision string, userId int32) (gitopsRepoName string, chartGitAttr *commonBean.ChartGitAttribute, err error)
 
 	// TODO: move inside reader service
@@ -168,7 +165,7 @@ func NewAppService(
 	deploymentConfigService common2.DeploymentConfigService,
 	envConfigOverrideReadService read.EnvConfigOverrideService,
 	cdWorkflowRunnerService cd.CdWorkflowRunnerService,
-	asyncRunnable *async.Runnable) *AppServiceImpl {
+	deploymentEventHandler DeploymentEventHandler) *AppServiceImpl {
 	appServiceImpl := &AppServiceImpl{
 		mergeUtil:                              mergeUtil,
 		pipelineOverrideRepository:             pipelineOverrideRepository,
@@ -199,7 +196,7 @@ func NewAppService(
 		deploymentConfigService:                deploymentConfigService,
 		envConfigOverrideReadService:           envConfigOverrideReadService,
 		cdWorkflowRunnerService:                cdWorkflowRunnerService,
-		asyncRunnable:                          asyncRunnable,
+		deploymentEventHandler:                 deploymentEventHandler,
 	}
 	return appServiceImpl
 }
@@ -309,6 +306,7 @@ func (impl *AppServiceImpl) UpdateDeploymentStatusForGitOpsPipelines(app *v1alph
 				impl.logger.Errorw("error on update cd workflow runner", "CdWorkflowId", pipelineOverride.CdWorkflowId, "status", cdWorkflow2.WorkflowTimedOut, "err", err)
 				return isSucceeded, isTimelineUpdated, pipelineOverride, err
 			}
+			impl.deploymentEventHandler.WriteCDNotificationEventAsync(cdPipeline.AppId, cdPipeline.EnvironmentId, pipelineOverride, eventUtil.Fail)
 			return isSucceeded, isTimelineUpdated, pipelineOverride, nil
 		}
 		if reconciledAt.IsZero() || (kubectlSyncedTimeline != nil && kubectlSyncedTimeline.Id > 0 && reconciledAt.After(kubectlSyncedTimeline.StatusTime)) {
@@ -325,7 +323,7 @@ func (impl *AppServiceImpl) UpdateDeploymentStatusForGitOpsPipelines(app *v1alph
 				}
 				if isSucceeded {
 					impl.logger.Infow("writing cd success event", "gitHash", gitHash, "pipelineOverride", pipelineOverride)
-					impl.asyncRunnable.Execute(func() { impl.WriteCDSuccessEvent(cdPipeline.AppId, cdPipeline.EnvironmentId, pipelineOverride) })
+					impl.deploymentEventHandler.WriteCDNotificationEventAsync(cdPipeline.AppId, cdPipeline.EnvironmentId, pipelineOverride, eventUtil.Success)
 				}
 			} else {
 				impl.logger.Debugw("event received for older triggered revision", "gitHash", gitHash)
@@ -773,23 +771,6 @@ func (impl *AppServiceImpl) UpdatePipelineStatusTimelineForApplicationChanges(ap
 	return isTimelineUpdated, isTimelineTimedOut, kubectlApplySyncedTimeline, nil
 }
 
-func (impl *AppServiceImpl) WriteCDSuccessEvent(appId int, envId int, override *chartConfig.PipelineOverride) {
-	event, _ := impl.eventFactory.Build(eventUtil.Success, &override.PipelineId, appId, &envId, eventUtil.CD)
-	impl.logger.Debugw("event WriteCDSuccessEvent", "event", event, "override", override)
-	event = impl.eventFactory.BuildExtraCDData(event, nil, override.Id, bean.CD_WORKFLOW_TYPE_DEPLOY)
-	_, evtErr := impl.eventClient.WriteNotificationEvent(event)
-	if evtErr != nil {
-		impl.logger.Errorw("error in writing event", "event", event, "err", evtErr)
-	}
-}
-
-func (impl *AppServiceImpl) BuildCDSuccessPayload(appName string, environmentName string) *client.Payload {
-	payload := &client.Payload{}
-	payload.AppName = appName
-	payload.EnvName = environmentName
-	return payload
-}
-
 type ValuesOverrideResponse struct {
 	MergedValues         string
 	ReleaseOverrideJSON  string
@@ -887,29 +868,6 @@ type PipelineMaterialInfo struct {
 	CommitHash         string
 }
 
-func buildCDTriggerEvent(impl *AppServiceImpl, overrideRequest *bean.ValuesOverrideRequest, pipeline *pipelineConfig.Pipeline,
-	envOverride *chartConfig.EnvConfigOverride, materialInfo map[string]string, artifact *repository.CiArtifact) client.Event {
-	event, _ := impl.eventFactory.Build(eventUtil.Trigger, &pipeline.Id, pipeline.AppId, &pipeline.EnvironmentId, eventUtil.CD)
-	return event
-}
-
-func (impl *AppServiceImpl) BuildPayload(overrideRequest *bean.ValuesOverrideRequest, pipeline *pipelineConfig.Pipeline,
-	envOverride *chartConfig.EnvConfigOverride, materialInfo map[string]string, artifact *repository.CiArtifact) *client.Payload {
-	payload := &client.Payload{}
-	payload.AppName = pipeline.App.AppName
-	payload.PipelineName = pipeline.Name
-	payload.EnvName = envOverride.Environment.Name
-
-	var revision string
-	for _, v := range materialInfo {
-		revision = v
-		break
-	}
-	payload.Source = url.PathEscape(revision)
-	payload.DockerImageUrl = artifact.Image
-	return payload
-}
-
 type ReleaseAttributes struct {
 	Name           string
 	Tag            string

pkg/app/AppService.go

@@ -0,0 +1,64 @@
+/*
+ * Copyright (c) 2020-2024. Devtron Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package app
+
+import (
+	"github.com/devtron-labs/common-lib/async"
+	"github.com/devtron-labs/devtron/api/bean"
+	client "github.com/devtron-labs/devtron/client/events"
+	"github.com/devtron-labs/devtron/internal/sql/repository/chartConfig"
+	util "github.com/devtron-labs/devtron/util/event"
+	"go.uber.org/zap"
+)
+
+type DeploymentEventHandler interface {
+	WriteCDNotificationEventAsync(appId int, envId int, override *chartConfig.PipelineOverride, eventType util.EventType)
+}
+
+type DeploymentEventHandlerImpl struct {
+	logger        *zap.SugaredLogger
+	eventFactory  client.EventFactory
+	eventClient   client.EventClient
+	asyncRunnable *async.Runnable
+}
+
+func NewDeploymentEventHandlerImpl(logger *zap.SugaredLogger, eventClient client.EventClient, eventFactory client.EventFactory, asyncRunnable *async.Runnable) *DeploymentEventHandlerImpl {
+	deploymentEventHandlerImpl := &DeploymentEventHandlerImpl{
+		logger:        logger,
+		eventClient:   eventClient,
+		eventFactory:  eventFactory,
+		asyncRunnable: asyncRunnable,
+	}
+	return deploymentEventHandlerImpl
+}
+
+func (impl *DeploymentEventHandlerImpl) writeCDNotificationEvent(appId int, envId int, override *chartConfig.PipelineOverride, eventType util.EventType) {
+	event, _ := impl.eventFactory.Build(eventType, &override.PipelineId, appId, &envId, util.CD)
+	impl.logger.Debugw("event WriteCDNotificationEvent", "event", event, "override", override)
+	event = impl.eventFactory.BuildExtraCDData(event, nil, override.Id, bean.CD_WORKFLOW_TYPE_DEPLOY)
+	_, evtErr := impl.eventClient.WriteNotificationEvent(event)
+	if evtErr != nil {
+		impl.logger.Errorw("error in writing event", "event", event, "err", evtErr)
+	}
+}
+
+// WriteCDNotificationEventAsync executes WriteCDNotificationEvent in a panic-safe goroutine
+func (impl *DeploymentEventHandlerImpl) WriteCDNotificationEventAsync(appId int, envId int, override *chartConfig.PipelineOverride, eventType util.EventType) {
+	impl.asyncRunnable.Execute(func() {
+		impl.writeCDNotificationEvent(appId, envId, override, eventType)
+	})
+}

pkg/app/DeploymentEventHandler.go

@@ -22,7 +22,6 @@ import (
 	"os"
 	"time"
 
-	"github.com/devtron-labs/common-lib/async"
 	pubsub "github.com/devtron-labs/common-lib/pubsub-lib"
 	util5 "github.com/devtron-labs/common-lib/utils/k8s"
 	bean3 "github.com/devtron-labs/devtron/api/bean"
@@ -168,7 +167,7 @@ type HandlerServiceImpl struct {
 	ciLogService                        pipeline.CiLogService
 	workflowService                     executor.WorkflowService
 	blobConfigStorageService            pipeline.BlobStorageConfigService
-	asyncRunnable                       *async.Runnable
+	deploymentEventHandler              app.DeploymentEventHandler
 }
 
 func NewHandlerServiceImpl(logger *zap.SugaredLogger,
@@ -230,7 +229,7 @@ func NewHandlerServiceImpl(logger *zap.SugaredLogger,
 	ciLogService pipeline.CiLogService,
 	workflowService executor.WorkflowService,
 	blobConfigStorageService pipeline.BlobStorageConfigService,
-	asyncRunnable *async.Runnable) (*HandlerServiceImpl, error) {
+	deploymentEventHandler app.DeploymentEventHandler) (*HandlerServiceImpl, error) {
 	impl := &HandlerServiceImpl{
 		logger:                              logger,
 		cdWorkflowCommonService:             cdWorkflowCommonService,
@@ -296,7 +295,7 @@ func NewHandlerServiceImpl(logger *zap.SugaredLogger,
 		ciLogService:             ciLogService,
 		workflowService:          workflowService,
 		blobConfigStorageService: blobConfigStorageService,
-		asyncRunnable:            asyncRunnable,
+		deploymentEventHandler:   deploymentEventHandler,
 	}
 	config, err := types.GetCdConfig()
 	if err != nil {

pkg/app/deploymentEventHandler.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	util2 "github.com/devtron-labs/devtron/util/event"
 	"net/http"
 	"path"
 	"regexp"
@@ -317,6 +318,12 @@ func (impl *HandlerServiceImpl) ManualCdTrigger(triggerContext bean.TriggerConte
 			if err != nil {
 				impl.logger.Errorw("error while updating current runner status to failed", "cdWfr", runner.Id, "err", err)
 			}
+			pipelineOverride, err := impl.pipelineOverrideRepository.FindLatestByCdWorkflowId(runner.CdWorkflowId)
+			if err != nil {
+				impl.logger.Errorw("error in getting latest pipeline override by cdWorkflowId", "err", err, "cdWorkflowId", runner.CdWorkflowId)
+				return 0, "", nil, err
+			}
+			impl.deploymentEventHandler.WriteCDNotificationEventAsync(pipelineOverride.Pipeline.AppId, pipelineOverride.Pipeline.EnvironmentId, pipelineOverride, util2.Fail)
 			return 0, "", nil, releaseErr
 		}