Remove leaderElection config

Author: ilyario

helm/dataflow-operator/templates/deployment.yaml

@@ -55,9 +55,6 @@ spec:
         args:
         - --metrics-bind-address=:{{ .Values.metrics.port }}
         - --health-probe-bind-address=:{{ .Values.health.probePort }}
-        {{- if .Values.leaderElection.enabled }}
-        - --leader-elect
-        {{- end }}
         ports:
         - name: metrics
           containerPort: {{ .Values.metrics.port }}

helm/dataflow-operator/values.yaml

@@ -56,13 +56,6 @@ rbac:
   # Specifies whether RBAC resources should be created
   create: true
 
-# Leader election configuration
-leaderElection:
-  enabled: false
-  # LeaderElectionID is the name of the resource that leader election
-  # will use for holding the leader lock.
-  leaderElectionID: "dataflow-operator.dataflow.io"
-
 # Metrics configuration
 metrics:
   enabled: true

internal/transformers/router.go

@@ -18,31 +18,39 @@ package transformers
 
 import (
 	"context"
-	"fmt"
 	"strings"
 
 	v1 "github.com/dataflow-operator/dataflow/api/v1"
 	"github.com/dataflow-operator/dataflow/internal/types"
+	"github.com/go-logr/logr"
 	"github.com/tidwall/gjson"
 )
 
 // RouterTransformer routes messages to different sinks based on conditions
 type RouterTransformer struct {
 	config *v1.RouterTransformation
+	logger logr.Logger
 }
 
 // NewRouterTransformer creates a new router transformer
 func NewRouterTransformer(config *v1.RouterTransformation) *RouterTransformer {
 	return &RouterTransformer{
 		config: config,
+		logger: logr.Discard(),
 	}
 }
 
