fix(controller): Enable dummy metrics server on non-leader workflow controller

cmd/workflow-controller/main.go

@@ -117,6 +117,7 @@ func NewRootCommand() *cobra.Command {
 				log.Info("Leader election is turned off. Running in single-instance mode")
 				log.WithField("id", "single-instance").Info("starting leading")
 				go wfController.Run(ctx, workflowWorkers, workflowTTLWorkers, podCleanupWorkers)
+				go wfController.RunMetricsServer(ctx, false)
 			} else {
 				nodeID, ok := os.LookupEnv("LEADER_ELECTION_IDENTITY")
 				if !ok {
@@ -128,6 +129,11 @@ func NewRootCommand() *cobra.Command {
 					leaderName = fmt.Sprintf("%s-%s", leaderName, wfController.Config.InstanceID)
 				}
 
+				// for controlling the dummy metrics server
+				dummyCtx, dummyCancel := context.WithCancel(context.Background())
+				defer dummyCancel()
+				go wfController.RunMetricsServer(dummyCtx, true)
+
 				go leaderelection.RunOrDie(ctx, leaderelection.LeaderElectionConfig{
 					Lock: &resourcelock.LeaseLock{
 						LeaseMeta: metav1.ObjectMeta{Name: leaderName, Namespace: namespace}, Client: kubeclientset.CoordinationV1(),
@@ -139,11 +145,14 @@ func NewRootCommand() *cobra.Command {
 					RetryPeriod:     env.LookupEnvDurationOr("LEADER_ELECTION_RETRY_PERIOD", 5*time.Second),
 					Callbacks: leaderelection.LeaderCallbacks{
 						OnStartedLeading: func(ctx context.Context) {
+							dummyCancel()
 							go wfController.Run(ctx, workflowWorkers, workflowTTLWorkers, podCleanupWorkers)
+							go wfController.RunMetricsServer(ctx, false)
 						},
 						OnStoppedLeading: func() {
 							log.WithField("id", nodeID).Info("stopped leading")
 							cancel()
+							go wfController.RunMetricsServer(dummyCtx, true)
 						},
 						OnNewLeader: func(identity string) {
 							log.WithField("leader", identity).Info("new leader")

workflow/controller/controller.go

@@ -298,9 +298,6 @@ func (wfc *WorkflowController) Run(ctx context.Context, wfWorkers, workflowTTLWo
 		log.Fatal("Timed out waiting for caches to sync")
 	}
 
-	// Start the metrics server
-	go wfc.metrics.RunServer(ctx)
-
 	for i := 0; i < podCleanupWorkers; i++ {
 		go wait.UntilWithContext(ctx, wfc.runPodCleanup, time.Second)
 	}
@@ -323,6 +320,10 @@ func (wfc *WorkflowController) Run(ctx context.Context, wfWorkers, workflowTTLWo
 	<-ctx.Done()
 }
 
+func (wfc *WorkflowController) RunMetricsServer(ctx context.Context, isDummy bool) {
+	go wfc.metrics.RunServer(ctx, isDummy)
+}
+
 // Create and the Synchronization Manager
 func (wfc *WorkflowController) createSynchronizationManager(ctx context.Context) {
 	getSyncLimit := func(lockKey string) (int, error) {

workflow/metrics/server.go

@@ -18,7 +18,8 @@ import (
 )
 
 // RunServer starts a metrics server
-func (m *Metrics) RunServer(ctx context.Context) {
+// If 'isDummy' is set to true, the dummy metrics server will be started. If it's false, the prometheus metrics server will be started
+func (m *Metrics) RunServer(ctx context.Context, isDummy bool) {
 	defer runtimeutil.HandleCrash(runtimeutil.PanicHandlers...)
 
 	if !m.metricsConfig.Enabled {
@@ -36,23 +37,33 @@ func (m *Metrics) RunServer(ctx context.Context) {
 		// If the telemetry server is different -- and it's enabled -- run each on its own instance
 		telemetryRegistry := prometheus.NewRegistry()
 		telemetryRegistry.MustRegister(collectors.NewGoCollector())
-		go runServer(m.telemetryConfig, telemetryRegistry, ctx)
+		go runServer(m.telemetryConfig, telemetryRegistry, ctx, isDummy)
 	}
 
 	// Run the metrics server
-	go runServer(m.metricsConfig, metricsRegistry, ctx)
+	go runServer(m.metricsConfig, metricsRegistry, ctx, isDummy)
 
 	go m.garbageCollector(ctx)
 }
 
-func runServer(config ServerConfig, registry *prometheus.Registry, ctx context.Context) {
+func runServer(config ServerConfig, registry *prometheus.Registry, ctx context.Context, isDummy bool) {
 	var handlerOpts promhttp.HandlerOpts
 	if config.IgnoreErrors {
 		handlerOpts.ErrorHandling = promhttp.ContinueOnError
 	}
 
+	name := ""
 	mux := http.NewServeMux()
-	mux.Handle(config.Path, promhttp.HandlerFor(registry, handlerOpts))
+	if isDummy {
+		// dummy metrics server responds to all requests with a 200 status, but without providing any metrics data
+		name = "dummy metrics server"
+		mux.HandleFunc(config.Path, func(w http.ResponseWriter, r *http.Request) {
+			w.WriteHeader(http.StatusOK)
+		})
+	} else {
+		name = "prometheus metrics server"
+		mux.Handle(config.Path, promhttp.HandlerFor(registry, handlerOpts))
+	}
 	srv := &http.Server{Addr: fmt.Sprintf(":%v", config.Port), Handler: mux}
 
 	if config.Secure {
@@ -67,15 +78,15 @@ func runServer(config ServerConfig, registry *prometheus.Registry, ctx context.C
 		}
 		srv.TLSConfig = tlsConfig
 		go func() {
-			log.Infof("Starting prometheus metrics server at localhost:%v%s", config.Port, config.Path)
-			if err := srv.ListenAndServeTLS("", ""); err != nil {
+			log.Infof("Starting %s at localhost:%v%s", name, config.Port, config.Path)
+			if err := srv.ListenAndServeTLS("", ""); err != http.ErrServerClosed {
 				panic(err)
 			}
 		}()
 	} else {
 		go func() {
-			log.Infof("Starting prometheus metrics server at localhost:%v%s", config.Port, config.Path)
-			if err := srv.ListenAndServe(); err != nil {
+			log.Infof("Starting %s at localhost:%v%s", name, config.Port, config.Path)
+			if err := srv.ListenAndServe(); err != http.ErrServerClosed {
 				panic(err)
 			}
 		}()
@@ -88,7 +99,9 @@ func runServer(config ServerConfig, registry *prometheus.Registry, ctx context.C
 	ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
 	defer cancel()
 	if err := srv.Shutdown(ctx); err != nil {
-		log.Infof("Unable to shutdown metrics server at localhost:%v%s", config.Port, config.Path)
+		log.Infof("Unable to shutdown %s at localhost:%v%s", name, config.Port, config.Path)
+	} else {
+		log.Infof("Successfully shutdown %s at localhost:%v%s", name, config.Port, config.Path)
 	}
 }
 

workflow/metrics/server_test.go

@@ -0,0 +1,87 @@
+package metrics
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net/http"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestDisableMetricsServer(t *testing.T) {
+	config := ServerConfig{
+		Enabled: false,
+		Path:    DefaultMetricsServerPath,
+		Port:    DefaultMetricsServerPort,
+	}
+	m := New(config, config)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	m.RunServer(ctx, false)
+	time.Sleep(1 * time.Second) // to confirm that the server doesn't start, even if we wait
+	resp, err := http.Get(fmt.Sprintf("http://localhost:%d%s", DefaultMetricsServerPort, DefaultMetricsServerPath))
+	if resp != nil {
+		defer resp.Body.Close()
+	}
+
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "connection refused") // expect that the metrics server not to start
+}
+
+func TestMetricsServer(t *testing.T) {
+	config := ServerConfig{
+		Enabled: true,
+		Path:    DefaultMetricsServerPath,
+		Port:    DefaultMetricsServerPort,
+	}
+	m := New(config, config)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	m.RunServer(ctx, false)
+	time.Sleep(1 * time.Second)
+	resp, err := http.Get(fmt.Sprintf("http://localhost:%d%s", DefaultMetricsServerPort, DefaultMetricsServerPath))
+	assert.NoError(t, err)
+	assert.Equal(t, http.StatusOK, resp.StatusCode)
+
+	defer resp.Body.Close()
+
+	bodyBytes, err := io.ReadAll(resp.Body)
+	assert.NoError(t, err)
+
+	bodyString := string(bodyBytes)
+	assert.NotEmpty(t, bodyString)
+}
+
+func TestDummyMetricsServer(t *testing.T) {
+	config := ServerConfig{
+		Enabled: true,
+		Path:    DefaultMetricsServerPath,
+		Port:    DefaultMetricsServerPort,
+	}
+	m := New(config, config)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	m.RunServer(ctx, true)
+	time.Sleep(1 * time.Second)
+	resp, err := http.Get(fmt.Sprintf("http://localhost:%d%s", DefaultMetricsServerPort, DefaultMetricsServerPath))
+	assert.NoError(t, err)
+	assert.Equal(t, http.StatusOK, resp.StatusCode)
+
+	defer resp.Body.Close()
+
+	bodyBytes, err := io.ReadAll(resp.Body)
+	assert.NoError(t, err)
+
+	bodyString := string(bodyBytes)
+
+	assert.Empty(t, bodyString) // expect the dummy metrics server to provide no metrics responses
+}