Update show metrics

Author: ilyario

internal/gui/api_handler.go

@@ -328,13 +328,64 @@ func (h *APIHandler) handleMetrics(w http.ResponseWriter, r *http.Request, parts
 		return
 	}
 
-	metrics := map[string]interface{}{
-		"namespace": namespace,
-		"name":      name,
-		"metrics":   map[string]interface{}{},
+	if h.server.operatorMetricsURL == "" {
+		metrics := map[string]interface{}{
+			"namespace": namespace,
+			"name":      name,
+			"metrics":   map[string]interface{}{},
+		}
+		json.NewEncoder(w).Encode(metrics)
+		return
+	}
+
+	metricsURL := strings.TrimSuffix(h.server.operatorMetricsURL, "/") + "/metrics"
+	req, err := http.NewRequestWithContext(r.Context(), http.MethodGet, metricsURL, nil)
+	if err != nil {
+		h.server.logger.Error(err, "Failed to create metrics request")
+		http.Error(w, "Failed to create metrics request", http.StatusInternalServerError)
+		return
+	}
+
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		h.server.logger.Error(err, "Failed to fetch metrics from operator")
+		http.Error(w, "Failed to fetch metrics from operator", http.StatusBadGateway)
+		return
 	}
+	defer resp.Body.Close()
 
