Skip to content

Commit

Permalink
Expose max timestamp for each topic and/or partition (#3)
Browse files Browse the repository at this point in the history
* Expose max timestamp for each topic and/or partition


Kafka 3.0+ added support to fetch offset with the max timestamp.
This is useful in monitoring the steady progress of producing to a topic/partion
but also to detect very high timestamps in the topic that can stuck topic retention.

This patch adds two new metrics:
- kafka_topic_partition_max_timestamp: for each topic partition reports the max timestamp in that partition
- kafka_topic_max_timestamp: for each topic reports the max timestamp in that topic, computed as the maximum timestamp across partitions


See: 
- https://issues.apache.org/jira/browse/KAFKA-12541
- https://cwiki.apache.org/confluence/display/KAFKA/KIP-734%3A+Improve+AdminClient.listOffsets+to+return+timestamp+and+offset+for+the+record+with+the+largest+timestamp

* Update prometheus/collect_topic_partition_offsets.go
  • Loading branch information
amuraru committed Dec 16, 2022
1 parent 205d739 commit 9ec9282
Show file tree
Hide file tree
Showing 2 changed files with 67 additions and 0 deletions.
51 changes: 51 additions & 0 deletions prometheus/collect_topic_partition_offsets.go
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,15 @@ func (e *Exporter) collectTopicPartitionOffsets(ctx context.Context, ch chan<- p
return false
}

// Highest Timestamp Offsets
// NB: this requires Kafka Brokers 3.0+ (see https://issues.apache.org/jira/browse/KAFKA-12541)
// In older versions this is returning the timestamp of the low watermarks (earliest offset)
maxTimestampOffsets, err := e.minionSvc.ListOffsetsCached(ctx, -3)
if err != nil {
e.logger.Error("failed to fetch offsets for max timestamp", zap.Error(err))
return false
}

// Process Low Watermarks
for _, topic := range lowWaterMarks.Topics {
if !e.minionSvc.IsTopicAllowed(topic.Topic) {
Expand Down Expand Up @@ -101,5 +110,47 @@ func (e *Exporter) collectTopicPartitionOffsets(ctx context.Context, ch chan<- p
}
}

// Process Max Timestamps
for _, topic := range maxTimestampOffsets.Topics {
if !e.minionSvc.IsTopicAllowed(topic.Topic) {
continue
}
topicMaxTimestamp := int64(0)
hasErrors := false
for _, partition := range topic.Partitions {
err := kerr.ErrorForCode(partition.ErrorCode)
if err != nil {
hasErrors = true
isOk = false
continue
}
if topicMaxTimestamp < partition.Timestamp {
topicMaxTimestamp = partition.Timestamp
}
// Let's end here if partition metrics shall not be exposed
if e.minionSvc.Cfg.Topics.Granularity == minion.TopicGranularityTopic {
continue
}
if partition.Timestamp > 0 {
ch <- prometheus.MustNewConstMetric(
e.partitionMaxTimestamp,
prometheus.GaugeValue,
float64(partition.Timestamp),
topic.Topic,
strconv.Itoa(int(partition.Partition)),
)
}
}
// We only want to report the max of all partition max timestamps if we receive results from all partitions
// and the topic is not empty
if !hasErrors && topicMaxTimestamp > 0 {
ch <- prometheus.MustNewConstMetric(
e.topicMaxTimestamp,
prometheus.GaugeValue,
float64(topicMaxTimestamp),
topic.Topic,
)
}
}
return isOk
}
16 changes: 16 additions & 0 deletions prometheus/exporter.go
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,8 @@ type Exporter struct {
partitionHighWaterMark *prometheus.Desc
topicLowWaterMarkSum *prometheus.Desc
partitionLowWaterMark *prometheus.Desc
topicMaxTimestamp *prometheus.Desc
partitionMaxTimestamp *prometheus.Desc

// Consumer Groups
consumerGroupInfo *prometheus.Desc
Expand Down Expand Up @@ -172,6 +174,20 @@ func (e *Exporter) InitializeMetrics() {
[]string{"topic_name"},
nil,
)
// Partition Max Timestamp
e.partitionMaxTimestamp = prometheus.NewDesc(
prometheus.BuildFQName(e.cfg.Namespace, "kafka", "topic_partition_max_timestamp"),
"Partition Max Timestamp",
[]string{"topic_name", "partition_id"},
nil,
)
// Topic Max Timestamp
e.topicMaxTimestamp = prometheus.NewDesc(
prometheus.BuildFQName(e.cfg.Namespace, "kafka", "topic_max_timestamp"),
"Topic Max Timestamp",
[]string{"topic_name"},
nil,
)

// Consumer Group Metrics
// Group Info
Expand Down

0 comments on commit 9ec9282

Please sign in to comment.