feat: GitOps async install for devtron applications (#5169)

* chore: context refactoring for gitops async

* chore: refactored context for argo app deploy operations

* tracer refactoring wip

* fixed: timeline status update issue and added force sync param - WIP

* fixed: undeclared varibale

* env gen updated

* common-lib version updated

* async deployment gitops and CEL expression

* chore: cel library added

* chore: go mod tidy

* feat: Async gitops deployment flow

* updated trigger pipeline scope

* triggerPipeline refactoring

* runner status update

* udpated pipelineOverride object in GetValuesOverrideForTrigger

* updated UserDeploymentRequest status

* cron method: incomplete deployment request

* updated alias naming

* updated alias naming

* updated go-bitbucket librabry

* added github.com/ash-exp/go-bitbucket

* chore: refactoring

* added migration script

* self review fixes - wip

* chore: ProcessIncompleteDeploymentReq updated

* updated migration and added util test cases

* updated test file name

* fix: trigger history audit step

* fix: status update for trigger audit

* chore: tracer refactoring

* removed comment

* removed duplicate status updates

* added NATS graceful shutdown

* fixed wire injection issue

* fixed: migration issues

* fix: deployment status update

* fix: nil UserDeploymentRequest model on save

* fix: UserDeploymentRequestRepositoryImpl.FindByCdWfIds nil model

* udpated validateStatusUpdate for user  deployment request

* removed duplicate flag

* fix: nil pointer in logger

* updated save userDeploymentRequest func

* updated save userDeploymentRequest func

* udpated: extractAsyncCdDeployRequestFromEventMsg

* fix: handled panic issues

* updated MarkCurrentDeploymentFailed func

* added tracer

* fix: unauthorized on async deployment

* fixed: updateTriggerEventForIncompleteRequest

* fixed envOverride.Environment data

* updateTriggerEventForIncompleteRequest for no gitops deployment types

* updated CreateHistoriesForDeploymentTrigger

* fix: updated history status

* fix: create history error

* fix: UpdateStatusForCdWfIds

* nil pointer check fix

* wire gen file

* removed unnecessary loggers

* fix: logger type

* create cm/cs history in transaction

* wirer gen file

* fix: nil pointer handling

* fix: TerminateForPipelineId func

* fix: UserDeploymentRequest db struct

* updated UpdateCDWorkflowRunnerStatus func

* fix: logger noise from markImageScanDeployed

* updated auto trigger loggers and ProcessConcurrentAsyncDeploymentReq context

* refactored: saveDeploymentTriggerHistory

* refactored: trigger async release

* fix: GetAllInCompleteRequests query

* fix: UNABLE_TO_FETCH timeline handling

* fix: async deployment request lock

* feat: gitops async with timeline (#5286)

* feat: replace user_deployment_request.status with pipeline_status_timelines

* updated schema migration script

* fix: migration script

* update readme to include multi arch flag (#4998)

* updated trigger event func and timeline names

* review comment incorporation

* fix: GetAllInCompleteRequests query

* fix: GetAllInCompleteRequests query

* fixed: UpdatePreviousQueuedRunnerStatus error

* fix: loogers

* doc: Added enhancements to security doc (#5203)

* Rough Draft Initiated

* Created Security Enhancement Docs + Other Adhoc Fixes

* Incorporated PM Feedback

* Modified Portforward Section to Kubectl Section (#5236)

* feat: cd pipeline deployment history refactoring (#5200)

* searchableKey service move

* migrated cd deployment history code

* idempotency of triggerPipeline func and minor fixes

* timeline fixes

* updated: NewTriggerEvent func

* updated ErrorDeploymentSuperseded err handling

* updated: GetArgoPipelinesHavingLatestTriggerStuckInNonTerminalStatuses statuses

* fix: timeline status

* refactored: UpdatePreviousQueuedRunnerStatus

* fix: GetPreviousQueuedRunners query

* fix: UpdateRunnerStatusToFailedForIds query

* fix: buildTriggerEventForOverrideRequest func

* fix: UpdateRunnerStatusToFailedForIds query

* fix: UpdateRunnerStatusToFailedForIds query

* updated: handleAsyncTriggerReleaseError func

* fix: concurrent case handlings

* updated folder name

* updated GetLastStatusPublishedTimeForWfrId func

* feat: removed audit timeline and error handlings

* fix: FetchTimelinesForWfrIdExcludingStatuses query

* fix: deleted pipeline flow

* timeline handling

* timeline handling fixed

* removed unused const

* chore: updated go-bitbucket version

* removed unnecessary alais

---------

Co-authored-by: ayu-devtron <[email protected]>
Co-authored-by: ashokdevtron <[email protected]>
Co-authored-by: kartik-579 <[email protected]>

* main branch merge

* vendor files updated

* udpated migration number

* fix: build issues

* chore: updated migration number

* fix: context canceling issue

* feat: git commit retry

* updated migration number

* fix: bitbucket git commit issue

* feat: env varibales for retry count and delay

* migration script updated

---------

Co-authored-by: ayu-devtron <[email protected]>
Co-authored-by: ashokdevtron <[email protected]>
Co-authored-by: kartik-579 <[email protected]>

App.go

@@ -21,7 +21,9 @@ import (
 	"crypto/tls"
 	"fmt"
 	"github.com/devtron-labs/common-lib/middlewares"
+	pubsub "github.com/devtron-labs/common-lib/pubsub-lib"
 	"github.com/devtron-labs/devtron/pkg/eventProcessor"
+	"github.com/devtron-labs/devtron/pkg/eventProcessor/in"
 	"log"
 	"net/http"
 	"os"
@@ -53,10 +55,12 @@ type App struct {
 	posthogClient         *telemetry.PosthogClient
 	centralEventProcessor *eventProcessor.CentralEventProcessor
 	// used for local dev only
-	serveTls           bool
-	sessionManager2    *authMiddleware.SessionManager
-	OtelTracingService *otel.OtelTracingServiceImpl
-	loggingMiddleware  util.LoggingMiddleware
+	serveTls                   bool
+	sessionManager2            *authMiddleware.SessionManager
+	OtelTracingService         *otel.OtelTracingServiceImpl
+	loggingMiddleware          util.LoggingMiddleware
+	pubSubClient               *pubsub.PubSubClientServiceImpl
+	workflowEventProcessorImpl *in.WorkflowEventProcessorImpl
 }
 
 func NewApp(router *router.MuxRouter,
@@ -68,21 +72,25 @@ func NewApp(router *router.MuxRouter,
 	posthogClient *telemetry.PosthogClient,
 	loggingMiddleware util.LoggingMiddleware,
 	centralEventProcessor *eventProcessor.CentralEventProcessor,
+	pubSubClient *pubsub.PubSubClientServiceImpl,
+	workflowEventProcessorImpl *in.WorkflowEventProcessorImpl,
 ) *App {
 	//check argo connection
 	//todo - check argo-cd version on acd integration installation
 	app := &App{
-		MuxRouter:             router,
-		Logger:                Logger,
-		SSE:                   sse,
-		Enforcer:              enforcer,
-		db:                    db,
-		serveTls:              false,
-		sessionManager2:       sessionManager2,
-		posthogClient:         posthogClient,
-		OtelTracingService:    otel.NewOtelTracingServiceImpl(Logger),
-		loggingMiddleware:     loggingMiddleware,
-		centralEventProcessor: centralEventProcessor,
+		MuxRouter:                  router,
+		Logger:                     Logger,
+		SSE:                        sse,
+		Enforcer:                   enforcer,
+		db:                         db,
+		serveTls:                   false,
+		sessionManager2:            sessionManager2,
+		posthogClient:              posthogClient,
+		OtelTracingService:         otel.NewOtelTracingServiceImpl(Logger),
+		loggingMiddleware:          loggingMiddleware,
+		centralEventProcessor:      centralEventProcessor,
+		pubSubClient:               pubSubClient,
+		workflowEventProcessorImpl: workflowEventProcessorImpl,
 	}
 	return app
 }
@@ -132,14 +140,19 @@ func (app *App) Start() {
 
 func (app *App) Stop() {
 	app.Logger.Info("orchestrator shutdown initiating")
+	err := app.pubSubClient.ShutDown()
+	if err != nil {
+		app.Logger.Errorw("error in NATS client shutdown", "err", err)
+	}
+	app.workflowEventProcessorImpl.ShutDownDevtronAppReleaseContext()
 	posthogCl := app.posthogClient.Client
 	if posthogCl != nil {
 		app.Logger.Info("flushing messages of posthog")
 		posthogCl.Close()
 	}
 	timeoutContext, _ := context.WithTimeout(context.Background(), 5*time.Second)
 	app.Logger.Infow("closing router")
-	err := app.server.Shutdown(timeoutContext)
+	err = app.server.Shutdown(timeoutContext)
 	if err != nil {
 		app.Logger.Errorw("error in mux router shutdown", "err", err)
 	}

Wire.go

@@ -72,6 +72,7 @@ import (
 	"github.com/devtron-labs/devtron/api/terminal"
 	util5 "github.com/devtron-labs/devtron/api/util"
 	webhookHelm "github.com/devtron-labs/devtron/api/webhook/helm"
+	"github.com/devtron-labs/devtron/cel"
 	"github.com/devtron-labs/devtron/client/argocdServer"
 	"github.com/devtron-labs/devtron/client/argocdServer/application"
 	cluster2 "github.com/devtron-labs/devtron/client/argocdServer/cluster"
@@ -121,6 +122,7 @@ import (
 	delete2 "github.com/devtron-labs/devtron/pkg/delete"
 	deployment2 "github.com/devtron-labs/devtron/pkg/deployment"
 	git2 "github.com/devtron-labs/devtron/pkg/deployment/gitOps/git"
+	"github.com/devtron-labs/devtron/pkg/deployment/manifest/publish"
 	"github.com/devtron-labs/devtron/pkg/deploymentGroup"
 	"github.com/devtron-labs/devtron/pkg/dockerRegistry"
 	"github.com/devtron-labs/devtron/pkg/eventProcessor"
@@ -926,8 +928,8 @@ func InitializeApp() (*App, error) {
 		wire.Bind(new(executors.SystemWorkflowExecutor), new(*executors.SystemWorkflowExecutorImpl)),
 		repository5.NewManifestPushConfigRepository,
 		wire.Bind(new(repository5.ManifestPushConfigRepository), new(*repository5.ManifestPushConfigRepositoryImpl)),
-		app.NewGitOpsManifestPushServiceImpl,
-		wire.Bind(new(app.GitOpsPushService), new(*app.GitOpsManifestPushServiceImpl)),
+		publish.NewGitOpsManifestPushServiceImpl,
+		wire.Bind(new(publish.GitOpsPushService), new(*publish.GitOpsManifestPushServiceImpl)),
 
 		// start: docker registry wire set injection
 		router.NewDockerRegRouterImpl,
@@ -965,6 +967,9 @@ func InitializeApp() (*App, error) {
 		wire.Bind(new(imageDigestPolicy.ImageDigestPolicyService), new(*imageDigestPolicy.ImageDigestPolicyServiceImpl)),
 
 		appStoreRestHandler.AppStoreWireSet,
+
+		cel.NewCELServiceImpl,
+		wire.Bind(new(cel.EvaluatorService), new(*cel.EvaluatorServiceImpl)),
 	)
 	return &App{}, nil
 }

api/bean/ValuesOverrideRequest.go

@@ -49,28 +49,38 @@ func (workflowType WorkflowType) WorkflowTypeToStageType() repository.PipelineSt
 	}
 }
 
+func (workflowType WorkflowType) IsStageTypeDeploy() bool {
+	switch workflowType {
+	case CD_WORKFLOW_TYPE_DEPLOY:
+		return true
+	}
+	return false
+}
+
 type ValuesOverrideRequest struct {
 	PipelineId                            int                         `json:"pipelineId" validate:"required"`
 	AppId                                 int                         `json:"appId" validate:"required"`
 	CiArtifactId                          int                         `json:"ciArtifactId" validate:"required"`
 	AdditionalOverride                    json.RawMessage             `json:"additionalOverride,omitempty"`
-	TargetDbVersion                       int                         `json:"targetDbVersion"`
 	ForceTrigger                          bool                        `json:"forceTrigger,notnull"`
 	DeploymentTemplate                    string                      `json:"strategy,omitempty"` // validate:"oneof=BLUE-GREEN ROLLING"`
 	DeploymentWithConfig                  DeploymentConfigurationType `json:"deploymentWithConfig"`
-	WfrIdForDeploymentWithSpecificTrigger int                         `json:"wfrIdForDeploymentWithSpecificTrigger"`
+	WfrIdForDeploymentWithSpecificTrigger int                         `json:"wfrIdForDeploymentWithSpecificTrigger"` // target cd_workflow_runner_id for rollback. Used in rollback deployment cases
 	CdWorkflowType                        WorkflowType                `json:"cdWorkflowType,notnull"`
 	WfrId                                 int                         `json:"wfrId,notnull"`
 	CdWorkflowId                          int                         `json:"cdWorkflowId"`
-	PipelineOverrideId                    int                         `json:"pipelineOverrideId"` //required for async install/upgrade event;
-	DeploymentType                        models.DeploymentType       `json:"deploymentType"`     //required for async install/upgrade handling; previously if was used internally
+	PipelineOverrideId                    int                         `json:"pipelineOverrideId"` // required for async install/upgrade event;
+	DeploymentType                        models.DeploymentType       `json:"deploymentType"`     // required for async install/upgrade handling; previously if was used internally
+	ForceSyncDeployment                   bool                        `json:"forceSyncDeployment,notnull"`
 	UserId                                int32                       `json:"-"`
 	EnvId                                 int                         `json:"-"`
 	EnvName                               string                      `json:"-"`
 	ClusterId                             int                         `json:"-"`
 	AppName                               string                      `json:"-"`
 	PipelineName                          string                      `json:"-"`
 	DeploymentAppType                     string                      `json:"-"`
+	Namespace                             string                      `json:"-"`
+	ReleaseName                           string                      `json:"-"`
 	Image                                 string                      `json:"-"`
 }
 

api/helm-app/gRPC/applicationClient.go

@@ -21,6 +21,7 @@ import (
 	"fmt"
 	"github.com/caarlos0/env"
 	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
+	"go.opentelemetry.io/otel"
 	"go.uber.org/zap"
 	"google.golang.org/grpc"
 	"time"
@@ -228,11 +229,13 @@ func (impl *HelmAppClientImpl) DeleteApplication(ctx context.Context, in *Releas
 }
 
 func (impl *HelmAppClientImpl) UpdateApplication(ctx context.Context, in *UpgradeReleaseRequest) (*UpgradeReleaseResponse, error) {
+	newCtx, span := otel.Tracer("orchestrator").Start(ctx, "HelmAppClientImpl.UpdateApplication")
+	defer span.End()
 	applicationClient, err := impl.getApplicationClient()
 	if err != nil {
 		return nil, err
 	}
-	manifest, err := applicationClient.UpgradeRelease(ctx, in)
+	manifest, err := applicationClient.UpgradeRelease(newCtx, in)
 	if err != nil {
 		return nil, err
 	}