feat: mvtx gauge metrics support (#2259)

Signed-off-by: veds-g <[email protected]>

config/advanced-install/namespaced-numaflow-server.yaml

@@ -137,12 +137,39 @@ metadata:
 ---
 apiVersion: v1
 data:
-  config.yaml: |-
+  config.yaml: |
     # url is a required field, it should be the url of the service to which the metrics proxy will connect
     # url: service_name + "." + service_namespace + ".svc.cluster.local" + ":" + port
     # example for local prometheus service
     # url: http://prometheus-operated.monitoring.svc.cluster.local:9090
     patterns:
+    - name: mono_vertex_gauge
+      object: mono-vertex
+      title: Pending Messages Lag
+      description: This query is the total number of pending messages for the mono vertex
+      expr: |
+        $metric_name{$filters}
+      params:
+        - name: start_time
+          required: false
+        - name: end_time
+          required: false
+      metrics:
+      - metric_name: monovtx_pending
+        required_filters:
+          - namespace
+          - mvtx_name
+        dimensions:
+          - name: pod
+            filters:
+              - name: pod
+                required: false
+              - name: period
+                required: false
+          - name: mono-vertex
+            filters:
+              - name: period
+                required: false
     - name: mono_vertex_histogram
       object: mono-vertex
       title: Processing Time Latency

config/advanced-install/numaflow-server.yaml

@@ -144,12 +144,39 @@ metadata:
 ---
 apiVersion: v1
 data:
-  config.yaml: |-
+  config.yaml: |
     # url is a required field, it should be the url of the service to which the metrics proxy will connect
     # url: service_name + "." + service_namespace + ".svc.cluster.local" + ":" + port
     # example for local prometheus service
     # url: http://prometheus-operated.monitoring.svc.cluster.local:9090
     patterns:
+    - name: mono_vertex_gauge
+      object: mono-vertex
+      title: Pending Messages Lag
+      description: This query is the total number of pending messages for the mono vertex
+      expr: |
+        $metric_name{$filters}
+      params:
+        - name: start_time
+          required: false
+        - name: end_time
+          required: false
+      metrics:
+      - metric_name: monovtx_pending
+        required_filters:
+          - namespace
+          - mvtx_name
+        dimensions:
+          - name: pod
+            filters:
+              - name: pod
+                required: false
+              - name: period
+                required: false
+          - name: mono-vertex
+            filters:
+              - name: period
+                required: false
     - name: mono_vertex_histogram
       object: mono-vertex
       title: Processing Time Latency

config/base/numaflow-server/numaflow-server-metrics-proxy-config.yaml

@@ -9,6 +9,33 @@ data:
     # example for local prometheus service
     # url: http://prometheus-operated.monitoring.svc.cluster.local:9090
     patterns:
+    - name: mono_vertex_gauge
+      object: mono-vertex
+      title: Pending Messages Lag
+      description: This query is the total number of pending messages for the mono vertex
+      expr: |
+        $metric_name{$filters}
+      params:
+        - name: start_time
+          required: false
+        - name: end_time
+          required: false
+      metrics:
+      - metric_name: monovtx_pending
+        required_filters:
+          - namespace
+          - mvtx_name
+        dimensions:
+          - name: pod
+            filters:
+              - name: pod
+                required: false
+              - name: period
+                required: false
+          - name: mono-vertex
+            filters:
+              - name: period
+                required: false
     - name: mono_vertex_histogram
       object: mono-vertex
       title: Processing Time Latency
@@ -77,4 +104,4 @@ data:
             - name: pod
               filters:
                 - name: pod
-                  required: false
+                  required: false

config/install.yaml

@@ -28557,12 +28557,39 @@ metadata:
 ---
 apiVersion: v1
 data:
-  config.yaml: |-
+  config.yaml: |
     # url is a required field, it should be the url of the service to which the metrics proxy will connect
     # url: service_name + "." + service_namespace + ".svc.cluster.local" + ":" + port
     # example for local prometheus service
     # url: http://prometheus-operated.monitoring.svc.cluster.local:9090
     patterns:
+    - name: mono_vertex_gauge
+      object: mono-vertex
+      title: Pending Messages Lag
+      description: This query is the total number of pending messages for the mono vertex
+      expr: |
+        $metric_name{$filters}
+      params:
+        - name: start_time
+          required: false
+        - name: end_time
+          required: false
+      metrics:
+      - metric_name: monovtx_pending
+        required_filters:
+          - namespace
+          - mvtx_name
+        dimensions:
+          - name: pod
+            filters:
+              - name: pod
+                required: false
+              - name: period
+                required: false
+          - name: mono-vertex
+            filters:
+              - name: period
+                required: false
     - name: mono_vertex_histogram
       object: mono-vertex
       title: Processing Time Latency

config/namespace-install.yaml

@@ -28445,12 +28445,39 @@ metadata:
 ---
 apiVersion: v1
 data:
-  config.yaml: |-
+  config.yaml: |
     # url is a required field, it should be the url of the service to which the metrics proxy will connect
     # url: service_name + "." + service_namespace + ".svc.cluster.local" + ":" + port
     # example for local prometheus service
     # url: http://prometheus-operated.monitoring.svc.cluster.local:9090
     patterns:
+    - name: mono_vertex_gauge
+      object: mono-vertex
+      title: Pending Messages Lag
+      description: This query is the total number of pending messages for the mono vertex
+      expr: |
+        $metric_name{$filters}
+      params:
+        - name: start_time
+          required: false
+        - name: end_time
+          required: false
+      metrics:
+      - metric_name: monovtx_pending
+        required_filters:
+          - namespace
+          - mvtx_name
+        dimensions:
+          - name: pod
+            filters:
+              - name: pod
+                required: false
+              - name: period
+                required: false
+          - name: mono-vertex
+            filters:
+              - name: period
+                required: false
     - name: mono_vertex_histogram
       object: mono-vertex
       title: Processing Time Latency

server/apis/v1/promql_service_test.go

@@ -273,6 +273,70 @@ func Test_PromQueryBuilder(t *testing.T) {
 			}
 		})
 	}
