Improve patch Deployment

Author: ilyario

helm/dataflow-operator/templates/deployment.yaml

@@ -52,6 +52,8 @@ spec:
           value: {{ .Release.Namespace | quote }}
         - name: LOG_LEVEL
           value: {{ .Values.logLevel | default "info" | quote }}
+        - name: PROCESSOR_LOG_LEVEL
+          value: {{ .Values.processorLogLevel | default "info" | quote }}
         args:
         - --metrics-bind-address=:{{ .Values.metrics.port }}
         - --health-probe-bind-address=:{{ .Values.health.probePort }}

helm/dataflow-operator/values.yaml

@@ -65,6 +65,9 @@ metrics:
 # Log level for the operator (sets env LOG_LEVEL: debug, info, warn, error)
 logLevel: "info"
 
+# Log level for processor pods (sets env PROCESSOR_LOG_LEVEL in the operator; operator passes it as LOG_LEVEL to each processor pod)
+processorLogLevel: "info"
+
 # Health probe configuration
 health:
   # The address the probe endpoint binds to

internal/controller/dataflow_controller.go

@@ -30,6 +30,7 @@ import (
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/api/equality"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -475,7 +476,10 @@ func (r *DataFlowReconciler) createOrUpdateDeployment(ctx context.Context, req c
 	} else if err != nil {
 		return fmt.Errorf("failed to get Deployment: %w", err)
 	} else {
-		// Обновляем существующий Deployment, если spec изменился
+		// Обновляем существующий Deployment только при реальном изменении spec
+		if equality.Semantic.DeepEqual(existing.Spec, deployment.Spec) {
+			return nil
+		}
 		existing.Spec = deployment.Spec
 		if err := r.Update(ctx, existing); err != nil {
 			return fmt.Errorf("failed to update Deployment: %w", err)

internal/controller/dataflow_controller_test.go

@@ -408,6 +408,151 @@ func TestDataFlowReconciler_Reconcile_WithResourcesAndNodeSelector(t *testing.T)
 	assert.Equal(t, ctrl.Result{}, result)
 }
 
+// TestCreateOrUpdateDeployment_NoUpdateWhenSpecUnchanged проверяет, что при второй реконсиляции
+// с неизменным spec DataFlow не вызывается Update Deployment (нет лишних PATCH и rolling update).
+func TestCreateOrUpdateDeployment_NoUpdateWhenSpecUnchanged(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"},
+	}
+
+	// Первая реконсиляция — создаётся Deployment (DeploymentCreated)
+	_, _ = reconciler.Reconcile(ctx, req)
+	drainRecorderEvents(fakeRecorder) // сбрасываем события создания
+
+	// Вторая реконсиляция без изменения spec — Update не должен вызываться (DeploymentUpdated не должно быть)
+	_, _ = 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, 0, deploymentUpdatedCount,
+		"expected no DeploymentUpdated event on second reconcile when spec unchanged")
+}
+
+// TestCreateOrUpdateDeployment_UpdateWhenSpecChanged проверяет, что при изменении spec DataFlow
+// вызывается Update Deployment и желаемое состояние применяется.
+func TestCreateOrUpdateDeployment_UpdateWhenSpecChanged(t *testing.T) {
+	scheme := runtime.NewScheme()
+	require.NoError(t, dataflowv1.AddToScheme(scheme))
+	require.NoError(t, clientgoscheme.AddToScheme(scheme))
+
+	fakeClient := fake.NewClientBuilder().WithScheme(scheme).Build()
+	reconciler := NewDataFlowReconciler(fakeClient, scheme, nil)
+
+	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"},
+	}
+
+	// Первая реконсиляция — создаётся Deployment
+	_, _ = reconciler.Reconcile(ctx, req)
+
+	deploymentName := types.NamespacedName{Name: "dataflow-test-dataflow", Namespace: "default"}
+	var deployment appsv1.Deployment
+	require.NoError(t, fakeClient.Get(ctx, deploymentName, &deployment))
+	generationBefore := deployment.Generation
+
+	// Меняем spec DataFlow (NodeSelector попадает в spec Deployment)
+	require.NoError(t, fakeClient.Get(ctx, req.NamespacedName, dataflow))
+	dataflow.Spec.NodeSelector = map[string]string{"node-type": "compute"}
+	require.NoError(t, fakeClient.Update(ctx, dataflow))
+
+	// Вторая реконсиляция — должен вызваться Update
+	_, _ = reconciler.Reconcile(ctx, req)
+
+	require.NoError(t, fakeClient.Get(ctx, deploymentName, &deployment))
+	assert.Greater(t, deployment.Generation, generationBefore,
+		"Deployment Generation should increase after spec change")
+	assert.Equal(t, "compute", deployment.Spec.Template.Spec.NodeSelector["node-type"],
+		"Deployment NodeSelector should reflect updated DataFlow spec")
+}
+
+func drainRecorderEvents(r *record.FakeRecorder) {
+	for {
+		select {
+		case <-r.Events:
+			// продолжаем вычитывать
+		default:
+			return
+		}
+	}
+}
+
 func TestDataFlowReconciler_Reconcile_InvalidSpec(t *testing.T) {
 	scheme := runtime.NewScheme()
 	err := dataflowv1.AddToScheme(scheme)