feat: add durable-functions integration

.github/workflows/serverless.yml

@@ -261,6 +261,25 @@ jobs:
         with:
           dd_api_key: ${{ secrets.DD_API_KEY }}
 
+  azure-durable-functions:
+    runs-on: ubuntu-latest
+    services:
+      azurite:
+        image: mcr.microsoft.com/azure-storage/azurite:3.34.0
+        ports:
+          - "127.0.0.1:10000:10000"
+          - "127.0.0.1:10001:10001"
+          - "127.0.0.1:10002:10002"
+    env:
+      PLUGINS: azure-durable-functions
+      SERVICES: azurite
+
+    steps:
+     - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
+     - uses: ./.github/actions/plugins/test
+       with:
+        dd_api_key: ${{ secrets.DD_API_KEY }}
+
   google-cloud-pubsub:
     runs-on: ubuntu-latest
     services:

docs/API.md

@@ -34,6 +34,7 @@ tracer.use('pg', {
 <h5 id="azure-event-hubs"></h5>
 <h5 id="azure-functions"></h5>
 <h5 id="azure-service-bus"></h5>
+<h5 id="azure-durable-functions"></h5>
 <h5 id="bullmq"></h5>
 <h5 id="bunyan"></h5>
 <h5 id="cassandra-driver"></h5>
@@ -112,6 +113,7 @@ tracer.use('pg', {
 * [azure-event-hubs](./interfaces/export_.plugins.azure_event_hubs.html)
 * [azure-functions](./interfaces/export_.plugins.azure_functions.html)
 * [azure-service-bus](./interfaces/export_.plugins.azure_service_bus.html)
+* [azure-durable-functions](./interfaces/export_.plugins.azure_durable_functions.html)
 * [bullmq](./interfaces/export_.plugins.bullmq.html)
 * [bunyan](./interfaces/export_.plugins.bunyan.html)
 * [cassandra-driver](./interfaces/export_.plugins.cassandra_driver.html)

index.d.ts

@@ -228,6 +228,7 @@ interface Plugins {
   "azure-event-hubs": tracer.plugins.azure_event_hubs;
   "azure-functions": tracer.plugins.azure_functions;
   "azure-service-bus": tracer.plugins.azure_service_bus;
+  "azure-durable-functions": tracer.plugins.azure_durable_functions
   "bullmq": tracer.plugins.bullmq;
   "bunyan": tracer.plugins.bunyan;
   "cassandra-driver": tracer.plugins.cassandra_driver;
@@ -2121,6 +2122,12 @@ declare namespace tracer {
      */
     interface azure_service_bus extends Integration {}
 
+    /**
+     * This plugin automatically instruments the
+     * durable-functions module
+     */
+      interface azure_durable_functions extends Integration {}
+
     /**
      * This plugin patches the [bunyan](https://github.com/trentm/node-bunyan)
      * to automatically inject trace identifiers in log records when the

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

@@ -0,0 +1,75 @@
+'use strict'
+
+const dc = require('dc-polyfill')
+const shimmer = require('../../datadog-shimmer')
+
+const {
+  addHook,
+} = require('./helpers/instrument')
+
+/**
+ * @type {import('diagnostics_channel').TracingChannel}
+ */
+const azureDurableFunctionsChannel = dc.tracingChannel('datadog:azure:durable-functions:invoke')
+
+addHook({ name: 'durable-functions', versions: ['>=3'], patchDefault: false }, (df) => {
+  const { app } = df
+
+  shimmer.wrap(app, 'entity', entityWrapper)
+  shimmer.wrap(app, 'activity', activityHandler)
+
+  return df
+})
+
+function entityWrapper (method) {
+  return function (entityName, arg) {
+    // because this method is overloaded, the second argument can either be an object
+    // with the handler or the handler itself, so first we figure which type it is
+    if (typeof arg === 'function') {
+      // if a function, this is the handler we want to wrap and trace
+      arguments[1] = shimmer.wrapFunction(arg, handler => entityHandler(handler, entityName))
+    } else {
+      // if an object, access the handler then trace it
+      shimmer.wrap(arg, 'handler', handler => entityHandler(handler, entityName))
+    }
+
+    return method.apply(this, arguments)
+  }
+}
+
+function entityHandler (handler, entityName) {
+  return function () {
+    if (!azureDurableFunctionsChannel.hasSubscribers) return handler.apply(this, arguments)
+
+    const entityContext = arguments[0]
+    return azureDurableFunctionsChannel.traceSync(
+      handler,
+      { trigger: 'Entity', functionName: entityName, operationName: entityContext?.df?.operationName },
+      this, ...arguments)
+  }
+}
+
+function activityHandler (method) {
+  return function (activityName, activityOptions) {
+    shimmer.wrap(activityOptions, 'handler', handler => {
+      const isAsync =
+        handler && handler.constructor && handler.constructor.name === 'AsyncFunction'
+
+      return function () {
+        if (!azureDurableFunctionsChannel.hasSubscribers) return handler.apply(this, arguments)
+
+        // use tracePromise if this is an async handler. otherwise, use traceSync
+        return isAsync
+          ? azureDurableFunctionsChannel.tracePromise(
+            handler,
+            { trigger: 'Activity', functionName: activityName },
+            this, ...arguments)
+          : azureDurableFunctionsChannel.traceSync(
+            handler,
+            { trigger: 'Activity', functionName: activityName },
+            this, ...arguments)
+      }
+    })
+    return method.apply(this, arguments)
+  }
+}

packages/datadog-instrumentations/src/helpers/hooks.js

@@ -8,6 +8,7 @@ module.exports = {
   '@aws-sdk/smithy-client': () => require('../aws-sdk'),
   '@azure/event-hubs': () => require('../azure-event-hubs'),
   '@azure/functions': () => require('../azure-functions'),
+  'durable-functions': () => require('../azure-durable-functions'),
   '@azure/service-bus': () => require('../azure-service-bus'),
   '@cucumber/cucumber': () => require('../cucumber'),
   '@playwright/test': () => require('../playwright'),

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

@@ -0,0 +1,49 @@
+'use strict'
+
+const TracingPlugin = require('../../dd-trace/src/plugins/tracing')
+
+class AzureDurableFunctionsPlugin extends TracingPlugin {
+  static get id () { return 'azure-durable-functions' }
+  static get operation () { return 'invoke' }
+  static get prefix () { return 'tracing:datadog:azure:durable-functions:invoke' }
+  static get type () { return 'serverless' }
+  static get kind () { return 'server' }
+
+  bindStart (ctx) {
+    const span = this.startSpan(this.operationName(), {
+      kind: 'internal',
+      type: 'serverless',
+
+      meta: {
+        component: 'azure-functions',
+        'aas.function.name': ctx.functionName,
+        'aas.function.trigger': ctx.trigger,
+        'resource.name': `${ctx.trigger} ${ctx.functionName}`,
+      },
+    }, ctx)
+
+    // in the case of entity functions, operationName should be available
+    if (ctx.operationName) {
+      span.setTag('aas.function.operation', ctx.operationName)
+      span.setTag('resource.name', `${ctx.trigger} ${ctx.functionName} ${ctx.operationName}`
+      )
+    }
+
+    ctx.span = span
+    return ctx.currentStore
+  }
+
+  end (ctx) {
+    // We only want to run finish here if this is a synchronous operation
+    // Only synchronous operations would have `result` or `error` on `end`
+    // So we skip operations that dont
+    if (!ctx.hasOwnProperty('result') && !ctx.hasOwnProperty('error')) return
+    super.finish(ctx)
+  }
+
+  asyncStart (ctx) {
+    super.finish(ctx)
+  }
+}
+
+module.exports = AzureDurableFunctionsPlugin

packages/datadog-plugin-azure-durable-functions/test/fixtures/host.json

@@ -0,0 +1,14 @@
+{
+  "version": "2.0",
+  "extensionBundle": {
+    "id": "Microsoft.Azure.Functions.ExtensionBundle",
+    "version": "[4.0.0, 4.28.0)"
+  },
+  "extensions": {
+    "durableTask": {
+      "storageProvider": {
+        "type": "AzureStorage"
+      }
+    }
+  }
+}

packages/datadog-plugin-azure-durable-functions/test/fixtures/local.settings.json

@@ -0,0 +1,8 @@
+{
+  "IsEncrypted": false,
+  "Values": {
+    "FUNCTIONS_WORKER_RUNTIME": "node",
+    "AzureWebJobsFeatureFlags": "EnableWorkerIndexing",
+    "AzureWebJobsStorage": "UseDevelopmentStorage=true"
+  }
+}

packages/datadog-plugin-azure-durable-functions/test/fixtures/package.json

@@ -0,0 +1,9 @@
+{
+  "name": "azure-durable-functions-tests",
+  "version": "1.0.0",
+  "description": "",
+  "main": "./server.mjs",
+  "scripts": {
+    "start": "func start"
+  }
+}

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

@@ -0,0 +1,145 @@
+'use strict'
+
+const assert = require('node:assert/strict')
+
+const { spawn } = require('child_process')
+const { describe, it } = require('mocha')
+const {
+  FakeAgent,
+  hookFile,
+  sandboxCwd,
+  useSandbox,
+  curlAndAssertMessage,
+  assertObjectContains,
+} = require('../../../../integration-tests/helpers')
+const { withVersions } = require('../../../dd-trace/test/setup/mocha')
+
+describe('esm', () => {
+  let agent
+  let proc
+
+  withVersions('azure-durable-functions', 'durable-functions', version => {
+    useSandbox([
+      `durable-functions@${version}`,
+      '@azure/functions',
+      'azure-functions-core-tools@4',
+    ],
+    false,
+    ['./packages/datadog-plugin-azure-durable-functions/test/integration-test/*',
+      './packages/datadog-plugin-azure-durable-functions/test/fixtures/*',
+    ])
+
+    beforeEach(async () => {
+      agent = await new FakeAgent().start()
+    })
+
+    afterEach(async () => {
+      // after each test, kill process and wait for exit before continuing
+      if (proc) {
+        proc.kill('SIGINT')
+        await new Promise(resolve => proc.on('exit', resolve))
+      }
+      await agent.stop()
+    })
+
+    it('is instrumented', async () => {
+      proc = await spawnPluginIntegrationTestProc(agent.port)
+      return await curlAndAssertMessage(agent, 'http://127.0.0.1:7071/api/httptest', ({ headers, payload }) => {
+        assert.strictEqual(headers.host, `127.0.0.1:${agent.port}`)
+        assert.ok(Array.isArray(payload))
+
+        // should expect spans for http.request, activity.hola, entity.counter.add_n, entity.counter.get_count
+        assert.strictEqual(payload.length, 4)
+
+        for (const maybeArray of payload) {
+          assert.ok(Array.isArray(maybeArray))
+        }
+
+        const [maybeHttpSpan, maybeHolaActivity, maybeAddNEntity, maybeGetCountEntity] = payload
+
+        assert.strictEqual(maybeHttpSpan.length, 2)
+        assert.strictEqual(maybeHttpSpan[0].resource, 'GET /api/httptest')
+
+        assert.strictEqual(maybeHolaActivity.length, 1)
+        assertObjectContains(maybeHolaActivity[0], {
+          name: 'azure.functions.invoke',
+          resource: 'Activity hola',
+          meta: {
+            'aas.function.trigger': 'Activity',
+            'aas.function.name': 'hola',
+          },
+        })
+
+        assert.strictEqual(maybeAddNEntity.length, 1)
+        assertObjectContains(maybeAddNEntity[0], {
+          name: 'azure.functions.invoke',
+          resource: 'Entity counter add_n',
+          meta: {
+            'aas.function.trigger': 'Entity',
+            'aas.function.name': 'counter',
+            'aas.function.operation': 'add_n',
+          },
+        })
+
+        assert.strictEqual(maybeGetCountEntity.length, 1)
+        assertObjectContains(maybeGetCountEntity[0], {
+          name: 'azure.functions.invoke',
+          resource: 'Entity counter get_count',
+          meta: {
+            'aas.function.trigger': 'Entity',
+            'aas.function.name': 'counter',
+            'aas.function.operation': 'get_count',
+          },
+        })
+      })
+    }).timeout(60_000)
+  })
+})
+
+/**
+ * - spawns process for azure func start commands
+ * - connects to azurite (running in container)
+ *    then runs the durable function locally
+ */
+async function spawnPluginIntegrationTestProc (agentPort) {
+  const cwd = sandboxCwd()
+  const env = {
+    NODE_OPTIONS: `--loader=${hookFile}`,
+    DD_TRACE_AGENT_PORT: agentPort,
+    DD_TRACE_DISABLED_PLUGINS: 'amqplib,amqp10,rhea,net',
+    PATH: `${cwd}/node_modules/azure-functions-core-tools/bin:${process.env.PATH}`,
+  }
+
+  const options = { cwd, env }
+
+  const proc = await spawnProc('func', ['start'], options)
+  return proc
+}
+
+function spawnProc (command, args, options = {}) {
+  const proc = spawn(command, args, { ...options, stdio: 'pipe' })
+  return new Promise((resolve, reject) => {
+    proc
+      .on('error', reject)
+      .on('exit', code => {
+        if (code !== 0) {
+          reject(new Error(`Process exited with status code ${code}.`))
+        }
+        resolve()
+      })
+
+    proc.stdout.on('data', data => {
+      // eslint-disable-next-line no-console
+      if (!options.silent) console.log(data.toString())
+
+      if (data.toString().includes('Host lock lease acquired by instance')) {
+        resolve(proc)
+      }
+    })
+
+    proc.stderr.on('data', data => {
+      // eslint-disable-next-line no-console
+      if (!options.silent) console.error(data.toString())
+    })
+  })
+}