Feat: support exit-on-lost-leader flag

pkg/server/config/config.go

@@ -58,6 +58,9 @@
 	PluginConfig PluginConfig
 
 	DexServerURL string
+
+	// ExitOnLostLeader will exit the process if this server lost the leader election, set this to true for debugging
+	ExitOnLostLeader bool
 }
 
 // PluginConfig the plugin directory config
@@ -95,7 +98,8 @@
 			CorePluginPath:   "core-plugins",
 			CustomPluginPath: []string{"plugins"},
 		},
-		DexServerURL: "http://dex.vela-system:5556",
+		DexServerURL:     "http://dex.vela-system:5556",
+		ExitOnLostLeader: true,
 	}
 }
 
@@ -127,5 +131,6 @@
 	fs.StringVar(&s.WorkflowVersion, "workflow-version", c.WorkflowVersion, "the version of workflow to meet controller requirement.")
 	fs.StringVar(&s.DexServerURL, "dex-server", c.DexServerURL, "the URL of the dex server.")
 	fs.StringArrayVar(&s.PluginConfig.CustomPluginPath, "plugin-path", c.PluginConfig.CustomPluginPath, "the path of the plugin directory")
+	fs.BoolVar(&s.ExitOnLostLeader, "exit-on-lost-leader", c.ExitOnLostLeader, "exit the process if this server lost the leader election")
 	profiling.AddFlags(fs)
 }

pkg/server/domain/service/workflow.go

@@ -445,7 +445,7 @@
 				record.Finished = "true"
 				record.Status = model.RevisionStatusFail
 				if err := w.Store.Put(ctx, record); err != nil {
-					return fmt.Errorf(("failed to set the record status to terminated: %s"), err.Error())
+					return fmt.Errorf("failed to set the record status to terminated: %s", err.Error())
 				}
 				return bcode.ErrApplicationRevisionNotExist
 			}

pkg/server/event/sync/worker.go

@@ -97,7 +97,14 @@ func (a *ApplicationSync) Start(ctx context.Context, errorChan chan error) {
 			}
 			app := item.(*v1beta1.Application)
 			if err := cu.AddOrUpdate(ctx, app); err != nil {
-				klog.Errorf("fail to add or update application %s: %s", app.Name, err.Error())
+				failTimes := a.Queue.NumRequeues(app)
+				klog.Errorf("fail to add or update application %s: %s, requeue times: %d", app.Name, err.Error(), failTimes)
+				if failTimes < 5 {
+					a.Queue.AddRateLimited(app)
+				} else {
+					klog.Errorf("fail to add or update application %s: %s, requeue times reach the limit(%d), give up", app.Name, err.Error(), failTimes)
+					a.Queue.Forget(app)
+				}
 			}
 			a.Queue.Done(app)
 		}

pkg/server/server.go

@@ -253,7 +253,9 @@
 				go event.StartEventWorker(ctx, errChan)
 			},
 			OnStoppedLeading: func() {
-				errChan <- fmt.Errorf("leader lost %s", s.cfg.LeaderConfig.ID)
+				if s.cfg.ExitOnLostLeader {
+					errChan <- fmt.Errorf("leader lost %s", s.cfg.LeaderConfig.ID)
+				}
 			},
 			OnNewLeader: func(identity string) {
 				if identity == s.cfg.LeaderConfig.ID {