fix: harden multi-pod delivery and load verification

server/kafka_runtime.go

@@ -8,12 +8,14 @@ import (
 	"regexp"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/segmentio/kafka-go"
 	"github.com/segmentio/kafka-go/sasl"
 	"github.com/segmentio/kafka-go/sasl/scram"
 	"github.com/tinode/chat/server/logs"
+	immetrics "github.com/tinode/chat/server/metrics"
 	"github.com/tinode/chat/server/store/types"
 )
 
@@ -155,6 +157,11 @@ func initKafkaRuntime(raw json.RawMessage) error {
 	return nil
 }
 
+func kafkaConfigured(raw json.RawMessage) bool {
+	var cfg kafkaConfig
+	return len(raw) > 0 && json.Unmarshal(raw, &cfg) == nil && cfg.Enabled
+}
+
 func normalizeKafkaConfig(cfg *kafkaConfig) {
 	if cfg.ClientID == "" {
 		cfg.ClientID = "im-server"
@@ -333,6 +340,7 @@ func startKafkaConsumerWithGroup(topicName, groupID string, handler kafkaMessage
 		MaxWait:     time.Second,
 		Dialer:      kafkaBus.dialer(),
 	})
+	readerMetrics := startKafkaReaderMetrics(kafkaBus.ctx, reader, topic, groupID)
 	kafkaBus.wg.Add(1)
 	go func() {
 		defer kafkaBus.wg.Done()
@@ -346,6 +354,7 @@ func startKafkaConsumerWithGroup(topicName, groupID string, handler kafkaMessage
 				logs.Err.Println("Kafka fetch failed:", err)
 				continue
 			}
+			readerMetrics.observe(msg.Time)
 			commit := func(ctx context.Context, message kafka.Message) error {
 				return reader.CommitMessages(ctx, message)
 			}
@@ -355,6 +364,7 @@ func startKafkaConsumerWithGroup(topicName, groupID string, handler kafkaMessage
 				}
 				logs.Err.Printf("Kafka message stopped without commit: topic=%s partition=%d offset=%d err=%v", msg.Topic, msg.Partition, msg.Offset, err)
 			}
+			readerMetrics.clear()
 		}
 	}()
 	logs.Info.Printf("Kafka consumer started: topic=%s group=%s", topic, groupID)
@@ -373,6 +383,7 @@ func startKafkaBatchConsumer(topicName string, handler kafkaBatchHandler) {
 		Brokers: kafkaBus.config.Brokers, GroupID: groupID, Topic: topic, StartOffset: kafka.FirstOffset,
 		MinBytes: 1, MaxBytes: 10e6, MaxWait: time.Second, Dialer: kafkaBus.dialer(),
 	})
+	readerMetrics := startKafkaReaderMetrics(kafkaBus.ctx, reader, topic, groupID)
 	kafkaBus.wg.Add(1)
 	go func() {
 		defer kafkaBus.wg.Done()
@@ -387,17 +398,75 @@ func startKafkaBatchConsumer(topicName string, handler kafkaBatchHandler) {
 				logs.Err.Println("Kafka batch fetch failed:", err)
 				continue
 			}
+			if len(messages) > 0 {
+				readerMetrics.observe(messages[0].Time)
+			}
 			if err = consumeKafkaMessageBatch(kafkaBus.ctx, messages, handler, reader.CommitMessages, waitKafkaRetry); err != nil {
 				if errors.Is(err, context.Canceled) || errors.Is(kafkaBus.ctx.Err(), context.Canceled) {
 					return
 				}
 				logs.Err.Printf("Kafka batch stopped without commit: topic=%s size=%d err=%v", topic, len(messages), err)
 			}
+			readerMetrics.clear()
 		}
 	}()
 	logs.Info.Printf("Kafka batch consumer started: topic=%s group=%s size=%d wait_ms=%d", topic, groupID, kafkaBus.config.MessageBatchSize, kafkaBus.config.MessageBatchWait)
 }
 
+type kafkaReaderMetrics struct {
+	reader     *kafka.Reader
+	topic      string
+	group      string
+	oldestNano atomic.Int64
+}
+
+func startKafkaReaderMetrics(ctx context.Context, reader *kafka.Reader, topic, group string) *kafkaReaderMetrics {
+	metrics := &kafkaReaderMetrics{reader: reader, topic: topic, group: group}
+	kafkaBus.wg.Add(1)
+	go func() {
+		defer kafkaBus.wg.Done()
+		ticker := time.NewTicker(5 * time.Second)
+		defer ticker.Stop()
+		metrics.update()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+				metrics.update()
+			}
+		}
+	}()
+	return metrics
+}
+
+func (m *kafkaReaderMetrics) observe(messageTime time.Time) {
+	if !messageTime.IsZero() {
+		m.oldestNano.Store(messageTime.UnixNano())
+	}
+	m.update()
+}
+
+func (m *kafkaReaderMetrics) clear() {
+	m.oldestNano.Store(0)
+	m.update()
+}
+
+func (m *kafkaReaderMetrics) update() {
+	stats := m.reader.Stats()
+	if stats.Lag >= 0 {
+		immetrics.KafkaConsumerLag.WithLabelValues(clusterPodName(), m.topic, m.group).Set(float64(stats.Lag))
+	}
+	age := float64(0)
+	if oldest := m.oldestNano.Load(); oldest > 0 {
+		age = time.Since(time.Unix(0, oldest)).Seconds()
+		if age < 0 {
+			age = 0
+		}
+	}
+	immetrics.KafkaBacklogOldestAge.WithLabelValues(clusterPodName(), m.topic, m.group).Set(age)
+}
+
 func fetchKafkaMessageBatch(ctx context.Context, reader *kafka.Reader, maxSize int, maxWait time.Duration) ([]kafka.Message, error) {
 	first, err := reader.FetchMessage(ctx)
 	if err != nil {