Merge pull request #72 from chemasan/master

Members in a consumer group

kafka_exporter.go

@@ -42,6 +42,7 @@ var (
 	consumergroupLag                   *prometheus.Desc
 	consumergroupLagSum                *prometheus.Desc
 	consumergroupLagZookeeper		   *prometheus.Desc
+	consumergroupMembers               *prometheus.Desc
 )
 
 // Exporter collects Kafka stats from the given server and exports them using
@@ -364,6 +365,9 @@ func (e *Exporter) Collect(ch chan<- prometheus.Metric) {
 					offsetFetchRequest.AddPartition(topic, partition)
 				}
 			}
+			ch <- prometheus.MustNewConstMetric(
+				consumergroupMembers, prometheus.GaugeValue, float64(len(group.Members)), group.GroupId,
+			)
 			if offsetFetchResponse, err := broker.FetchOffset(&offsetFetchRequest); err != nil {
 				plog.Errorf("Cannot get offset of group %s: %v", group.GroupId, err)
 			} else {
@@ -564,6 +568,12 @@ func main() {
 		[]string{"consumergroup", "topic"}, labels,
 	)
 
+	consumergroupMembers = prometheus.NewDesc(
+		prometheus.BuildFQName(namespace, "consumergroup", "members"),
+		"Amount of members in a consumer group",
+		[]string{"consumergroup"}, labels,
+	)
+
 	if *logSarama {
 		sarama.Logger = log.New(os.Stdout, "[sarama] ", log.LstdFlags)
 	}