Add consumer_lag in Kafka consumergroup metricset

metricbeat/docs/fields.asciidoc

@@ -15791,6 +15791,15 @@ type: keyword
 
 --
 
+*`kafka.consumergroup.consumer_lag`*::
++
+--
+consumer lag for partition/topic
+
+type: long
+
+--
+
 *`kafka.consumergroup.error.code`*::
 +
 --

metricbeat/module/kafka/broker.go

@@ -43,6 +43,7 @@ func Version(version string) kafka.Version {
 type Broker struct {
 	broker *sarama.Broker
 	cfg    *sarama.Config
+	client sarama.Client
 
 	advertisedAddr string
 	id             int32
@@ -96,6 +97,7 @@ func NewBroker(host string, settings BrokerSettings) *Broker {
 	return &Broker{
 		broker:  sarama.NewBroker(host),
 		cfg:     cfg,
+		client:  nil,
 		id:      noID,
 		matchID: settings.MatchID,
 	}
@@ -104,6 +106,7 @@ func NewBroker(host string, settings BrokerSettings) *Broker {
 // Close the broker connection
 func (b *Broker) Close() error {
 	closeBroker(b.broker)
+	b.client.Close()
 	return nil
 }
 
@@ -134,6 +137,13 @@ func (b *Broker) Connect() error {
 	debugf("found matching broker %v with id %v", other.Addr(), other.ID())
 	b.id = other.ID()
 	b.advertisedAddr = other.Addr()
+
+	c, err := getClusteWideClient(b.Addr(), b.cfg)
+	if err != nil {
+		closeBroker(b.broker)
+		return fmt.Errorf("Could not get cluster client for advertised broker with address %v", b.Addr())
+	}
+	b.client = c
 	return nil
 }
 
@@ -270,6 +280,15 @@ func (b *Broker) FetchGroupOffsets(group string, partitions map[string][]int32)
 	return b.broker.FetchOffset(requ)
 }
 
+// GetPartitionOffsetFromTheLeader fetches the OffsetNewest from the leader.
+func (b *Broker) GetPartitionOffsetFromTheLeader(topic string, partitionID int32) (int64, error) {
+	offset, err := b.client.GetOffset(topic, partitionID, sarama.OffsetNewest)
+	if err != nil {
+		return -1, err
+	}
+	return offset, nil
+}
+
 // ID returns the broker or -1 if the broker id is unknown.
 func (b *Broker) ID() int32 {
 	if b.id == noID {
@@ -524,6 +543,14 @@ func anyIPsMatch(as, bs []net.IP) bool {
 	return false
 }
 
+func getClusteWideClient(addr string, cfg *sarama.Config) (sarama.Client, error) {
+	client, err := sarama.NewClient([]string{addr}, cfg)
+	if err != nil {
+		return nil, err
+	}
+	return client, nil
+}
+
 func brokerAddresses(brokers []*sarama.Broker) []string {
 	addresses := make([]string, len(brokers))
 	for i, b := range brokers {

metricbeat/module/kafka/consumergroup/_meta/data.json

@@ -1,45 +1,49 @@
 {
     "@timestamp": "2017-10-12T08:05:34.853Z",
-    "beat": {
-        "hostname": "host.example.com",
-        "name": "host.example.com"
+    "event": {
+        "dataset": "kafka.consumergroup",
+        "duration": 115000,
+        "module": "kafka"
     },
     "kafka": {
         "broker": {
-            "address": "172.18.0.2:9092",
+            "address": "localhost:32768",
             "id": 0
         },
         "consumergroup": {
             "broker": {
-                "address": "172.18.0.2:9092",
+                "address": "localhost:32768",
                 "id": 0
             },
             "client": {
-                "host": "172.18.0.1",
-                "id": "sarama",
-                "member_id": "sarama-fcb5a5db-0474-4f3a-a5af-29e2f14549c5"
+                "host": "127.0.0.1",
+                "id": "consumer-1",
+                "member_id": "consumer-1-a12ac7d4-00aa-45a0-8b35-0a9c6e880bf4"
             },
+            "consumer_lag": 1059,
             "error": {
                 "code": 0
             },
-            "id": "test-group",
+            "id": "console-consumer-50413",
             "meta": "",
             "offset": -1,
             "partition": 0,
-            "topic": "metricbeat-test"
+            "topic": "test"
         },
         "partition": {
             "id": 0,
-            "topic_id": "0-metricbeat-test"
+            "topic_id": "0-test"
         },
         "topic": {
-            "name": "metricbeat-test"
+            "name": "test"
         }
     },
     "metricset": {
-        "host": "kafka:9092",
-        "module": "kafka",
         "name": "consumergroup",
-        "rtt": 115
+        "period": 10000
+    },
+    "service": {
+        "address": "localhost:32768",
+        "type": "kafka"
     }
-}
+}

metricbeat/module/kafka/consumergroup/_meta/fields.yml

@@ -43,6 +43,10 @@
       type: keyword
       description: custom consumer meta data string
 
+    - name: consumer_lag
+      type: long
+      description: consumer lag for partition/topic
+
     - name: error.code
       type: long
       description: >

metricbeat/module/kafka/consumergroup/consumergroup.go

@@ -80,6 +80,7 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 // Fetch consumer group metrics from kafka
 func (m *MetricSet) Fetch(r mb.ReporterV2) error {
 	broker, err := m.Connect()
+
 	if err != nil {
 		return errors.Wrap(err, "error in connect")
 	}

metricbeat/module/kafka/consumergroup/mock_test.go

@@ -27,9 +27,10 @@ import (
 )
 
 type mockClient struct {
-	listGroups        func() ([]string, error)
-	describeGroups    func(group []string) (map[string]kafka.GroupDescription, error)
-	fetchGroupOffsets func(group string) (*sarama.OffsetFetchResponse, error)
+	listGroups                      func() ([]string, error)
+	describeGroups                  func(group []string) (map[string]kafka.GroupDescription, error)
+	fetchGroupOffsets               func(group string) (*sarama.OffsetFetchResponse, error)
+	getPartitionOffsetFromTheLeader func(topic string, partitionID int32) (int64, error)
 }
 
 type mockState struct {
@@ -45,6 +46,9 @@ func defaultMockClient(state mockState) *mockClient {
 		listGroups:        makeListGroups(state),
 		describeGroups:    makeDescribeGroups(state),
 		fetchGroupOffsets: makeFetchGroupOffsets(state),
+		getPartitionOffsetFromTheLeader: func(topic string, partitionID int32) (int64, error) {
+			return 42, nil
+		},
 	}
 }
 
@@ -145,3 +149,6 @@ func (c *mockClient) DescribeGroups(groups []string) (map[string]kafka.GroupDesc
 func (c *mockClient) FetchGroupOffsets(group string, partitions map[string][]int32) (*sarama.OffsetFetchResponse, error) {
 	return c.fetchGroupOffsets(group)
 }
+func (c *mockClient) GetPartitionOffsetFromTheLeader(topic string, partitionID int32) (int64, error) {
+	return c.getPartitionOffsetFromTheLeader(topic, partitionID)
+}

metricbeat/module/kafka/consumergroup/query.go

@@ -29,6 +29,7 @@ type client interface {
 	ListGroups() ([]string, error)
 	DescribeGroups(group []string) (map[string]kafka.GroupDescription, error)
 	FetchGroupOffsets(group string, partitions map[string][]int32) (*sarama.OffsetFetchResponse, error)
+	GetPartitionOffsetFromTheLeader(topic string, partitionID int32) (int64, error)
 }
 
 func fetchGroupInfo(
@@ -113,12 +114,22 @@ func fetchGroupInfo(
 
 		for topic, partitions := range ret.off.Blocks {
 			for partition, info := range partitions {
+				partitionOffset, err := getPartitionOffsetFromTheLeader(b, topic, partition)
+				if err != nil {
+					logp.Err("failed to fetch offset for (topic, partition): ('%v', %v)", topic, partition)
+					continue
+				}
+				consumerLag := partitionOffset - info.Offset
+				if consumerLag < 0 {
+					consumerLag = 0
+				}
 				event := common.MapStr{
-					"id":        ret.group,
-					"topic":     topic,
-					"partition": partition,
-					"offset":    info.Offset,
-					"meta":      info.Metadata,
+					"id":           ret.group,
+					"topic":        topic,
+					"partition":    partition,
+					"offset":       info.Offset,
+					"meta":         info.Metadata,
+					"consumer_lag": consumerLag,
 					"error": common.MapStr{
 						"code": info.Err,
 					},
@@ -133,7 +144,6 @@ func fetchGroupInfo(
 						}
 					}
 				}
-
 				emit(event)
 			}
 		}
@@ -145,6 +155,14 @@ func fetchGroupInfo(
 	return err
 }
 
+func getPartitionOffsetFromTheLeader(b client, topic string, partitionID int32) (int64, error) {
+	offset, err := b.GetPartitionOffsetFromTheLeader(topic, partitionID)
+	if err != nil {
+		return -1, err
+	}
+	return offset, nil
+}
+
 func listGroups(b client, filter func(string) bool) ([]string, error) {
 	groups, err := b.ListGroups()
 	if err != nil {

metricbeat/module/kafka/consumergroup/query_test.go

@@ -68,35 +68,35 @@ func TestFetchGroupInfo(t *testing.T) {
 			expected: []common.MapStr{
 				testEvent("group1", "topic1", 0, common.MapStr{
 					"client": clientMeta(0),
-					"offset": int64(10),
+					"offset": int64(10), "consumer_lag": int64(42) - int64(10),
 				}),
 				testEvent("group1", "topic1", 1, common.MapStr{
 					"client": clientMeta(1),
-					"offset": int64(11),
+					"offset": int64(11), "consumer_lag": int64(42) - int64(11),
 				}),
 				testEvent("group1", "topic1", 2, common.MapStr{
 					"client": clientMeta(0),
-					"offset": int64(12),
+					"offset": int64(12), "consumer_lag": int64(42) - int64(12),
 				}),
 				testEvent("group1", "topic3", 0, common.MapStr{
 					"client": clientMeta(1),
-					"offset": int64(6),
+					"offset": int64(6), "consumer_lag": int64(42) - int64(6),
 				}),
 				testEvent("group1", "topic3", 1, common.MapStr{
 					"client": clientMeta(0),
-					"offset": int64(7),
+					"offset": int64(7), "consumer_lag": int64(42) - int64(7),
 				}),
 				testEvent("group2", "topic2", 0, common.MapStr{
 					"client": clientMeta(0),
-					"offset": int64(3),
+					"offset": int64(3), "consumer_lag": int64(42) - int64(3),
 				}),
 				testEvent("group2", "topic3", 0, common.MapStr{
 					"client": clientMeta(0),
-					"offset": int64(9),
+					"offset": int64(9), "consumer_lag": int64(42) - int64(9),
 				}),
 				testEvent("group2", "topic3", 1, common.MapStr{
 					"client": clientMeta(0),
-					"offset": int64(10),
+					"offset": int64(10), "consumer_lag": int64(42) - int64(10),
 				}),
 			},
 		},
@@ -128,11 +128,11 @@ func TestFetchGroupInfo(t *testing.T) {
 			expected: []common.MapStr{
 				testEvent("group1", "topic1", 0, common.MapStr{
 					"client": clientMeta(0),
-					"offset": int64(1),
+					"offset": int64(1), "consumer_lag": int64(42) - int64(1),
 				}),
 				testEvent("group1", "topic1", 1, common.MapStr{
 					"client": clientMeta(0),
-					"offset": int64(2),
+					"offset": int64(2), "consumer_lag": int64(42) - int64(2),
 				}),
 			},
 		},
@@ -224,7 +224,7 @@ func TestFetchGroupInfo(t *testing.T) {
 		for key, expected := range indexEvents(test.expected) {
 			event, found := indexed[key]
 			if !found {
-				t.Errorf("Missing event: %v", key)
+				t.Errorf("Missing key %v from events: %v", key, events)
 				continue
 			}
 			assertEvent(t, expected, event)