Defer OTEP-4947 endpoint until routing tags land; suppress metadata on unsupported runtimes

Addresses two P2 review points on the OTEP-4947 branch.

Endpoint refresh (#2). HTTP server plugins set 'span.type=web' and
'http.method' when a request arrives and only add 'http.route' /
'resource.name' later, once framework routing has resolved the URL. The
writer used to treat any non-undefined webTags result as a fully formed
answer at build time — 'needsEndpoint' would flip to false immediately —
so the OTEP-4947 record's endpoint attribute got frozen as e.g. 'GET'
with no route, and later route arrival was silently ignored.

Defer writing the endpoint until 'isEndpointFinal(tags)' is true (either
'resource.name' is set, or both 'http.method' and 'http.route' are), and
subscribe onTagsUpdated to the raw 'dd-trace:span:tags:update' channel
(not webTagsCache.resolvedCh) so we also see content changes on
already-cached web-server spans. The shared cache still runs first
(module-load subscribe), so its cached webTags is up to date by the time
our handler queries it. Once the endpoint has been appended, further
tags-update fires are no-ops (we can't overwrite a written entry — OTEP
readers honor the first occurrence).

Process-context metadata gate (#4). On non-Linux or non-ACF runtimes,
otel-thread-ctx.start() returns false but tracer_metadata.js was still
publishing the threadlocal_metadata block in the OTel process context —
advertising a decodable OTEP-4947 stream that no writer was producing
records for. Add the same 'process.platform === linux' and 'isACFActive'
gates to getThreadLocalMetadata that start() applies, so the block is
only advertised when the writer can actually run.

Tests: three new cases for the defer-and-append endpoint path and two
new cases for the platform/ACF gates on getThreadLocalMetadata. 24/24
passing.

Author: szegedi

packages/dd-trace/src/otel-thread-ctx.js

@@ -25,11 +25,13 @@
 
 const { isMainThread, threadId } = require('worker_threads')
 
+const { HTTP_METHOD, HTTP_ROUTE, RESOURCE_NAME } = require('../../../ext/tags')
 const { isACFActive } = require('../../datadog-core/src/storage')
 const log = require('./log')
 const {
   enterCh,
   spanFinishCh,
+  tagsUpdateCh,
   getActiveSpan,
   ensureChannelsActivated,
 } = require('./storage-channels')
@@ -39,6 +41,20 @@ const {
 } = require('./profiling/webspan-utils')
 const webTagsCache = require('./web-tags-cache')
 
+// The endpoint label the writer computes from a web-server span's tag bag
+// can change over the span's lifetime: HTTP plugins commonly set
+// `http.method` when the request arrives and only add `http.route` (and
+// often `resource.name`) once framework routing has resolved the URL.
+// The on-the-wire OTEP-4947 record only lets us write each attribute once
+// (readers honor the first occurrence and ignore duplicates), so we defer
+// writing the endpoint until the value looks stable — either `resource.name`
+// is set, or both `http.method` and `http.route` are — and re-check on
+// every subsequent tags update until we can commit.
+function isEndpointFinal (tags) {
+  return tags != null && (tags[RESOURCE_NAME] != null ||
+    (tags[HTTP_METHOD] != null && tags[HTTP_ROUTE] != null))
+}
+
 // Positional attribute layout. The local root span ID stays at index 0 by
 // convention (mirrors libdatadog's libdd-otel-thread-ctx, where
 // `local_root_span_id` is always the first entry in
@@ -100,17 +116,21 @@ function getOrBuildContext (span) {
   const startedSpans = getStartedSpans(spanContext)
   const rootContext = startedSpans.length ? startedSpans[0].context() : spanContext
   const webTags = webTagsCache.getCachedWebTags(span)
+  // Only publish the endpoint when the value is stable — see isEndpointFinal.
+  // Otherwise leave a hole and let onTagsUpdated fill it in when the
+  // remaining routing tags arrive.
+  const endpointReady = isEndpointFinal(webTags)
   const attrs = []
   attrs[LOCAL_ROOT_SPAN_ID_IDX] = rootContext.toSpanId(true)
-  if (webTags) attrs[ENDPOINT_IDX] = endpointNameFromTags(webTags)
+  if (endpointReady) attrs[ENDPOINT_IDX] = endpointNameFromTags(webTags)
   attrs[THREAD_NAME_IDX] = THREAD_NAME
   attrs[THREAD_ID_IDX] = THREAD_ID
   if (cached === undefined) {
     cached = {}
     span[CachedSym] = cached
   }
   cached.context = new ThreadContext(traceId, spanId, attrs)
-  cached.needsEndpoint = webTags === undefined
+  cached.needsEndpoint = !endpointReady
   return cached.context
 }
 
@@ -145,16 +165,21 @@ function onSpanFinished (span) {
 
 function onTagsUpdated (span) {
   if (!started) return
-  // Invoked (via webTagsCache.resolvedCh) once per span at the moment the
-  // shared cache promotes a previously-undefined webTags answer into a
-  // real value.
+  // Subscribed to `dd-trace:span:tags:update` (not to webTagsCache.resolvedCh)
+  // because we need to catch content changes — e.g. `http.route` arriving on
+  // an already-cached web-server span — not just presence transitions.
+  // web-tags-cache subscribes to the same channel at module load and always
+  // runs before us (module init happens before start()), so its cache is
+  // already up to date when we query it here.
   const cached = span[CachedSym]
   if (cached === undefined || !cached.needsEndpoint || cached.context === undefined) return
+  const webTags = webTagsCache.getCachedWebTags(span)
+  if (!isEndpointFinal(webTags)) return
   // Append the endpoint in place. The record buffer is shared across every
   // async-context frame holding this context, so the endpoint becomes
   // visible everywhere at once.
   const append = []
-  append[ENDPOINT_IDX] = endpointNameFromTags(webTagsCache.getCachedWebTags(span))
+  append[ENDPOINT_IDX] = endpointNameFromTags(webTags)
   cached.context.appendAttributes(append)
   cached.needsEndpoint = false
 }
@@ -194,7 +219,7 @@ function start () {
   ensureChannelsActivated(isACFActive)
   enterCh.subscribe(onEnter)
   spanFinishCh.subscribe(onSpanFinished)
-  webTagsCache.resolvedCh.subscribe(onTagsUpdated)
+  tagsUpdateCh.subscribe(onTagsUpdated)
 
   started = true
   log.info('OTEP-4947 thread context writer started')
@@ -210,9 +235,14 @@ function start () {
 //   { attributeKeys, schemaVersion, extraAttributes: [{ key, intValue|stringValue }] }
 //
 // Returns undefined if @datadog/pprof isn't installed or doesn't expose the
-// otelThreadCtx.getProcessContextAttributes helper; callers should treat that
-// as "no threadlocal block" (equivalent to the flag being off).
+// otelThreadCtx.getProcessContextAttributes helper, or if the runtime
+// can't actually run the writer (non-Linux, or Linux without
+// AsyncContextFrame). Callers should treat that as "no threadlocal
+// block" (equivalent to the flag being off) — otherwise we'd publish
+// process-context metadata advertising a decodable OTEP-4947 stream
+// while no writer is producing records.
 function getThreadLocalMetadata () {
+  if (process.platform !== 'linux' || !isACFActive) return
   let pprofMod
   try {
     pprofMod = require('@datadog/pprof')

packages/dd-trace/test/otel-thread-ctx.spec.js

@@ -307,30 +307,64 @@ describe('otel-thread-ctx', () => {
       sinon.assert.notCalled(setActive)
     })
 
-    it('web-tags resolvedCh appends endpoint to the cached context when one was already built', () => {
+    it('tagsUpdate appends endpoint when the shared cache now has final routing tags', () => {
       activeSpan = makeSpan({ tags: {} })
       // No web tags cached at build time → context is built without an
       // endpoint and left with `needsEndpoint = true`.
       enterCh.publish()
       const context = constructedContexts[0]
       sinon.assert.notCalled(context.appendAttributes)
 
-      // Shared cache promoted this span to a web-server span and published
-      // on its resolvedCh. Seed the cache and fire the event.
+      // Simulate the shared cache picking up http.method + http.route on
+      // this span; on the next tagsUpdate we should append the endpoint.
       cachedWebTags.set(activeSpan,
         { 'span.type': 'web', 'http.method': 'GET', 'http.route': '/x' })
-      webTagsResolvedCh.publish(activeSpan)
+      tagsUpdateCh.publish(activeSpan)
       sinon.assert.calledOnce(context.appendAttributes)
       // Endpoint lands at index 1 (local-root-span id occupies index 0).
       const appended = context.appendAttributes.firstCall.args[0]
       assert.equal(appended[1], 'GET /x')
     })
 
-    it('web-tags resolvedCh is a no-op when the span has not been entered yet', () => {
+    it('defers the endpoint until http.route arrives (method-only is not final)', () => {
+      // Simulates the common HTTP server flow: `span.type=web` and
+      // `http.method` set on request start, `http.route` added later by
+      // the routing plugin.
+      const webTags = { 'span.type': 'web', 'http.method': 'GET' }
+      activeSpan = makeSpan({ tags: webTags })
+      cachedWebTags.set(activeSpan, webTags)
+      enterCh.publish()
+      const context = constructedContexts[0]
+      // Endpoint is not final yet — no endpoint in the initial attrs.
+      assert.strictEqual(context.attributes[1], undefined)
+
+      // Route arrives on a subsequent tagsUpdate.
+      webTags['http.route'] = '/x'
+      tagsUpdateCh.publish(activeSpan)
+      sinon.assert.calledOnce(context.appendAttributes)
+      assert.equal(context.appendAttributes.firstCall.args[0][1], 'GET /x')
+    })
+
+    it('does not re-append the endpoint once it has been written', () => {
+      // Final endpoint at build time → nothing to append on subsequent
+      // tagsUpdate events, even if the tag bag continues to mutate.
+      const webTags = { 'span.type': 'web', 'http.method': 'GET', 'http.route': '/x' }
+      activeSpan = makeSpan({ tags: webTags })
+      cachedWebTags.set(activeSpan, webTags)
+      enterCh.publish()
+      const context = constructedContexts[0]
+      sinon.assert.notCalled(context.appendAttributes)
+
+      webTags['http.status_code'] = '200'
+      tagsUpdateCh.publish(activeSpan)
+      sinon.assert.notCalled(context.appendAttributes)
+    })
+
+    it('tagsUpdate is a no-op when the span has not been entered yet', () => {
       const span = makeSpan({ tags: { 'span.type': 'web' } })
       cachedWebTags.set(span,
         { 'span.type': 'web', 'http.method': 'GET', 'http.route': '/x' })
-      webTagsResolvedCh.publish(span)
+      tagsUpdateCh.publish(span)
       assert.equal(constructedContexts.length, 0)
     })
   })
@@ -365,6 +399,17 @@ describe('otel-thread-ctx', () => {
       sinon.assert.calledWithMatch(log.warn, /pprof unavailable|does not expose/)
     })
 
+    it('returns undefined on non-Linux platforms', () => {
+      Object.defineProperty(process, 'platform', { value: 'darwin', configurable: true })
+      const m = loadModule()
+      assert.equal(m.getThreadLocalMetadata(), undefined)
+    })
+
+    it('returns undefined when AsyncContextFrame is inactive', () => {
+      const m = loadModule({ storage: { '@noCallThru': true, isACFActive: false } })
+      assert.equal(m.getThreadLocalMetadata(), undefined)
+    })
+
     it('returns undefined when otelThreadCtx.getProcessContextAttributes is missing', () => {
       const m = loadModule({
         pprof: {