Add consumer_lag in Kafka consumergroup metricset #14822

ChrsMark · 2019-11-27T14:26:15Z

This PR adds consumer_lag field in consumergroup metricset. This is calculated by subtracting groupOffset from partitionOffset for a partition-topic pair.

partitionOffset is retrieved from the cluster directly using https://github.com/Shopify/sarama/blob/afedecade3c6d8e99ab6dfeeea7814bf800b90a4/client.go#L62

May resolve_ #3608.

Signed-off-by: chrismark [email protected]

Signed-off-by: chrismark <[email protected]>

jsoriano

@ChrsMark thanks for taking this! I think that this solves one of the main issues we had with this module. We can decide in future PRs if we could further refactor this to have a single client, but I think that we still need to connect to non-leaders to monitor certain things.

Could you also take a look to the dashboard? The consumer lag visualization can be surely simplified with this new field 🙂

jsoriano · 2019-11-28T11:22:12Z

metricbeat/module/kafka/broker.go

@@ -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) {


Nit. Use Fetch for consistency with other methods here.

Suggested change

func (b *Broker) GetPartitionOffsetFromTheLeader(topic string, partitionID int32) (int64, error) {

func (b *Broker) FetchPartitionOffsetFromTheLeader(topic string, partitionID int32) (int64, error) {

jsoriano · 2019-11-28T11:27:45Z

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),


Nit.

Suggested change

"offset": int64(10), "consumer_lag": int64(42) - int64(10),

"offset": int64(10),

"consumer_lag": int64(42) - int64(10),

jsoriano · 2019-11-28T11:29:29Z

metricbeat/module/kafka/broker.go

@@ -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)


I was thinking that we could use this client for everything, but not, we may still need to fetch offsets from non-leader partition replicas for monitoring purpouses.

Though we can also use client.Leader(topic, partitionID) for that.

+1 on revisiting this in a followup PR with refactoring purpose

metricbeat/module/kafka/broker.go

metricbeat/module/kafka/consumergroup/query.go

metricbeat/module/kafka/consumergroup/consumergroup.go

jsoriano · 2019-11-28T12:18:27Z

metricbeat/module/kafka/consumergroup/query.go

@@ -113,12 +114,22 @@ func fetchGroupInfo(

 		for topic, partitions := range ret.off.Blocks {
 			for partition, info := range partitions {
+				partitionOffset, err := getPartitionOffsetFromTheLeader(b, topic, partition)


Something to explore here.

I guess that it may happen that the partition offset here is always going to be ahead of the group offset, because we get first the group offset, and then the partition offset. Between both operations the partition offset may have changed.

Starting on version 4 of ListOffsets (the API method used to get partition offsets), it is possible to indicate a current_leader_epoch to retrieve "old" metadata.

Starting on version 5 of OffsetFetch (the API method used to get consumer group offsets), its response contains a leader_epoch field.

I wonder if we can use the leader_epoch contained in the response of OffsetFetch when available to query for the offset of the partition in the same epoch. This way we could have a more accurate value.

What are the implications for the support matrix? I'm not very familiar with API version vs Kafka versioning.

@jsoriano not sure if this can be achieved with the current implementation of GetOffset

What are the implications for the support matrix? I'm not very familiar with API version vs Kafka versioning.

All messages in kafka protocol are versioned, each client and broker can support a different range of versions for each message. There is a message (ApiVersionsRequest) to query the versions supported by the broker, we could use this method to decide if we can use the methods aware of the epoch.

@jsoriano not sure if this can be achieved with the current implementation of GetOffset

No, we would need to forge our own request as we do to request partition offsets to the leader. Or we could contribute to Sarama the support for these versions. 🙂

metricbeat/module/kafka/broker.go

exekias · 2019-11-28T12:36:13Z

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


probably worth it explaining what this is, as an important metric: the difference between the partition offset and consumer offset

Signed-off-by: chrismark <[email protected]>

ChrsMark · 2019-11-28T13:32:15Z

@jsoriano, @mtojek , @exekias thank you all for reviewing! Almost everything is now addressed.

@jsoriano regarding some special points:

about dashboard I can sure add it but I would propose in a different PR after this one is merged so as to keep PRs more compact ;)
about the general refactoring and the reuse of the new client I would propose to create a refactoring issue to keep track of that needs. I can take care of it by collecting all the issues we discussed here and all related issues we currently have.
about leader_epoch I checked it really quickly and answered inline in your comment. Let me know if I understood it correctly. However I'm full positive in investigating further as part of the refactoring story.

Let me know what you think!

mtojek · 2019-11-28T13:42:13Z

Regarding @ChrsMark 's last comment - this PR is already medium sized one, I'm for pushing refactoring and dashboards to two next independent PRs.

Thanks for addressing comments. LGTM!

jsoriano

Yep, dashboard and further refactors can be left for future changes.

ChrsMark · 2019-11-28T22:00:59Z

Failing tests are irrelevant and already addressed on #14849.
Merging this.

(cherry picked from commit 23aaf5c)

Add consumer_lag in Kafka consumergroup metricset

d9b5704

Signed-off-by: chrismark <[email protected]>

ChrsMark added enhancement module Metricbeat Metricbeat [zube]: In Review v7.6.0 labels Nov 27, 2019

ChrsMark requested a review from a team as a code owner November 27, 2019 14:26

ChrsMark self-assigned this Nov 27, 2019

ChrsMark added 7 commits November 27, 2019 17:44

Fix sub and tests

558a39b

Signed-off-by: chrismark <[email protected]>

Include consumer_lag in unit_test checks

3f8a29e

Signed-off-by: chrismark <[email protected]>

Refactor methods to reuse struct fields

2eec175

Signed-off-by: chrismark <[email protected]>

Merge remote-tracking branch 'upstream/master' into add_consumer_lag

5379558

Close client connection

85a29aa

Signed-off-by: chrismark <[email protected]>

Store cluster client in Broker struct for reusability

0f6a8ef

Signed-off-by: chrismark <[email protected]>

Update data.json

6500bf7

Signed-off-by: chrismark <[email protected]>

jsoriano reviewed Nov 28, 2019

View reviewed changes

mtojek reviewed Nov 28, 2019

View reviewed changes

metricbeat/module/kafka/broker.go Outdated Show resolved Hide resolved

mtojek reviewed Nov 28, 2019

View reviewed changes

metricbeat/module/kafka/consumergroup/query.go Outdated Show resolved Hide resolved

mtojek reviewed Nov 28, 2019

View reviewed changes

metricbeat/module/kafka/consumergroup/consumergroup.go Outdated Show resolved Hide resolved

jsoriano reviewed Nov 28, 2019

View reviewed changes

mtojek reviewed Nov 28, 2019

View reviewed changes

metricbeat/module/kafka/broker.go Outdated Show resolved Hide resolved

exekias reviewed Nov 28, 2019

View reviewed changes

ChrsMark added 2 commits November 28, 2019 15:06

Review fixes

7061004

Signed-off-by: chrismark <[email protected]>

Add changelog

240c16e

Signed-off-by: chrismark <[email protected]>

mtojek approved these changes Nov 28, 2019

View reviewed changes

jsoriano approved these changes Nov 28, 2019

View reviewed changes

ChrsMark merged commit 23aaf5c into elastic:master Nov 28, 2019

ChrsMark added a commit to ChrsMark/beats that referenced this pull request Nov 29, 2019

Add consumer_lag in Kafka consumergroup metricset (elastic#14822)

9226f08

(cherry picked from commit 23aaf5c)

This was referenced Nov 29, 2019

Cherry-pick #14822 to 7.x: Add consumer_lag in Kafka consumergroup metricset #14851

Merged

[Metricbeat] Kafka module refactoring #14852

Closed

Fix cluster-wide Kafka client bug #14857

Merged

ChrsMark added a commit that referenced this pull request Nov 29, 2019

Add consumer_lag in Kafka consumergroup metricset (#14822) (#14851)

311bab6

(cherry picked from commit 23aaf5c)

This was referenced Nov 29, 2019

Cherry-pick #14857 to 7.x: Fix cluster-wide Kafka client bug #14861

Closed

Cherry-pick #14857 to 7.x: Fix cluster-wide Kafka client bug #14862

Merged

Make use of consumer_lag in dashboard #14863

Merged

This was referenced Dec 5, 2019

Cherry-pick #14863 to 7.x: Make use of consumer_lag in dashboard #14948

Merged

Metricbeat - Kafka consumer lag per partition stats #3608

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add consumer_lag in Kafka consumergroup metricset #14822

Add consumer_lag in Kafka consumergroup metricset #14822

ChrsMark commented Nov 27, 2019 •

edited

Loading

jsoriano left a comment

jsoriano Nov 28, 2019

jsoriano Nov 28, 2019

jsoriano Nov 28, 2019

jsoriano Nov 28, 2019

ChrsMark Nov 28, 2019 •

edited

Loading

jsoriano Nov 28, 2019

exekias Nov 28, 2019

ChrsMark Nov 28, 2019 •

edited

Loading

jsoriano Nov 28, 2019

ChrsMark Nov 28, 2019 •

edited

Loading

exekias Nov 28, 2019

ChrsMark commented Nov 28, 2019 •

edited

Loading

mtojek commented Nov 28, 2019

jsoriano left a comment

ChrsMark commented Nov 28, 2019 •

edited

Loading

	func (b *Broker) GetPartitionOffsetFromTheLeader(topic string, partitionID int32) (int64, error) {
	func (b *Broker) FetchPartitionOffsetFromTheLeader(topic string, partitionID int32) (int64, error) {

	"offset": int64(10), "consumer_lag": int64(42) - int64(10),
	"offset": int64(10),
	"consumer_lag": int64(42) - int64(10),

Add consumer_lag in Kafka consumergroup metricset #14822

Add consumer_lag in Kafka consumergroup metricset #14822

Conversation

ChrsMark commented Nov 27, 2019 • edited Loading

jsoriano left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ChrsMark Nov 28, 2019 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ChrsMark Nov 28, 2019 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ChrsMark Nov 28, 2019 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ChrsMark commented Nov 28, 2019 • edited Loading

mtojek commented Nov 28, 2019

jsoriano left a comment

Choose a reason for hiding this comment

ChrsMark commented Nov 28, 2019 • edited Loading

ChrsMark commented Nov 27, 2019 •

edited

Loading

ChrsMark Nov 28, 2019 •

edited

Loading

ChrsMark Nov 28, 2019 •

edited

Loading

ChrsMark Nov 28, 2019 •

edited

Loading

ChrsMark commented Nov 28, 2019 •

edited

Loading

ChrsMark commented Nov 28, 2019 •

edited

Loading