Repository: argoproj/argo (go)

Issue:
Expose internal queue metrics via Prometheus
# Summary

Right now, we have metrics around how long operations are taking, custom metrics, etc., however very little visibility into the internals of Argo to determine how long any given item is sitting in the workqueue, the depth of the workqueue, etc.

All of these metrics seem to be readily available since we're using the work queue from client-go, we just need to expose them.

https://godoc.org/k8s.io/client-go/util/workqueue#MetricsProvider

# Motivation

Additional visibility into the state of the queues in Argo for help debugging performance issues that might indicate the need to change resource requests/limits, parallelism, etc..

# Proposal

Expose the metrics already provided by https://godoc.org/k8s.io/client-go/util/workqueue#MetricsProvider for each of the internal work queues

---
<!-- Issue Author: Don't delete this message to encourage other users to support your issue! -->
**Message from the maintainers**:

If you wish to see this enhancement implemented please add a 👍 reaction to this issue! We often sort issues this way to know what to prioritize.


Pull request description:
feat: Expose certain queue metrics
Closes #3366

Patch:
diff --git a/workflow/controller/config.go b/workflow/controller/config.go
index 14d756d04..4ebd64775 100644
--- a/workflow/controller/config.go
+++ b/workflow/controller/config.go
@@ -74,7 +74,6 @@ func (wfc *WorkflowController) updateConfig(config config.Config) error {
 		log.Info("Persistence configuration disabled")
 	}
 	wfc.hydrator = hydrator.New(wfc.offloadNodeStatusRepo)
-	wfc.throttler.SetParallelism(config.Parallelism)
 	return nil
 }
 
diff --git a/workflow/controller/controller.go b/workflow/controller/controller.go
index 7811eb9fc..29c308f30 100644
--- a/workflow/controller/controller.go
+++ b/workflow/controller/controller.go
@@ -103,17 +103,22 @@ func NewWorkflowController(restConfig *rest.Config, kubeclientset kubernetes.Int
 		cliExecutorImage:           executorImage,
 		cliExecutorImagePullPolicy: executorImagePullPolicy,
 		containerRuntimeExecutor:   containerRuntimeExecutor,
-		wfQueue:                    workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter()),
-		podQueue:                   workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter()),
 		configController:           config.NewController(namespace, configMap, kubeclientset),
 		completedPods:              make(chan string, 512),
 		gcPods:                     make(chan string, 512),
 		eventRecorderManager:       newEventRecorderManager(kubeclientset),
 	}
-	wfc.throttler = NewThrottler(0, wfc.wfQueue)
+
 	wfc.UpdateConfig()
 
 	wfc.metrics = metrics.New(wfc.getMetricsServerConfig())
+
+	workqueue.SetProvider(wfc.metrics)
+	wfc.wfQueue = workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "workflow_queue")
+	wfc.throttler = NewThrottler(0, wfc.wfQueue)
+	wfc.throttler.SetParallelism(wfc.getParallelism())
+	wfc.podQueue = workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "pod_queue")
+
 	return &wfc
 }
 
@@ -721,6 +726,10 @@ func (wfc *WorkflowController) GetContainerRuntimeExecutor() string {
 	return wfc.Config.ContainerRuntimeExecutor
 }
 
+func (wfc *WorkflowController) getParallelism() int {
+	return wfc.Config.Parallelism
+}
+
 func (wfc *WorkflowController) getMetricsServerConfig() (metrics.ServerConfig, metrics.ServerConfig) {
 	// Metrics config
 	path := wfc.Config.MetricsConfig.Path
diff --git a/workflow/controller/dag.go b/workflow/controller/dag.go
index ac286b2c7..efb3a19e2 100644
--- a/workflow/controller/dag.go
+++ b/workflow/controller/dag.go
@@ -366,11 +366,12 @@ func (woc *wfOperationCtx) executeDAGTask(dagCtx *dagContext, taskName string) {
 			connectDependencies(nodeName)
 			return
 		}
+		// Recurse into all of this node's dependencies
+		for _, dep := range taskDependencies {
+			woc.executeDAGTask(dagCtx, dep)
+		}
 		if !proceed {
-			// This node's dependencies are not completed yet, recurse into them, then return
-			for _, dep := range taskDependencies {
-				woc.executeDAGTask(dagCtx, dep)
-			}
+			// This node's dependencies are not completed yet, return
 			return
 		}
 		if !execute {
diff --git a/workflow/cron/controller.go b/workflow/cron/controller.go
index 19999203d..57b7a35e9 100644
--- a/workflow/cron/controller.go
+++ b/workflow/cron/controller.go
@@ -68,8 +68,8 @@ func NewCronController(
 		restConfig:         restConfig,
 		nameEntryIDMap:     make(map[string]cron.EntryID),
 		nameEntryIDMapLock: &sync.Mutex{},
-		wfQueue:            workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter()),
-		cronWfQueue:        workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter()),
+		wfQueue:            workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "wf_cron_queue"),
+		cronWfQueue:        workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "cron_wf_queue"),
 		metrics:            metrics,
 	}
 }
