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 <[email protected]>
  • Loading branch information
astraw99 authored and terrytangyuan committed Sep 5, 2023
1 parent aab2160 commit e731cc0
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 @@ -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)
}
Expand Down
5 changes: 5 additions & 0 deletions workflow/controller/controller.go
Original file line number Diff line number Diff line change
Expand Up @@ -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()

Expand Down

0 comments on commit e731cc0

Please sign in to comment.