feat(otel): connect durable orchestration spans across worker processes

Persist one orchestration span identity per instance in a shared store so
activities on any worker parent correctly, seed metadata from the HTTP
span at startNew, and export a single orchestration span when the instance
completes instead of one span per replay turn.

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: chemystery09

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

@@ -25,6 +25,18 @@ addHook({ name: 'durable-functions', versions: ['>=3'], patchDefault: false }, (
   return df
 })
 
+addHook({
+  name: 'durable-functions',
+  versions: ['>=3'],
+  file: 'lib/src/durableClient/DurableClient.js',
+}, (durableClientModule) => {
+  if (require('./helpers/otel-azure-enabled').isOtelAzureInstrumentationEnabled()) {
+    require('./helpers/otel-orchestration-http-link').patchDurableClient(durableClientModule.DurableClient)
+  }
+
+  return durableClientModule
+})
+
 function entityWrapper (method) {
   return function (entityName, arg) {
     // because this method is overloaded, the second argument can either be an object

packages/datadog-instrumentations/src/helpers/azure-trace-context.js

@@ -23,6 +23,81 @@ function extractContext (traceContext) {
   return api.propagation.extract(ROOT_CONTEXT, carrier, api.defaultTextMapGetter)
 }
 
+function getInstanceId (invocationContext) {
+  const attributes = invocationContext?.traceContext?.attributes
+  if (!attributes) return
+
+  return attributes['durabletask.task.instance_id'] || attributes.DurableFunctionsInstanceId
+}
+
+function parentContextFromOrchestrationMeta (meta) {
+  const { traceContextFromMeta } = require('./otel-orchestration-meta')
+  return extractContext(traceContextFromMeta(meta))
+}
+
+function resolveActivityParentContext (invocationContext) {
+  const instanceId = getInstanceId(invocationContext)
+  const traceContext = invocationContext?.traceContext
+
+  const { getOrchestrationSpan } = require('./otel-orchestration-registry')
+  const orchestrationSpan = getOrchestrationSpan(instanceId)
+  if (orchestrationSpan) {
+    return api.trace.setSpan(extractContext(traceContext), orchestrationSpan)
+  }
+
+  const { readOrchestrationSpanMetaSync } = require('./otel-orchestration-store')
+  const meta = readOrchestrationSpanMetaSync(instanceId, traceContext)
+  if (meta) {
+    return parentContextFromOrchestrationMeta(meta)
+  }
+
+  return extractContext(traceContext)
+}
+
+async function resolveActivityParentContextAsync (invocationContext) {
+  const instanceId = getInstanceId(invocationContext)
+  const traceContext = invocationContext?.traceContext
+
+  const { getOrchestrationSpan } = require('./otel-orchestration-registry')
+  const orchestrationSpan = getOrchestrationSpan(instanceId)
+  if (orchestrationSpan) {
+    return api.trace.setSpan(extractContext(traceContext), orchestrationSpan)
+  }
+
+  const {
+    readOrchestrationSpanMetaAsync,
+    readOrchestrationSpanMetaSync,
+  } = require('./otel-orchestration-store')
+
+  let meta = readOrchestrationSpanMetaSync(instanceId, traceContext)
+  if (!meta) {
+    meta = await readOrchestrationSpanMetaAsync(instanceId, traceContext)
+  }
+  if (meta) {
+    return parentContextFromOrchestrationMeta(meta)
+  }
+
+  return extractContext(traceContext)
+}
+
+function buildSpanParentContext (args, trigger) {
+  const invocationContext = getInvocationContext(args, trigger)
+  if (trigger === 'durable-activity') {
+    return resolveActivityParentContext(invocationContext)
+  }
+
+  return extractContext(invocationContext?.traceContext)
+}
+
+function buildSpanParentContextAsync (args, trigger) {
+  const invocationContext = getInvocationContext(args, trigger)
+  if (trigger === 'durable-activity') {
+    return resolveActivityParentContextAsync(invocationContext)
+  }
+
+  return Promise.resolve(extractContext(invocationContext?.traceContext))
+}
+
 function runWithTraceContext (traceContext, fn) {
   return api.context.with(extractContext(traceContext), fn)
 }
@@ -52,9 +127,14 @@ function runWithInvocationContext (args, trigger, fn) {
 }
 
 module.exports = {
+  buildSpanParentContext,
+  buildSpanParentContextAsync,
   carrierFromTraceContext,
   extractContext,
+  getInstanceId,
   getInvocationContext,
+  parentContextFromOrchestrationMeta,
+  resolveActivityParentContext,
   runWithInvocationContext,
   runWithTraceContext,
 }

packages/datadog-instrumentations/src/helpers/otel-azure-span.js

@@ -25,13 +25,26 @@ function spanAttributes (functionName, trigger, operationName) {
   }
 }
 
+function startChildSpan (tracerName, trigger, functionName, operationName, args, parentContext) {
+  return getTracer(tracerName).startSpan(
+    `${trigger} ${functionName}`,
+    { attributes: spanAttributes(functionName, trigger, operationName) },
+    parentContext,
+  )
+}
+
 function wrapSyncWithTraceContext (tracerName, trigger, handler, functionName, operationName) {
   return function (...args) {
-    const { runWithInvocationContext } = require('./azure-trace-context')
+    const { runWithInvocationContext, buildSpanParentContext } = require('./azure-trace-context')
     return runWithInvocationContext(args, trigger, () => {
-      const span = getTracer(tracerName).startSpan(`${trigger} ${functionName}`, {
-        attributes: spanAttributes(functionName, trigger, operationName),
-      })
+      const span = startChildSpan(
+        tracerName,
+        trigger,
+        functionName,
+        operationName,
+        args,
+        buildSpanParentContext(args, trigger),
+      )
       try {
         const result = handler.apply(this, args)
         endSpan(span)
@@ -46,11 +59,17 @@ function wrapSyncWithTraceContext (tracerName, trigger, handler, functionName, o
 
 function wrapAsyncWithTraceContext (tracerName, trigger, handler, functionName) {
   return function (...args) {
-    const { runWithInvocationContext } = require('./azure-trace-context')
-    return runWithInvocationContext(args, trigger, () =>
-      getTracer(tracerName).startActiveSpan(
+    const {
+      runWithInvocationContext,
+      buildSpanParentContextAsync,
+    } = require('./azure-trace-context')
+
+    return runWithInvocationContext(args, trigger, async () => {
+      const parentContext = await buildSpanParentContextAsync(args, trigger)
+      return getTracer(tracerName).startActiveSpan(
         `${trigger} ${functionName}`,
         { attributes: spanAttributes(functionName, trigger) },
+        parentContext,
         async (span) => {
           try {
             const result = await handler.apply(this, args)
@@ -61,7 +80,8 @@ function wrapAsyncWithTraceContext (tracerName, trigger, handler, functionName)
             throw error
           }
         },
-      ))
+      )
+    })
   }
 }
 

packages/datadog-instrumentations/src/helpers/otel-orchestration-export.js

@@ -0,0 +1,115 @@
+'use strict'
+
+const api = require('@opentelemetry/api')
+const createId = require('../../../dd-trace/src/id')
+const DatadogSpanContext = require('../../../dd-trace/src/opentracing/span_context')
+const OtelSpan = require('../../../dd-trace/src/opentelemetry/span')
+const OtelSpanContext = require('../../../dd-trace/src/opentelemetry/span_context')
+const { extractContext } = require('./azure-trace-context')
+const { getTracer, spanAttributes, endSpan } = require('./otel-azure-span')
+const { normalizeSpanId, normalizeTraceId } = require('./otel-orchestration-meta')
+const { resolveHttpParentForOrchestration } = require('./otel-orchestration-http-link')
+
+function getParentFromTraceContext (traceContext) {
+  const traceParent = traceContext?.traceParent
+  if (!traceParent) return
+
+  const parts = traceParent.split('-')
+  if (parts.length < 4) return
+
+  return {
+    traceId: normalizeTraceId(parts[1]),
+    parentId: normalizeSpanId(parts[2]),
+  }
+}
+
+function createOrchestrationMeta (instanceId, invocationContext, functionName) {
+  const traceContext = invocationContext?.traceContext
+  const httpParent = resolveHttpParentForOrchestration(instanceId, traceContext)
+  const fromHeader = getParentFromTraceContext(traceContext)
+
+  let traceId = httpParent?.traceId ?? fromHeader?.traceId
+  let parentId = httpParent?.spanId ?? fromHeader?.parentId
+
+  if (!traceId) {
+    const parentContext = extractContext(traceContext)
+    const parentSpan = api.trace.getSpan(parentContext)
+    const parentDdContext = parentSpan?.spanContext()?._ddContext
+
+    if (parentDdContext) {
+      traceId = normalizeTraceId(parentDdContext._traceId)
+      parentId ??= normalizeSpanId(parentDdContext._spanId)
+    }
+  }
+
+  if (!traceId) {
+    traceId = normalizeTraceId(createId())
+  }
+
+  return {
+    instanceId,
+    functionName,
+    traceId,
+    spanId: normalizeSpanId(createId()),
+    parentId,
+    startTime: Date.now(),
+    status: 'open',
+  }
+}
+
+// Build orchestration metadata from the HTTP span that called `startNew`. The
+// orchestration runs later, often in another worker process, so its identity has
+// to be decided here while the HTTP span is still known.
+function createOrchestrationMetaFromHttpParent (instanceId, httpParent, functionName) {
+  if (!httpParent?.traceId || !httpParent.spanId) return
+
+  return {
+    instanceId,
+    functionName,
+    traceId: normalizeTraceId(httpParent.traceId),
+    spanId: normalizeSpanId(createId()),
+    parentId: normalizeSpanId(httpParent.spanId),
+    httpParentSpanId: normalizeSpanId(httpParent.spanId),
+    startTime: Date.now(),
+    // Replaced with the real start time on the first orchestration turn.
+    pendingStart: true,
+    status: 'open',
+  }
+}
+
+function exportOrchestrationSpanFromMeta (tracerName, meta, { error, endTime } = {}) {
+  if (!meta?.traceId || !meta?.spanId) return false
+
+  const tracer = getTracer(tracerName)
+  const ddContext = new DatadogSpanContext({
+    traceId: createId(meta.traceId, 16),
+    spanId: createId(meta.spanId, 16),
+    parentId: meta.parentId ? createId(meta.parentId, 16) : null,
+  })
+
+  const span = new OtelSpan(
+    tracer,
+    api.ROOT_CONTEXT,
+    `orchestration ${meta.functionName || 'orchestration'}`,
+    new OtelSpanContext(ddContext),
+    api.SpanKind.INTERNAL,
+    [],
+    meta.startTime,
+    spanAttributes(meta.functionName || 'orchestration', 'durable-orchestration'),
+  )
+
+  if (error) {
+    endSpan(span, error)
+  } else {
+    span.end(endTime ?? Date.now())
+  }
+
+  return true
+}
+
+module.exports = {
+  createOrchestrationMeta,
+  createOrchestrationMetaFromHttpParent,
+  exportOrchestrationSpanFromMeta,
+  getParentFromTraceContext,
+}