Skip to content

Commit

Permalink
fix: Argo DB init conflict when deploy workflow-controller with multi…
Browse files Browse the repository at this point in the history
…ple replicas #11177 (#11569)

Signed-off-by: astraw99 <wangchengiscool@gmail.com>
  • Loading branch information
astraw99 authored Sep 5, 2023
1 parent 633c5e9 commit 297bea6
Show file tree
Hide file tree
Showing 2 changed files with 29 additions and 8 deletions.
32 changes: 24 additions & 8 deletions workflow/controller/config.go
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand All @@ -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)
Expand Down Expand Up @@ -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()
Expand All @@ -87,6 +82,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 {
rateLimiter := wfc.Config.GetResourceRateLimit()
return rate.NewLimiter(rate.Limit(rateLimiter.Limit), rateLimiter.Burst)
Expand Down
5 changes: 5 additions & 0 deletions workflow/controller/controller.go
Original file line number Diff line number Diff line change
Expand Up @@ -247,6 +247,11 @@ var indexers = cache.Indexers{
func (wfc *WorkflowController) Run(ctx context.Context, wfWorkers, workflowTTLWorkers, podCleanupWorkers, cronWorkflowWorkers 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()

Expand Down

0 comments on commit 297bea6

Please sign in to comment.