fix: dedupe realtime metrics and handle delete tombstone (cherry-pick #15216 for 3.7) (#15224)

Signed-off-by: oninowang <oninowang@tencent.com>
Co-authored-by: jswxstw <jswxstw@gmail.com>
Co-authored-by: oninowang <oninowang@tencent.com>

Author: 3 people

workflow/controller/controller.go

@@ -1014,8 +1014,14 @@ func (wfc *WorkflowController) addWorkflowInformerHandlers(ctx context.Context)
 	}
 	_, err = wfc.wfInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
 		DeleteFunc: func(obj interface{}) {
-			wf, ok := obj.(*unstructured.Unstructured)
-			if ok { // maybe cache.DeletedFinalStateUnknown
+			var wf *unstructured.Unstructured
+			switch x := obj.(type) {
+			case *unstructured.Unstructured:
+				wf = x
+			case cache.DeletedFinalStateUnknown:
+				wf, _ = x.Obj.(*unstructured.Unstructured)
+			}
+			if wf != nil {
 				wfc.metrics.DeleteRealtimeMetricsForWfUID(string(wf.GetUID()))
 			}
 		},

workflow/metrics/metrics_custom.go

@@ -227,20 +227,21 @@ func (m *Metrics) UpsertCustomMetric(ctx context.Context, metricSpec *wfv1.Prome
 }
 
 func (m *Metrics) attachCustomMetricToWorkflow(metricSpec *wfv1.Prometheus, ownerKey string) {
-	if metricSpec.IsRealtime() {
-		m.realtimeMutex.Lock()
-		defer m.realtimeMutex.Unlock()
-		// Must move to run each workflowkey
-		for key := range m.realtimeWorkflows {
-			if key == ownerKey {
-				return
-			}
+	if !metricSpec.IsRealtime() {
+		return
+	}
+	m.realtimeMutex.Lock()
+	defer m.realtimeMutex.Unlock()
+	metricKey := metricSpec.GetKey()
+	for _, tracker := range m.realtimeWorkflows[ownerKey] {
+		if tracker.key == metricKey {
+			return
 		}
-		m.realtimeWorkflows[ownerKey] = append(m.realtimeWorkflows[ownerKey], realtimeTracker{
-			inst: m.GetInstrument(metricSpec.Name),
-			key:  metricSpec.GetKey(),
-		})
 	}
+	m.realtimeWorkflows[ownerKey] = append(m.realtimeWorkflows[ownerKey], realtimeTracker{
+		inst: m.GetInstrument(metricSpec.Name),
+		key:  metricSpec.GetKey(),
+	})
 }
 
 func (m *Metrics) createCustomMetric(metricSpec *wfv1.Prometheus) error {