[Перевод] Ваш Kafka lag врёт: как настроить алерты по реальной задержке, а не по числу сообщений

• latest_offset – позиция следующего сообщения, которое будет записано
  
• committed_offset– позиция, до которой потребитель уже обработал сообщения

• Создать потребителя
  
• Переместиться к нужному смещению 
  
• Считать сообщение
  
• Извлечь его таймстемп
  
• Сделать это для каждой партиции с лагом

• Проблема холодного старта:таблица соответствий изначально пустая. Пока не накопится достаточное количество точек, метрика фактически не имеет значения.
  
• Граничный случай с простаивающим продюсером: когда производство сообщений останавливается, таблица «схлопывает» повторяющиеся смещения в плоский участок. Интерполяция внутри такого участка приводит к делению на ноль (dy = 0 в формуле), что дает NaN для time lag.
  
• Переменная пропускная способность:линейная интерполяция предполагает постоянную скорость поступления сообщений между точками. Если внутри одного интервала опроса (по умолчанию 30 секунд) происходят всплески, оценки времени становятся неточными, поскольку реальный нелинейный характер нагрузки не учитывается.

• Создать потребителя и перейти к committed offset
  
• Считать одно сообщение
  
• Извлечь его таймстемп
  
• Посчитать: time_lag = now - message_timestamp
  
• Закешировать результат, чтобы не перегружать Kafka

• Точный time lag независимо от пропускной способности топика
  
• Корректное поведение при простое продюсера (time lag продолжает расти, как и должен)
  
• Контролируемую нагрузку на Kafka за счет кэширования

kafka_consumergroup_group_lag{group="orders", topic="orders", partition="0"} 1523
kafka_consumergroup_group_lag_seconds{group="orders", topic="orders", partition="0"} 12.5

kafka_consumergroup_group_max_lag{group="orders"} 2847
kafka_consumergroup_group_max_lag_seconds{group="orders"} 45.2

kafka_consumergroup_group_max_lag_seconds{group="orders-processor"} > 60

• Создайте базовую конфигурацию

# Создайте базовый конфиг
cat > config.toml << 'EOF'
[exporter]
poll_interval = "30s"
http_port = 8000
[[clusters]]
name = "my-cluster"
bootstrap_servers = "kafka:9092"
EOF

docker run -d \
  --name klag-exporter \
  -p 8000:8000 \
  -v $(pwd)/config.toml:/etc/klag-exporter/config.toml \
  --network your-kafka-network \
  ghcr.io/your-org/klag-exporter:latest

# Проверить состояние сервиса
curl http://localhost:8000/health
# Проверить готовность сервиса
curl http://localhost:8000/ready
# Посмотреть метрики
curl http://localhost:8000/metrics | grep kafka_consumergroup

kafka_consumergroup_group_lag{cluster_name="my-cluster",group="my-group",topic="my-topic",partition="0"} 1523
kafka_consumergroup_group_lag_seconds{cluster_name="my-cluster",consumer_group="my-group",topic="my-topic",partition="0"} 12.5

git clone https://github.com/softwaremill/klag-exporter
cd klag-exporter/test-stack
docker-compose up -d

• Настройки экспортера

[exporter]
poll_interval = "30s"       # Как часто собирать метрики
http_port = 8000            # Порт для endpoint с метриками
http_host = "0.0.0.0"       # Адрес прослушивания
granularity = "partition"   # "topic" или "partition"
log_level = "info"          # debug, info, warn, error
[exporter.timestamp_sampling]
enabled = true              # Рассчитывать временной лаг, рекомендуется
cache_ttl = "60s"           # Как долго кэшировать timestamp
max_concurrent_fetches = 10 # Параллельные запросы timestamp

[[clusters]]
name = "production"
bootstrap_servers = "kafka1:9092,kafka2:9092,kafka3:9092"
# Фильтровать группы и топики для мониторинга
group_whitelist = ["^prod-.*"]             # Regex: только группы, начинающиеся с "prod-"
group_blacklist = ["^test-.*", "^__.*"]    # Исключить тестовые и внутренние группы
topic_whitelist = [".*"]                   # Все топики
topic_blacklist = ["^__.*"]                # Исключить внутренние топики
# Добавить пользовательские labels к метрикам
[clusters.labels]
environment = "production"
datacenter = "us-east-1"

[clusters.consumer_properties]
"security.protocol" = "SASL_PLAINTEXT"
"sasl.mechanism" = "PLAIN"
"sasl.username" = "${KAFKA_USER}"
"sasl.password" = "${KAFKA_PASSWORD}"

