Fix metrics and autoupdate

Author: ilyario

cmd/processor/main.go

@@ -20,29 +20,35 @@ import (
 	"context"
 	"encoding/json"
 	"flag"
+	"net/http"
 	"os"
 	"os/signal"
 	"strings"
 	"syscall"
 
+	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
 	ctrl "sigs.k8s.io/controller-runtime"
 	zaprctrl "sigs.k8s.io/controller-runtime/pkg/log/zap"
+	"sigs.k8s.io/controller-runtime/pkg/metrics"
 
 	dataflowv1 "github.com/dataflow-operator/dataflow/api/v1"
 	"github.com/dataflow-operator/dataflow/internal/constants"
 	"github.com/dataflow-operator/dataflow/internal/logkeys"
+	_ "github.com/dataflow-operator/dataflow/internal/metrics" // Register metrics
 	"github.com/dataflow-operator/dataflow/internal/processor"
 )
 
 func main() {
 	var specPath string
 	var namespace string
 	var name string
+	var metricsPort string
 	flag.StringVar(&specPath, "spec-path", "/etc/dataflow/spec.json", "Path to DataFlow spec JSON file")
 	flag.StringVar(&namespace, "namespace", "", "Namespace of the DataFlow resource")
 	flag.StringVar(&name, "name", "", "Name of the DataFlow resource")
+	flag.StringVar(&metricsPort, "metrics-port", ":9090", "Address for the metrics HTTP server")
 	opts := zaprctrl.Options{
 		Development: true,
 	}
@@ -82,6 +88,16 @@ func main() {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
+	// Start metrics HTTP server (must be before proc.Start so /metrics is available from the start)
+	mux := http.NewServeMux()
+	mux.Handle("/metrics", promhttp.HandlerFor(metrics.Registry, promhttp.HandlerOpts{}))
+	metricsServer := &http.Server{Addr: metricsPort, Handler: mux}
+	go func() {
+		if err := metricsServer.ListenAndServe(); err != nil && err != http.ErrServerClosed {
+			logger.Error(err, "Metrics server exited")
+		}
+	}()
+
 	// Signal handling for graceful shutdown
 	sigChan := make(chan os.Signal, constants.DefaultSingleValueChannelBufferSize)
 	signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)

cmd/processor/metrics_test.go

@@ -0,0 +1,59 @@
+/*
+Copyright 2024.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package main
+
+import (
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+	"sigs.k8s.io/controller-runtime/pkg/metrics"
+
+	_ "github.com/dataflow-operator/dataflow/internal/metrics" // Register metrics
+)
+
+func TestMetricsEndpoint(t *testing.T) {
+	mux := http.NewServeMux()
+	mux.Handle("/metrics", promhttp.HandlerFor(metrics.Registry, promhttp.HandlerOpts{}))
+	server := httptest.NewServer(mux)
+	defer server.Close()
+
+	resp, err := http.Get(server.URL + "/metrics")
+	if err != nil {
+		t.Fatalf("GET /metrics: %v", err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		t.Errorf("GET /metrics status = %d, want 200", resp.StatusCode)
+	}
+
+	// Read body
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		t.Fatalf("Read body: %v", err)
+	}
+	content := string(body)
+
+	// Should contain dataflow_ metric definitions (from internal/metrics init)
+	if !strings.Contains(content, "dataflow_") {
+		t.Errorf("metrics output should contain dataflow_ metrics, got: %s", content)
+	}
+}

config/rbac/role.yaml

@@ -4,6 +4,14 @@ kind: ClusterRole
 metadata:
   name: manager-role
 rules:
+- apiGroups:
+  - ""
+  resources:
+  - pods
+  verbs:
+  - get
+  - list
+  - watch
 - apiGroups:
   - ""
   resources:

internal/controller/dataflow_controller.go

@@ -28,6 +28,7 @@ import (
 	"time"
 
 	crand "crypto/rand"
+	"crypto/sha256"
 
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
@@ -320,7 +321,7 @@ func (r *DataFlowReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 	}
 
 	// Create or update Deployment
-	if err := r.createOrUpdateDeployment(ctx, req, &dataflow); err != nil {
+	if err := r.createOrUpdateDeployment(ctx, req, &dataflow, resolvedSpec); err != nil {
 		log.Error(err, "failed to create or update Deployment")
 		if r.Recorder != nil {
 			r.Recorder.Eventf(&dataflow, corev1.EventTypeWarning, "DeploymentFailed", "Failed to create or update Deployment: %v", err)
@@ -478,13 +479,25 @@ func (r *DataFlowReconciler) processorImageFor(dataflow *dataflowv1.DataFlow) st
 	return r.processorImage
 }
 
+// specHashAnnotation is the pod template annotation key for spec content hash.
+// When spec changes, the hash changes, triggering a Deployment rollout.
+const specHashAnnotation = "dataflow.dataflow.io/spec-hash"
+
 // createOrUpdateDeployment creates or updates Deployment for the processor.
-func (r *DataFlowReconciler) createOrUpdateDeployment(ctx context.Context, req ctrl.Request, dataflow *dataflowv1.DataFlow) error {
+func (r *DataFlowReconciler) createOrUpdateDeployment(ctx context.Context, req ctrl.Request, dataflow *dataflowv1.DataFlow, resolvedSpec *dataflowv1.DataFlowSpec) error {
 	log := log.FromContext(ctx)
 
 	deploymentName := fmt.Sprintf("dataflow-%s", dataflow.Name)
 	configMapName := fmt.Sprintf("dataflow-%s-spec", dataflow.Name)
 
+	// Compute spec hash so pod template changes when ConfigMap content changes, triggering rollout.
+	specJSON, err := json.Marshal(resolvedSpec)
+	if err != nil {
+		return fmt.Errorf("failed to marshal spec for hash: %w", err)
+	}
+	hash := sha256.Sum256(specJSON)
+	specHash := hex.EncodeToString(hash[:])
+
 	labels := map[string]string{
 		"app":                        "dataflow-processor",
 		"dataflow.dataflow.io/name":  dataflow.Name,
@@ -506,6 +519,9 @@ func (r *DataFlowReconciler) createOrUpdateDeployment(ctx context.Context, req c
 			Template: corev1.PodTemplateSpec{
 				ObjectMeta: metav1.ObjectMeta{
 					Labels: labels,
+					Annotations: map[string]string{
+						specHashAnnotation: specHash,
+					},
 				},
 				Spec: corev1.PodSpec{
 					TerminationGracePeriodSeconds: ptr.To(int64(30)),
@@ -522,6 +538,9 @@ func (r *DataFlowReconciler) createOrUpdateDeployment(ctx context.Context, req c
 							Env: []corev1.EnvVar{
 								{Name: "LOG_LEVEL", Value: processorLogLevel()},
 							},
+							Ports: []corev1.ContainerPort{
+								{Name: "metrics", ContainerPort: 9090, Protocol: corev1.ProtocolTCP},
+							},
 							Lifecycle: &corev1.Lifecycle{
 								PreStop: &corev1.LifecycleHandler{
 									Exec: &corev1.ExecAction{
@@ -567,7 +586,7 @@ func (r *DataFlowReconciler) createOrUpdateDeployment(ctx context.Context, req c
 
 	// Check if Deployment exists
 	existing := &appsv1.Deployment{}
-	err := r.Get(ctx, types.NamespacedName{Name: deploymentName, Namespace: req.Namespace}, existing)
+	err = r.Get(ctx, types.NamespacedName{Name: deploymentName, Namespace: req.Namespace}, existing)
 	if err != nil && apierrors.IsNotFound(err) {
 		// Ensure finalizer before creating first child so deletion is coordinated
 		if err := r.ensureDataFlowFinalizer(ctx, req); err != nil {

internal/controller/dataflow_controller_test.go

@@ -520,6 +520,10 @@ func TestDataFlowReconciler_Reconcile_CreateDeployment(t *testing.T) {
 	err = fakeClient.Get(ctx, deploymentName, &deployment)
 	assert.NoError(t, err, "Deployment should be created")
 	assert.Equal(t, "dataflow-test-dataflow", deployment.Name)
+	assert.Contains(t, deployment.Spec.Template.Annotations, specHashAnnotation,
+		"Deployment pod template should have spec-hash annotation for ConfigMap change detection")
+	assert.NotEmpty(t, deployment.Spec.Template.Annotations[specHashAnnotation],
+		"spec-hash annotation should be non-empty")
 	require.Len(t, deployment.Spec.Template.Spec.Containers, 1)
 	assert.Equal(t, version.DefaultProcessorImage(), deployment.Spec.Template.Spec.Containers[0].Image, "default processor image should match controller")
 	var hasLogLevel bool
@@ -1139,6 +1143,88 @@ func TestCreateOrUpdateDeployment_UpdateWhenSpecChanged(t *testing.T) {
 		"Deployment NodeSelector should reflect updated DataFlow spec")
 }
 
+// TestCreateOrUpdateDeployment_UpdateWhenSpecContentChanged verifies that when DataFlow spec content
+// changes (e.g. Kafka brokers, SecretRef), the spec-hash annotation changes and Deployment is updated,
+// triggering a pod restart for the new config.
+func TestCreateOrUpdateDeployment_UpdateWhenSpecContentChanged(t *testing.T) {
+	scheme := runtime.NewScheme()
+	require.NoError(t, dataflowv1.AddToScheme(scheme))
+	require.NoError(t, clientgoscheme.AddToScheme(scheme))
+
+	fakeRecorder := record.NewFakeRecorder(10)
+	fakeClient := fake.NewClientBuilder().WithScheme(scheme).Build()
+	reconciler := NewDataFlowReconciler(fakeClient, scheme, fakeRecorder)
+
+	ctx := context.Background()
+	dataflow := &dataflowv1.DataFlow{
+		TypeMeta: metav1.TypeMeta{
+			APIVersion: "dataflow.dataflow.io/v1",
+			Kind:       "DataFlow",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "test-dataflow",
+			Namespace: "default",
+		},
+		Spec: dataflowv1.DataFlowSpec{
+			Source: dataflowv1.SourceSpec{
+				Type: "kafka",
+				Kafka: &dataflowv1.KafkaSourceSpec{
+					Brokers:       []string{"localhost:9092"},
+					Topic:         "test-topic",
+					ConsumerGroup: "test-group",
+				},
+			},
+			Sink: dataflowv1.SinkSpec{
+				Type:  "kafka",
+				Kafka: &dataflowv1.KafkaSinkSpec{Brokers: []string{"localhost:9092"}, Topic: "output-topic"},
+			},
+		},
+	}
+	require.NoError(t, fakeClient.Create(ctx, dataflow))
+
+	req := ctrl.Request{
+		NamespacedName: types.NamespacedName{Name: "test-dataflow", Namespace: "default"},
+	}
+
+	// First reconcile — Deployment created
+	_, _ = reconciler.Reconcile(ctx, req)
+	drainRecorderEvents(fakeRecorder)
+
+	deploymentName := types.NamespacedName{Name: "dataflow-test-dataflow", Namespace: "default"}
+	var deployment appsv1.Deployment
+	require.NoError(t, fakeClient.Get(ctx, deploymentName, &deployment))
+	hashBefore := deployment.Spec.Template.Annotations[specHashAnnotation]
+	require.NotEmpty(t, hashBefore, "initial Deployment should have spec-hash annotation")
+
+	// Change DataFlow spec content (Kafka brokers) — same ConfigMap name, but content changes
+	require.NoError(t, fakeClient.Get(ctx, req.NamespacedName, dataflow))
+	dataflow.Spec.Source.Kafka.Brokers = []string{"kafka-1:9092", "kafka-2:9092"}
+	require.NoError(t, fakeClient.Update(ctx, dataflow))
+
+	// Second reconcile — Deployment should be updated (spec-hash changed)
+	_, _ = reconciler.Reconcile(ctx, req)
+
+	var deploymentUpdatedCount int
+	for {
+		select {
+		case e := <-fakeRecorder.Events:
+			if strings.Contains(e, "DeploymentUpdated") {
+				deploymentUpdatedCount++
+			}
+		default:
+			goto done
+		}
+	}
+done:
+	assert.Equal(t, 1, deploymentUpdatedCount,
+		"expected DeploymentUpdated event when spec content (Kafka brokers) changed")
+
+	require.NoError(t, fakeClient.Get(ctx, deploymentName, &deployment))
+	hashAfter := deployment.Spec.Template.Annotations[specHashAnnotation]
+	assert.NotEqual(t, hashBefore, hashAfter,
+		"spec-hash should change when Kafka brokers change, triggering pod restart")
+}
+
 // TestCreateOrUpdateDeployment_RetryOnConflict verifies that when Deployment Update returns 409 Conflict,
 // the controller retries and succeeds on the next attempt (no extra rollout: spec comparison skips redundant Update).
 func TestCreateOrUpdateDeployment_RetryOnConflict(t *testing.T) {