Extract shared web-tags cache used by wall profiler and OTEP-4947 writer

The wall profiler and the OTEP-4947 thread-context writer were both
walking each span's started-spans chain to find the nearest web-server
ancestor, each caching the answer under its own Symbol. Move the walk
and cache into packages/dd-trace/src/web-tags-cache.js.

- getCachedWebTags(span): lazy parent-chain walk, cached on a shared
  Symbol.
- onSpanTagsUpdated(span): call from a tagsUpdate subscriber; if the
  walk previously came up empty and the span is now a web-server span,
  promote its tags into the cache. Returns true iff the cache
  transitioned from undefined to a real value — a signal to consumers
  that they should snapshot the new value into whatever they built
  while the answer was undefined.

wall.js still snapshots webTags into its per-sample ProfilingContext
because label generation reads it from the sample-context ref (not from
the span); its tagsUpdate handler now refreshes that snapshot only when
the shared cache signals a transition.

otel-thread-ctx.js's cached-per-span record drops the local walk plus
the webTags / webTagsResolved fields; it now just tracks whether the
ThreadContext was built without an endpoint (needsEndpoint) and appends
one when the shared cache reports a late transition.

No functional change. CODEOWNERS gets the new file scoped to
@DataDog/profiling-js (same as its two callers).

Author: szegedi

.github/CODEOWNERS