+// SetLogger sets the logger for the transformer (used by processor to inject logr)
+func (r *RouterTransformer) SetLogger(logger logr.Logger) {
+	r.logger = logger
+}
+
 // Transform routes messages based on conditions
 // Returns messages with routing metadata
 func (r *RouterTransformer) Transform(ctx context.Context, message *types.Message) ([]*types.Message, error) {
-	fmt.Printf("DEBUG Router: Processing message, routes count: %d\n", len(r.config.Routes))
-	fmt.Printf("DEBUG Router: Message data: %s\n", string(message.Data))
+	r.logger.V(1).Info("Router processing message",
+		"routesCount", len(r.config.Routes),
+		"dataSize", len(message.Data))
 
 	for i, route := range r.config.Routes {
 		// Check if condition contains comparison operator (==)
@@ -51,19 +59,24 @@ func (r *RouterTransformer) Transform(ctx context.Context, message *types.Messag
 		var expectedValue string
 		var isComparison bool
 
-		fmt.Printf("DEBUG Router: Checking route %d, condition: '%s'\n", i, condition)
+		r.logger.V(1).Info("Router checking route",
+			"routeIndex", i,
+			"condition", condition)
 
 		// Parse condition like "$.type == 'order'" or "$.type"
 		if idx := findComparisonOperator(condition); idx >= 0 {
 			// Trim spaces from field path
 			fieldPath = strings.TrimSpace(condition[:idx])
 			expectedValue = extractStringValue(condition[idx:])
 			isComparison = true
-			fmt.Printf("DEBUG Router: Parsed comparison - fieldPath: '%s', expectedValue: '%s'\n", fieldPath, expectedValue)
+			r.logger.V(1).Info("Router parsed comparison",
+				"fieldPath", fieldPath,
+				"expectedValue", expectedValue)
 		} else {
 			fieldPath = condition
 			isComparison = false
-			fmt.Printf("DEBUG Router: No comparison operator found, using fieldPath: '%s'\n", fieldPath)
+			r.logger.V(1).Info("Router using field path (no comparison)",
+				"fieldPath", fieldPath)
 		}
 
 		// Remove $. prefix if present (gjson doesn't need it for root fields)
@@ -77,19 +90,25 @@ func (r *RouterTransformer) Transform(ctx context.Context, message *types.Messag
 		result := gjson.GetBytes(message.Data, fieldPath)
 
 		if !result.Exists() {
-			fmt.Printf("DEBUG Router: Field '%s' does not exist in message\n", fieldPath)
+			r.logger.V(1).Info("Router field does not exist",
+				"fieldPath", fieldPath)
 			continue
 		}
 
-		fmt.Printf("DEBUG Router: Field '%s' exists, value: '%s' (raw: %v)\n", fieldPath, result.String(), result.Value())
+		r.logger.V(1).Info("Router field exists",
+			"fieldPath", fieldPath,
+			"value", result.String())
 
 		// Check if condition is true
 		var isTrue bool
 		if isComparison {
 			// For comparison, check if value matches expected
 			value := result.String()
 			isTrue = value == expectedValue
-			fmt.Printf("DEBUG Router: Comparison result: value='%s' == expected='%s' = %v\n", value, expectedValue, isTrue)
+			r.logger.V(1).Info("Router comparison result",
+				"value", value,
+				"expectedValue", expectedValue,
+				"match", isTrue)
 		} else {
 			// For simple existence check, use truthiness
 			value := result.Value()
@@ -116,14 +135,16 @@ func (r *RouterTransformer) Transform(ctx context.Context, message *types.Messag
 			}
 			newMsg.Metadata["routed_condition"] = route.Condition
 			newMsg.Timestamp = message.Timestamp
-			// Debug: log routing decision
-			fmt.Printf("DEBUG Router: Message routed to condition '%s', value='%s', expected='%s'\n",
-				route.Condition, result.String(), expectedValue)
+			r.logger.V(1).Info("Router message routed",
+				"condition", route.Condition,
+				"value", result.String(),
+				"expectedValue", expectedValue)
 			return []*types.Message{newMsg}, nil
 		} else if isComparison {
-			// Debug: log why condition didn't match
-			fmt.Printf("DEBUG Router: Condition '%s' didn't match: value='%s', expected='%s'\n",
-				route.Condition, result.String(), expectedValue)
+			r.logger.V(1).Info("Router condition did not match",
+				"condition", route.Condition,
+				"value", result.String(),
+				"expectedValue", expectedValue)
 		}
 	}
 

internal/transformers/router_test.go

@@ -0,0 +1,91 @@
+/*
+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 transformers
+
+import (
+	"context"
+	"encoding/json"
+	"testing"
+
+	v1 "github.com/dataflow-operator/dataflow/api/v1"
+	"github.com/dataflow-operator/dataflow/internal/types"
+	"github.com/go-logr/logr"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// Ensure RouterTransformer implements SetLogger (used by processor).
+var _ interface{ SetLogger(logr.Logger) } = (*RouterTransformer)(nil)
+
+func TestRouterTransformer_SetLogger_DoesNotPanic(t *testing.T) {
+	transformer := NewRouterTransformer(&v1.RouterTransformation{
+		Routes: []v1.RouteRule{
+			{Condition: "$.type == 'order'", Sink: v1.SinkSpec{Type: "kafka"}},
+		},
+	})
+	transformer.SetLogger(logr.Discard())
+
+	msg := types.NewMessage([]byte(`{"type":"order"}`))
+	result, err := transformer.Transform(context.Background(), msg)
+	require.NoError(t, err)
+	require.Len(t, result, 1)
+	assert.Contains(t, result[0].Metadata, "routed_condition")
+	assert.Equal(t, "$.type == 'order'", result[0].Metadata["routed_condition"])
+}
+
+func TestRouterTransformer_Transform_WithLogger_SameBehavior(t *testing.T) {
+	transformer := NewRouterTransformer(&v1.RouterTransformation{
+		Routes: []v1.RouteRule{
+			{Condition: "type == 'user'", Sink: v1.SinkSpec{Type: "kafka"}},
+			{Condition: "type == 'order'", Sink: v1.SinkSpec{Type: "kafka"}},
+		},
+	})
+	transformer.SetLogger(logr.Discard())
+	ctx := context.Background()
+
+	t.Run("match first route", func(t *testing.T) {
+		data, _ := json.Marshal(map[string]interface{}{"id": 1, "type": "user", "name": "alice"})
+		msg := types.NewMessage(data)
+		result, err := transformer.Transform(ctx, msg)
+		require.NoError(t, err)
+		require.Len(t, result, 1)
+		assert.Equal(t, "type == 'user'", result[0].Metadata["routed_condition"])
+		assert.Equal(t, data, result[0].Data)
+	})
+
+	t.Run("match second route", func(t *testing.T) {
+		data, _ := json.Marshal(map[string]interface{}{"id": 2, "type": "order", "amount": 100})
+		msg := types.NewMessage(data)
+		result, err := transformer.Transform(ctx, msg)
+		require.NoError(t, err)
+		require.Len(t, result, 1)
+		assert.Equal(t, "type == 'order'", result[0].Metadata["routed_condition"])
+		assert.Equal(t, data, result[0].Data)
+	})
+
+	t.Run("no match returns original message", func(t *testing.T) {
+		data, _ := json.Marshal(map[string]interface{}{"id": 3, "type": "unknown"})
+		msg := types.NewMessage(data)
+		msg.Metadata["custom"] = "value"
+		result, err := transformer.Transform(ctx, msg)
+		require.NoError(t, err)
+		require.Len(t, result, 1)
+		assert.NotContains(t, result[0].Metadata, "routed_condition")
+		assert.Equal(t, data, result[0].Data)
+		assert.Equal(t, "value", result[0].Metadata["custom"])
+	})
+}

main.go

@@ -71,14 +71,10 @@ func levelFromEnvOrOptions(envLevel string, optsLevel zapcore.LevelEnabler) zapc
 
 func main() {
 	var metricsAddr string
-	var enableLeaderElection bool
 	var probeAddr string
 	var logFile string
 	flag.StringVar(&metricsAddr, "metrics-bind-address", ":9090", "The address the metric endpoint binds to.")
 	flag.StringVar(&probeAddr, "health-probe-bind-address", ":9091", "The address the probe endpoint binds to.")
-	flag.BoolVar(&enableLeaderElection, "leader-elect", false,
-		"Enable leader election for controller manager. "+
-			"Enabling this will ensure there is only one active controller manager.")
 	flag.StringVar(&logFile, "log-file", "", "Path to log file. If empty, logs will be written to stdout.")
 	opts := zaprctrl.Options{
 		Development: true,
@@ -143,7 +139,7 @@ func main() {
 			Port: 9443,
 		}),
 		HealthProbeBindAddress: probeAddr,
-		LeaderElection:         enableLeaderElection,
+		LeaderElection:         true, // Always HA-ready: only one active controller across replicas
 		LeaderElectionID:       "dataflow-operator.dataflow.io",
 	})
 	if err != nil {