fix(azure-durable-functions): link activity and entity spans to the HTTP trace

.github/workflows/serverless.yml

@@ -293,6 +293,9 @@ jobs:
 
     steps:
       - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+      - uses: ./.github/actions/plugins/test
+        env:
+          SPEC: index
       - uses: ./.github/actions/plugins/integration-test-newest-lts
         with:
           flags: serverless-azure-durable-functions

packages/datadog-instrumentations/src/azure-durable-functions.js

@@ -42,9 +42,16 @@ function entityHandler (handler, entityName) {
     if (!azureDurableFunctionsChannel.hasSubscribers) return handler.apply(this, args)
 
     const entityContext = args[0]
+    const traceContext = entityContext?.traceContext
     return azureDurableFunctionsChannel.traceSync(
       handler,
-      { trigger: 'Entity', functionName: entityName, operationName: entityContext?.df?.operationName },
+      {
+        trigger: 'Entity',
+        functionName: entityName,
+        operationName: entityContext?.df?.operationName,
+        traceparent: traceContext?.traceParent,
+        tracestate: traceContext?.traceState,
+      },
       this, ...args)
   }
 }
@@ -58,16 +65,18 @@ function activityHandler (method) {
       return function (...args) {
         if (!azureDurableFunctionsChannel.hasSubscribers) return handler.apply(this, args)
 
+        const traceContext = args[1]?.traceContext
+        const channelCtx = {
+          trigger: 'Activity',
+          functionName: activityName,
+          traceparent: traceContext?.traceParent,
+          tracestate: traceContext?.traceState,
+        }
+
         // use tracePromise if this is an async handler. otherwise, use traceSync
         return isAsync
-          ? azureDurableFunctionsChannel.tracePromise(
-            handler,
-            { trigger: 'Activity', functionName: activityName },
-            this, ...args)
-          : azureDurableFunctionsChannel.traceSync(
-            handler,
-            { trigger: 'Activity', functionName: activityName },
-            this, ...args)
+          ? azureDurableFunctionsChannel.tracePromise(handler, channelCtx, this, ...args)
+          : azureDurableFunctionsChannel.traceSync(handler, channelCtx, this, ...args)
       }
     })
     return method.apply(this, arguments)

packages/datadog-plugin-azure-durable-functions/src/index.js

@@ -1,6 +1,8 @@
 'use strict'
 
 const TracingPlugin = require('../../dd-trace/src/plugins/tracing')
