fix: Argo DB init conflict when deploy workflow-controller with multi…

…ple replicas argoproj#11177 (argoproj#11569) Signed-off-by: astraw99 <wangchengiscool@gmail.com> Signed-off-by: Dillen Padhiar <dillen_padhiar@intuit.com>
dpadhiar · May 9, 2024 · 9a71ba6 · 9a71ba6
1 parent debb21f
commit 9a71ba6
Show file tree

Hide file tree

Showing 2 changed files with 29 additions and 8 deletions.
diff --git a/workflow/controller/config.go b/workflow/controller/config.go
@@ -27,6 +27,7 @@ func (wfc *WorkflowController) updateConfig() error {
 	wfc.offloadNodeStatusRepo = sqldb.ExplosiveOffloadNodeStatusRepo
 	wfc.wfArchive = sqldb.NullWorkflowArchive
 	wfc.archiveLabelSelector = labels.Everything()
+
 	persistence := wfc.Config.Persistence
 	if persistence != nil {
 		log.Info("Persistence configuration enabled")
@@ -40,14 +41,7 @@ func (wfc *WorkflowController) updateConfig() error {
 				return err
 			}
 			log.Info("Persistence Session created successfully")
-			if !persistence.SkipMigration {
-				err = sqldb.NewMigrate(session, persistence.GetClusterName(), tableName).Exec(context.Background())
-				if err != nil {
-					return err
-				}
-			} else {
-				log.Info("DB migration is disabled")
-			}
+
 			wfc.session = session
 		}
 		sqldb.ConfigureDBSession(wfc.session, persistence.ConnectionPool)
@@ -75,6 +69,7 @@ func (wfc *WorkflowController) updateConfig() error {
 	} else {
 		log.Info("Persistence configuration disabled")
 	}
+
 	wfc.hydrator = hydrator.New(wfc.offloadNodeStatusRepo)
 	wfc.updateEstimatorFactory()
 	wfc.rateLimiter = wfc.newRateLimiter()
@@ -86,6 +81,27 @@ func (wfc *WorkflowController) updateConfig() error {
 	return nil
 }
 
+// initDB inits argo DB tables
+func (wfc *WorkflowController) initDB() error {
+	persistence := wfc.Config.Persistence
+	if persistence != nil {
+		tableName, err := sqldb.GetTableName(persistence)
+		if err != nil {
+			return err
+		}
+		if !persistence.SkipMigration {
+			err = sqldb.NewMigrate(wfc.session, persistence.GetClusterName(), tableName).Exec(context.Background())
+			if err != nil {
+				return err
+			}
+		} else {
+			log.Info("DB migration is disabled")
+		}
+	}
+
+	return nil
+}
+
 func (wfc *WorkflowController) newRateLimiter() *rate.Limiter {
 	return rate.NewLimiter(rate.Limit(wfc.Config.GetResourceRateLimit().Limit), wfc.Config.GetResourceRateLimit().Burst)
 }

diff --git a/workflow/controller/controller.go b/workflow/controller/controller.go
@@ -239,6 +239,11 @@ var indexers = cache.Indexers{
 func (wfc *WorkflowController) Run(ctx context.Context, wfWorkers, workflowTTLWorkers, podCleanupWorkers int) {
 	defer runtimeutil.HandleCrash(runtimeutil.PanicHandlers...)
 
+	// init DB after leader election (if enabled)
+	if err := wfc.initDB(); err != nil {
+		log.Fatalf("Failed to init db: %v", err)
+	}
+
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()