@@ -320,6 +320,7 @@
 /packages/dd-trace/*/profiling/ @DataDog/profiling-js
 /packages/dd-trace/src/otel-thread-ctx.js @DataDog/profiling-js
 /packages/dd-trace/src/storage-channels.js @DataDog/profiling-js @DataDog/apm-sdk-capabilities-js
+/packages/dd-trace/src/web-tags-cache.js @DataDog/profiling-js
 /packages/dd-trace/test/exporters/common/form-data.spec.js @DataDog/profiling-js
 /packages/dd-trace/test/otel-thread-ctx.spec.js @DataDog/profiling-js
 

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

@@ -30,15 +30,14 @@ const log = require('./log')
 const {
   enterCh,
   spanFinishCh,
-  tagsUpdateCh,
   getActiveSpan,
   ensureChannelsActivated,
 } = require('./storage-channels')
 const {
-  isWebServerSpan,
   endpointNameFromTags,
   getStartedSpans,
 } = require('./profiling/webspan-utils')
+const webTagsCache = require('./web-tags-cache')
 
 // Positional attribute layout. The local root span ID stays at index 0 by
 // convention (mirrors libdatadog's libdd-otel-thread-ctx, where
@@ -75,59 +74,23 @@ const THREAD_ID = String(threadId)
 // reference verbatim, and the context's record buffer is mutated in
 // place by appendAttributes, so all frames observe the same record.
 //
-// Fields populated lazily:
-//   context: ThreadContext from @datadog/pprof.otelThreadCtx — built the first
-//     time onEnter activates the span.
-//   webTagsResolved + webTags: true once the parent-chain walk has
-//     run; webTags is the resolved tag bag (or undefined when no web
-//     ancestor was found).
+// Fields:
+//   context:        ThreadContext from @datadog/pprof.otelThreadCtx —
+//                   built the first time onEnter activates the span.
+//   needsEndpoint:  true when the ThreadContext was built without an
+//                   endpoint attribute (the shared web-tags cache came
+//                   up empty at build time). Cleared when a late
+//                   tagsUpdate lets us append the endpoint.
 const CachedSym = Symbol('OtelThreadCtx.cached')
 
 let started = false
 let ThreadContext
 let getContext
 let clearContext
 
-function getOrCreateCache (span) {
-  let cached = span[CachedSym]
-  if (cached === undefined) {
-    cached = {}
-    span[CachedSym] = cached
-  }
-  return cached
-}
-
-// Walks up the started-spans stack to find the nearest ancestor whose
-// tags identify it as a web-server span. Mirrors the same walk in
-// profiling/profilers/wall.js (which keeps its own cache under a
-// different Symbol). If the two ever drift we should extract.
-function getCachedWebTags (span) {
-  const cached = getOrCreateCache(span)
-  if (cached.webTagsResolved) return cached.webTags
-  const spanContext = span.context()
-  const tags = spanContext.getTags()
-  let webTags
-  if (isWebServerSpan(tags)) {
-    webTags = tags
-  } else {
-    const parentId = spanContext._parentId
-    const startedSpans = getStartedSpans(spanContext)
-    for (let i = startedSpans.length; --i >= 0;) {
-      const ispan = startedSpans[i]
-      if (ispan.context()._spanId === parentId) {
-        webTags = getCachedWebTags(ispan)
-        break
-      }
-    }
-  }
-  cached.webTags = webTags
-  cached.webTagsResolved = true
-  return webTags
-}
-
 function getOrBuildContext (span) {
-  const cached = getOrCreateCache(span)
-  if (cached.context !== undefined) return cached.context
+  let cached = span[CachedSym]
+  if (cached !== undefined && cached.context !== undefined) return cached.context
   const spanContext = span.context()
   const traceId = Uint8Array.from(Buffer.from(spanContext.toTraceId(true), 'hex'))
   const spanId = Uint8Array.from(Buffer.from(spanContext.toSpanId(true), 'hex'))
@@ -136,13 +99,18 @@ function getOrBuildContext (span) {
   // per the libdatadog convention.
   const startedSpans = getStartedSpans(spanContext)
   const rootContext = startedSpans.length ? startedSpans[0].context() : spanContext
-  const webTags = getCachedWebTags(span)
+  const webTags = webTagsCache.getCachedWebTags(span)
   const attrs = []
   attrs[LOCAL_ROOT_SPAN_ID_IDX] = rootContext.toSpanId(true)
   if (webTags) 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
   return cached.context
 }
 
@@ -177,24 +145,18 @@ 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.
   const cached = span[CachedSym]
-  // Skip unless the prior parent-chain walk already ran and came up
-  // empty. If the walk hasn't happened yet (cached.webTagsResolved
-  // false), onEnter will resolve it the natural way. If it ran and
-  // found a web span, we already have the endpoint.
-  if (cached === undefined || !cached.webTagsResolved || cached.webTags !== undefined) return
-  const tags = span.context().getTags()
-  if (!isWebServerSpan(tags)) return
-  cached.webTags = tags
-  if (cached.context !== undefined) {
-    // The context was already built without an endpoint; append it 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(tags)
-    cached.context.appendAttributes(append)
-  }
+  if (cached === undefined || !cached.needsEndpoint || cached.context === undefined) 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))
+  cached.context.appendAttributes(append)
+  cached.needsEndpoint = false
 }
 
 function start () {
@@ -232,7 +194,7 @@ function start () {
   ensureChannelsActivated(isACFActive)
   enterCh.subscribe(onEnter)
   spanFinishCh.subscribe(onSpanFinished)
-  tagsUpdateCh.subscribe(onTagsUpdated)
+  webTagsCache.resolvedCh.subscribe(onTagsUpdated)
 
   started = true
   log.info('OTEP-4947 thread context writer started')

packages/dd-trace/src/profiling/profilers/wall.js

@@ -3,16 +3,16 @@
 const log = require('../../log')
 const runtimeMetrics = require('../../runtime_metrics')
 const telemetryMetrics = require('../../telemetry/metrics')
-const { isWebServerSpan, endpointNameFromTags, getStartedSpans } = require('../webspan-utils')
+const { endpointNameFromTags, getStartedSpans } = require('../webspan-utils')
 const { SAMPLING_INTERVAL } = require('../constants')
 const {
   enterCh,
   beforeCh,
   spanFinishCh,
-  tagsUpdateCh,
   getActiveSpan,
   ensureChannelsActivated,
 } = require('../../storage-channels')
+const webTagsCache = require('../../web-tags-cache')
 
 const {
   END_TIMESTAMP_LABEL,
@@ -162,7 +162,10 @@ class NativeWallProfiler {
         enterCh.subscribe(this.#boundEnter)
         spanFinishCh.subscribe(this.#boundSpanFinished)
         if (this.#endpointCollectionEnabled) {
-          tagsUpdateCh.subscribe(this.#boundSpanTagsUpdated)
+          // Web-tags cache publishes once per span at the moment its
+          // walk-result transitions from undefined to a real value —
+          // exactly when we need to refresh the ProfilingContext snapshot.
+          webTagsCache.resolvedCh.subscribe(this.#boundSpanTagsUpdated)
         }
       }
     }
@@ -245,32 +248,20 @@ class NativeWallProfiler {
     let profilingContext = span[ProfilingContext]
     if (profilingContext === undefined) {
       const context = span.context()
-      const startedSpans = getStartedSpans(context)
 
       let spanId
       let rootSpanId
       if (this.#codeHotspotsEnabled) {
+        const startedSpans = getStartedSpans(context)
         spanId = context._spanId
         rootSpanId = startedSpans.length ? startedSpans[0].context()._spanId : context._spanId
       }
 
-      let webTags
-      if (this.#endpointCollectionEnabled) {
-        const tags = context.getTags()
-        if (isWebServerSpan(tags)) {
-          webTags = tags
-        } else {
-          // Get parent's context's web tags
-          const parentId = context._parentId
-          for (let i = startedSpans.length; --i >= 0;) {
-            const ispan = startedSpans[i]
-            if (ispan.context()._spanId === parentId) {
-              webTags = this.#getProfilingContext(ispan).webTags
-              break
-            }
-          }
-        }
-      }
+      // webTags is snapshotted into the sample context at getProfilingContext
+      // time; if the answer turns out to be undefined and the span later gets
+      // web-server tags, #spanTagsUpdated refreshes this field via the shared
+      // cache (see web-tags-cache.js).
+      const webTags = this.#endpointCollectionEnabled ? webTagsCache.getCachedWebTags(span) : undefined
 
       profilingContext = { spanId, rootSpanId, webTags }
       span[ProfilingContext] = profilingContext
@@ -291,14 +282,15 @@ class NativeWallProfiler {
     }
   }
 
+  // Invoked (via webTagsCache.resolvedCh) once per span at the moment the
+  // shared cache promotes a previously-undefined webTags answer into a
+  // real value. Refresh the ProfilingContext snapshot so future samples
+  // pick it up.
   #spanTagsUpdated (span) {
     if (!this.#started) return
     const profilingContext = span[ProfilingContext]
-    if (profilingContext === undefined || profilingContext.webTags !== undefined) return
-    const tags = span.context().getTags()
-    if (isWebServerSpan(tags)) {
-      profilingContext.webTags = tags
-    }
+    if (profilingContext === undefined) return
+    profilingContext.webTags = webTagsCache.getCachedWebTags(span)
   }
 
   #reportV8bug (maybeBug) {
@@ -347,7 +339,7 @@ class NativeWallProfiler {
         enterCh.unsubscribe(this.#boundEnter)
         spanFinishCh.unsubscribe(this.#boundSpanFinished)
         if (this.#endpointCollectionEnabled) {
-          tagsUpdateCh.unsubscribe(this.#boundSpanTagsUpdated)
+          webTagsCache.resolvedCh.unsubscribe(this.#boundSpanTagsUpdated)
         }
         this.#profilerState = undefined
       }

packages/dd-trace/src/web-tags-cache.js

@@ -0,0 +1,82 @@
+'use strict'
+
+// Per-span cache of "which tag bag from the started-spans chain identifies
+// this span (or its nearest web-server ancestor) as a web-server span?"
+// Populated lazily on first `getCachedWebTags(span)`, refreshed
+// automatically when a `dd-trace:span:tags:update` event promotes a
+// previously-empty answer for that span into a real value.
+//
+// Used by the wall profiler (endpoint-collection label on samples) and by
+// the OTEP-4947 thread-context writer (endpoint attribute in the record);
+// having a single cache means the parent-chain walk happens once per span
+// no matter how many consumers ask.
+//
+// Consumers that want to react to late web-server-span discovery
+// subscribe to `resolvedCh` — a diagnostics channel we publish on once
+// per span at the moment its cached webTags transitions from undefined
+// to a real value. Doing it via a channel (rather than exposing a
+// stateful "did the transition happen?" query) means each consumer sees
+// every transition exactly once, regardless of subscription order or
+// how many other consumers are attached.
+
+const dc = require('dc-polyfill')
+const { isWebServerSpan, getStartedSpans } = require('./profiling/webspan-utils')
+
+// Fields on the cache entry:
+//   resolved: true once the parent-chain walk has run.
+//   webTags:  the resolved tag bag, or undefined when the walk came up
+//             empty (no web-server span found in the started-spans chain).
+const CachedSym = Symbol('WebTagsCache')
+
+const tagsUpdateCh = dc.channel('dd-trace:span:tags:update')
+const resolvedCh = dc.channel('dd-trace:web-tags:resolved')
+
+function getCache (span) {
+  let cached = span[CachedSym]
+  if (cached === undefined) {
+    cached = {}
+    span[CachedSym] = cached
+  }
+  return cached
+}
+
+// Returns the web-server tag bag for this span or its nearest web-server
+// ancestor in the started-spans chain, or undefined if none is a
+// web-server span. Lazy: walks the parent chain on the first call, caches
+// the result on the span.
+function getCachedWebTags (span) {
+  const cached = getCache(span)
+  if (cached.resolved) return cached.webTags
+  const spanContext = span.context()
+  const tags = spanContext.getTags()
+  let webTags
+  if (isWebServerSpan(tags)) {
+    webTags = tags
+  } else {
+    const parentId = spanContext._parentId
+    const startedSpans = getStartedSpans(spanContext)
+    for (let i = startedSpans.length; --i >= 0;) {
+      const ispan = startedSpans[i]
+      if (ispan.context()._spanId === parentId) {
+        webTags = getCachedWebTags(ispan)
+        break
+      }
+    }
+  }
+  cached.webTags = webTags
+  cached.resolved = true
+  return webTags
+}
+
+// Own the tagsUpdate → transition promotion here. Subscribed at module
+// load; inert (an O(1) Symbol check) for any span nobody has queried yet.
+tagsUpdateCh.subscribe((span) => {
+  const cached = span[CachedSym]
+  if (cached === undefined || !cached.resolved || cached.webTags !== undefined) return
+  const tags = span.context().getTags()
+  if (!isWebServerSpan(tags)) return
+  cached.webTags = tags
+  resolvedCh.publish(span)
+})
+
+module.exports = { getCachedWebTags, resolvedCh }