Add consumer_lag in Kafka consumergroup metricset

CHANGELOG.next.asciidoc

@@ -350,6 +350,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Add `fingerprint` processor. {issue}11173[11173] {pull}14205[14205]
 - Add support for API keys in Elasticsearch outputs. {pull}14324[14324]
 - Ensure that init containers are no longer tailed after they stop {pull}14394[14394]
+- Add consumer_lag in Kafka consumergroup metricset {pull}14822[14822]
 
 *Auditbeat*
 

metricbeat/docs/fields.asciidoc

@@ -15791,6 +15791,15 @@ type: keyword
 
 --
 
+*`kafka.consumergroup.consumer_lag`*::
++
+--
+consumer lag for partition/topic calculated as the difference between the partition offset and consumer offset
+
+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 := getClusterWideClient(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,7 +280,16 @@ func (b *Broker) FetchGroupOffsets(group string, partitions map[string][]int32)
 	return b.broker.FetchOffset(requ)
 }
 
-// ID returns the broker or -1 if the broker id is unknown.
+// FetchPartitionOffsetFromTheLeader fetches the OffsetNewest from the leader.
+func (b *Broker) FetchPartitionOffsetFromTheLeader(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 ID or -1 if the broker id is unknown.
 func (b *Broker) ID() int32 {
 	if b.id == noID {
 		return b.broker.ID()
@@ -524,6 +543,14 @@ func anyIPsMatch(as, bs []net.IP) bool {
 	return false
 }
 
+func getClusterWideClient(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 calculated as the difference between the partition offset and consumer offset
+
     - name: error.code
       type: long
       description: >

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) FetchPartitionOffsetFromTheLeader(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)
+	FetchPartitionOffsetFromTheLeader(topic string, partitionID int32) (int64, error)
 }
 
 func fetchGroupInfo(
@@ -113,12 +114,19 @@ 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
 				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 +141,6 @@ func fetchGroupInfo(
 						}
 					}
 				}
-
 				emit(event)
 			}
 		}
@@ -145,6 +152,14 @@ func fetchGroupInfo(
 	return err
 }
 
+func getPartitionOffsetFromTheLeader(b client, topic string, partitionID int32) (int64, error) {
+	offset, err := b.FetchPartitionOffsetFromTheLeader(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

@@ -67,36 +67,44 @@ func TestFetchGroupInfo(t *testing.T) {
 			}),
 			expected: []common.MapStr{
 				testEvent("group1", "topic1", 0, common.MapStr{
-					"client": clientMeta(0),
-					"offset": int64(10),
+					"client":       clientMeta(0),
+					"offset":       int64(10),
+					"consumer_lag": int64(42) - int64(10),
 				}),
 				testEvent("group1", "topic1", 1, common.MapStr{
-					"client": clientMeta(1),
-					"offset": int64(11),
+					"client":       clientMeta(1),
+					"offset":       int64(11),
+					"consumer_lag": int64(42) - int64(11),
 				}),
 				testEvent("group1", "topic1", 2, common.MapStr{
-					"client": clientMeta(0),
-					"offset": int64(12),
+					"client":       clientMeta(0),
+					"offset":       int64(12),
+					"consumer_lag": int64(42) - int64(12),
 				}),
 				testEvent("group1", "topic3", 0, common.MapStr{
-					"client": clientMeta(1),
-					"offset": int64(6),
+					"client":       clientMeta(1),
+					"offset":       int64(6),
+					"consumer_lag": int64(42) - int64(6),
 				}),
 				testEvent("group1", "topic3", 1, common.MapStr{
-					"client": clientMeta(0),
-					"offset": int64(7),
+					"client":       clientMeta(0),
+					"offset":       int64(7),
+					"consumer_lag": int64(42) - int64(7),
 				}),
 				testEvent("group2", "topic2", 0, common.MapStr{
-					"client": clientMeta(0),
-					"offset": int64(3),
+					"client":       clientMeta(0),
+					"offset":       int64(3),
+					"consumer_lag": int64(42) - int64(3),
 				}),
 				testEvent("group2", "topic3", 0, common.MapStr{
-					"client": clientMeta(0),
-					"offset": int64(9),
+					"client":       clientMeta(0),
+					"offset":       int64(9),
+					"consumer_lag": int64(42) - int64(9),
 				}),
 				testEvent("group2", "topic3", 1, common.MapStr{
-					"client": clientMeta(0),
-					"offset": int64(10),
+					"client":       clientMeta(0),
+					"offset":       int64(10),
+					"consumer_lag": int64(42) - int64(10),
 				}),
 			},
 		},
@@ -127,12 +135,14 @@ func TestFetchGroupInfo(t *testing.T) {
 			topics: []string{"topic1"},
 			expected: []common.MapStr{
 				testEvent("group1", "topic1", 0, common.MapStr{
-					"client": clientMeta(0),
-					"offset": int64(1),
+					"client":       clientMeta(0),
+					"offset":       int64(1),
+					"consumer_lag": int64(42) - int64(1),
 				}),
 				testEvent("group1", "topic1", 1, common.MapStr{
-					"client": clientMeta(0),
-					"offset": int64(2),
+					"client":       clientMeta(0),
+					"offset":       int64(2),
+					"consumer_lag": int64(42) - int64(2),
 				}),
 			},
 		},
@@ -224,7 +234,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)