fix(controller): race condition when starting Prometheus metrics serv…

…er. Fixes argoproj#10807

While investigating the flaky `MetricsSuite/TestMetricsEndpoint` test
that's been failing periodically for awhile now, I noticed this in the
controller logs
([example](https://github.com/argoproj/argo-workflows/actions/runs/11221357877/job/31191811077)):

```
controller: time="2024-10-07T18:22:14.793Z" level=info msg="Starting dummy metrics server at localhost:9090/metrics"
server: time="2024-10-07T18:22:14.793Z" level=info msg="Creating event controller" asyncDispatch=false operationQueueSize=16 workerCount=4
server: time="2024-10-07T18:22:14.800Z" level=info msg="GRPC Server Max Message Size, MaxGRPCMessageSize, is set" GRPC_MESSAGE_SIZE=104857600
server: time="2024-10-07T18:22:14.800Z" level=info msg="Argo Server started successfully on http://localhost:2746" url="http://localhost:2746"
controller: I1007 18:22:14.800947   25045 leaderelection.go:260] successfully acquired lease argo/workflow-controller
controller: time="2024-10-07T18:22:14.801Z" level=info msg="new leader" leader=local
controller: time="2024-10-07T18:22:14.801Z" level=info msg="Generating Self Signed TLS Certificates for Telemetry Servers"
controller: time="2024-10-07T18:22:14.802Z" level=info msg="Starting prometheus metrics server at localhost:9090/metrics"
controller: panic: listen tcp :9090: bind: address already in use
controller:
controller: goroutine 37 [running]:
controller: github.com/argoproj/argo-workflows/v3/util/telemetry.(*Metrics).RunPrometheusServer.func2()
controller: 	/home/runner/work/argo-workflows/argo-workflows/util/telemetry/exporter_prometheus.go:94 +0x16a
controller: created by github.com/argoproj/argo-workflows/v3/util/telemetry.(*Metrics).RunPrometheusServer in goroutine 36
controller: 	/home/runner/work/argo-workflows/argo-workflows/util/telemetry/exporter_prometheus.go:91 +0x53c
2024/10/07 18:22:14 controller: process exited 25045: exit status 2
controller: exit status 2
2024/10/07 18:22:14 controller: backing off 4s
```

I believe this is a race condition introduced in
argoproj#11295. Here's
the sequence of events that trigger this:
1. Controller starts
2. Dummy metrics server started on port 9090
3. Leader election takes place and controller starts leading
4. Context for dummy metrics server cancelled
5. Metrics server shuts down
6. Prometheus metrics server started on 9090

The problems is steps 5-6 can happen out-of-order, because the shutdown
happens after the contxt is cancelled. Per the docs, "a CancelFunc does
not wait for the work to stop" (https://pkg.go.dev/context#CancelFunc).

The controller needs to explicitly wait for the dummy metrics server to
shut down properly before starting the Prometheus metrics server.
There's many ways of doing that, and this uses a `WaitGroup`, as that's
the simplest approach I could think of.

Signed-off-by: Mason Malone <651224+MasonM@users.noreply.github.com>

cmd/workflow-controller/main.go

@@ -121,7 +121,7 @@ func NewRootCommand() *cobra.Command {
 				log.WithField("id", "single-instance").Info("starting leading")
 
 				go wfController.Run(ctx, workflowWorkers, workflowTTLWorkers, podCleanupWorkers, cronWorkflowWorkers, workflowArchiveWorkers)
-				go wfController.RunPrometheusServer(ctx, false)
+				wfController.RunPrometheusServer(ctx, false)
 			} else {
 				nodeID, ok := os.LookupEnv("LEADER_ELECTION_IDENTITY")
 				if !ok {
@@ -136,7 +136,7 @@ func NewRootCommand() *cobra.Command {
 				// for controlling the dummy metrics server
 				dummyCtx, dummyCancel := context.WithCancel(context.Background())
 				defer dummyCancel()
-				go wfController.RunPrometheusServer(dummyCtx, true)
+				wg := wfController.RunPrometheusServer(dummyCtx, true)
 
 				go leaderelection.RunOrDie(ctx, leaderelection.LeaderElectionConfig{
 					Lock: &resourcelock.LeaseLock{
@@ -150,12 +150,14 @@ func NewRootCommand() *cobra.Command {
 					Callbacks: leaderelection.LeaderCallbacks{
 						OnStartedLeading: func(ctx context.Context) {
 							dummyCancel()
+							wg.Wait()
 							go wfController.Run(ctx, workflowWorkers, workflowTTLWorkers, podCleanupWorkers, cronWorkflowWorkers, workflowArchiveWorkers)
 							go wfController.RunPrometheusServer(ctx, false)
 						},
 						OnStoppedLeading: func() {
 							log.WithField("id", nodeID).Info("stopped leading")
 							cancel()
+							wg.Wait()
 							go wfController.RunPrometheusServer(dummyCtx, true)
 						},
 						OnNewLeader: func(identity string) {

util/telemetry/exporter_prometheus.go

@@ -5,6 +5,7 @@ import (
 	"crypto/tls"
 	"fmt"
 	"net/http"
+	"sync"
 	"time"
 
 	promgo "github.com/prometheus/client_golang/prometheus"
@@ -53,7 +54,8 @@ func (config *Config) port() int {
 
 // RunPrometheusServer starts a prometheus metrics server
 // 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) RunPrometheusServer(ctx context.Context, isDummy bool) {
+func (m *Metrics) RunPrometheusServer(ctx context.Context, isDummy bool, wg *sync.WaitGroup) {
+	defer wg.Done()
 	if !m.config.Enabled {
 		return
 	}

util/telemetry/exporter_prometheus_test.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"io"
 	"net/http"
+	"sync"
 	"testing"
 	"time"
 
@@ -27,7 +28,7 @@ func TestDisablePrometheusServer(t *testing.T) {
 	defer cancel()
 	m, err := NewMetrics(ctx, testScopeName, testScopeName, &config)
 	require.NoError(t, err)
-	go m.RunPrometheusServer(ctx, false)
+	go m.RunPrometheusServer(ctx, false, &sync.WaitGroup{})
 	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", DefaultPrometheusServerPort, DefaultPrometheusServerPath))
 	if resp != nil {
@@ -38,6 +39,7 @@ func TestDisablePrometheusServer(t *testing.T) {
 }
 
 func TestPrometheusServer(t *testing.T) {
+	var wg sync.WaitGroup
 	config := Config{
 		Enabled: true,
 		Path:    DefaultPrometheusServerPath,
@@ -47,7 +49,7 @@ func TestPrometheusServer(t *testing.T) {
 	defer cancel()
 	m, err := NewMetrics(ctx, testScopeName, testScopeName, &config)
 	require.NoError(t, err)
-	go m.RunPrometheusServer(ctx, false)
+	go m.RunPrometheusServer(ctx, false, &wg)
 	time.Sleep(1 * time.Second)
 	resp, err := http.Get(fmt.Sprintf("http://localhost:%d%s", DefaultPrometheusServerPort, DefaultPrometheusServerPath))
 	require.NoError(t, err)
@@ -61,11 +63,11 @@ func TestPrometheusServer(t *testing.T) {
 	bodyString := string(bodyBytes)
 	assert.NotEmpty(t, bodyString)
 
-	cancel()                    // Explicit cancel as sometimes in github CI port 9090 is still busy
-	time.Sleep(1 * time.Second) // Wait for prometheus server
+	wg.Wait() // wait for server shutdown to prevent port conflicts with subsequent tests
 }
 
 func TestDummyPrometheusServer(t *testing.T) {
+	var wg sync.WaitGroup
 	config := Config{
 		Enabled: true,
 		Path:    DefaultPrometheusServerPath,
@@ -76,7 +78,7 @@ func TestDummyPrometheusServer(t *testing.T) {
 	defer cancel()
 	m, err := NewMetrics(ctx, testScopeName, testScopeName, &config)
 	require.NoError(t, err)
-	go m.RunPrometheusServer(ctx, true)
+	go m.RunPrometheusServer(ctx, true, &wg)
 	time.Sleep(1 * time.Second)
 	resp, err := http.Get(fmt.Sprintf("http://localhost:%d%s", DefaultPrometheusServerPort, DefaultPrometheusServerPath))
 	require.NoError(t, err)
@@ -91,6 +93,5 @@ func TestDummyPrometheusServer(t *testing.T) {
 
 	assert.Empty(t, bodyString) // expect the dummy metrics server to provide no metrics responses
 
-	cancel()                    // Explicit cancel as sometimes in github CI port 9090 is still busy
-	time.Sleep(1 * time.Second) // Wait for prometheus server
+	wg.Wait() // wait for server shutdown to prevent port conflicts with subsequent tests
 }

workflow/controller/controller.go

@@ -379,8 +379,11 @@ func (wfc *WorkflowController) Run(ctx context.Context, wfWorkers, workflowTTLWo
 	<-ctx.Done()
 }
 
-func (wfc *WorkflowController) RunPrometheusServer(ctx context.Context, isDummy bool) {
-	go wfc.metrics.RunPrometheusServer(ctx, isDummy)
+func (wfc *WorkflowController) RunPrometheusServer(ctx context.Context, isDummy bool) *gosync.WaitGroup {
+	var wg gosync.WaitGroup
+	wg.Add(1)
+	go wfc.metrics.RunPrometheusServer(ctx, isDummy, &wg)
+	return &wg
 }
 
 // Create and the Synchronization Manager