+const { AUTO_KEEP } = require('../../../ext/priority')
+const TraceState = require('../../dd-trace/src/opentracing/propagation/tracestate')
 
 class AzureDurableFunctionsPlugin extends TracingPlugin {
   static get id () { return 'azure-durable-functions' }
@@ -10,7 +12,22 @@ class AzureDurableFunctionsPlugin extends TracingPlugin {
   static get kind () { return 'server' }
 
   bindStart (ctx) {
+    // Continue the trace propagated by the Durable Functions host (W3C traceparent
+    // supplied on the invocation's traceContext) so activity/entity invocations join
+    // the same trace as the HTTP root instead of each starting a new root.
+    let childOf
+    if (ctx.traceparent) {
+      // extract() returns null when the carrier can't be parsed. Normalize to
+      // undefined so startSpan still falls back to any active in-process parent
+      // rather than being forced to start a brand new root span.
+      childOf = this.tracer.extract('text_map', {
+        traceparent: ctx.traceparent,
+        tracestate: ctx.tracestate,
+      }) ?? undefined
+    }
+
     const span = this.startSpan(this.operationName(), {
+      childOf,
       kind: 'internal',
       type: 'serverless',
 
@@ -29,6 +46,14 @@ class AzureDurableFunctionsPlugin extends TracingPlugin {
       )
     }
 
+    // The host clears the W3C sampled flag in traceparent while datadog tracestate
+    // still says keep, so extraction would drop this chunk. Re-apply the propagated
+    // `s` priority when it indicates keep; upstream drop decisions are left untouched.
+    const propagatedPriority = propagatedSamplingPriority(ctx.tracestate)
+    if (childOf && sampledFlagCleared(ctx.traceparent) && propagatedPriority >= AUTO_KEEP) {
+      span._prioritySampler?.setPriority(span, propagatedPriority)
+    }
+
     ctx.span = span
     return ctx.currentStore
   }
@@ -46,4 +71,25 @@ class AzureDurableFunctionsPlugin extends TracingPlugin {
   }
 }
 
+// True when the W3C traceparent's sampled flag is cleared (flags & 0x01 === 0),
+// i.e. the carrier says "drop". Format: version-traceId-spanId-flags.
+function sampledFlagCleared (traceparent) {
+  if (typeof traceparent !== 'string') return false
+  const flags = traceparent.split('-')[3]
+  return flags !== undefined && (Number.parseInt(flags, 16) & 1) === 0
+}
+
+// Read the datadog-propagated sampling priority (`dd=...;s:<n>`) from a W3C
+// tracestate. Returns undefined when there is no datadog tracestate or no valid
+// `s` value, so callers can distinguish "no propagated decision" from a drop.
+function propagatedSamplingPriority (tracestate) {
+  if (typeof tracestate !== 'string' || !tracestate) return
+  let priority
+  TraceState.fromString(tracestate).forVendor('dd', state => {
+    const parsed = Number.parseInt(state.get('s'), 10)
+    if (Number.isInteger(parsed)) priority = parsed
+  })
+  return priority
+}
+
 module.exports = AzureDurableFunctionsPlugin

packages/datadog-plugin-azure-durable-functions/test/index.spec.js

@@ -0,0 +1,179 @@
+'use strict'
+
+const assert = require('node:assert/strict')
+
+const { afterEach, beforeEach, describe, it } = require('mocha')
+const sinon = require('sinon')
+
+require('../../dd-trace/test/setup/core')
+
+const { AUTO_KEEP, USER_KEEP } = require('../../../ext/priority')
+const AzureDurableFunctionsPlugin = require('../src')
+
+describe('azure-durable-functions plugin', () => {
+  let plugin
+  let extract
+  let startSpan
+  let setPriority
+  let span
+
+  beforeEach(() => {
+    setPriority = sinon.stub()
+    span = {
+      _prioritySampler: { setPriority },
+      setTag: sinon.stub(),
+    }
+
+    extract = sinon.stub()
+    startSpan = sinon.stub().returns(span)
+
+    plugin = new AzureDurableFunctionsPlugin({
+      extract,
+      startSpan,
+      _service: 'test-service',
+      _nomenclature: {
+        opName: () => 'azure.functions.invoke',
+        serviceName: () => ({ name: 'test-service' }),
+      },
+    })
+    plugin.configure({})
+  })
+
+  afterEach(() => {
+    sinon.restore()
+  })
+
+  function bindStart (overrides = {}) {
+    const ctx = {
+      trigger: 'Activity',
+      functionName: 'hola',
+      currentStore: {},
+      ...overrides,
+    }
+
+    plugin.bindStart(ctx)
+    return ctx
+  }
+
+  it('continues the host trace when traceparent is provided', () => {
+    const parent = { _traceId: 'parent' }
+    extract.returns(parent)
+
+    bindStart({
+      traceparent: '00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01',
+      tracestate: 'dd=s:1',
+    })
+
+    sinon.assert.calledOnceWithExactly(extract, 'text_map', {
+      traceparent: '00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01',
+      tracestate: 'dd=s:1',
+    })
+    sinon.assert.calledWith(
+      startSpan,
+      'azure.functions.invoke',
+      sinon.match({ childOf: parent })
+    )
+  })
+
+  it('normalizes a failed extract to undefined childOf', () => {
+    extract.returns(null)
+
+    bindStart({
+      traceparent: 'not-a-valid-traceparent',
+    })
+
+    sinon.assert.calledWith(
+      startSpan,
+      'azure.functions.invoke',
+      sinon.match({ childOf: undefined })
+    )
+  })
+
+  it('does not extract when traceparent is missing', () => {
+    bindStart()
+
+    sinon.assert.notCalled(extract)
+    sinon.assert.calledWith(
+      startSpan,
+      'azure.functions.invoke',
+      sinon.match({ childOf: undefined })
+    )
+  })
+
+  it('tags entity operation metadata when operationName is present', () => {
+    bindStart({
+      trigger: 'Entity',
+      functionName: 'counter',
+      operationName: 'add_n',
+    })
+
+    sinon.assert.calledWith(span.setTag, 'aas.function.operation', 'add_n')
+    sinon.assert.calledWith(span.setTag, 'resource.name', 'Entity counter add_n')
+  })
+
+  it('re-applies propagated keep when the host cleared the sampled flag', () => {
+    const parent = { _traceId: 'parent' }
+    extract.returns(parent)
+
+    bindStart({
+      traceparent: '00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-00',
+      tracestate: 'dd=s:1',
+    })
+
+    sinon.assert.calledOnceWithExactly(setPriority, span, AUTO_KEEP)
+  })
+
+  it('preserves stronger propagated keep priorities', () => {
+    const parent = { _traceId: 'parent' }
+    extract.returns(parent)
+
+    bindStart({
+      traceparent: '00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-00',
+      tracestate: 'dd=s:2',
+    })
+
+    sinon.assert.calledOnceWithExactly(setPriority, span, USER_KEEP)
+  })
+
+  it('does not override sampling when the sampled flag is still set', () => {
+    const parent = { _traceId: 'parent' }
+    extract.returns(parent)
+
+    bindStart({
+      traceparent: '00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01',
+      tracestate: 'dd=s:1',
+    })
+
+    sinon.assert.notCalled(setPriority)
+  })
+
+  it('does not override sampling when propagated priority is a drop', () => {
+    const parent = { _traceId: 'parent' }
+    extract.returns(parent)
+
+    bindStart({
+      traceparent: '00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-00',
+      tracestate: 'dd=s:-1',
+    })
+
+    sinon.assert.notCalled(setPriority)
+  })
+
+  it('does not override sampling when tracestate has no datadog decision', () => {
+    const parent = { _traceId: 'parent' }
+    extract.returns(parent)
+
+    bindStart({
+      traceparent: '00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-00',
+      tracestate: 'other=vendor',
+    })
+
+    sinon.assert.notCalled(setPriority)
+  })
+
+  it('binds the started span on the invocation context', () => {
+    const ctx = bindStart()
+
+    assert.strictEqual(ctx.span, span)
+  })
+})