fix(standalone): stamp _dd.apm.enabled on every exported chunk (#9483)

Author: IlyasShabi

integration-tests/appsec/standalone-asm.spec.js

@@ -45,6 +45,14 @@ describe('Standalone ASM', () => {
     }
   }
 
+  function isLateOutboundSpan (span) {
+    return span.name === 'http.request' && span.meta['http.url']?.endsWith('/intake/v2/events')
+  }
+
+  function isLateOutboundRoot (span) {
+    return span.resource === 'GET /late-outbound'
+  }
+
   describe('enabled', () => {
     beforeEach(async () => {
       agent = await new FakeAgent().start()
@@ -81,6 +89,45 @@ describe('Standalone ASM', () => {
       })
     })
 
+    it('should add _dd.apm.enabled tag to delayed local child chunks', async () => {
+      const seenGroups = []
+      const groups = await agent.collectGroups({
+        trigger: () => curl(`${proc.url}/late-outbound`),
+        predicate: group => {
+          seenGroups.push(group.map(span => ({
+            name: span.name,
+            resource: span.resource,
+            url: span.meta['http.url'],
+            apmEnabled: span.metrics['_dd.apm.enabled'],
+          })))
+
+          return group.some(isLateOutboundRoot) || group.some(isLateOutboundSpan)
+        },
+        expectedCount: 2,
+      }).catch(error => {
+        error.message += `\nSeen groups: ${inspect(seenGroups, { depth: null })}`
+        throw error
+      })
+
+      const rootGroup = groups.find(group => group.some(isLateOutboundRoot))
+      const outboundGroup = groups.find(group => group.some(isLateOutboundSpan))
+      const rootSpan = rootGroup?.find(isLateOutboundRoot)
+      const outboundSpan = outboundGroup?.find(isLateOutboundSpan)
+
+      // Two distinct chunks: parent flushes first, delayed child flushes later.
+      assert.notStrictEqual(rootGroup, undefined)
+      assert.notStrictEqual(outboundGroup, undefined)
+      assert.notStrictEqual(outboundGroup, rootGroup)
+      assert.ok(groups.indexOf(rootGroup) < groups.indexOf(outboundGroup))
+      assert.strictEqual(String(outboundSpan.parent_id), String(rootSpan.span_id))
+
+      // Load-bearing: the delayed child chunk must carry the billing marker,
+      // even though its parent is a local (non-remote) span.
+      assert.strictEqual(rootSpan.metrics['_dd.apm.enabled'], 0)
+      assert.strictEqual(outboundGroup[0], outboundSpan)
+      assert.strictEqual(outboundSpan.metrics['_dd.apm.enabled'], 0)
+    })
+
     it('should keep fifth req because RateLimiter allows 1 req/min', async () => {
       const promise = curlAndAssertMessage(agent, proc, ({ headers, payload }) => {
         assert.strictEqual(headers['datadog-client-computed-stats'], 'yes')

integration-tests/standalone-asm/index.js

@@ -23,11 +23,14 @@ tracer.init(options)
 
 const crypto = require('crypto')
 const http = require('http')
+const net = require('net')
 
 const express = require('express')
 const app = express()
 
 const valueToHash = 'iast-showcase-demo'
+let delayedOutboundPort
+let server
 
 async function makeRequest (url) {
   return new Promise((resolve, reject) => {
@@ -66,6 +69,20 @@ app.get('/vulnerableHash', (req, res) => {
   res.status(200).send(result)
 })
 
+app.get('/late-outbound', (req, res) => {
+  const url = `http://localhost:${delayedOutboundPort}/intake/v2/events`
+  const activeSpan = tracer.scope().active()
+  const rootSpan = activeSpan?.context()._trace.started[0] || activeSpan
+
+  setTimeout(() => {
+    tracer.scope().activate(rootSpan, () => {
+      makeRequest(url).catch(() => {})
+    })
+  }, 250)
+
+  res.status(200).send('late-outbound')
+})
+
 app.get('/propagation-with-event', async (req, res) => {
   tracer.appsec.trackCustomEvent('custom-event')
 
@@ -109,7 +126,17 @@ app.get('/propagation-after-drop-and-call-sdk', async (req, res) => {
   res.status(200).send(`drop-and-call-sdk ${sdkRes}`)
 })
 
-const server = http.createServer(app).listen(0, () => {
-  const port = (/** @type {import('net').AddressInfo} */ (server.address())).port
-  process.send?.({ port })
+const delayedOutboundServer = net.createServer(socket => {
+  socket.once('data', () => {
+    socket.end('HTTP/1.1 202 Accepted\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok')
+  })
+})
+
+delayedOutboundServer.listen(0, () => {
+  delayedOutboundPort = (/** @type {import('net').AddressInfo} */ (delayedOutboundServer.address())).port
+
+  server = http.createServer(app).listen(0, () => {
+    const port = (/** @type {import('net').AddressInfo} */ (server.address())).port
+    process.send?.({ port })
+  })
 })

packages/dd-trace/src/span_processor.js

@@ -6,6 +6,7 @@ const SpanSampler = require('./span_sampler')
 const GitMetadataTagger = require('./git_metadata_tagger')
 const processTags = require('./process-tags')
 const { applyHttpOtelSemantics } = require('./plugins/util/http-otel-semantics')
+const { APM_TRACING_ENABLED_KEY } = require('./constants')
 
 const startedSpans = new WeakSet()
 const finishedSpans = new WeakSet()
@@ -54,12 +55,16 @@ class SpanProcessor {
       this._gitMetadataTagger.tagGitMetadata(spanContext)
 
       let isFirstSpanInChunk = true
+      const stampApmDisabled = this._config.apmTracingEnabled === false
 
       for (const span of started) {
         if (span._duration === undefined) {
           active.push(span)
         } else {
           const formattedSpan = spanFormat(span, isFirstSpanInChunk, this._processTags)
+          if (isFirstSpanInChunk && stampApmDisabled) {
+            formattedSpan.metrics[APM_TRACING_ENABLED_KEY] = 0
+          }
           isFirstSpanInChunk = false
           // Span stats read Datadog HTTP tag names from the formatted span, so
           // record them before the OTel rename — an export-only transform.

packages/dd-trace/src/standalone/index.js

@@ -2,38 +2,24 @@
 
 const { channel } = require('dc-polyfill')
 const { USER_KEEP } = require('../../../../ext/priority')
-const { APM_TRACING_ENABLED_KEY } = require('../constants')
 const TraceSourcePrioritySampler = require('./tracesource_priority_sampler')
 const { hasTraceSourcePropagationTag } = require('./tracesource')
 
-const startCh = channel('dd-trace:span:start')
 const extractCh = channel('dd-trace:span:extract')
 
 /**
  * @param {import('../config/config-base')} config - Tracer configuration
  */
 function configure (config) {
-  if (startCh.hasSubscribers) startCh.unsubscribe(onSpanStart)
   if (extractCh.hasSubscribers) extractCh.unsubscribe(onSpanExtract)
 
   if (config.apmTracingEnabled !== false) return
 
-  startCh.subscribe(onSpanStart)
   extractCh.subscribe(onSpanExtract)
 
   return new TraceSourcePrioritySampler(config.env)
 }
 
-function onSpanStart ({ span, fields }) {
-  const context = span.context?.()
-  if (!context) return
-
-  const { parent } = fields
-  if (!parent || parent._isRemote) {
-    context.setTag(APM_TRACING_ENABLED_KEY, 0)
-  }
-}
-
 function onSpanExtract ({ spanContext = {} }) {
   if (!spanContext._trace?.tags || !spanContext._sampling) return
 

packages/dd-trace/test/span_processor.spec.js

@@ -9,6 +9,8 @@ const proxyquire = require('proxyquire')
 
 require('./setup/core')
 
+const { APM_TRACING_ENABLED_KEY } = require('../src/constants')
+
 describe('SpanProcessor', () => {
   let prioritySampler
   let processor
@@ -233,6 +235,67 @@ describe('SpanProcessor', () => {
     sinon.assert.calledWith(spanFormat.getCall(3), finishedSpan, false, processor._processTags)
   })
 
+  it('should add APM disabled marker to first span in a chunk when APM tracing is disabled', () => {
+    config.apmTracingEnabled = false
+    config.flushMinSpans = 2
+    const processor = new SpanProcessor(exporter, prioritySampler, config)
+    const firstFormatted = { metrics: {} }
+    const secondFormatted = { metrics: {} }
+    spanFormat.onFirstCall().returns(firstFormatted)
+    spanFormat.onSecondCall().returns(secondFormatted)
+    trace.started = [activeSpan, finishedSpan, finishedSpan]
+    trace.finished = [finishedSpan, finishedSpan]
+
+    processor.process(finishedSpan)
+
+    assert.strictEqual(firstFormatted.metrics[APM_TRACING_ENABLED_KEY], 0)
+    assert.ok(!Object.hasOwn(secondFormatted.metrics, APM_TRACING_ENABLED_KEY))
+    sinon.assert.calledWith(exporter.export, [firstFormatted, secondFormatted])
+  })
+
+  it('should add APM disabled marker to every chunk when a delayed child flushes alone', () => {
+    // Reproduces the standalone-ASM billing regression: the entry span flushes
+    // in one chunk, then a long-lived child (e.g. delayed http.request) flushes
+    // later in its own chunk. Both chunks must carry _dd.apm.enabled:0.
+    config.apmTracingEnabled = false
+    const processor = new SpanProcessor(exporter, prioritySampler, config)
+    const parentFormatted = { metrics: {} }
+    const childFormatted = { metrics: {} }
+    spanFormat.onFirstCall().returns(parentFormatted)
+    spanFormat.onSecondCall().returns(childFormatted)
+
+    const parentSpan = { ...finishedSpan }
+    const childSpan = { ...finishedSpan }
+    trace.started = [parentSpan]
+    trace.finished = [parentSpan]
+
+    processor.process(parentSpan)
+
+    assert.strictEqual(parentFormatted.metrics[APM_TRACING_ENABLED_KEY], 0)
+    sinon.assert.calledWith(exporter.export, [parentFormatted])
+
+    trace.started = [childSpan]
+    trace.finished = [childSpan]
+
+    processor.process(childSpan)
+
+    assert.strictEqual(childFormatted.metrics[APM_TRACING_ENABLED_KEY], 0)
+    sinon.assert.calledWith(exporter.export.secondCall, [childFormatted])
+  })
+
+  it('should not add APM disabled marker when APM tracing is enabled', () => {
+    config.apmTracingEnabled = true
+    const processor = new SpanProcessor(exporter, prioritySampler, config)
+    const formattedSpan = { metrics: {} }
+    spanFormat.returns(formattedSpan)
+    trace.started = [finishedSpan]
+    trace.finished = [finishedSpan]
+
+    processor.process(finishedSpan)
+
+    assert.ok(!Object.hasOwn(formattedSpan.metrics, APM_TRACING_ENABLED_KEY))
+  })
+
   describe('with DD_TRACE_OTEL_SEMANTICS_ENABLED', () => {
     function formattedHttpSpan () {
       return {

packages/dd-trace/test/standalone/index.spec.js

@@ -14,7 +14,6 @@ const standalone = require('../../src/standalone')
 const DatadogSpan = require('../../src/opentracing/span')
 
 const {
-  APM_TRACING_ENABLED_KEY,
   SAMPLING_MECHANISM_APPSEC,
   DECISION_MAKER_KEY,
   TRACE_SOURCE_PROPAGATION_KEY,
@@ -24,7 +23,6 @@ const TextMapPropagator = require('../../src/opentracing/propagation/text_map')
 const TraceState = require('../../src/opentracing/propagation/tracestate')
 const TraceSourcePrioritySampler = require('../../src/standalone/tracesource_priority_sampler')
 
-const startCh = channel('dd-trace:span:start')
 const extractCh = channel('dd-trace:span:extract')
 
 describe('Disabled APM Tracing or Standalone', () => {
@@ -49,33 +47,26 @@ describe('Disabled APM Tracing or Standalone', () => {
   afterEach(() => { sinon.restore() })
 
   describe('configure', () => {
-    let startChSubscribe
-    let startChUnsubscribe
     let extractChSubscribe
     let extractChUnsubscribe
 
     beforeEach(() => {
-      startChSubscribe = sinon.stub(startCh, 'subscribe')
-      startChUnsubscribe = sinon.stub(startCh, 'unsubscribe')
       extractChSubscribe = sinon.stub(extractCh, 'subscribe')
       extractChUnsubscribe = sinon.stub(extractCh, 'unsubscribe')
     })
 
-    it('should subscribe to start span if apmTracing disabled', () => {
+    it('should subscribe to extract if apmTracing disabled', () => {
       standalone.configure(config)
 
-      sinon.assert.calledOnce(startChSubscribe)
       sinon.assert.calledOnce(extractChSubscribe)
     })
 
-    it('should not subscribe to start span if apmTracing enabled', () => {
+    it('should not subscribe to extract if apmTracing enabled', () => {
       config.apmTracingEnabled = true
 
       standalone.configure(config)
 
-      sinon.assert.notCalled(startChSubscribe)
       sinon.assert.notCalled(extractChSubscribe)
-      sinon.assert.notCalled(startChUnsubscribe)
       sinon.assert.notCalled(extractChUnsubscribe)
     })
 
@@ -120,66 +111,6 @@ describe('Disabled APM Tracing or Standalone', () => {
     })
   })
 
-  describe('onStartSpan', () => {
-    it('should not add _dd.apm.enabled tag when standalone is disabled', () => {
-      config.apmTracingEnabled = true
-      standalone.configure(config)
-
-      const span = new DatadogSpan(tracer, processor, prioritySampler, {
-        operationName: 'operation',
-      })
-
-      assert.ok(!span.context().hasTag(APM_TRACING_ENABLED_KEY))
-    })
-
-    it('should add _dd.apm.enabled tag when standalone is enabled', () => {
-      standalone.configure(config)
-
-      const span = new DatadogSpan(tracer, processor, prioritySampler, {
-        operationName: 'operation',
-      })
-
-      assert.ok(
-        span.context().hasTag(APM_TRACING_ENABLED_KEY),
-        `Available keys: ${inspect(Object.keys(span.context().getTags()))}`
-      )
-    })
-
-    it('should not add _dd.apm.enabled tag in child spans with local parent', () => {
-      standalone.configure(config)
-
-      const parent = new DatadogSpan(tracer, processor, prioritySampler, {
-        operationName: 'operation',
-      })
-
-      assert.strictEqual(parent.context().getTag(APM_TRACING_ENABLED_KEY), 0)
-
-      const child = new DatadogSpan(tracer, processor, prioritySampler, {
-        operationName: 'operation',
-        parent,
-      })
-
-      assert.ok(!child.context().hasTag(APM_TRACING_ENABLED_KEY))
-    })
-
-    it('should add _dd.apm.enabled tag in child spans with remote parent', () => {
-      standalone.configure(config)
-
-      const parent = new DatadogSpan(tracer, processor, prioritySampler, {
-        operationName: 'operation',
-      })
-
-      parent._isRemote = true
-
-      const child = new DatadogSpan(tracer, processor, prioritySampler, {
-        operationName: 'operation',
-        parent,
-      })
-
-      assert.strictEqual(child.context().getTag(APM_TRACING_ENABLED_KEY), 0)
-    })
-  })
-
   describe('onSpanExtract', () => {
     it('should reset priority if _dd.p.ts not present', () => {
       standalone.configure(config)