Add Data Streams Monitoring support for Racecar#5962
Open
dasch wants to merge 8 commits into
Open
Conversation
Instrument Racecar consumers and producers for Data Streams Monitoring. Consume checkpoints are set from the existing ActiveSupport notification events: single-message consume extracts the upstream pathway context from the message headers carried in the event payload, while batch consume sets a topic-level checkpoint (the batch payload carries no per-message headers). Both also track the consumed offset for consumer lag. Produce checkpoints are injected by prepending instrumentation onto Racecar::Consumer#produce and the standalone Racecar::Producer, mirroring the ruby-kafka integration.
Move consume-side DSM from the ActiveSupport event handlers to a prepend on Racecar::Runner#process and #process_batch, where the individual rdkafka messages and their headers are available. This lets batch consumption set a checkpoint per message rather than a single topic-level checkpoint, so each message keeps its own upstream pathway context. Pipelines that merge multiple input sources into one processing step (N:M topology) are now represented as distinct fan-in edges in Data Streams Monitoring.
dasch
commented
Jun 29, 2026
Match stringified header keys instead of converting the lookup key to a symbol, so the lookup does not depend on symbol garbage collection.
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: bee4c77ad3
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
Racecar 1.x consumes through Kafka::Consumer#each_message/#each_batch (ruby-kafka) rather than Runner#process, so the Runner prepend records nothing there. Gate the Runner prepend on the rdkafka-based runner and fall back to the kafka integration's consumer instrumentation on older, ruby-kafka-based versions.
vpellan
reviewed
Jul 1, 2026
vpellan
left a comment
vpellan
left a comment
Contributor
There was a problem hiding this comment.
Hey! Thanks for your contribution! I added some comments and will launch the CI, but I will also ask for a review from the DSM team.
This comment has been minimized.
This comment has been minimized.
Replace untyped signatures for the racecar instrumentation with concrete types, and split the DSM specs into consumer_spec.rb and producer_spec.rb to mirror the lib layout.
Contributor
Author
|
Thanks! I got my little helper to do the work – any other improvements? CI should be green, tests are working locally at least. |
ericfirth
approved these changes
Jul 2, 2026
The method returns its headers argument unchanged when DSM is disabled, so the declared return type must be nilable.
vpellan
approved these changes
Jul 3, 2026
Comment on lines
+8
to
+16
| interface _Message | ||
| def topic: () -> ::String | ||
|
|
||
| def partition: () -> ::Integer | ||
|
|
||
| def offset: () -> ::Integer | ||
|
|
||
| def headers: () -> ::Hash[(::String | ::Symbol), ::String]? | ||
| end |
Contributor
There was a problem hiding this comment.
Minor: I think this interface should live in the vendor/rbs folder
Contributor
Author
There was a problem hiding this comment.
Unfortunately I'll be traveling for the next month and can't continue work until I'm back. Just so you don't think it odd that I don't respond 😅
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What does this PR do?
Adds Data Streams Monitoring (DSM) support to the Racecar integration, for both consuming and producing messages.
Consume checkpoints are set by prepending
Racecar::Runner#processand#process_batch, where the individual rdkafka messages (with their headers) are available. Each message gets its own checkpoint, with its upstream pathway context extracted from the message headers. Both single and batch consumption also track the consumed offset for consumer lag.Produce checkpoints are injected into outgoing message headers by prepending
Racecar::Consumer#produceand the standaloneRacecar::Producer(#produce_async/#produce_sync), mirroring the existingruby-kafkaintegration.Motivation:
Racecar consumers and producers previously had tracing but no DSM, so pathways flowing through Racecar apps were invisible in the Data Streams product. This closes that gap, matching the support already shipped for
ruby-kafkaandkarafka/waterdrop.Tradeoffs and decisions:
parent_hash → hashedges, so it natively supports steps that merge multiple input sources and/or fan out to multiple outputs. By checkpointing each consumed message individually — rather than once per batch — a step that consumes a batch drawn from several source topics records a distinct fan-in edge per source, preserving the real topology.Racecar::Runnerrather than the notification events. The individual messages and their headers only exist inside the runner's consume path; theprocess_batch.racecarActiveSupport payload carries aggregate offsets but no per-message headers. Prepending the runner is what makes per-message batch checkpoints possible. (Produce uses a prepend for the same reason — headers must be injected before librdkafka sends the message.)Racecar::Produceronly exists in newer versions, so prepends are guarded bydefined?. All DSM calls are wrapped inDataStreams.enabled?guards and rescue any error at debug level, so DSM can never disrupt message processing.Change log entry
Yes. Add Data Streams Monitoring support for the Racecar integration (consumers and producers), including per-message checkpoints for batch consumption.
Additional Notes:
N/A
How to test the change?
Covered by a new
spec/datadog/tracing/contrib/racecar/data_streams_spec.rb: single and batch consume checkpoints (including per-message fan-in from distinct sources), upstream context extraction with both String and Symbol header keys, nil/missing headers, consumer-lag tracking, resilience when a DSM call raises, produce injection on both the consumer and standalone producer, and DSM-disabled no-ops. Verified viabundle exec rake test:racecaron Ruby 3.2 (racecar 2.8.2 / rdkafka 0.12) and Ruby 3.4 (racecar 2.11 / rdkafka 0.15);standardandsteepboth pass.