[clusters.consumer_properties]
"security.protocol" = "SASL_SSL"
"sasl.mechanism" = "SCRAM-SHA-512"
"sasl.username" = "${KAFKA_USER}"
"sasl.password" = "${KAFKA_PASSWORD}"
"ssl.ca.location" = "/certs/ca.crt"

[clusters.consumer_properties]
"security.protocol" = "SSL"
"ssl.ca.location" = "/certs/ca.crt"
"ssl.certificate.location" = "/certs/client.crt"
"ssl.key.location" = "/certs/client.key"

[clusters.consumer_properties]
"security.protocol" = "SASL_SSL"
"sasl.mechanism" = "AWS_MSK_IAM"
"sasl.jaas.config" = "software.amazon.msk.auth.iam.IAMLoginModule required;"

bootstrap_servers = "${KAFKA_BROKERS}"
"sasl.password" = "${KAFKA_PASSWORD}"

# Из репозитория
cp test-stack/grafana/provisioning/dashboards/kafka-lag.json \
  /your/grafana/dashboards/

• Перейдите в Dashboards → Import
  
• Загрузите kafka-lag.json или вставьте его содержимое
  
• Выберите ваш источник данных Prometheus
  
• Нажмите Import

• Total Consumer Lag – общий лаг потребителей, сумма по всем группам
  
• Max Time Lag – максимальный временной лаг, задержка в худшем случае
  
• Exporter Status – статус экспортера, UP/DOWN
  
• Scrape Duration – длительность сбора метрик, показатель производительности сбора
  
• Consumer Groups – количество отслеживаемых групп потребителей

• Consumer Group Total Lag – динамика общего лага группы потребителей во времени
  
• Time Lag in Seconds – временной лаг в секундах, главная метрика
  
• Per-Partition Lag – лаг по партициям для отладки конкретных проблем

• Partition Latest Offsets – распределение последних смещений по партициям
  
• Topic Message Rate – скорость записи сообщений в топик во времени

[[clusters]]
name = "production-us"
bootstrap_servers = "kafka-us.example.com:9092"
[clusters.labels]
region = "us-east-1"
environment = "production"
[[clusters]]
name = "production-eu"
bootstrap_servers = "kafka-eu.example.com:9092"
[clusters.labels]
region = "eu-west-1"
environment = "production"
[[clusters]]
name = "staging"
bootstrap_servers = "kafka-staging.internal:9092"
[clusters.labels]
environment = "staging"

[exporter.otlp]
enabled = true
endpoint = "http://otel-collector:4317"
push_interval = "15s"

• Проверьте подключение к Kafka:

# Из контейнера/pod экспортера
nc -zv kafka-broker 9092

docker logs klag-exporter
# или
kubectl logs deployment/klag-exporter

• Проверьте паттерны whitelist/blacklist: если у вас строгие паттерны, группы могут быть отфильтрованы.
  
• Проверьте, что группы потребителей существуют: kafka-consumer-groups --bootstrap-server kafka:9092 --list
  
• Проверьте, есть ли у потребителей зафиксированные смещения: klag-exporter показывает только группы, у которых есть committed offsets.

• Проверьте, включено ли считывание таймстемпов:

[exporter.timestamp_sampling]
enabled = true

• Проверьте targets в Prometheus: http://localhost:9090/targets – klag-exporter в статусе UP?
  
• Проверьте источник данных: Grafana → Configuration → Data Sources → Test
  
• Проверьте названия метрик: дашборд ожидает конкретные имена метрик. Если вы их изменили, обновите запросы.

• klag-exporter, который собирает метрики offset lag и time lag
  
• Prometheus, который забирает и хранит эти метрики
  
• дашборды Grafana, показывающие динамику и распределение лага
  
• алерты, настроенные под ваши SLA
  
• понимание граничных случаев: обнаружение compaction и retention

• Offset lag показывает, на сколько сообщений вы отстали; time lag показывает, насколько вы отстали в реальном времени.Важны обе метрики, но ваши SLA измеряются именно во временном лаге.
  
• Инструменты на основе интерполяции ломаются на простаивающих или низконагруженных топиках.Они могут показывать нулевой лаг, когда реальное отставание уже измеряется часами.
  
• Прямое считывание таймстемпов точнее, но требует кэширования. Получайте данные только для партиций с лагом и кэшируйте результаты с TTL.
  
• Настраивайте алерты по time lag, а не по offset lag. Дежурные инженеры скажут спасибо, когда смогут сразу понять критичность ситуации.