Add structured logging

Author: ilyario

cmd/processor/main.go

@@ -22,12 +22,16 @@ import (
 	"flag"
 	"os"
 	"os/signal"
+	"strings"
 	"syscall"
 
+	"go.uber.org/zap"
+	"go.uber.org/zap/zapcore"
 	ctrl "sigs.k8s.io/controller-runtime"
 	zaprctrl "sigs.k8s.io/controller-runtime/pkg/log/zap"
 
 	dataflowv1 "github.com/dataflow-operator/dataflow/api/v1"
+	"github.com/dataflow-operator/dataflow/internal/logkeys"
 	"github.com/dataflow-operator/dataflow/internal/processor"
 )
 
@@ -44,9 +48,14 @@ func main() {
 	opts.BindFlags(flag.CommandLine)
 	flag.Parse()
 
-	// Настройка логгера
-	ctrl.SetLogger(zaprctrl.New(zaprctrl.UseFlagOptions(&opts)))
-	logger := ctrl.Log.WithName("processor").WithValues("namespace", namespace, "name", name)
+	// Уровень логирования: переменная окружения LOG_LEVEL (debug, info, warn, error) или флаги
+	levelEnabler := processorLevelFromEnv(os.Getenv("LOG_LEVEL"), opts.Level)
+	zapOpts := []zaprctrl.Opts{zaprctrl.UseFlagOptions(&opts)}
+	if levelEnabler != nil {
+		zapOpts = append(zapOpts, zaprctrl.Level(levelEnabler))
+	}
+	ctrl.SetLogger(zaprctrl.New(zapOpts...))
+	logger := ctrl.Log.WithName("processor").WithValues(logkeys.DataflowNamespace, namespace, logkeys.DataflowName, name)
 
 	// Читаем spec из файла
 	specData, err := os.ReadFile(specPath)
@@ -102,3 +111,16 @@ func main() {
 
 	logger.Info("Processor stopped successfully")
 }
+
+// processorLevelFromEnv returns zap LevelEnabler from LOG_LEVEL env if set, otherwise optsLevel.
+func processorLevelFromEnv(envLevel string, optsLevel zapcore.LevelEnabler) zapcore.LevelEnabler {
+	s := strings.TrimSpace(strings.ToLower(envLevel))
+	if s == "" {
+		return optsLevel
+	}
+	var l zapcore.Level
+	if err := l.UnmarshalText([]byte(s)); err != nil {
+		return optsLevel
+	}
+	return zap.NewAtomicLevelAt(l)
+}

helm/dataflow-operator/templates/deployment.yaml

@@ -50,6 +50,8 @@ spec:
           value: {{ include "dataflow-operator.fullname" . | quote }}
         - name: OPERATOR_NAMESPACE
           value: {{ .Release.Namespace | quote }}
+        - name: LOG_LEVEL
+          value: {{ .Values.logLevel | default "info" | quote }}
         args:
         - --metrics-bind-address=:{{ .Values.metrics.port }}
         - --health-probe-bind-address=:{{ .Values.health.probePort }}

helm/dataflow-operator/values.yaml

@@ -69,6 +69,9 @@ metrics:
   # The address the metric endpoint binds to
   port: 9090
 
+# Log level for the operator (sets env LOG_LEVEL: debug, info, warn, error)
+logLevel: "info"
+
 # Health probe configuration
 health:
   # The address the probe endpoint binds to

internal/connectors/kafka.go

@@ -36,6 +36,7 @@ import (
 
 	"github.com/IBM/sarama"
 	v1 "github.com/dataflow-operator/dataflow/api/v1"
+	"github.com/dataflow-operator/dataflow/internal/logkeys"
 	"github.com/dataflow-operator/dataflow/internal/metrics"
 	"github.com/dataflow-operator/dataflow/internal/retry"
 	"github.com/dataflow-operator/dataflow/internal/types"
@@ -237,6 +238,7 @@ func (k *KafkaSourceConnector) Connect(ctx context.Context) error {
 			}(), err)
 	}
 	k.consumer = consumer
+	k.logger.Info("Successfully connected to Kafka", "brokers", k.config.Brokers, "topic", k.config.Topic, "group", consumerGroup)
 
 	// Record connection status
 	if k.namespace != "" && k.name != "" {
@@ -607,6 +609,7 @@ func (k *KafkaSourceConnector) Close() error {
 		return nil
 	}
 
+	k.logger.Info("Closing Kafka source connection", "brokers", k.config.Brokers, "topic", k.config.Topic)
 	k.closed = true
 	if k.consumer != nil {
 		// Record connection status
@@ -668,6 +671,7 @@ func (h *kafkaConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSes
 				msgData, err = h.connector.deserializeAvro(session.Context(), message.Value)
 				if err != nil {
 					h.connector.logger.Error(err, "Failed to deserialize Avro message",
+						logkeys.MessageID, fmt.Sprintf("%d/%d", message.Partition, message.Offset),
 						"topic", message.Topic,
 						"partition", message.Partition,
 						"offset", message.Offset)
@@ -685,11 +689,12 @@ func (h *kafkaConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSes
 			msg.Metadata["partition"] = message.Partition
 			msg.Metadata["offset"] = message.Offset
 			msg.Metadata["key"] = string(message.Key)
+			// Commit offset only after the message is successfully written to the sink (called by sink connectors)
+			msg.Ack = func() { session.MarkMessage(message, "") }
 
 			select {
 			case h.msgChan <- msg:
-				session.MarkMessage(message, "")
-				// Record metrics
+				// Record metrics (offset will be committed by sink after successful write)
 				if h.connector.namespace != "" && h.connector.name != "" {
 					metrics.RecordConnectorMessageRead(h.connector.namespace, h.connector.name, "kafka", "source")
 				}
@@ -887,6 +892,7 @@ func (k *KafkaSinkConnector) Connect(ctx context.Context) error {
 			}(), err)
 	}
 	k.producer = producer
+	k.logger.Info("Successfully connected to Kafka", "brokers", k.config.Brokers, "topic", k.config.Topic)
 
 	// Record connection status
 	if k.namespace != "" && k.name != "" {
@@ -981,6 +987,10 @@ func (k *KafkaSinkConnector) Write(ctx context.Context, messages <-chan *types.M
 
 			msg.Metadata["partition"] = partition
 			msg.Metadata["offset"] = offset
+
+			if msg.Ack != nil {
+				msg.Ack()
+			}
 		}
 	}
 }
@@ -994,6 +1004,7 @@ func (k *KafkaSinkConnector) Close() error {
 		return nil
 	}
 
+	k.logger.Info("Closing Kafka sink connection", "brokers", k.config.Brokers, "topic", k.config.Topic)
 	k.closed = true
 	if k.producer != nil {
 		// Record connection status