Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Add committed offset metric per CG and partition #255

Open
wants to merge 2 commits into
base: master
Choose a base branch
from
Open

Add committed offset metric per CG and partition #255

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
478f6ee
Add committed offset metric per CG and partition
tkornai Mar 26, 2024
ffcb3e0
consumer_group_topic_partition_offset should be a gauge
tkornai Mar 26, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions docs/metrics.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -95,6 +95,10 @@ kminion_kafka_consumer_group_topic_offset_sum{group_id="bigquery-sink",topic_nam
# TYPE kminion_kafka_consumer_group_topic_partition_lag gauge
kminion_kafka_consumer_group_topic_partition_lag{group_id="bigquery-sink",partition_id="10",topic_name="shop-activity"} 147481

# HELP kminion_kafka_consumer_group_topic_partition_offset The committed offset of a consumer group for a given partition
# TYPE kminion_kafka_consumer_group_topic_partition_offset gauge
kminion_kafka_consumer_group_topic_partition_offset{group_id="bigquery-sink",partition_id="10",topic_name="shop-activity"} 427

# HELP kminion_kafka_consumer_group_topic_lag The number of messages a consumer group is lagging behind across all partitions in a topic
# TYPE kminion_kafka_consumer_group_topic_lag gauge
kminion_kafka_consumer_group_topic_lag{group_id="bigquery-sink",topic_name="shop-activity"} 147481
Expand Down
16 changes: 16 additions & 0 deletions prometheus/collect_consumer_group_lags.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -84,6 +84,14 @@ func (e *Exporter) collectConsumerGroupLagsOffsetTopic(_ context.Context, ch cha
if e.minionSvc.Cfg.ConsumerGroups.Granularity == minion.ConsumerGroupGranularityTopic {
continue
}
ch <- prometheus.MustNewConstMetric(
e.consumerGroupTopicPartitionOffset,
prometheus.GaugeValue,
float64(partition.Value.Offset),
groupName,
topicName,
strconv.Itoa(int(partitionID)),
)
ch <- prometheus.MustNewConstMetric(
e.consumerGroupTopicPartitionLag,
prometheus.GaugeValue,
Expand Down Expand Up @@ -176,6 +184,14 @@ func (e *Exporter) collectConsumerGroupLagsAdminAPI(ctx context.Context, ch chan
if e.minionSvc.Cfg.ConsumerGroups.Granularity == minion.ConsumerGroupGranularityTopic {
continue
}
ch <- prometheus.MustNewConstMetric(
e.consumerGroupTopicPartitionOffset,
prometheus.GaugeValue,
float64(partition.Offset),
groupName,
topic.Topic,
strconv.Itoa(int(partition.Partition)),
)
ch <- prometheus.MustNewConstMetric(
e.consumerGroupTopicPartitionLag,
prometheus.GaugeValue,
Expand Down
10 changes: 9 additions & 1 deletion prometheus/exporter.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,7 @@ type Exporter struct {
consumerGroupMembersEmpty *prometheus.Desc
consumerGroupTopicMembers *prometheus.Desc
consumerGroupAssignedTopicPartitions *prometheus.Desc
consumerGroupTopicPartitionOffset *prometheus.Desc
consumerGroupTopicOffsetSum *prometheus.Desc
consumerGroupTopicPartitionLag *prometheus.Desc
consumerGroupTopicLag *prometheus.Desc
Expand Down Expand Up @@ -179,7 +180,14 @@ func (e *Exporter) InitializeMetrics() {
[]string{"group_id", "topic_name"},
nil,
)
// Topic / Partition Offset Sum (useful for calculating the consumed messages / sec on a topic)
// Topic Partition Offsets (useful for calculating the consumed messages / sec on a topic partition)
e.consumerGroupTopicPartitionOffset = prometheus.NewDesc(
prometheus.BuildFQName(e.cfg.Namespace, "kafka", "consumer_group_topic_partition_offset"),
"The committed offset of a consumer group for a given partition",
[]string{"group_id", "topic_name", "partition_id"},
nil,
)
// Topic Offset Sum (useful for calculating the consumed messages / sec on a topic)
e.consumerGroupTopicOffsetSum = prometheus.NewDesc(
prometheus.BuildFQName(e.cfg.Namespace, "kafka", "consumer_group_topic_offset_sum"),
"The sum of all committed group offsets across all partitions in a topic",
Expand Down