[SVLS-7168] Create inferred Span and Span links for GCP PubSub - #6415
Conversation
Overall package sizeSelf size: 13.67 MB Dependency sizes| name | version | self size | total size | |------|---------|-----------|------------| | @datadog/libdatadog | 0.7.0 | 35.02 MB | 35.02 MB | | @datadog/native-appsec | 10.3.0 | 20.73 MB | 20.74 MB | | @datadog/pprof | 5.12.0 | 11.19 MB | 11.57 MB | | @datadog/native-iast-taint-tracking | 4.1.0 | 9.01 MB | 9.02 MB | | @opentelemetry/resources | 1.30.1 | 557.67 kB | 7.71 MB | | @opentelemetry/core | 1.30.1 | 908.66 kB | 7.16 MB | | protobufjs | 7.5.4 | 2.95 MB | 5.83 MB | | @datadog/wasm-js-rewriter | 5.0.1 | 2.82 MB | 3.53 MB | | @datadog/native-metrics | 3.1.1 | 1.02 MB | 1.43 MB | | @opentelemetry/api-logs | 0.208.0 | 199.48 kB | 1.42 MB | | @opentelemetry/api | 1.9.0 | 1.22 MB | 1.22 MB | | jsonpath-plus | 10.3.0 | 617.18 kB | 1.08 MB | | import-in-the-middle | 1.15.0 | 127.66 kB | 856.24 kB | | lru-cache | 10.4.3 | 804.3 kB | 804.3 kB | | @datadog/openfeature-node-server | 0.2.0 | 118.51 kB | 437.19 kB | | opentracing | 0.14.7 | 194.81 kB | 194.81 kB | | source-map | 0.7.6 | 185.63 kB | 185.63 kB | | pprof-format | 2.2.1 | 163.06 kB | 163.06 kB | | @datadog/sketches-js | 2.1.1 | 109.9 kB | 109.9 kB | | @isaacs/ttlcache | 2.1.3 | 90.79 kB | 90.79 kB | | lodash.sortby | 4.7.0 | 75.76 kB | 75.76 kB | | ignore | 7.0.5 | 63.38 kB | 63.38 kB | | istanbul-lib-coverage | 3.2.2 | 34.37 kB | 34.37 kB | | rfdc | 1.4.1 | 27.15 kB | 27.15 kB | | dc-polyfill | 0.1.10 | 26.73 kB | 26.73 kB | | tlhunter-sorted-set | 0.1.0 | 24.94 kB | 24.94 kB | | shell-quote | 1.8.3 | 23.74 kB | 23.74 kB | | limiter | 1.1.5 | 23.17 kB | 23.17 kB | | retry | 0.13.1 | 18.85 kB | 18.85 kB | | semifies | 1.0.0 | 15.84 kB | 15.84 kB | | jest-docblock | 29.7.0 | 8.99 kB | 12.76 kB | | crypto-randomuuid | 1.0.0 | 11.18 kB | 11.18 kB | | ttl-set | 1.0.0 | 4.61 kB | 9.69 kB | | mutexify | 1.4.0 | 5.71 kB | 8.74 kB | | path-to-regexp | 0.1.12 | 6.6 kB | 6.6 kB | | module-details-from-path | 1.0.4 | 3.96 kB | 3.96 kB | | escape-string-regexp | 5.0.0 | 3.66 kB | 3.66 kB |🤖 This report was automatically generated by heaviest-objects-in-the-universe |
BenchmarksBenchmark execution time: 2025-12-12 06:20:53 Comparing candidate commit 805f38e in PR branch Found 0 performance improvements and 1 performance regressions! Performance is the same for 291 metrics, 28 unstable metrics. scenario:plugin-http-server-querystring-obfuscation-24
|
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## nina.rei/SVLS-7168/gcp-pubsub-batch-plugin #6415 +/- ##
==============================================================================
- Coverage 84.14% 83.98% -0.17%
==============================================================================
Files 517 516 -1
Lines 22264 22376 +112
==============================================================================
+ Hits 18735 18793 +58
- Misses 3529 3583 +54 ☔ View full report in Codecov by Sentry. 🚀 New features to boost your workflow:
|
3efb6be to
128343d
Compare
|
d7f0d5e to
3fdb180
Compare
128343d to
ee0b34e
Compare
3fdb180 to
6393626
Compare
6ec410a to
c8815fe
Compare
d76ed3b to
f638388
Compare
c8815fe to
929f281
Compare
9b2c547 to
4ee3b98
Compare
f638388 to
8e3014c
Compare
- Collect span links from messages 2-N (first becomes parent) - Extract parent context from first message trace context - Create pubsub.request span with span links metadata - Inject batch metadata into all messages (_dd.pubsub_request.*, _dd.batch.*) - Add 128-bit trace ID support (_dd.p.tid) - Add operation tag for batched vs single requests
- Add ack context map to preserve trace context across batched acknowledges - Update producer to use batchSpan._startTime for accurate publish time - Add explicit parent span support in client plugin - Wrap Message.ack() to store context before batched gRPC acknowledge - Update Subscription.emit to properly handle storage context - Sync auto-load improvements from Branch 1
8e3014c to
10eb97c
Compare
1ad601a to
601a630
Compare
nina9753
changed the title
…i/SVLS-7168/gcp-pubsub-batch-plugin
…i/SVLS-7168/gcp-pubsub-batch-plugin
![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.
ℹ️ About Codex in GitHub
Your team has set up Codex to 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 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
ccc9ae5 to
f7af73c
Compare
| const receiveFinishCh = channel('apm:google-cloud-pubsub:receive:finish') | ||
| const receiveErrorCh = channel('apm:google-cloud-pubsub:receive:error') | ||
|
|
||
| const ackContextMap = new Map() |
There was a problem hiding this comment.
Maps like these which correlate to user-operations (such as HTTP requests) are scary because they can cause memory leaks whenever we don't clean up properly.
For that reason we try to use a WeakMap instead. Those hold a "weak reference" to the key and cause the value to be deleted automatically when the key is garbage collected.
However it only works when the key is an object and your Map is using a string as a key. We could theoretically do something where the request is the key but I don't think that is available later when the delete happens?
Can you explain more about the lifecycle of these ack IDs? Maybe we can figure out something that way.
There was a problem hiding this comment.
The lifecycle has a critical gap that prevents WeakMap usage:
// 1. MESSAGE RECEIVED: User gets a Message object
subscription.on('message', (message) => {
// message.ackId = "abc123" (a string property)
// 2. USER ACKS: We intercept message.ack() and store context
message.ack()
// → ackContextMap.set("abc123", context) // Key is STRING, not object
// 3. MESSAGE OBJECT GOES OUT OF SCOPE: User code finishes
// The Message object may be GC'd here
// 4. BATCHED ACKNOWLEDGE CALL: SDK batches acks into gRPC call (seconds later)
pubsubClient.acknowledge({ ackIds: ["abc123", "def456", "ghi789"] })
// → We only have STRING ackIds, NOT the original Message objects
// → Need to look up context by string: ackContextMap.get("abc123")
})
Between steps 2 and 4, the Message object is gone, but we need to keep the context alive and retrieve it using only the string ackId. The acknowledge gRPC call creates a single pubsub.request span for the batch, and we need to ensure that span runs in the right async context. Even though N messages were acked, there's only 1 acknowledge API call. So, because the Message Object is not available during retrieval, we must use a Map. To prevent memory leaks, I added a TTL to handle manual cleanup, a graceful shutdown, and cases where acknowledgment never happens (network failure, crash, etc.).
|
|
||
| // Wrap Topic.publish (legacy API) | ||
| if (obj.Topic.prototype.publish) { | ||
| if (typeof obj.Topic.prototype.publish === 'function') { |
There was a problem hiding this comment.
Did you find that sometimes obj.Topic.prototype.publish was a truthy value but wasn't a function?
There was a problem hiding this comment.
No, I have not found evidence that obj.Topic.prototype.publish is ever a truthy non-function value in practice.
The typeof obj.Topic.prototype.publish === 'function' check is following the pattern that's used throughout the dd-trace-js codebase for instrumentation you can see other examples in the same file that i did not add myself
| const span = this.startSpan({ | ||
| childOf, | ||
| resource: topic, | ||
| resource: `Message from ${topicName}`, |
There was a problem hiding this comment.
Like in the previous PR this one also looks like a breaking change.
There was a problem hiding this comment.
There shouldn't be any users for this currently, but I can always revert to what is in prod. I wanted to make the spans more descriptive for the blog post and product release.
…168/gcp-pubsub-synthetic-span
| }, 60_000) // Run cleanup every 60 seconds | ||
|
|
||
| // Allow process to exit cleanly | ||
| if (ackContextCleanupInterval.unref) { | ||
| ackContextCleanupInterval.unref() | ||
| } |
There was a problem hiding this comment.
.unref() will always be present (and it returns the timer instance which is nice for method chaining). But I'm pretty concerned about using a timer like this for cleanup since these types of solutions can often cause memory leaks. E.g. with a 60 second timer there could be a saw tooth memory increase with a 60 second period.
| }, 60_000) // Run cleanup every 60 seconds | |
| // Allow process to exit cleanly | |
| if (ackContextCleanupInterval.unref) { | |
| ackContextCleanupInterval.unref() | |
| } | |
| }, 60_000).unref() // Run cleanup every 60 seconds |
…ei/SVLS-7168/gcp-pubsub-synthetic-span
) * feat: add producer-side batch message handling with span linking - Collect span links from messages 2-N (first becomes parent) - Extract parent context from first message trace context - Create pubsub.request span with span links metadata - Inject batch metadata into all messages (_dd.pubsub_request.*, _dd.batch.*) - Add 128-bit trace ID support (_dd.p.tid) - Add operation tag for batched vs single requests * feat: add span linking from delivery span to pubsub.request - Add _reconstructPubSubContext to extract pubsub.request span ID from headers - Add span link to original trace context if different from pubsub.request - Supports same-trace parenting for better trace continuity * feat: check for pubsub.delivery span in AsyncLocalStorage before extracting from headers - HTTP plugin now checks if a delivery span is active in storage - If found, uses delivery span as parent for http.request - Ensures proper span hierarchy for push subscriptions * feat: add span linking and batch metadata to pull-based consumer - Extract pubsub.request span ID from message attributes - Add span link correlation tags - Calculate delivery duration from publish start time - Add batch size and index tags for batched messages * remove comments * new test file * Fix push subscription test module resolution for CI * Implement full HTTP+Pub/Sub integration tests * run linter * test other frameworks * Fix push subscription tests
87ecea4
into
nina.rei/SVLS-7168/gcp-pubsub-batch-plugin
4 participants
What does this PR do?
This PR completes the distributed tracing story by adding bidirectional span linking between producer and consumer spans. It enables full trace visualization across the entire message lifecycle: producer → Pub/Sub topic → consumer. The implementation includes intelligent span linking that adapts based on whether messages originated from a single trace or multiple traces.
Consumer Span Linking (Pull Subscriptions)
Enhanced the consumer plugin to create span links back to the producer's batch span. When a consumer receives a message, it reconstructs the producer's batch span context from the metadata injected by the producer plugin. This context becomes a span link, creating a visible connection in the APM UI between the consumer operation and the batch that produced it.
Push Subscription Advanced Linking
Enhanced the push subscription plugin with conditional span linking based on trace ID comparison.
This also follows the industry standard OpenTelemetry concept, representing causal relationships between spans that aren't parent-child.
Web (framework) Plugin IntegrationUpdated the web framework span creation logic to check AsyncLocalStorage for an active pubsub.delivery span before extracting context from HTTP headers. This ensures web framework spans (Express, Fastify, etc.) correctly parented to the delivery span when handling push subscription requests.Enhanced Batch Metadata
All consumer spans (both pull and push) now include rich batch metadata:
Followup PR with tests can be found here #6414
Motivation
With the previous PRs we made it so we get proper trace context propagation from producer to consumer, but the relationship between the producer's batch span and each individual consumer span wasn't visible in the APM UI. This PR adds span links that let you:
Plugin Checklist
Additional Notes
Follow-up PR to #6260
Additional information can be found in this doc