Repository navigation
feat(sidecar): track Data Streams Monitoring backlogs in shared memory - #2729
Open
arjunguhaswe wants to merge 10 commits into
Open
arjunguhaswe wants to merge 10 commits into
arjunguhaswe wants to merge 10 commits into
Conversation
Adds a second DSM shared memory table, next to the stats table, holding the highest offset per backlog (a Kafka partition's produce offset, a consumer group's committed one) for each 10 s bucket. PHP workers cannot keep that maximum across requests themselves. - libdd-ipc: the `DataStreamsBacklogs` schema. Backlog tags have fixed, append-only positions (`BacklogTag`, in the Java tracer's order); a slot stores each present tag's value and a presence mask, and keys are compared byte for byte so that collisions cannot merge backlogs. - datadog-sidecar: both tables are created, drained (at one instant) and retired together; backlogs go out in the `Backlogs` field of the bucket for their interval, omitted when empty as in the Java tracer. An `add_dsm_backlog` IPC call is the fallback until a worker has the table open. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
…rjun.guha/dsm2-450-php-tracer-complete-the-kafka-integration Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
…s table key Merging arjun.guha/dsm-shm-table brought two changes the backlog code predates: - Schema is no longer an unsafe trait: its Key and Value must be ShmSafe instead (#2673). ShmBacklogKey and ShmBacklogValue now implement ShmSafe, and DataStreamsBacklogs implements Schema without unsafe, as DataStreams does. - DsmTableKey has no product any more (#2674), so add_dsm_backlog builds its key without one. Also name ShmDsmBacklogTable::add_backlog in code spans. It aliases the shared table type of the private shm_table module, so links to its methods fail the doc lint, as #2674 found for ShmDsmTable. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
…tracer-complete-the-kafka-integration Carry the stats-to-checkpoints rename over to the backlog table and the flusher that drains both: Drained::stats -> Drained::checkpoints, and the wording that still called the checkpoint table the stats table. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Cut the default backlogs per bucket from 1024 to 128, and size the string pool from the slot count to keep 256 bytes of tag values per backlog (32 KiB per bucket). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
…tracer-complete-the-kafka-integration Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2 tasks
Contributor
🔒 Cargo Deny Results📦
|
Contributor
📚 Documentation Check Results📦
|
BenchmarksComparisonBenchmark execution time: 2026-10-27 16:27:04 Comparing candidate commit d47ba93 in PR branch Found 1 performance improvements and 2 performance regressions! Performance is the same for 154 metrics, 2 unstable metrics.
|
Contributor
Artifact Size Benchmark Reportaarch64-alpine-linux-musl
aarch64-unknown-linux-gnu
libdatadog-x64-windows
libdatadog-x86-windows
x86_64-alpine-linux-musl
x86_64-unknown-linux-gnu
|
arjunguhaswe
marked this pull request as ready for review
October 26, 2026 18:01
Codex Review SummaryThis comment shows the latest Codex review activity on this pull request.
ℹ️ About Codex in GitHubYour team has set up Codex to review pull requests in this repo. Reviews are triggered when you
Codex reacts with 👀 while any review is running, comments if it has suggestions, and reacts with 👍 once all reviews finish with no findings. |
![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: 52c0f49740
ℹ️ 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".
| DEFAULT_DSM_CHECKPOINT_SLOT_COUNT, | ||
| DEFAULT_DSM_CHECKPOINT_STRING_POOL_BYTES, | ||
| ) | ||
| .inspect_err(|_| backlogs.unlink())?; |
There was a problem hiding this comment.
If checkpoint-table creation fails after another worker has opened the newly published backlog segment, unlink() removes only its name and leaves that worker's existing mapping marked ready. The worker therefore keeps accepting backlog writes as successful instead of falling back to IPC, while no sidecar state owns or drains the mapping, silently losing those offsets. Signal reload on backlogs before unlinking it on this error path.
Useful? React with 👍 / 👎.
…rjun.guha/dsm2-450-php-tracer-complete-the-kafka-integration
pool_string now returns None for a reference that does not resolve or does not fit the flush budget. A backlog with such a value is dropped rather than exported with that value blank, which would merge it with, or pass it off as, another backlog. A present tag with an empty value is still reported. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
🎉 All green!🧪 All tests passed 🎯 Code Coverage (details) 🔗 Commit SHA: d47ba93 | Docs | View more details | Give us feedback! |
…rjun.guha/dsm2-450-php-tracer-complete-the-kafka-integration
…rjun.guha/dsm2-450-php-tracer-complete-the-kafka-integration
yannham
approved these changes
Oct 28, 2026
yannham
left a comment
yannham
left a comment
Contributor
There was a problem hiding this comment.
Some questions, nothing blocking 👍
Comment on lines
+31
to
+35
| /// Positions are durable. A new tag is appended at the end, after the last one; none is ever | ||
| /// reordered, removed or reused, so adding one leaves every existing backlog's `Tags` as they | ||
| /// were. (Appending one changes the slot, so it also bumps [`DataStreamsBacklogs`]'s version.) | ||
| /// The order of the first five is the Java tracer's (`DataStreamsTags`), so that a backlog is | ||
| /// reported with the same `Tags` whichever tracer sent it. |
Contributor
There was a problem hiding this comment.
Nit: I'm not sure to understand this part at this point. BacklogTag is a simple enum, but we're talking about positions and re-ordering. Should this move to the code that actually append or do stuff with tags?
| Partition = 4, | ||
| } | ||
|
|
||
| impl BacklogTag { |
Contributor
There was a problem hiding this comment.
We tend to use strum_macros::EnumIter and AsRefStr to do that automatically instead of defining ALL etc. manually
| h.write_u8(input.present()); | ||
| for value in input.tags.iter().flatten() { | ||
| value.hash(&mut h); | ||
| } |
Contributor
There was a problem hiding this comment.
Same as in the previous PR: is this different from the Hash implementation of DsmBacklogInput?
| /// | ||
| /// `None` when the slot's key is not one this writer could have produced; the offset is | ||
| /// cleared either way, so that it is not left for whichever key next occupies the slot. | ||
| fn drain_entry(entry: &ShmEntry, pool: &Pool, budget: &mut usize) -> Option<DsmBacklog> { |
Contributor
There was a problem hiding this comment.
My clanker relates that there might be a race with a concurrent writer:
- Writer find a slot with a matching hash. Has loaded it and checked for equality
- Drainer comes, nullify
value, and storesSLOT_EMPTYin the hash - A second writer comes, reclaim the empty slot, write the key and publish the hash
- The first writer resumes and stores an unrelated value in the slot
To be honest the whole shmem structure is kinda complex, so I don't have a precise enough mental model to know if 1. that's actually possible and 2. we should care, but reporting just in case.
| drained: &Drained, | ||
| ) -> anyhow::Result<Vec<u8>> { | ||
| // Both tables' buckets for an interval go out as one, ordered by start. | ||
| let mut buckets: BTreeMap<(u64, u64), StatsBucket<'_>> = BTreeMap::new(); |
Contributor
There was a problem hiding this comment.
Not very important, but is there a reason to use a BTreeMap here over a hashmap?
This branch has not been deployed
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
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?
Stacked on #2674. Split back out of it (previously #2709).
libdd-ipc: the
DataStreamsBacklogsschema (shm_dsm_backlog.rs)BacklogTagit carries:Topic,Type,KafkaClusterId,ConsumerGroup,Partition. Positions are append-only and in Java's order (DataStreamsTags), soTagsmatch Java's. Tags can be given in any order. An unknown key drops the offset.fetch_max, as in Java. Negative offsets, librdkafka's "unknown" markers, are dropped.datadog-sidecar (
data_streams_flusher.rs)DsmTableKey, at/dddsb…(backlogs), next to/dddsm…(stats).add_dsm_backlogcall.nowso both date a bucket the same way, and are merged by(start, duration)into oneStatsBucket. They're retired together once both drains come back empty.Backlogs: [{Tags, Value}], omitted when empty.Motivation
Kafka offsets go into a second table the same way, for consumer lag (DSM2-450).
Additional Notes
kafka_high_watermarkbacklog type, which Go reports and Java doesn't.How to test the change?
The DSM tests cover:
Backlogsfield layout, and the IPC fallbackManual Testing:
Many test apps were created, all functionally managed to use the tables to produce correct and consistent data, screenshots from one such test app have been provided below:
Lag Metrics
🤖 Generated with Claude Code