Repository: argoproj/argo
Pull request #3350 (2020-06-30 16:50:51)

fix: Don't double-count metric events
The K8s informer [may send multiple delete events](https://github.com/kubernetes/kubernetes/issues/22150) for the same deletion, causing the metrics service to double-count object deletions and resulting in negative workflow count numbers. To fix, keep a list of observed Workflow keys and ignore double events.


Patch:
diff --git a/USERS.md b/USERS.md
index fcbe8d0aa..4d764e333 100644
--- a/USERS.md
+++ b/USERS.md
@@ -74,3 +74,4 @@ Currently, the following organizations are **officially** using Argo Workflows:
 1. [Threekit](https://www.threekit.com/)
 1. [Tiger Analytics](https://www.tigeranalytics.com/)
 1. [Wavefront](https://www.wavefront.com/)
+1. [Wellcome Trust](https://wellcome.ac.uk/)
diff --git a/ui/src/app/workflows/components/workflow-details/workflow-details.tsx b/ui/src/app/workflows/components/workflow-details/workflow-details.tsx
index 89812739a..ec2713aae 100644
--- a/ui/src/app/workflows/components/workflow-details/workflow-details.tsx
+++ b/ui/src/app/workflows/components/workflow-details/workflow-details.tsx
@@ -177,14 +177,10 @@ export class WorkflowDetails extends React.Component<RouteComponentProps<any>, W
         );
     }
 
-    private confirmAction(title: string): void {
+    private performAction(action: (name: string, namespace: string) => Promise<any>, title: string, redirect: string, ctx: ContextApis): void {
         if (!confirm(`Are you sure you want to ${title.toLowerCase()} this workflow?`)) {
             return;
         }
-    }
-
-    private performAction(action: (name: string, namespace: string) => Promise<any>, title: string, redirect: string, ctx: ContextApis): void {
-        this.confirmAction(title);
         action(this.props.match.params.name, this.props.match.params.namespace)
             .then(() => ctx.navigation.goto(uiUrl(redirect)))
             .catch(() => {
diff --git a/ui/src/app/workflows/components/workflows-toolbar/workflows-toolbar.tsx b/ui/src/app/workflows/components/workflows-toolbar/workflows-toolbar.tsx
index ef394521f..ff7ddc932 100644
--- a/ui/src/app/workflows/components/workflows-toolbar/workflows-toolbar.tsx
+++ b/ui/src/app/workflows/components/workflows-toolbar/workflows-toolbar.tsx
@@ -50,7 +50,9 @@ export class WorkflowsToolbar extends React.Component<WorkflowsToolbarProps, {}>
     }
 
     private performActionOnSelectedWorkflows(ctx: any, title: string, action: (name: string, namespace: string) => Promise<any>): Promise<any> {
-        this.confirmAction(title);
+        if (!confirm(`Are you sure you want to ${title.toLowerCase()} all selected workflows?`)) {
+            return;
+        }
         const promises = [];
         for (const wfUID of Object.keys(this.props.selectedWorkflows)) {
             const wf = this.props.selectedWorkflows[wfUID];
@@ -67,13 +69,6 @@ export class WorkflowsToolbar extends React.Component<WorkflowsToolbarProps, {}>
         return Promise.all(promises);
     }
 
-    private confirmAction(title: string): void {
-        if (!confirm(`Are you sure you want to ${title.toLowerCase()} all selected workflows?`)) {
-            return;
-        }
-        return;
-    }
-
     private renderActions(ctx: any): JSX.Element[] {
         const actionButtons = [];
         const actions: any = Actions.WorkflowOperations;
diff --git a/workflow/controller/controller.go b/workflow/controller/controller.go
index 8a0d8ecb1..7811eb9fc 100644
--- a/workflow/controller/controller.go
+++ b/workflow/controller/controller.go
@@ -80,7 +80,7 @@ type WorkflowController struct {
 	offloadNodeStatusRepo sqldb.OffloadNodeStatusRepo
 	hydrator              hydrator.Interface
 	wfArchive             sqldb.WorkflowArchive
-	metrics               metrics.Metrics
+	metrics               *metrics.Metrics
 	eventRecorderManager  EventRecorderManager
 	archiveLabelSelector  labels.Selector
 }
@@ -127,7 +127,7 @@ func (wfc *WorkflowController) runTTLController(ctx context.Context) {
 }
 
 func (wfc *WorkflowController) runCronController(ctx context.Context) {
-	cronController := cron.NewCronController(wfc.wfclientset, wfc.restConfig, wfc.namespace, wfc.GetManagedNamespace(), wfc.Config.InstanceID, &wfc.metrics)
+	cronController := cron.NewCronController(wfc.wfclientset, wfc.restConfig, wfc.namespace, wfc.GetManagedNamespace(), wfc.Config.InstanceID, wfc.metrics)
 	cronController.Run(ctx)
 }
 
@@ -587,13 +587,22 @@ func (wfc *WorkflowController) addWorkflowInformerHandlers() {
 	)
 	wfc.wfInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
 		AddFunc: func(obj interface{}) {
-			wfc.metrics.WorkflowAdded(getWfPhase(obj))
+			key, err := cache.MetaNamespaceKeyFunc(obj)
+			if err == nil {
+				wfc.metrics.WorkflowAdded(key, getWfPhase(obj))
+			}
 		},
 		UpdateFunc: func(old, new interface{}) {
-			wfc.metrics.WorkflowUpdated(getWfPhase(old), getWfPhase(new))
+			key, err := cache.MetaNamespaceKeyFunc(new)
+			if err == nil {
+				wfc.metrics.WorkflowUpdated(key, getWfPhase(old), getWfPhase(new))
+			}
 		},
 		DeleteFunc: func(obj interface{}) {
-			wfc.metrics.WorkflowDeleted(getWfPhase(obj))
+			key, err := cache.MetaNamespaceKeyFunc(obj)
+			if err == nil {
+				wfc.metrics.WorkflowDeleted(key, getWfPhase(obj))
+			}
 		},
 	})
 }
diff --git a/workflow/metrics/metrics.go b/workflow/metrics/metrics.go
index 92343dd31..a61ec250c 100644
--- a/workflow/metrics/metrics.go
+++ b/workflow/metrics/metrics.go
@@ -39,6 +39,7 @@ type Metrics struct {
 
 	workflowsProcessed prometheus.Counter
 	workflowsByPhase   map[v1alpha1.NodePhase]prometheus.Gauge
+	workflows          map[string]bool
 	operationDurations prometheus.Histogram
 	errors             map[ErrorCause]prometheus.Counter
 	customMetrics      map[string]metric
@@ -47,14 +48,15 @@ type Metrics struct {
 	defaultMetricDescs map[string]bool
 }
 
-var _ prometheus.Collector = Metrics{}
+var _ prometheus.Collector = &Metrics{}
 
-func New(metricsConfig, telemetryConfig ServerConfig) Metrics {
-	metrics := Metrics{
+func New(metricsConfig, telemetryConfig ServerConfig) *Metrics {
+	metrics := &Metrics{
 		metricsConfig:      metricsConfig,
 		telemetryConfig:    telemetryConfig,
 		workflowsProcessed: newCounter("workflows_processed_count", "Number of workflow updates processed", nil),
 		workflowsByPhase:   getWorkflowPhaseGauges(),
+		workflows:          make(map[string]bool),
 		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),
@@ -68,7 +70,7 @@ func New(metricsConfig, telemetryConfig ServerConfig) Metrics {
 	return metrics
 }
 
-func (m Metrics) allMetrics() []prometheus.Metric {
+func (m *Metrics) allMetrics() []prometheus.Metric {
 	allMetrics := []prometheus.Metric{
 		m.workflowsProcessed,
 		m.operationDurations,
@@ -86,33 +88,44 @@ func (m Metrics) allMetrics() []prometheus.Metric {
 	return allMetrics
 }
 
-func (m Metrics) WorkflowAdded(phase v1alpha1.NodePhase) {
+func (m *Metrics) WorkflowAdded(key string, phase v1alpha1.NodePhase) {
+	if m.workflows[key] {
+		return
+	}
+	m.workflows[key] = true
 	if _, ok := m.workflowsByPhase[phase]; ok {
 		m.workflowsByPhase[phase].Inc()
 	}
 }
 
-func (m Metrics) WorkflowUpdated(fromPhase, toPhase v1alpha1.NodePhase) {
-	m.WorkflowDeleted(fromPhase)
-	m.WorkflowAdded(toPhase)
+func (m *Metrics) WorkflowUpdated(key string, fromPhase, toPhase v1alpha1.NodePhase) {
+	if fromPhase == toPhase || !m.workflows[key] {
+		return
+	}
+	m.WorkflowDeleted(key, fromPhase)
+	m.WorkflowAdded(key, toPhase)
 }
 
-func (m Metrics) WorkflowDeleted(phase v1alpha1.NodePhase) {
+func (m *Metrics) WorkflowDeleted(key string, phase v1alpha1.NodePhase) {
+	if !m.workflows[key] {
+		return
+	}
+	delete(m.workflows, key)
 	if _, ok := m.workflowsByPhase[phase]; ok {
 		m.workflowsByPhase[phase].Dec()
 	}
 }
 
-func (m Metrics) OperationCompleted(durationSeconds float64) {
+func (m *Metrics) OperationCompleted(durationSeconds float64) {
 	m.operationDurations.Observe(durationSeconds)
 }
 
-func (m Metrics) GetCustomMetric(key string) prometheus.Metric {
+func (m *Metrics) GetCustomMetric(key string) prometheus.Metric {
 	// It's okay to return nil metrics in this function
 	return m.customMetrics[key].metric
 }
 
-func (m Metrics) UpsertCustomMetric(key string, newMetric prometheus.Metric) error {
+func (m *Metrics) UpsertCustomMetric(key string, newMetric prometheus.Metric) error {
 	if _, inUse := m.defaultMetricDescs[newMetric.Desc().String()]; inUse {
 		return fmt.Errorf("metric '%s' is already in use by the system, please use a different name", newMetric.Desc())
 	}
@@ -127,10 +140,10 @@ const (
 	ErrorCauseCronWorkflowSubmissionError ErrorCause = "CronWorkflowSubmissionError"
 )
 
-func (m Metrics) OperationPanic() {
+func (m *Metrics) OperationPanic() {
 	m.errors[ErrorCauseOperationPanic].Inc()
 }
 
-func (m Metrics) CronWorkflowSubmissionError() {
+func (m *Metrics) CronWorkflowSubmissionError() {
 	m.errors[ErrorCauseCronWorkflowSubmissionError].Inc()
 }
diff --git a/workflow/metrics/server.go b/workflow/metrics/server.go
index dcc62730d..eb4285623 100644
--- a/workflow/metrics/server.go
+++ b/workflow/metrics/server.go
@@ -13,7 +13,7 @@ import (
 )
 
 // RunServer starts a metrics server
-func (m Metrics) RunServer(ctx context.Context) {
+func (m *Metrics) RunServer(ctx context.Context) {
 	if !m.metricsConfig.Enabled {
 		// If metrics aren't enabled, return
 		return
@@ -66,19 +66,19 @@ func runServer(config ServerConfig, registry *prometheus.Registry, ctx context.C
 	}
 }
 
-func (m Metrics) Describe(ch chan<- *prometheus.Desc) {
+func (m *Metrics) Describe(ch chan<- *prometheus.Desc) {
 	for _, metric := range m.allMetrics() {
 		ch <- metric.Desc()
 	}
 }
 
-func (m Metrics) Collect(ch chan<- prometheus.Metric) {
+func (m *Metrics) Collect(ch chan<- prometheus.Metric) {
 	for _, metric := range m.allMetrics() {
 		ch <- metric
 	}
 }
 
-func (m Metrics) garbageCollector(ctx context.Context) {
+func (m *Metrics) garbageCollector(ctx context.Context) {
 	if m.metricsConfig.TTL == 0 {
 		return
 	}


Test patch:
diff --git a/workflow/cron/operator_test.go b/workflow/cron/operator_test.go
index 31ed71cb0..b27640e69 100644
--- a/workflow/cron/operator_test.go
+++ b/workflow/cron/operator_test.go
@@ -176,7 +176,7 @@ func TestCronWorkflowConditionSubmissionError(t *testing.T) {
 		wfLister:    &fakeLister{},
 		cronWf:      &cronWf,
 		log:         logrus.WithFields(logrus.Fields{}),
-		metrics:     &testMetrics,
+		metrics:     testMetrics,
 	}
 	woc.Run()
 
@@ -232,7 +232,7 @@ func TestSpecError(t *testing.T) {
 		wfLister:    &fakeLister{},
 		cronWf:      &cronWf,
 		log:         logrus.WithFields(logrus.Fields{}),
-		metrics:     &testMetrics,
+		metrics:     testMetrics,
 	}
 
 	err = woc.validateCronWorkflow()
@@ -261,7 +261,7 @@ func TestReapplyUpdate(t *testing.T) {
 		origCronWf:  cronWf.DeepCopy(),
 		name:        cronWf.Name,
 		log:         logrus.WithFields(logrus.Fields{}),
-		metrics:     &testMetrics,
+		metrics:     testMetrics,
 	}
 
 	cronWf.Spec.Schedule = "1 * * * *"
diff --git a/workflow/metrics/metrics_test.go b/workflow/metrics/metrics_test.go
index ac9101d6e..2862e7539 100644
--- a/workflow/metrics/metrics_test.go
+++ b/workflow/metrics/metrics_test.go
@@ -47,14 +47,21 @@ func TestMetrics(t *testing.T) {
 	}
 	m := New(config, config)
 
-	m.WorkflowAdded(v1alpha1.NodeRunning)
+	m.WorkflowAdded("wf", v1alpha1.NodeRunning)
 	var metric dto.Metric
 	err := m.workflowsByPhase[v1alpha1.NodeRunning].Write(&metric)
 	if assert.NoError(t, err) {
 		assert.Equal(t, float64(1), *metric.Gauge.Value)
 	}
 
-	m.WorkflowUpdated(v1alpha1.NodeRunning, v1alpha1.NodeSucceeded)
+	// Test that we don't double add
+	m.WorkflowAdded("wf", v1alpha1.NodeRunning)
+	err = m.workflowsByPhase[v1alpha1.NodeRunning].Write(&metric)
+	if assert.NoError(t, err) {
+		assert.Equal(t, float64(1), *metric.Gauge.Value)
+	}
+
+	m.WorkflowUpdated("wf", v1alpha1.NodeRunning, v1alpha1.NodeSucceeded)
 	err = m.workflowsByPhase[v1alpha1.NodeRunning].Write(&metric)
 	if assert.NoError(t, err) {
 		assert.Equal(t, float64(0), *metric.Gauge.Value)
@@ -64,12 +71,30 @@ func TestMetrics(t *testing.T) {
 		assert.Equal(t, float64(1), *metric.Gauge.Value)
 	}
 
-	m.WorkflowDeleted(v1alpha1.NodeSucceeded)
+	m.WorkflowDeleted("wf", v1alpha1.NodeSucceeded)
+	err = m.workflowsByPhase[v1alpha1.NodeRunning].Write(&metric)
+	if assert.NoError(t, err) {
+		assert.Equal(t, float64(0), *metric.Gauge.Value)
+	}
+
+	// Test that we don't double delete
+	m.WorkflowDeleted("wf", v1alpha1.NodeSucceeded)
 	err = m.workflowsByPhase[v1alpha1.NodeRunning].Write(&metric)
 	if assert.NoError(t, err) {
 		assert.Equal(t, float64(0), *metric.Gauge.Value)
 	}
 
+	// Test that we don't update workflows that we're not tracking
+	m.WorkflowUpdated("does-not-exist", v1alpha1.NodeRunning, v1alpha1.NodeSucceeded)
+	err = m.workflowsByPhase[v1alpha1.NodeRunning].Write(&metric)
+	if assert.NoError(t, err) {
+		assert.Equal(t, float64(0), *metric.Gauge.Value)
+	}
+	err = m.workflowsByPhase[v1alpha1.NodeSucceeded].Write(&metric)
+	if assert.NoError(t, err) {
+		assert.Equal(t, float64(0), *metric.Gauge.Value)
+	}
+
 	m.OperationCompleted(0.05)
 	err = m.operationDurations.Write(&metric)
 	if assert.NoError(t, err) {