add metrics for partitions

[jsurloppe]

Hi,

This PR add 3 metrics:

• kminion_kafka_topic_partition_count: give the partition_count as value for being able to used in prom queries.
• kminion_kafka_topic_under_replicated_partition: set to 1 if a partition is currently under replicated.
• kminion_kafka_topic_partition_leader: current leader (broker ID)

That's 3 metrics that we miss from kminion, it would be nice to have them in all in one place. ;)

[CLAassistant]

All committers have signed the CLA.

[vst-name]

@weeco Hello Martin!

Can you please review this PR?
It add's much needed features(partition count series & partition leader).

My org. plans to move away from kafka-exporter to kminion, these are the only metrics we're missing.
I've tested it locally against kminion master, it looks promising.

Honestly, unless these metrics are added, we would have to fork it for a while(

Thank you!

[weeco]

Hey @jsurloppe and @vst-name , apologies for the very delayed review.

Thanks for the PR! A few points:

1. Granularity config not respected

The new partition-level metrics (partitionUnderReplicated and partitionLeader) don't check the topics.granularity config. Please add this check in the partition loop,
following the pattern in collect_topic_partition_offsets.go:50-51:

  for _, partition := range topic.Partitions {
      // Skip partition metrics if granularity is set to topic level
      if e.minionSvc.Cfg.Topics.Granularity == minion.TopicGranularityTopic {
          continue
      }
      // ... rest of the code
  }

2. Consider adding a partition_info metric

To make it easier to query "under-replicated partitions by leader" without complex PromQL joins, I'd suggest adding a fourth metric following the existing topicInfo /
brokerInfo pattern:

  e.partitionInfo = prometheus.NewDesc(
      prometheus.BuildFQName(e.cfg.Namespace, "kafka", "topic_partition_info"),
      "Partition metadata with leader as label for joining",
      []string{"topic_name", "partition_id", "leader_id"},
      nil,
  )

With value always = 1. This allows queries like:
sum by (leader_id) (
kminion_kafka_topic_under_replicated_partition{} == 1
* on (topic_name, partition_id) group_left(leader_id)
kminion_kafka_topic_partition_info{}
)