Merge pull request linkedin#783 from javishere/ADD-partition-status-m…

…etric

Add partition status metric

core/internal/httpserver/prometheus.go

@@ -24,11 +24,19 @@ var (
 	consumerStatusGauge = promauto.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Name: "burrow_kafka_consumer_status",
-			Help: "The status of the consumer group. It is calculated from the highest status for the individual partitions. Statuses are an index list from NOTFOUND, OK, WARN, ERR, STOP, STALL, REWIND",
+			Help: "The status of the consumer group. It is calculated from the highest status for the individual partitions. Statuses are an index list from NOTFOUND, OK, WARN, or ERR",
 		},
 		[]string{"cluster", "consumer_group"},
 	)
 
+	partitionStatusGauge = promauto.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Name: "burrow_kafka_topic_partition_status",
+			Help: "The status of topic partition. It is calculated from the highest status for the individual partitions. Statuses are an index list from OK, WARN, STOP, STALL, REWIND",
+		},
+		[]string{"cluster", "consumer_group", "topic", "partition"},
+	)
+
 	consumerPartitionCurrentOffset = promauto.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Name: "burrow_kafka_consumer_current_offset",
@@ -117,6 +125,7 @@ func (hc *Coordinator) handlePrometheusMetrics() http.HandlerFunc {
 
 					if partition.Complete == 1.0 {
 						consumerPartitionCurrentOffset.With(labels).Set(float64(partition.End.Offset))
+						partitionStatusGauge.With(labels).Set(float64(partition.Status))
 					}
 				}
 			}