contrib/segmentio/kafka.go.v0: add DSM support #2625

adrien-f · 2024-03-21T14:22:04Z

What does this PR do?

Add Data Streams Monitoring support to segmentio/kafka integration

Motivation

We currently use segmentio/kafka for our Go projects and it was easier to update the integration than change library.

Reviewer's Checklist

Changed code has unit tests for its functionality at or near 100% coverage.
System-Tests covering this feature have been added and enabled with the va.b.c-dev version tag.
There is a benchmark for any new code, or changes to existing code.
If this interacts with the agent in a new way, a system test has been added.
Add an appropriate team label so this PR gets put in the right place for the release notes.
Non-trivial go.mod changes, e.g. adding new modules, are reviewed by @DataDog/dd-trace-go-guild.

For Datadog employees:

If this PR touches code that handles credentials of any kind, such as Datadog API keys, I've requested a review from @DataDog/security-design-and-guidance.
This PR doesn't touch any of that.

Unsure? Have a question? Request a review!

adrien-f · 2024-03-21T14:44:28Z

contrib/segmentio/kafka.go.v0/kafka.go

+	datastreams.InjectToBase64Carrier(ctx, carrier)
+}
+
+func getProducerMsgSize(msg *kafka.Message) (size int64) {


There is an alternative private implementation here:

https://github.com/segmentio/kafka-go/blob/ebca72eaee918d303c532feb7ff29afdcd8c2efa/message.go#L48-L50

I've kept with the one in the other integrations.

darccio · 2024-03-21T14:53:36Z

Hi @adrien-f, thanks for contributing this feature. I approved running the CI pipeline and there are some things to fix:

It seems to fail in kafka_go.go:210 with this error.
A little nitpick here: contrib/segmentio/kafka.go.v0/kafka.go:61:2: var-naming: struct field groupId should be groupID (revive)

adrien-f · 2024-03-21T16:03:06Z

Many thanks @darccio !

I've pushed the groupID change 👍

Regarding the issue reported in the CI, it appears to be caused by the -race flag, I will investigate what's up with that.

adrien-f · 2024-03-22T13:48:00Z

I believe I corrected the issue 👍

Happy weekend,

darccio · 2024-03-25T15:49:05Z

contrib/segmentio/kafka.go.v0/kafka.go

@@ -46,13 +48,17 @@ func WrapReader(c *kafka.Reader, opts ...Option) *Reader {
 	if c.Config().Brokers != nil {
 		wrapped.bootstrapServers = strings.Join(c.Config().Brokers, ",")
 	}
+
+	wrapped.groupID = c.Config().GroupID


@adrien-f Comparing with Shopify/sarama contrib, groupID comes from WithGroupID instead. I think it'll be better to be coherent with the other contrib, and avoid changing the default behaviour (working without GroupID).

Do you mind adding the option and removing this line?

For sure, I'm also looking at the confluent-kafka-go integration and they have a WithConfig:

dd-trace-go/contrib/confluentinc/confluent-kafka-go/kafka/option.go

Lines 114 to 116 in 4904f57

if groupID, err := cg.Get("group.id", ""); err == nil {

cfg.groupID = groupID.(string)

}

Which is called through this:

dd-trace-go/contrib/confluentinc/confluent-kafka-go/kafka/kafka.go

Lines 43 to 44 in 4904f57

opts = append(opts, WithConfig(conf))

return WrapConsumer(c, opts...), nil

And I like this approach of enhancing the constructor transparently as users adopting DSM will only need to update and turn on the environment variable instead of having to explicitly pass the groupID.

This might be too much of a broad change for this PR though, what do you think?

@adrien-f If you want, I can contribute the change. It won't take me too much time. I'd rather avoid introducing a subtle breaking change.

No worries, I just wanted to be sure :) I've pushed the change!

@adrien-f I don't see any new commit.

@darccio @adrien-f the reason sarama has this group ID option is because the consumer group is not available otherwise (the usage of consumer groups is through a different api that we currently don't support, see: #2133).

In general, the implementation that we should be looking for reference is the confluent-kafka one: https://github.com/DataDog/dd-trace-go/blob/main/contrib/confluentinc/confluent-kafka-go/kafka.v2/option.go#L114-L116 -> the groupID is fetched from the kafka config actually used by the consumer.

darccio

LGTM

adrien-f · 2024-03-27T08:41:49Z

LGTM

Thanks for the review !

darccio · 2024-03-27T08:47:39Z

Thanks for the review !

You are welcome. I'll get a second review from Ecosystems, @rarguelloF, but it won't be done until next week (Holy week is close, so some Datadogs are already OOO).

rarguelloF · 2024-04-01T12:47:39Z

contrib/segmentio/kafka.go.v0/option.go

+// WithGroupID tags the produced data streams metrics with the given groupID (aka consumer group)
+func WithGroupID(groupID string) Option {
+	return func(cfg *config) {
+		cfg.groupID = groupID
+	}
+}


I think this should be fetched from the actual consumer config right?

You can access it in the WrapReader function using kafka.Reader.Config().GroupID.

Hey :) This is what I had at first before changing it following #2625 (comment)

How would you see it?

I replied here: #2625 (comment)

Sorry for the confusion! @adrien-f but TLDR; the reason we have that option in Sarama is because we don't have a way to access the groupID there, but in this contrib and the confluent kafka one we do have the option to access it from the config, so this would make this WithGroupID option unnecessary.

Thanks for the explanation! I've restored this behavior 👍

darccio · 2024-04-04T08:41:04Z

@adrien-f Please, merge main in your branch. This will trigger the CI and the automerging that I just activated. Thanks!

adrien-f · 2024-04-04T17:35:26Z

@darccio I used the rebase button/feature but looks like it removed the auto-merge, sorry about that 🙏

Signed-off-by: Adrien Fillon <[email protected]>

darccio · 2024-04-05T12:41:47Z

/merge

dd-devflow · 2024-04-05T12:41:54Z

🚂 MergeQueue

This merge request is not mergeable yet, because of pending checks/missing approvals. It will be added to the queue as soon as checks pass and/or get approvals.
Note: if you pushed new commits since the last approval, you may need additional approval.
You can remove it from the waiting list with /remove command.

Use /merge -c to cancel this operation!

dd-devflow · 2024-04-05T12:50:05Z

🚂 MergeQueue

Pull request added to the queue.

This build is going to start soon! (estimated merge in less than 0s)

Use /merge -c to cancel this operation!

darccio · 2024-04-05T12:51:28Z

/remove

dd-devflow · 2024-04-05T12:51:33Z

🚂 Devflow: /remove

dd-devflow · 2024-04-05T12:51:35Z

⚠️ MergeQueue

This merge request build was cancelled

If you need support, contact us on Slack #devflow!

adrien-f requested a review from a team as a code owner March 21, 2024 14:22

darccio added apm:ecosystem contrib/* related feature requests or bugs needs-triage New issues that have not yet been triaged labels Mar 21, 2024

adrien-f commented Mar 21, 2024

View reviewed changes

adrien-f force-pushed the segment-dsm branch from a2600a1 to 7e49cca Compare March 21, 2024 16:00

adrien-f force-pushed the segment-dsm branch from 7e49cca to 7a9c514 Compare March 22, 2024 13:32

darccio removed the needs-triage New issues that have not yet been triaged label Mar 22, 2024

darccio requested changes Mar 25, 2024

View reviewed changes

adrien-f force-pushed the segment-dsm branch from 7a9c514 to a547875 Compare March 26, 2024 19:39

darccio approved these changes Mar 27, 2024

View reviewed changes

darccio requested a review from rarguelloF March 27, 2024 08:26

rarguelloF reviewed Apr 1, 2024

View reviewed changes

adrien-f force-pushed the segment-dsm branch from a547875 to 1b7510b Compare April 2, 2024 14:32

piochelepiotr approved these changes Apr 3, 2024

View reviewed changes

rarguelloF approved these changes Apr 3, 2024

View reviewed changes

darccio enabled auto-merge (squash) April 4, 2024 08:40

auto-merge was automatically disabled April 4, 2024 17:33
Head branch was pushed to by a user without write access

adrien-f force-pushed the segment-dsm branch from 1b7510b to e6759cf Compare April 4, 2024 17:33

contrib/segmentio/kafka.go.v0: add DSM support

bb7d7a1

Signed-off-by: Adrien Fillon <[email protected]>

adrien-f force-pushed the segment-dsm branch from e6759cf to bb7d7a1 Compare April 5, 2024 12:40

dd-devflow bot added the mergequeue-status: waiting label Apr 5, 2024

dd-devflow bot added mergequeue-status: queued and removed mergequeue-status: waiting labels Apr 5, 2024

dd-devflow bot added mergequeue-status: in_progress and removed mergequeue-status: queued labels Apr 5, 2024

dd-devflow bot removed the mergequeue-status: in_progress label Apr 5, 2024

dd-devflow bot added the mergequeue-status: removed label Apr 5, 2024

darccio merged commit 531cae3 into DataDog:main Apr 5, 2024
98 of 101 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

contrib/segmentio/kafka.go.v0: add DSM support #2625

contrib/segmentio/kafka.go.v0: add DSM support #2625

adrien-f commented Mar 21, 2024

adrien-f Mar 21, 2024

darccio commented Mar 21, 2024

adrien-f commented Mar 21, 2024

adrien-f commented Mar 22, 2024

darccio Mar 25, 2024

adrien-f Mar 26, 2024

darccio Mar 26, 2024

adrien-f Mar 26, 2024

darccio Mar 26, 2024

rarguelloF Apr 2, 2024

darccio left a comment

adrien-f commented Mar 27, 2024

darccio commented Mar 27, 2024

rarguelloF Apr 1, 2024

adrien-f Apr 2, 2024 •

edited

Loading

rarguelloF Apr 2, 2024

adrien-f Apr 2, 2024

darccio commented Apr 4, 2024

adrien-f commented Apr 4, 2024

darccio commented Apr 5, 2024

dd-devflow bot commented Apr 5, 2024

dd-devflow bot commented Apr 5, 2024

darccio commented Apr 5, 2024

dd-devflow bot commented Apr 5, 2024

dd-devflow bot commented Apr 5, 2024

	if groupID, err := cg.Get("group.id", ""); err == nil {
	cfg.groupID = groupID.(string)
	}

	opts = append(opts, WithConfig(conf))
	return WrapConsumer(c, opts...), nil

contrib/segmentio/kafka.go.v0: add DSM support #2625

contrib/segmentio/kafka.go.v0: add DSM support #2625

Conversation

adrien-f commented Mar 21, 2024

What does this PR do?

Motivation

Reviewer's Checklist

Choose a reason for hiding this comment

darccio commented Mar 21, 2024

adrien-f commented Mar 21, 2024

adrien-f commented Mar 22, 2024

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

Choose a reason for hiding this comment

darccio left a comment

Choose a reason for hiding this comment

adrien-f commented Mar 27, 2024

darccio commented Mar 27, 2024

Choose a reason for hiding this comment

adrien-f Apr 2, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

darccio commented Apr 4, 2024

adrien-f commented Apr 4, 2024

darccio commented Apr 5, 2024

dd-devflow bot commented Apr 5, 2024

dd-devflow bot commented Apr 5, 2024

darccio commented Apr 5, 2024

dd-devflow bot commented Apr 5, 2024

dd-devflow bot commented Apr 5, 2024

adrien-f Apr 2, 2024 •

edited

Loading