fix: resolve resourcesDuration (#7299)

Signed-off-by: Kenny Trytek <kenneth.g.trytek@gmail.com>

workflow/controller/operator.go

@@ -260,10 +260,8 @@ func (woc *wfOperationCtx) operate(ctx context.Context) {
 	}
 
 	if woc.execWf.Spec.Metrics != nil {
-		realTimeScope := map[string]func() float64{common.GlobalVarWorkflowDuration: func() float64 {
-			return time.Since(woc.wf.Status.StartedAt.Time).Seconds()
-		}}
-		woc.computeMetrics(woc.execWf.Spec.Metrics.Prometheus, woc.globalParams, realTimeScope, true)
+		localScope, realTimeScope := woc.prepareDefaultMetricScope()
+		woc.computeMetrics(woc.execWf.Spec.Metrics.Prometheus, localScope, realTimeScope, true)
 	}
 
 	if woc.wf.Status.Phase == wfv1.WorkflowUnknown {
@@ -467,11 +465,9 @@ func (woc *wfOperationCtx) operate(ctx context.Context) {
 	}
 
 	if woc.execWf.Spec.Metrics != nil {
-		realTimeScope := map[string]func() float64{common.GlobalVarWorkflowDuration: func() float64 {
-			return node.FinishedAt.Sub(node.StartedAt.Time).Seconds()
-		}}
 		woc.globalParams[common.GlobalVarWorkflowStatus] = string(workflowStatus)
-		woc.computeMetrics(woc.execWf.Spec.Metrics.Prometheus, woc.globalParams, realTimeScope, false)
+		localScope, realTimeScope := woc.prepareMetricScope(node)
+		woc.computeMetrics(woc.execWf.Spec.Metrics.Prometheus, localScope, realTimeScope, false)
 	}
 
 	err = woc.deletePVCs(ctx)

workflow/controller/steps.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/Knetic/govaluate"
 	log "github.com/sirupsen/logrus"
+	v1 "k8s.io/api/core/v1"
 
 	"github.com/argoproj/argo-workflows/v3/errors"
 	wfv1 "github.com/argoproj/argo-workflows/v3/pkg/apis/workflow/v1alpha1"
@@ -490,6 +491,25 @@ func (woc *wfOperationCtx) expandStep(step wfv1.WorkflowStep) ([]wfv1.WorkflowSt
 	return expandedStep, nil
 }
 
+func (woc *wfOperationCtx) prepareDefaultMetricScope() (map[string]string, map[string]func() float64) {
+	durationCPU := fmt.Sprintf("%s.%s", common.LocalVarResourcesDuration, v1.ResourceCPU)
+	durationMem := fmt.Sprintf("%s.%s", common.LocalVarResourcesDuration, v1.ResourceMemory)
+
+	localScope := woc.globalParams.DeepCopy()
+	localScope[common.LocalVarDuration] = "0"
+	localScope[common.LocalVarStatus] = string(wfv1.NodePending)
+	localScope[durationCPU] = "0"
+	localScope[durationMem] = "0"
+
+	var realTimeScope = map[string]func() float64{
+		common.GlobalVarWorkflowDuration: func() float64 {
+			return time.Since(woc.wf.Status.StartedAt.Time).Seconds()
+		},
+	}
+
+	return localScope, realTimeScope
+}
+
 func (woc *wfOperationCtx) prepareMetricScope(node *wfv1.NodeStatus) (map[string]string, map[string]func() float64) {
 	realTimeScope := make(map[string]func() float64)
 	localScope := woc.globalParams.DeepCopy()

workflow/controller/steps_test.go

@@ -3,6 +3,7 @@ package controller
 import (
 	"context"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -170,6 +171,22 @@ func TestResourceDurationMetric(t *testing.T) {
 	assert.Equal(t, "0", localScope["exitCode"])
 }
 
+func TestResourceDurationMetricDefaultMetricScope(t *testing.T) {
+	wf := wfv1.Workflow{Status: wfv1.WorkflowStatus{StartedAt: metav1.NewTime(time.Now())}}
+	woc := wfOperationCtx{
+		globalParams: make(common.Parameters),
+		wf:           &wf,
+	}
+
+	localScope, realTimeScope := woc.prepareDefaultMetricScope()
+
+	assert.Equal(t, "0", localScope["resourcesDuration.cpu"])
+	assert.Equal(t, "0", localScope["resourcesDuration.memory"])
+	assert.Equal(t, "0", localScope["duration"])
+	assert.Equal(t, "Pending", localScope["status"])
+	assert.Less(t, realTimeScope["workflow.duration"](), 1.0)
+}
+
 var optionalArgumentAndParameter = `
 apiVersion: argoproj.io/v1alpha1
 kind: Workflow

workflow/metrics/util.go

@@ -1,6 +1,7 @@
 package metrics
 
 import (
+	"errors"
 	"fmt"
 	"regexp"
 	"strconv"
@@ -221,6 +222,26 @@ func IsValidMetricName(name string) bool {
 	return model.IsValidMetricName(model.LabelValue(name))
 }
 
+func ValidateMetricValues(metric *wfv1.Prometheus) error {
+	if metric.Gauge != nil {
+		if metric.Gauge.Value == "" {
+			return errors.New("missing gauge.value")
+		}
+		if metric.Gauge.Realtime != nil && *metric.Gauge.Realtime {
+			if strings.Contains(metric.Gauge.Value, "resourcesDuration.") {
+				return errors.New("'resourcesDuration.*' metrics cannot be used in real-time")
+			}
+		}
+	}
+	if metric.Counter != nil && metric.Counter.Value == "" {
+		return errors.New("missing counter.value")
+	}
+	if metric.Histogram != nil && metric.Histogram.Value == "" {
+		return errors.New("missing histogram.value")
+	}
+	return nil
+}
+
 func ValidateMetricLabels(metrics map[string]string) error {
 	for name := range metrics {
 		if !IsValidMetricName(name) {

workflow/validate/validate.go

@@ -429,6 +429,9 @@ func (ctx *templateValidationCtx) validateTemplate(tmpl *wfv1.Template, tmplCtx
 			if metric.Help == "" {
 				return errors.Errorf(errors.CodeBadRequest, "templates.%s metric '%s' must contain a help string under 'help: ' field", tmpl.Name, metric.Name)
 			}
+			if err := metrics.ValidateMetricValues(metric); err != nil {
+				return errors.Errorf(errors.CodeBadRequest, "templates.%s metric '%s' error: %s", tmpl.Name, metric.Name, err)
+			}
 		}
 	}
 	return nil

workflow/validate/validate_test.go

@@ -2234,6 +2234,89 @@ func TestInvalidMetricHelp(t *testing.T) {
 	assert.EqualError(t, err, "templates.whalesay metric 'metric_name' must contain a help string under 'help: ' field")
 }
 
+var invalidRealtimeMetricGauge = `
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  generateName: hello-world-
+spec:
+  entrypoint: whalesay
+  templates:
+  - name: whalesay
+    metrics:
+      prometheus:
+        - name: metric_name
+          help: please
+          gauge:
+            realtime: true
+            value: "{{resourcesDuration.cpu}}/{{resourcesDuration.memory}}"
+    container:
+      image: docker/whalesay:latest
+`
+
+func TestInvalidMetricGauge(t *testing.T) {
+	_, err := validate(invalidRealtimeMetricGauge)
+	assert.EqualError(t, err, "templates.whalesay metric 'metric_name' error: 'resourcesDuration.*' metrics cannot be used in real-time")
+}
+
+var invalidNoValueMetricGauge = `
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  generateName: hello-world-
+spec:
+  entrypoint: whalesay
+  templates:
+  - name: whalesay
+    metrics:
+      prometheus:
+        - name: metric_name
+          help: please
+          gauge:
+            realtime: false
+    container:
+      image: docker/whalesay:latest
+`
+
+func TestInvalidNoValueMetricGauge(t *testing.T) {
+	_, err := validate(invalidNoValueMetricGauge)
+	assert.EqualError(t, err, "templates.whalesay metric 'metric_name' error: missing gauge.value")
+}
+
+var validMetricGauges = `
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  generateName: hello-world-
+spec:
+  entrypoint: whalesay
+  templates:
+  - name: whalesay
+    metrics:
+      prometheus:
+        - name: metric_one
+          help: please
+          gauge:
+            realtime: true
+            value: "{{duration}}/{{workflow.duration}}"
+        - name: metric_two
+          help: please
+          gauge:
+            realtime: false
+            value: "{{resourcesDuration.cpu}}/{{resourcesDuration.memory}}/{{duration}}/{{workflow.duration}}"
+        - name: metric_three
+          help: please
+          gauge:
+            value: "{{resourcesDuration.cpu}}/{{resourcesDuration.memory}}/{{duration}}/{{workflow.duration}}"
+    container:
+      image: docker/whalesay:latest
+`
+
+func TestValidMetricGauge(t *testing.T) {
+	_, err := validate(validMetricGauges)
+	assert.NoError(t, err)
+}
+
 var globalVariables = `
 apiVersion: argoproj.io/v1alpha1
 kind: Workflow