+
+	// tests for gauge metrics
+	var gauge_service = &PromQlService{
+		PlaceHolders: map[string]map[string][]string{
+			"monovtx_pending": {
+				"mono-vertex": {"$dimension", "$metric_name", "$filters"},
+			},
+		},
+		Expression: map[string]map[string]string{
+			"monovtx_pending": {
+				"mono-vertex": "$metric_name{$filters}",
+			},
+		},
+	}
+
+	gauge_metrics_tests := []struct {
+		name          string
+		requestBody   MetricsRequestBody
+		expectedQuery string
+		expectError   bool
+	}{
+		{
+			name: "Successful gauge metrics template substitution",
+			requestBody: MetricsRequestBody{
+				MetricName: "monovtx_pending",
+				Dimension:  "mono-vertex",
+				Filters: map[string]string{
+					"namespace": "test_namespace",
+					"mvtx_name": "test_mvtx",
+					"period":    "5m",
+				},
+			},
+			expectedQuery: `monovtx_pending{namespace= "test_namespace", mvtx_name= "test_mvtx", period= "5m"}`,
+		},
+		{
+			name: "Missing metric name in service config",
+			requestBody: MetricsRequestBody{
+				MetricName: "non_existent_metric",
+				Dimension:  "mono-vertex",
+				Filters: map[string]string{
+					"namespace": "test_namespace",
+					"mvtx_name": "test_mvtx",
+					"period":    "5m",
+				},
+			},
+			expectError: true,
+		},
+	}
+
+	for _, tt := range gauge_metrics_tests {
+		t.Run(tt.name, func(t *testing.T) {
+			actualQuery, err := gauge_service.BuildQuery(tt.requestBody)
+			if tt.expectError {
+				assert.Error(t, err)
+			} else {
+				assert.NoError(t, err)
+				if !comparePrometheusQueries(tt.expectedQuery, actualQuery) {
+					t.Errorf("Prometheus queries do not match.\nExpected: %s\nGot: %s", tt.expectedQuery, actualQuery)
+				} else {
+					t.Log("Prometheus queries match!")
+				}
+			}
+		})
+	}
 }
 
 func Test_QueryPrometheus(t *testing.T) {
@@ -322,6 +386,29 @@ func Test_QueryPrometheus(t *testing.T) {
 		assert.Equal(t, 1, matrix.Len())
 	})
 
+	t.Run("Successful gauge query", func(t *testing.T) {
+		mockAPI := &MockPrometheusAPI{}
+		promQlService := &PromQlService{
+			PrometheusClient: &Prometheus{
+				Api: mockAPI,
+			},
+		}
+		query := `monovtx_pending{namespace="default", mvtx_name="test-mvtx", pending="5m"}`
+		startTime := time.Now().Add(-30 * time.Minute)
+		endTime := time.Now()
+
+		ctx := context.Background()
+		result, err := promQlService.QueryPrometheus(ctx, query, startTime, endTime)
+
+		assert.NoError(t, err)
+		assert.NotNil(t, result)
+
+		// for query range , response should be a matrix
+		matrix, ok := result.(model.Matrix)
+		assert.True(t, ok)
+		assert.Equal(t, 1, matrix.Len())
+	})
+
 	t.Run("Prometheus client is nil", func(t *testing.T) {
 		service := &PromQlService{
 			PrometheusClient: nil,