-	json.NewEncoder(w).Encode(metrics)
+	if resp.StatusCode != http.StatusOK {
+		h.server.logger.Info("Operator metrics returned non-200", "status", resp.StatusCode)
+		http.Error(w, fmt.Sprintf("Operator metrics returned %d", resp.StatusCode), http.StatusBadGateway)
+		return
+	}
+
+	w.Header().Set("Content-Type", "text/plain; version=0.0.4; charset=utf-8")
+	filterPrometheusByDataFlow(resp.Body, w, namespace, name)
+}
+
+// filterPrometheusByDataFlow reads Prometheus text format from src and writes to dst
+// only lines for dataflow_* metrics with namespace and name labels matching the given DataFlow.
+func filterPrometheusByDataFlow(src io.Reader, dst io.Writer, namespace, name string) {
+	nsMatch := `namespace="` + namespace + `"`
+	nameMatch := `name="` + name + `"`
+	scanner := bufio.NewScanner(src)
+	const maxLineSize = 64 * 1024
+	buf := make([]byte, 0, maxLineSize)
+	scanner.Buffer(buf, maxLineSize)
+
+	for scanner.Scan() {
+		line := scanner.Text()
+		if strings.HasPrefix(line, "#") {
+			if strings.Contains(line, "dataflow_") {
+				fmt.Fprintln(dst, line)
+			}
+			continue
+		}
+		if strings.HasPrefix(line, "dataflow_") && strings.Contains(line, nsMatch) && strings.Contains(line, nameMatch) {
+			fmt.Fprintln(dst, line)
+		}
+	}
 }
 
 func (h *APIHandler) handleStatus(w http.ResponseWriter, r *http.Request, parts []string) {

internal/gui/api_handler_test.go

@@ -34,16 +34,21 @@ import (
 var ctx = context.Background()
 
 func setupTestServer() (*Server, error) {
+	return setupTestServerWithMetricsURL("")
+}
+
+func setupTestServerWithMetricsURL(operatorMetricsURL string) (*Server, error) {
 	scheme := runtime.NewScheme()
 	utilruntime.Must(clientgoscheme.AddToScheme(scheme))
 	utilruntime.Must(dataflowv1.AddToScheme(scheme))
 
 	fakeClient := fake.NewClientBuilder().WithScheme(scheme).Build()
 
 	server := &Server{
-		client:    fakeClient,
-		k8sClient: nil, // not needed for dataflow CRUD tests
-		logger:    ctrl.Log.WithName("test"),
+		client:              fakeClient,
+		k8sClient:           nil, // not needed for dataflow CRUD tests
+		logger:              ctrl.Log.WithName("test"),
+		operatorMetricsURL:   operatorMetricsURL,
 	}
 
 	return server, nil
@@ -208,6 +213,126 @@ func TestAPIHandler_DeleteDataFlow(t *testing.T) {
 	}
 }
 
+func TestAPIHandler_Metrics_RequiresNamespaceAndName(t *testing.T) {
+	server, err := setupTestServer()
+	if err != nil {
+		t.Fatalf("Failed to setup test server: %v", err)
+	}
+	handler := NewAPIHandler(server)
+
+	for _, tc := range []struct {
+		query string
+	}{
+		{"?"},
+		{"?namespace=default"},
+		{"?name=foo"},
+	} {
+		req := httptest.NewRequest("GET", "/metrics"+tc.query, nil)
+		w := httptest.NewRecorder()
+		handler.ServeHTTP(w, req)
+		if w.Code != http.StatusBadRequest {
+			t.Errorf("query %q: expected 400, got %d", tc.query, w.Code)
+		}
+	}
+}
+
+func TestAPIHandler_Metrics_StubWhenNoOperatorURL(t *testing.T) {
+	server, err := setupTestServer()
+	if err != nil {
+		t.Fatalf("Failed to setup test server: %v", err)
+	}
+	handler := NewAPIHandler(server)
+
+	req := httptest.NewRequest("GET", "/metrics?namespace=default&name=myflow", nil)
+	w := httptest.NewRecorder()
+	handler.ServeHTTP(w, req)
+
+	if w.Code != http.StatusOK {
+		t.Errorf("Expected 200, got %d: %s", w.Code, w.Body.String())
+	}
+	if w.Header().Get("Content-Type") != "application/json" {
+		t.Errorf("Expected application/json, got %s", w.Header().Get("Content-Type"))
+	}
+	var out map[string]interface{}
+	if err := json.NewDecoder(w.Body).Decode(&out); err != nil {
+		t.Fatalf("Failed to decode: %v", err)
+	}
+	if out["namespace"] != "default" || out["name"] != "myflow" {
+		t.Errorf("Expected namespace=default, name=myflow, got %v", out)
+	}
+	metrics, ok := out["metrics"].(map[string]interface{})
+	if !ok || len(metrics) != 0 {
+		t.Errorf("Expected empty metrics map, got %v", out["metrics"])
+	}
+}
+
+func TestAPIHandler_Metrics_ProxyFromOperator(t *testing.T) {
+	promOutput := `# HELP dataflow_messages_received_total Total messages received
+# TYPE dataflow_messages_received_total counter
+dataflow_messages_received_total{namespace="default",name="myflow",source_type="kafka"} 100
+dataflow_messages_received_total{namespace="other",name="otherflow",source_type="kafka"} 50
+# HELP go_goroutines Number of goroutines
+# TYPE go_goroutines gauge
+go_goroutines 10
+`
+	mockOperator := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/metrics" {
+			http.NotFound(w, r)
+			return
+		}
+		w.Header().Set("Content-Type", "text/plain; version=0.0.4")
+		w.Write([]byte(promOutput))
+	}))
+	defer mockOperator.Close()
+
+	server, err := setupTestServerWithMetricsURL(mockOperator.URL)
+	if err != nil {
+		t.Fatalf("Failed to setup test server: %v", err)
+	}
+	handler := NewAPIHandler(server)
+
+	req := httptest.NewRequest("GET", "/metrics?namespace=default&name=myflow", nil)
+	w := httptest.NewRecorder()
+	handler.ServeHTTP(w, req)
+
+	if w.Code != http.StatusOK {
+		t.Errorf("Expected 200, got %d: %s", w.Code, w.Body.String())
+	}
+	body := w.Body.String()
+	if !strings.Contains(body, `dataflow_messages_received_total{namespace="default",name="myflow"`) {
+		t.Errorf("Expected filtered metrics for default/myflow, got: %s", body)
+	}
+	if strings.Contains(body, `namespace="other"`) {
+		t.Errorf("Should not include metrics from other namespace: %s", body)
+	}
+	if strings.Contains(body, "go_goroutines") {
+		t.Errorf("Should not include non-dataflow metrics: %s", body)
+	}
+}
+
+func TestFilterPrometheusByDataFlow(t *testing.T) {
+	input := `# HELP dataflow_foo A metric
+# TYPE dataflow_foo counter
+dataflow_foo{namespace="ns1",name="n1"} 1
+dataflow_foo{namespace="ns2",name="n2"} 2
+# HELP go_goroutines Goroutines
+# TYPE go_goroutines gauge
+go_goroutines 5
+`
+	var buf bytes.Buffer
+	filterPrometheusByDataFlow(strings.NewReader(input), &buf, "ns1", "n1")
+	out := buf.String()
+	if !strings.Contains(out, `dataflow_foo{namespace="ns1",name="n1"} 1`) {
+		t.Errorf("Expected ns1/n1 metric: %s", out)
+	}
+	if strings.Contains(out, `namespace="ns2"`) {
+		t.Errorf("Should not include ns2: %s", out)
+	}
+	if strings.Contains(out, "go_goroutines") {
+		t.Errorf("Should not include go_goroutines: %s", out)
+	}
+}
+
 // TestScannerReadsLongLogLine verifies that a scanner with an increased buffer
 // reads lines longer than 64 KB without "token too long" error.
 func TestScannerReadsLongLogLine(t *testing.T) {

internal/gui/server.go

@@ -85,11 +85,12 @@ func (h staticHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 
 // Server represents the web server for the GUI.
 type Server struct {
-	bindAddr   string
-	httpServer *http.Server
-	client     client.Client
-	k8sClient  *kubernetes.Clientset
-	logger     logr.Logger
+	bindAddr          string
+	httpServer        *http.Server
+	client            client.Client
+	k8sClient         *kubernetes.Clientset
+	logger            logr.Logger
+	operatorMetricsURL string
 }
 
 // NewServer creates a new GUI server.
@@ -130,10 +131,11 @@ func NewServer(bindAddr, kubeconfig string) (*Server, error) {
 	}
 
 	server := &Server{
-		bindAddr:  bindAddr,
-		client:    k8sClient,
-		k8sClient: clientset,
-		logger:    logger,
+		bindAddr:          bindAddr,
+		client:            k8sClient,
+		k8sClient:         clientset,
+		logger:            logger,
+		operatorMetricsURL: os.Getenv("OPERATOR_METRICS_URL"),
 	}
 
 	mux := http.NewServeMux()

web/src/api/client.js

@@ -86,7 +86,7 @@ export async function getStatus(namespace, name) {
 }
 
 export async function getMetrics(namespace, name) {
-  return request(
+  return requestText(
     `/metrics?namespace=${encodeURIComponent(namespace)}&name=${encodeURIComponent(name)}`
   )
 }