feat: manual sync for argocd apps (#4392)

* argocd manual sync code

* removing unused function

* changing sync timeline time

* sync time update

* creating constant for application template

* updating sync time in cron

* adding argocd sync timeline in helm apps

* fixing wire issue

* making autosync mode dynamic in application template

* dev testing changes

* adding manual sync support for rollback and chart group

* fix make error

* fixing cron

* fix: helm app sync condition in cron

* fixing error handling in rollback

* wip: returning err

* adding context in rollback funtion

* wip

* fix githash update

* wip

* removing argocd_sync_initiated timeline

* updating sync time

* pr review changes

* fix wire

* adding check for argo manual sync disabled

Dockerfile

@@ -21,7 +21,7 @@ COPY --from=build-env  /go/src/github.com/devtron-labs/devtron/vendor/github.com
 COPY --from=build-env  /go/src/github.com/devtron-labs/devtron/scripts/devtron-reference-helm-charts scripts/devtron-reference-helm-charts
 COPY --from=build-env  /go/src/github.com/devtron-labs/devtron/scripts/sql scripts/sql
 COPY --from=build-env  /go/src/github.com/devtron-labs/devtron/scripts/casbin scripts/casbin
-COPY --from=build-env  /go/src/github.com/devtron-labs/devtron/scripts/argo-assets/APPLICATION_TEMPLATE.JSON scripts/argo-assets/APPLICATION_TEMPLATE.JSON
+COPY --from=build-env  /go/src/github.com/devtron-labs/devtron/scripts/argo-assets/APPLICATION_TEMPLATE.tmpl scripts/argo-assets/APPLICATION_TEMPLATE.tmpl
 
 COPY ./git-ask-pass.sh /git-ask-pass.sh
 RUN chmod +x /git-ask-pass.sh

api/appStore/deployment/CommonDeploymentRestHandler.go

@@ -32,6 +32,7 @@ import (
 	"github.com/devtron-labs/devtron/pkg/user"
 	"github.com/devtron-labs/devtron/pkg/user/casbin"
 	util2 "github.com/devtron-labs/devtron/util"
+	"github.com/devtron-labs/devtron/util/argo"
 	"github.com/devtron-labs/devtron/util/rbac"
 	"github.com/gorilla/mux"
 	"go.opentelemetry.io/otel"
@@ -59,12 +60,13 @@ type CommonDeploymentRestHandlerImpl struct {
 	validator                  *validator.Validate
 	helmAppService             client.HelmAppService
 	helmAppRestHandler         client.HelmAppRestHandler
+	argoUserService            argo.ArgoUserService
 }
 
 func NewCommonDeploymentRestHandlerImpl(Logger *zap.SugaredLogger, userAuthService user.UserService,
 	enforcer casbin.Enforcer, enforcerUtil rbac.EnforcerUtil, enforcerUtilHelm rbac.EnforcerUtilHelm, appStoreDeploymentService service.AppStoreDeploymentService,
 	validator *validator.Validate, helmAppService client.HelmAppService, appStoreDeploymentServiceC appStoreDeploymentCommon.AppStoreDeploymentCommonService,
-	helmAppRestHandler client.HelmAppRestHandler) *CommonDeploymentRestHandlerImpl {
+	helmAppRestHandler client.HelmAppRestHandler, argoUserService argo.ArgoUserService) *CommonDeploymentRestHandlerImpl {
 	return &CommonDeploymentRestHandlerImpl{
 		Logger:                     Logger,
 		userAuthService:            userAuthService,
@@ -76,6 +78,7 @@ func NewCommonDeploymentRestHandlerImpl(Logger *zap.SugaredLogger, userAuthServi
 		helmAppService:             helmAppService,
 		appStoreDeploymentServiceC: appStoreDeploymentServiceC,
 		helmAppRestHandler:         helmAppRestHandler,
+		argoUserService:            argoUserService,
 	}
 }
 func (handler *CommonDeploymentRestHandlerImpl) getAppOfferingMode(installedAppId string, appId string) (string, *appStoreBean.InstallAppVersionDTO, error) {
@@ -287,8 +290,28 @@ func (handler *CommonDeploymentRestHandlerImpl) RollbackApplication(w http.Respo
 		return
 	}
 	//rbac block ends here
+	ctx, cancel := context.WithCancel(r.Context())
+	if cn, ok := w.(http.CloseNotifier); ok {
+		go func(done <-chan struct{}, closed <-chan bool) {
+			select {
+			case <-done:
+			case <-closed:
+				cancel()
+			}
+		}(ctx.Done(), cn.CloseNotify())
+	}
+	if util2.IsBaseStack() || util2.IsHelmApp(appOfferingMode) {
+		ctx = context.WithValue(r.Context(), "token", token)
+	} else {
+		acdToken, err := handler.argoUserService.GetLatestDevtronArgoCdUserToken()
+		if err != nil {
+			handler.Logger.Errorw("error in getting acd token", "err", err)
+			common.WriteJsonResp(w, err, nil, http.StatusInternalServerError)
+			return
+		}
+		ctx = context.WithValue(r.Context(), "token", acdToken)
+	}
 
-	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
 	defer cancel()
 	success, err := handler.appStoreDeploymentService.RollbackApplication(ctx, request, installedAppDto, userId)
 	if err != nil {

api/appStore/deployment/wire_appStoreDeployment.go

@@ -1,6 +1,7 @@
 package appStoreDeployment
 
 import (
+	"github.com/devtron-labs/devtron/client/argocdServer"
 	appStoreDeploymentCommon "github.com/devtron-labs/devtron/pkg/appStore/deployment/common"
 	"github.com/devtron-labs/devtron/pkg/appStore/deployment/repository"
 	"github.com/devtron-labs/devtron/pkg/appStore/deployment/service"
@@ -29,4 +30,5 @@ var AppStoreDeploymentWireSet = wire.NewSet(
 	wire.Bind(new(CommonDeploymentRestHandler), new(*CommonDeploymentRestHandlerImpl)),
 	NewCommonDeploymentRouterImpl,
 	wire.Bind(new(CommonDeploymentRouter), new(*CommonDeploymentRouterImpl)),
+	argocdServer.GetACDDeploymentConfig,
 )

client/argocdServer/ArgoClientWrapperService.go

@@ -3,26 +3,53 @@ package argocdServer
 import (
 	"context"
 	application2 "github.com/argoproj/argo-cd/v2/pkg/apiclient/application"
+	"github.com/argoproj/argo-cd/v2/pkg/apis/application/v1alpha1"
+	"github.com/caarlos0/env"
 	"github.com/devtron-labs/devtron/client/argocdServer/application"
 	"github.com/devtron-labs/devtron/client/argocdServer/bean"
 	"go.uber.org/zap"
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
+type ACDConfig struct {
+	ArgoCDAutoSyncEnabled bool `env:"ARGO_AUTO_SYNC_ENABLED" envDefault:"true"` //will gradually switch this flag to false in enterprise
+}
+
+func GetACDDeploymentConfig() (*ACDConfig, error) {
+	cfg := &ACDConfig{}
+	err := env.Parse(cfg)
+	if err != nil {
+		return nil, err
+	}
+	return cfg, err
+}
+
 type ArgoClientWrapperService interface {
+
+	//GetArgoAppWithNormalRefresh - refresh app at argocd side
 	GetArgoAppWithNormalRefresh(context context.Context, argoAppName string) error
+
+	//SyncArgoCDApplicationIfNeededAndRefresh - if ARGO_AUTO_SYNC_ENABLED=true, app will be refreshed to initiate refresh at argoCD side or else it will be synced and refreshed
+	SyncArgoCDApplicationIfNeededAndRefresh(context context.Context, argoAppName string) error
+
+	// UpdateArgoCDSyncModeIfNeeded - if ARGO_AUTO_SYNC_ENABLED=true and app is in manual sync mode or vice versa update app
+	UpdateArgoCDSyncModeIfNeeded(ctx context.Context, argoApplication *v1alpha1.Application) (err error)
 }
 
 type ArgoClientWrapperServiceImpl struct {
 	logger    *zap.SugaredLogger
 	acdClient application.ServiceClient
+	ACDConfig *ACDConfig
 }
 
 func NewArgoClientWrapperServiceImpl(logger *zap.SugaredLogger,
 	acdClient application.ServiceClient,
+	ACDConfig *ACDConfig,
 ) *ArgoClientWrapperServiceImpl {
 	return &ArgoClientWrapperServiceImpl{
 		logger:    logger,
 		acdClient: acdClient,
+		ACDConfig: ACDConfig,
 	}
 }
 
@@ -37,3 +64,69 @@ func (impl *ArgoClientWrapperServiceImpl) GetArgoAppWithNormalRefresh(context co
 	impl.logger.Debugw("done getting the application with refresh with no error", "argoAppName", argoAppName)
 	return nil
 }
+
+func (impl *ArgoClientWrapperServiceImpl) SyncArgoCDApplicationIfNeededAndRefresh(context context.Context, argoAppName string) error {
+	impl.logger.Info("argocd manual sync for app started", "argoAppName", argoAppName)
+	if !impl.ACDConfig.ArgoCDAutoSyncEnabled {
+		impl.logger.Debugw("syncing argocd app as manual sync is enabled", "argoAppName", argoAppName)
+		revision := "master"
+		pruneResources := true
+		_, syncErr := impl.acdClient.Sync(context, &application2.ApplicationSyncRequest{Name: &argoAppName, Revision: &revision, Prune: &pruneResources})
+		if syncErr != nil {
+			impl.logger.Errorw("cannot get application with refresh", "app", argoAppName)
+			return syncErr
+		}
+		impl.logger.Debugw("argocd sync completed", "argoAppName", argoAppName)
+	}
+	refreshErr := impl.GetArgoAppWithNormalRefresh(context, argoAppName)
+	if refreshErr != nil {
+		impl.logger.Errorw("error in refreshing argo app", "err", refreshErr)
+	}
+	return nil
+}
+
+func (impl *ArgoClientWrapperServiceImpl) UpdateArgoCDSyncModeIfNeeded(ctx context.Context, argoApplication *v1alpha1.Application) (err error) {
+	if impl.isArgoAppSyncModeMigrationNeeded(argoApplication) {
+		syncModeUpdateRequest := impl.CreateRequestForArgoCDSyncModeUpdateRequest(argoApplication)
+		validate := false
+		_, err = impl.acdClient.Update(ctx, &application2.ApplicationUpdateRequest{Application: syncModeUpdateRequest, Validate: &validate})
+		if err != nil {
+			impl.logger.Errorw("error in creating argo pipeline ", "name", argoApplication.Name, "err", err)
+			return err
+		}
+	}
+	return nil
+}
+
+func (impl *ArgoClientWrapperServiceImpl) isArgoAppSyncModeMigrationNeeded(argoApplication *v1alpha1.Application) bool {
+	if !impl.ACDConfig.ArgoCDAutoSyncEnabled && argoApplication.Spec.SyncPolicy.Automated != nil {
+		return true
+	}
+	if impl.ACDConfig.ArgoCDAutoSyncEnabled && argoApplication.Spec.SyncPolicy.Automated == nil {
+		return true
+	}
+	return false
+}
+
+func (impl *ArgoClientWrapperServiceImpl) CreateRequestForArgoCDSyncModeUpdateRequest(argoApplication *v1alpha1.Application) *v1alpha1.Application {
+	// set automated field in update request
+	var automated *v1alpha1.SyncPolicyAutomated
+	if impl.ACDConfig.ArgoCDAutoSyncEnabled {
+		automated = &v1alpha1.SyncPolicyAutomated{
+			Prune: true,
+		}
+	}
+	return &v1alpha1.Application{
+		ObjectMeta: v1.ObjectMeta{
+			Name:      argoApplication.Name,
+			Namespace: DevtronInstalationNs,
+		},
+		Spec: v1alpha1.ApplicationSpec{
+			Destination: argoApplication.Spec.Destination,
+			Source:      argoApplication.Spec.Source,
+			SyncPolicy: &v1alpha1.SyncPolicy{
+				Automated:   automated,
+				SyncOptions: argoApplication.Spec.SyncPolicy.SyncOptions,
+				Retry:       argoApplication.Spec.SyncPolicy.Retry,
+			}}}
+}

client/argocdServer/k8sClient.go

@@ -26,12 +26,16 @@ type AppTemplate struct {
 	ValuesFile      string
 	RepoPath        string
 	RepoUrl         string
+	AutoSyncEnabled bool
 }
 
-const TimeoutSlow = 30 * time.Second
+const (
+	TimeoutSlow                 = 30 * time.Second
+	ARGOCD_APPLICATION_TEMPLATE = "./scripts/argo-assets/APPLICATION_TEMPLATE.tmpl"
+)
 
 type ArgoK8sClient interface {
-	CreateAcdApp(appRequest *AppTemplate, cluster *repository.Cluster) (string, error)
+	CreateAcdApp(appRequest *AppTemplate, cluster *repository.Cluster, applicationTemplatePath string) (string, error)
 	GetArgoApplication(namespace string, appName string, cluster *repository.Cluster) (map[string]interface{}, error)
 }
 type ArgoK8sClientImpl struct {
@@ -58,8 +62,8 @@ func (impl ArgoK8sClientImpl) tprintf(tmpl string, data interface{}) (string, er
 	return buf.String(), nil
 }
 
-func (impl ArgoK8sClientImpl) CreateAcdApp(appRequest *AppTemplate, cluster *repository.Cluster) (string, error) {
-	chartYamlContent, err := ioutil.ReadFile(filepath.Clean("./scripts/argo-assets/APPLICATION_TEMPLATE.JSON"))
+func (impl ArgoK8sClientImpl) CreateAcdApp(appRequest *AppTemplate, cluster *repository.Cluster, applicationTemplatePath string) (string, error) {
+	chartYamlContent, err := ioutil.ReadFile(filepath.Clean(applicationTemplatePath))
 	if err != nil {
 		impl.logger.Errorw("err in reading template", "err", err)
 		return "", err

client/cron/CdApplicationStatusUpdateHandler.go

@@ -131,7 +131,7 @@ func (impl *CdApplicationStatusUpdateHandlerImpl) Subscribe() error {
 		impl.logger.Debugw("ARGO_PIPELINE_STATUS_UPDATE_REQ", "stage", "subscribeDataUnmarshal", "data", statusUpdateEvent)
 
 		if statusUpdateEvent.IsAppStoreApplication {
-			installedApp, err = impl.installedAppVersionRepository.GetInstalledAppByInstalledAppVersionId(statusUpdateEvent.PipelineId)
+			installedApp, err = impl.installedAppVersionRepository.GetInstalledAppByInstalledAppVersionId(statusUpdateEvent.InstalledAppVersionId)
 			if err != nil {
 				impl.logger.Errorw("error in getting installedAppVersion by id", "err", err, "id", statusUpdateEvent.PipelineId)
 				return
@@ -221,7 +221,7 @@ func (impl *CdApplicationStatusUpdateHandlerImpl) SyncPipelineStatusForResourceT
 		return nil
 	}
 	if !util.IsTerminalStatus(cdWfr.Status) {
-		impl.CdHandler.CheckAndSendArgoPipelineStatusSyncEventIfNeeded(pipeline.Id, 1, false)
+		impl.CdHandler.CheckAndSendArgoPipelineStatusSyncEventIfNeeded(pipeline.Id, 0, 1, false)
 	}
 	return nil
 }
@@ -234,7 +234,7 @@ func (impl *CdApplicationStatusUpdateHandlerImpl) SyncPipelineStatusForAppStoreF
 		return nil
 	}
 	if !util.IsTerminalStatus(installedAppVersionHistory.Status) {
-		impl.CdHandler.CheckAndSendArgoPipelineStatusSyncEventIfNeeded(installedAppVersion.Id, 1, true)
+		impl.CdHandler.CheckAndSendArgoPipelineStatusSyncEventIfNeeded(0, installedAppVersion.Id, 1, true)
 	}
 	return nil
 }

internal/sql/repository/pipelineConfig/PipelineStatusTimelineRepository.go

@@ -15,6 +15,8 @@ const (
 	TIMELINE_STATUS_DEPLOYMENT_INITIATED   TimelineStatus = "DEPLOYMENT_INITIATED"
 	TIMELINE_STATUS_GIT_COMMIT             TimelineStatus = "GIT_COMMIT"
 	TIMELINE_STATUS_GIT_COMMIT_FAILED      TimelineStatus = "GIT_COMMIT_FAILED"
+	TIMELINE_STATUS_ARGOCD_SYNC_INITIATED  TimelineStatus = "ARGOCD_SYNC_INITIATED"
+	TIMELINE_STATUS_ARGOCD_SYNC_COMPLETED  TimelineStatus = "ARGOCD_SYNC_COMPLETED"
 	TIMELINE_STATUS_KUBECTL_APPLY_STARTED  TimelineStatus = "KUBECTL_APPLY_STARTED"
 	TIMELINE_STATUS_KUBECTL_APPLY_SYNCED   TimelineStatus = "KUBECTL_APPLY_SYNCED"
 	TIMELINE_STATUS_APP_HEALTHY            TimelineStatus = "HEALTHY"
@@ -50,6 +52,7 @@ type PipelineStatusTimelineRepository interface {
 	DeleteByCdWfrIdAndTimelineStatusesWithTxn(cdWfrId int, status []TimelineStatus, tx *pg.Tx) error
 	FetchTimelinesByInstalledAppVersionHistoryId(installedAppVersionHistoryId int) ([]*PipelineStatusTimeline, error)
 	FetchLatestTimelinesByInstalledAppVersionHistoryId(installedAppVersionHistoryId int) (*PipelineStatusTimeline, error)
+	GetConnection() *pg.DB
 }
 
 type PipelineStatusTimelineRepositoryImpl struct {
@@ -125,9 +128,10 @@ func (impl *PipelineStatusTimelineRepositoryImpl) FetchTimelinesByPipelineId(pip
 }
 
 func (impl *PipelineStatusTimelineRepositoryImpl) FetchTimelinesByWfrId(wfrId int) ([]*PipelineStatusTimeline, error) {
+	//ignoring 'ARGOCD_SYNC_INITIATED' in sql query as it is not handled at FE
 	var timelines []*PipelineStatusTimeline
 	err := impl.dbConnection.Model(&timelines).
-		Where("cd_workflow_runner_id = ?", wfrId).
+		Where("cd_workflow_runner_id = ? and status !='ARGOCD_SYNC_INITIATED' ", wfrId).
 		Order("status_time ASC").Select()
 	if err != nil {
 		impl.logger.Errorw("error in getting timelines by wfrId", "err", err, "wfrId", wfrId)
@@ -270,9 +274,10 @@ func (impl *PipelineStatusTimelineRepositoryImpl) DeleteByCdWfrIdAndTimelineStat
 }
 
 func (impl *PipelineStatusTimelineRepositoryImpl) FetchTimelinesByInstalledAppVersionHistoryId(installedAppVersionHistoryId int) ([]*PipelineStatusTimeline, error) {
+	//ignoring 'ARGOCD_SYNC_INITIATED' in sql query as it is not handled at FE
 	var timelines []*PipelineStatusTimeline
 	err := impl.dbConnection.Model(&timelines).
-		Where("installed_app_version_history_id = ?", installedAppVersionHistoryId).
+		Where("installed_app_version_history_id = ? and status !='ARGOCD_SYNC_INITIATED'", installedAppVersionHistoryId).
 		Order("status_time ASC").Select()
 	if err != nil {
 		impl.logger.Errorw("error in getting timelines by installAppVersionHistoryId", "err", err, "wfrId", installedAppVersionHistoryId)
@@ -293,3 +298,7 @@ func (impl *PipelineStatusTimelineRepositoryImpl) FetchLatestTimelinesByInstalle
 	}
 	return timeline, nil
 }
+
+func (impl *PipelineStatusTimelineRepositoryImpl) GetConnection() *pg.DB {
+	return impl.dbConnection
+}