diff --git a/workflow/metrics/metrics.go b/workflow/metrics/metrics.go
index 819c3ce8b..4775ae1f9 100644
--- a/workflow/metrics/metrics.go
+++ b/workflow/metrics/metrics.go
@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"github.com/prometheus/client_golang/prometheus"
+	"k8s.io/client-go/util/workqueue"
 
 	"github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
 )
@@ -46,6 +47,7 @@ type Metrics struct {
 	operationDurations prometheus.Histogram
 	errors             map[ErrorCause]prometheus.Counter
 	customMetrics      map[string]metric
+	workqueueMetrics   map[string]prometheus.Metric
 
 	// Used to quickly check if a metric desc is already used by the system
 	defaultMetricDescs map[string]bool
@@ -63,6 +65,7 @@ func New(metricsConfig, telemetryConfig ServerConfig) *Metrics {
 		operationDurations: newHistogram("operation_duration_seconds", "Histogram of durations of operations", nil, []float64{0.1, 0.25, 0.5, 0.75, 1.0, 1.25, 1.5, 1.75, 2.0, 2.5, 3.0}),
 		errors:             getErrorCounters(),
 		customMetrics:      make(map[string]metric),
+		workqueueMetrics:   make(map[string]prometheus.Metric),
 		defaultMetricDescs: make(map[string]bool),
 	}
 
@@ -84,6 +87,9 @@ func (m *Metrics) allMetrics() []prometheus.Metric {
 	for _, metric := range m.errors {
 		allMetrics = append(allMetrics, metric)
 	}
+	for _, metric := range m.workqueueMetrics {
+		allMetrics = append(allMetrics, metric)
+	}
 	for _, metric := range m.customMetrics {
 		allMetrics = append(allMetrics, metric.metric)
 	}
@@ -159,3 +165,47 @@ func (m *Metrics) OperationPanic() {
 func (m *Metrics) CronWorkflowSubmissionError() {
 	m.errors[ErrorCauseCronWorkflowSubmissionError].Inc()
 }
+
+// Act as a metrics provider for a workflow queue
+var _ workqueue.MetricsProvider = &Metrics{}
+
+func (m *Metrics) NewDepthMetric(name string) workqueue.GaugeMetric {
+	key := fmt.Sprintf("%s-depth", name)
+	if _, ok := m.workqueueMetrics[key]; !ok {
+		m.workqueueMetrics[key] = newGauge("queue_depth_count", "Depth of the queue", map[string]string{"queue_name": name})
+	}
+	return m.workqueueMetrics[key].(prometheus.Gauge)
+}
+
+func (m *Metrics) NewAddsMetric(name string) workqueue.CounterMetric {
+	key := fmt.Sprintf("%s-adds", name)
+	if _, ok := m.workqueueMetrics[key]; !ok {
+		m.workqueueMetrics[key] = newCounter("queue_adds_count", "Adds to the queue", map[string]string{"queue_name": name})
+	}
+	return m.workqueueMetrics[key].(prometheus.Counter)
+}
+
+func (m *Metrics) NewLatencyMetric(name string) workqueue.HistogramMetric {
+	key := fmt.Sprintf("%s-latency", name)
+	if _, ok := m.workqueueMetrics[key]; !ok {
+		m.workqueueMetrics[key] = newHistogram("queue_latency", "Time objects spend waiting in the queue", map[string]string{"queue_name": name}, []float64{1.0, 5.0, 20.0, 60.0, 180.0})
+	}
+	return m.workqueueMetrics[key].(prometheus.Histogram)
+}
+
+// These metrics are not relevant to be exposed
+type noopMetric struct{}
+
+func (noopMetric) Inc()            {}
+func (noopMetric) Dec()            {}
+func (noopMetric) Set(float64)     {}
+func (noopMetric) Observe(float64) {}
+
+func (m *Metrics) NewRetriesMetric(name string) workqueue.CounterMetric        { return noopMetric{} }
+func (m *Metrics) NewWorkDurationMetric(name string) workqueue.HistogramMetric { return noopMetric{} }
+func (m *Metrics) NewUnfinishedWorkSecondsMetric(name string) workqueue.SettableGaugeMetric {
+	return noopMetric{}
+}
+func (m *Metrics) NewLongestRunningProcessorSecondsMetric(name string) workqueue.SettableGaugeMetric {
+	return noopMetric{}
+}
diff --git a/workflow/ttlcontroller/ttlcontroller.go b/workflow/ttlcontroller/ttlcontroller.go
index eb18f7f4b..6374aad5f 100644
--- a/workflow/ttlcontroller/ttlcontroller.go
+++ b/workflow/ttlcontroller/ttlcontroller.go
@@ -37,7 +37,7 @@ func NewController(wfClientset wfclientset.Interface, wfInformer cache.SharedInd
 	controller := &Controller{
 		wfclientset:  wfClientset,
 		wfInformer:   wfInformer,
-		workqueue:    workqueue.NewNamedDelayingQueue("workflow-ttl"),
+		workqueue:    workqueue.NewDelayingQueue(),
 		resyncPeriod: workflowTTLResyncPeriod,
 		clock:        clock.RealClock{},
 	}


Test patch:
diff --git a/workflow/controller/dag_test.go b/workflow/controller/dag_test.go
index 7ec11902e..9cb21e162 100644
--- a/workflow/controller/dag_test.go
+++ b/workflow/controller/dag_test.go
@@ -1643,3 +1643,130 @@ func TestOnExitDAGPhase(t *testing.T) {
 
 	assert.Equal(t, wfv1.NodeRunning, woc.wf.Status.Phase)
 }
+
+var testOnExitNonLeaf = `
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  name: exit-handler-bug-example
+spec:
+  arguments: {}
+  entrypoint: dag
+  templates:
+  - arguments: {}
+    dag:
+      tasks:
+      - arguments: {}
+        name: step-2
+        onExit: on-exit
+        template: step-template
+      - arguments: {}
+        dependencies:
+        - step-2
+        name: step-3
+        onExit: on-exit
+        template: step-template
+    inputs: {}
+    metadata: {}
+    name: dag
+    outputs: {}
+  - arguments: {}
+    container:
+      args:
+      - echo exit-handler-step-{{pod.name}}
+      command:
+      - sh
+      - -c
+      image: alpine:latest
+      name: ""
+      resources: {}
+    inputs: {}
+    metadata: {}
+    name: on-exit
+    outputs: {}
+  - arguments: {}
+    container:
+      args:
+      - echo step {{pod.name}}
+      command:
+      - sh
+      - -c
+      image: alpine:latest
+      name: ""
+      resources: {}
+    inputs: {}
+    metadata: {}
+    name: step-template
+    outputs: {}
+status:
+  nodes:
+    exit-handler-bug-example:
+      children:
+      - exit-handler-bug-example-3054913383
+      displayName: exit-handler-bug-example
+      finishedAt: "2020-07-07T16:15:54Z"
+      id: exit-handler-bug-example
+      name: exit-handler-bug-example
+      outboundNodes:
+      - exit-handler-bug-example-3038135764
+      phase: Running
+      startedAt: "2020-07-07T16:15:33Z"
+      templateName: dag
+      templateScope: local/exit-handler-bug-example
+      type: DAG
+    exit-handler-bug-example-3054913383:
+      boundaryID: exit-handler-bug-example
+      displayName: step-2
+      finishedAt: "2020-07-07T16:15:37Z"
+      hostNodeName: minikube
+      id: exit-handler-bug-example-3054913383
+      name: exit-handler-bug-example.step-2
+      outputs:
+        artifacts:
+        - archiveLogs: true
+          name: main-logs
+          s3:
+            accessKeySecret:
+              key: accesskey
+              name: my-minio-cred
+            bucket: my-bucket
+            endpoint: minio:9000
+            insecure: true
+            key: exit-handler-bug-example/exit-handler-bug-example-3054913383/main.log
+            secretKeySecret:
+              key: secretkey
+              name: my-minio-cred
+        exitCode: "0"
+      phase: Succeeded
+      resourcesDuration:
+        cpu: 4
+        memory: 2
+      startedAt: "2020-07-07T16:15:33Z"
+      templateName: step-template
+      templateScope: local/exit-handler-bug-example
+      type: Pod
+  phase: Running
+  startedAt: "2020-07-07T16:15:33Z"
+`
+
+func TestOnExitNonLeaf(t *testing.T) {
+	cancel, controller := newController()
+	defer cancel()
+	wfcset := controller.wfclientset.ArgoprojV1alpha1().Workflows("")
+
+	wf := unmarshalWF(testOnExitNonLeaf)
+	wf, err := wfcset.Create(wf)
+	assert.NoError(t, err)
+	woc := newWorkflowOperationCtx(wf, controller)
+
+	woc.operate()
+	retryNode := woc.getNodeByName("step-2.onExit")
+	if assert.NotNil(t, retryNode) {
+		assert.Equal(t, wfv1.NodePending, retryNode.Phase)
+	}
+	retryNode = woc.getNodeByName("exit-handler-bug-example.step-3")
+	if assert.NotNil(t, retryNode) {
+		assert.Equal(t, wfv1.NodePending, retryNode.Phase)
+	}
+	assert.Equal(t, wfv1.NodeRunning, woc.wf.Status.Phase)
+}
diff --git a/workflow/metrics/metrics_test.go b/workflow/metrics/metrics_test.go
index 2862e7539..b7e60ff38 100644
--- a/workflow/metrics/metrics_test.go
+++ b/workflow/metrics/metrics_test.go
@@ -7,6 +7,7 @@ import (
 
 	dto "github.com/prometheus/client_model/go"
 	"github.com/stretchr/testify/assert"
+	"k8s.io/client-go/util/workqueue"
 
 	"github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
 )
@@ -151,3 +152,29 @@ func TestMetricGC(t *testing.T) {
 
 	assert.Len(t, m.customMetrics, 0)
 }
+
+func TestWorkflowQueueMetrics(t *testing.T) {
+	config := ServerConfig{
+		Enabled: true,
+		Path:    DefaultMetricsServerPath,
+		Port:    DefaultMetricsServerPort,
+		TTL:     1 * time.Second,
+	}
+	m := New(config, config)
+	workqueue.SetProvider(m)
+	wfQueue := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "workflow_queue")
+	assert.NotNil(t, m.workqueueMetrics["workflow_queue-depth"])
+	assert.NotNil(t, m.workqueueMetrics["workflow_queue-adds"])
+	assert.NotNil(t, m.workqueueMetrics["workflow_queue-latency"])
+
+	wfQueue.Add("hello")
+
+	if assert.NotNil(t, m.workqueueMetrics["workflow_queue-adds"]) {
+		var metric dto.Metric
+		err := m.workqueueMetrics["workflow_queue-adds"].Write(&metric)
+		if assert.NoError(t, err) {
+			assert.Equal(t, 1.0, *metric.Counter.Value)
+		}
+
+	}
+}
diff --git a/workflow/ttlcontroller/ttlcontroller_test.go b/workflow/ttlcontroller/ttlcontroller_test.go
index a4c515459..7951c97bc 100644
--- a/workflow/ttlcontroller/ttlcontroller_test.go
+++ b/workflow/ttlcontroller/ttlcontroller_test.go
@@ -348,7 +348,7 @@ func newTTLController() *Controller {
 		wfInformer:   wfInformer,
 		resyncPeriod: workflowTTLResyncPeriod,
 		clock:        clock,
-		workqueue:    workqueue.NewNamedDelayingQueue("workflow-ttl"),
+		workqueue:    workqueue.NewDelayingQueue(),
 	}
 }