Merge pull request #1507 from grafana/report_per_partition_readiness

report readiness per kafka partition

Author: Dieterbe

input/kafkamdm/kafkamdm.go

@@ -360,6 +360,7 @@ func (k *KafkaMdm) trackStats(topic string, partition int32) {
 			kafkaStats.Lag.Set(lag)
 			k.lagMonitor.StoreOffsets(partition, currentOffset, newest, ts)
 			kafkaStats.Priority.Set(k.lagMonitor.GetPartitionPriority(partition))
+			kafkaStats.Ready.Set(cluster.Manager.IsReady())
 		}
 	}
 }

stats/kafka.go

@@ -19,6 +19,7 @@ type KafkaPartition struct {
 	LogSize  Gauge64
 	Lag      Gauge64
 	Priority Gauge64
+	Ready    Bool
 }
 
 func NewKafkaPartition(prefix string) *KafkaPartition {
@@ -27,5 +28,6 @@ func NewKafkaPartition(prefix string) *KafkaPartition {
 	registry.getOrAdd(prefix+".log_size", &k.LogSize)
 	registry.getOrAdd(prefix+".lag", &k.Lag)
 	registry.getOrAdd(prefix+".priority", &k.Priority)
+	registry.getOrAdd(prefix+".ready", &k.Ready)
 	return &k
 }