DSM Implementation for Ruby with Karafka instrumented

[ericfirth]

What does this PR do?
This implements Data Streams Monitoring in Ruby. It autoinstruments 2 libraries:

• ruby-kafka
• karafka (though only the consumer, as waterdrop, the producer library is not instrumented)

Data Streams Monitoring works by setting a checkpoint with topic (i.e. queue), direction (i.e. produce/consume = in/out) and service (plus a few other tags). Produces then set the context of the checkpoint in the message (usually via the headers, but for SQS and non-header libraries, it will be in the message body) and consumers take the message context and set a checkpoint with a link to the previous checkpoint. Thus, we can give latency stats (i.e. non-sampled) through queues and long pipelines.

Motivation:
There are have been many requests for DSM in Ruby over the years and there is an issue on the repo to add it.

Change log entry
Yes. DataStreams: Adds Data Streams Monitoring capability, instrumenting ruby-kafka and karafka consumers with manual instrumentation available

Additional Notes:

How to test the change?
The following branch is being tested on staging with our test apps. You can see them here

They are defined here:
https://github.com/DataDog/data-streams-dev/pull/252

[datadog-official]

✅ Tests

🎉 All green!

❄️ No new flaky tests detected
🧪 All tests passed

🎯 Code Coverage
• Patch Coverage: 97.48%
• Total Coverage: 98.58% (+0.03%)

View detailed report

_{This comment will be updated automatically if new data arrives.
🔗 Commit SHA: 728b9c7 | Docs | Datadog PR Page | Was this helpful? Give us feedback!}

[github-actions]

Typing analysis

Ignored files

This PR introduces 1 ignored file. It increases the percentage of typed files from 35.94% to 36.87% (+0.93%).

Ignored files (+1-0)

❌ Introduced:

lib/datadog/data_streams/configuration/settings.rb

Note: Ignored files are excluded from the next sections.

Untyped methods

This PR introduces 7 untyped methods and 8 partially typed methods. It increases the percentage of typed methods from 51.66% to 53.09% (+1.43%).

Untyped methods (+7-0)

❌ Introduced:

sig/datadog/data_streams.rbs:19
└── def self.components: () -> untyped
sig/datadog/tracing/contrib/kafka/instrumentation/consumer.rbs:7
└── def self.included: (untyped base) -> untyped
sig/datadog/tracing/contrib/kafka/instrumentation/consumer.rbs:9
└── def each_message: (**untyped kwargs) { (untyped) -> untyped } -> untyped
sig/datadog/tracing/contrib/kafka/instrumentation/consumer.rbs:10
└── def each_batch: (**untyped kwargs) { (untyped) -> untyped } -> untyped
sig/datadog/tracing/contrib/kafka/instrumentation/producer.rbs:7
└── def self.included: (untyped base) -> untyped
sig/datadog/tracing/contrib/kafka/instrumentation/producer.rbs:9
└── def deliver_messages: (**untyped kwargs) -> untyped
sig/datadog/tracing/contrib/kafka/instrumentation/producer.rbs:10
└── def send_messages: (untyped messages, **untyped kwargs) -> untyped

Partially typed methods (+8-0)

❌ Introduced:

sig/datadog/data_streams/processor.rbs:52
└── def process_kafka_produce_event: (::Hash[::Symbol, untyped] event) -> void
sig/datadog/data_streams/processor.rbs:53
└── def process_kafka_consume_event: (::Hash[::Symbol, untyped] event) -> void
sig/datadog/data_streams/processor.rbs:54
└── def process_checkpoint_event: (::Hash[::Symbol, untyped] event) -> void
sig/datadog/data_streams/processor.rbs:67
└── def send_stats_to_agent: (::Hash[::String, untyped] payload) -> void
sig/datadog/data_streams/processor.rbs:68
└── def send_dsm_payload: (::String data, ::Hash[::String, ::String] headers) -> untyped
sig/datadog/data_streams/processor.rbs:71
└── def serialize_buckets: () -> ::Array[::Hash[::String, untyped]]
sig/datadog/data_streams/processor.rbs:72
└── def serialize_consumer_backlogs: () -> ::Array[::Hash[::String, untyped]]
sig/datadog/data_streams/transport/stats.rbs:36
└── def send_stats: (::Hash[::String, untyped] payload) -> (Core::Transport::HTTP::Response | Core::Transport::InternalErrorResponse)

If you believe a method or an attribute is rightfully untyped or partially typed, you can add # untyped:accept to the end of the line to remove it from the stats.

[pr-commenter]

Benchmarks

Benchmark execution time: 2025-11-03 18:46:49

Comparing candidate commit 2ae6bf8 in PR branch eric.firth/dsm-ruby with baseline commit 0808070 in branch master.

Found 2 performance improvements and 1 performance regressions! Performance is the same for 41 metrics, 2 unstable metrics.

scenario:profiling - Allocations (baseline)

• 🟩 throughput [+272517.572op/s; +286292.951op/s] or [+5.450%; +5.726%]

scenario:tracing - Propagation - Datadog

• 🟩 throughput [+2702.065op/s; +2773.593op/s] or [+9.407%; +9.656%]

scenario:tracing - Propagation - Trace Context

• 🟥 throughput [-3545.328op/s; -3447.408op/s] or [-9.502%; -9.240%]

[github-actions]

Thank you for updating Change log entry section 👏

^{Visited at: 2025-10-23 12:59:15 UTC}

[ericfirth]

Removed a ton of worthless comments and de-namespaced it from tracing. It is ready for re-review @y9v

[vpellan]

LGTM for tracing part! (non blocker comment)

[p-datadog]

I have the follow-up PR to improve ddsketch loading: #5008. I think it should stay as a separate PR because I would like the guild to review it since it is a bit of a conceptual change in how loading/referencing of components from libdatadog is accomplished.

This present PR does not add any circular dependencies as far as I can tell, I think it's OK to be merged in its present state.