fix(controller): handle race when starting metrics server. Fixes #10807

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"
 
@@ -18,6 +19,7 @@ import (
 const testScopeName string = "argo-workflows-test"
 
 func TestDisablePrometheusServer(t *testing.T) {
+ var wg sync.WaitGroup
  config := Config{
  Enabled: false,
  Path: DefaultPrometheusServerPath,
@@ -27,8 +29,9 @@ func TestDisablePrometheusServer(t *testing.T) {
  defer cancel()
  m, err := NewMetrics(ctx, testScopeName, testScopeName, &config)
  require.NoError(t, err)
- go m.RunPrometheusServer(ctx, false)
- time.Sleep(1 * time.Second) // to confirm that the server doesn't start, even if we wait
+ wg.Add(1)
+ go m.RunPrometheusServer(ctx, false, &wg)
+ wg.Wait()
  resp, err := http.Get(fmt.Sprintf("http://localhost:%d%s", DefaultPrometheusServerPort, DefaultPrometheusServerPath))
  if resp != nil {
  defer resp.Body.Close()
@@ -38,6 +41,7 @@ func TestDisablePrometheusServer(t *testing.T) {
 }
 
 func TestPrometheusServer(t *testing.T) {
+ var wg sync.WaitGroup
  config := Config{
  Enabled: true,
  Path: DefaultPrometheusServerPath,
@@ -47,7 +51,8 @@ func TestPrometheusServer(t *testing.T) {
  defer cancel()
  m, err := NewMetrics(ctx, testScopeName, testScopeName, &config)
  require.NoError(t, err)
- go m.RunPrometheusServer(ctx, false)
+ wg.Add(1)
+ 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 +66,12 @@ 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
+ cancel() // cancel and wait for server shutdown to prevent port conflicts with subsequent tests
+ wg.Wait()
 }
 
 func TestDummyPrometheusServer(t *testing.T) {
+ var wg sync.WaitGroup
  config := Config{
  Enabled: true,
  Path: DefaultPrometheusServerPath,
@@ -76,7 +82,8 @@ func TestDummyPrometheusServer(t *testing.T) {
  defer cancel()
  m, err := NewMetrics(ctx, testScopeName, testScopeName, &config)
  require.NoError(t, err)
- go m.RunPrometheusServer(ctx, true)
+ wg.Add(1)
+ 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 +98,6 @@ 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
+ cancel() // cancel and wait for server shutdown to prevent port conflicts with subsequent tests
+ wg.Wait()
 }

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