From ecea8c1b23aadaac6117c91ea4f02fe8f2d15077 Mon Sep 17 00:00:00 2001 From: Jean-Paul COSAL Date: Mon, 17 Jun 2024 15:40:15 +0200 Subject: [PATCH] Added metric kafka_consumer_topic_lag --- core/internal/httpserver/prometheus.go | 21 +++++++++++++++++++++ core/internal/httpserver/prometheus_test.go | 4 ++++ 2 files changed, 25 insertions(+) diff --git a/core/internal/httpserver/prometheus.go b/core/internal/httpserver/prometheus.go index 1ba1059e..1975c005 100644 --- a/core/internal/httpserver/prometheus.go +++ b/core/internal/httpserver/prometheus.go @@ -53,6 +53,14 @@ var ( []string{"cluster", "consumer_group", "topic", "partition"}, ) + consumerTopicLagGauge = promauto.NewGaugeVec( + prometheus.GaugeOpts{ + Name: "burrow_kafka_consumer_topic_lag", + Help: "Number of messages the consumer group is behind by for a topic as reported by Burrow", + }, + []string{"cluster", "consumer_group", "topic"}, + ) + topicPartitionOffsetGauge = promauto.NewGaugeVec( prometheus.GaugeOpts{ Name: "burrow_kafka_topic_partition_offset", @@ -127,6 +135,8 @@ func (hc *Coordinator) handlePrometheusMetrics() http.HandlerFunc { consumerTotalLagGauge.With(labels).Set(float64(consumerStatus.TotalLag)) consumerStatusGauge.With(labels).Set(float64(consumerStatus.Status)) + topicLags := make(map[string]uint64) + for _, partition := range consumerStatus.Partitions { labels := map[string]string{ "cluster": cluster, @@ -136,12 +146,23 @@ func (hc *Coordinator) handlePrometheusMetrics() http.HandlerFunc { } consumerPartitionLagGauge.With(labels).Set(float64(partition.CurrentLag)) + topicLags[partition.Topic] += partition.CurrentLag if partition.Complete == 1.0 { consumerPartitionCurrentOffset.With(labels).Set(float64(partition.End.Offset)) partitionStatusGauge.With(labels).Set(float64(partition.Status)) } } + + for topic, lag := range topicLags { + labels := map[string]string{ + "cluster": cluster, + "consumer_group": consumer, + "topic": topic, + } + + consumerTopicLagGauge.With(labels).Set(float64(lag)) + } } // Topics diff --git a/core/internal/httpserver/prometheus_test.go b/core/internal/httpserver/prometheus_test.go index 0b751ec7..2dc8598e 100644 --- a/core/internal/httpserver/prometheus_test.go +++ b/core/internal/httpserver/prometheus_test.go @@ -155,6 +155,10 @@ func TestHttpServer_handlePrometheusMetrics(t *testing.T) { assert.Contains(t, promExp, `burrow_kafka_consumer_partition_lag{cluster="testcluster",consumer_group="testgroup",partition="0",topic="incomplete"} 0`) assert.Contains(t, promExp, `burrow_kafka_consumer_partition_lag{cluster="testcluster",consumer_group="testgroup",partition="1",topic="incomplete"} 10`) + assert.Contains(t, promExp, `burrow_kafka_consumer_topic_lag{cluster="testcluster",consumer_group="testgroup",topic="testtopic"} 110`) + assert.Contains(t, promExp, `burrow_kafka_consumer_topic_lag{cluster="testcluster",consumer_group="testgroup",topic="testtopic1"} 50`) + assert.Contains(t, promExp, `burrow_kafka_consumer_topic_lag{cluster="testcluster",consumer_group="testgroup",topic="incomplete"} 10`) + assert.Contains(t, promExp, `burrow_kafka_consumer_current_offset{cluster="testcluster",consumer_group="testgroup",partition="0",topic="testtopic"} 22663`) assert.Contains(t, promExp, `burrow_kafka_consumer_current_offset{cluster="testcluster",consumer_group="testgroup",partition="1",topic="testtopic"} 2488`) assert.Contains(t, promExp, `burrow_kafka_consumer_current_offset{cluster="testcluster",consumer_group="testgroup",partition="0",topic="testtopic1"} 99888`)