Skip to content

Commit c6975fa

Browse files
ojproductionstlhunter
authored andcommitted
feat: add durable-functions integration (#7535)
* azure durable functions * debug logs * debug logs * debug logs * debug logs * add durable-functions hook * add consistent plugin naming * update supported configs * add spans * add entity instrumentation * remove debug logs * tests * tests pt2 * add df plugin interface and fix esm issues * add azure-durable-function to api.md * add ci * remove azurite proc * add envs to ci * remove azurite dep and run yarn * add asserts on span meta data * remove alreadyPatched flag * lint * use asyncStart * update supported conviguration version * use same span name as azure-functions * rename tracing channel * add hasSubscribers guard * typo * move hasSubscribers guard --------- Co-authored-by: Thomas Hunter II <tlhunter@datadog.com>
1 parent f6593d2 commit c6975fa

17 files changed

Lines changed: 412 additions & 1 deletion

File tree

.github/workflows/serverless.yml

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -261,6 +261,25 @@ jobs:
261261
with:
262262
dd_api_key: ${{ secrets.DD_API_KEY }}
263263

264+
azure-durable-functions:
265+
runs-on: ubuntu-latest
266+
services:
267+
azurite:
268+
image: mcr.microsoft.com/azure-storage/azurite:3.34.0
269+
ports:
270+
- "127.0.0.1:10000:10000"
271+
- "127.0.0.1:10001:10001"
272+
- "127.0.0.1:10002:10002"
273+
env:
274+
PLUGINS: azure-durable-functions
275+
SERVICES: azurite
276+
277+
steps:
278+
- uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
279+
- uses: ./.github/actions/plugins/test
280+
with:
281+
dd_api_key: ${{ secrets.DD_API_KEY }}
282+
264283
google-cloud-pubsub:
265284
runs-on: ubuntu-latest
266285
services:

docs/API.md

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -34,6 +34,7 @@ tracer.use('pg', {
3434
<h5 id="azure-event-hubs"></h5>
3535
<h5 id="azure-functions"></h5>
3636
<h5 id="azure-service-bus"></h5>
37+
<h5 id="azure-durable-functions"></h5>
3738
<h5 id="bullmq"></h5>
3839
<h5 id="bunyan"></h5>
3940
<h5 id="cassandra-driver"></h5>
@@ -112,6 +113,7 @@ tracer.use('pg', {
112113
* [azure-event-hubs](./interfaces/export_.plugins.azure_event_hubs.html)
113114
* [azure-functions](./interfaces/export_.plugins.azure_functions.html)
114115
* [azure-service-bus](./interfaces/export_.plugins.azure_service_bus.html)
116+
* [azure-durable-functions](./interfaces/export_.plugins.azure_durable_functions.html)
115117
* [bullmq](./interfaces/export_.plugins.bullmq.html)
116118
* [bunyan](./interfaces/export_.plugins.bunyan.html)
117119
* [cassandra-driver](./interfaces/export_.plugins.cassandra_driver.html)

index.d.ts

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -228,6 +228,7 @@ interface Plugins {
228228
"azure-event-hubs": tracer.plugins.azure_event_hubs;
229229
"azure-functions": tracer.plugins.azure_functions;
230230
"azure-service-bus": tracer.plugins.azure_service_bus;
231+
"azure-durable-functions": tracer.plugins.azure_durable_functions
231232
"bullmq": tracer.plugins.bullmq;
232233
"bunyan": tracer.plugins.bunyan;
233234
"cassandra-driver": tracer.plugins.cassandra_driver;
@@ -2121,6 +2122,12 @@ declare namespace tracer {
21212122
*/
21222123
interface azure_service_bus extends Integration {}
21232124

2125+
/**
2126+
* This plugin automatically instruments the
2127+
* durable-functions module
2128+
*/
2129+
interface azure_durable_functions extends Integration {}
2130+
21242131
/**
21252132
* This plugin patches the [bunyan](https://github.com/trentm/node-bunyan)
21262133
* to automatically inject trace identifiers in log records when the
Lines changed: 75 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,75 @@
1+
'use strict'
2+
3+
const dc = require('dc-polyfill')
4+
const shimmer = require('../../datadog-shimmer')
5+
6+
const {
7+
addHook,
8+
} = require('./helpers/instrument')
9+
10+
/**
11+
* @type {import('diagnostics_channel').TracingChannel}
12+
*/
13+
const azureDurableFunctionsChannel = dc.tracingChannel('datadog:azure:durable-functions:invoke')
14+
15+
addHook({ name: 'durable-functions', versions: ['>=3'], patchDefault: false }, (df) => {
16+
const { app } = df
17+
18+
shimmer.wrap(app, 'entity', entityWrapper)
19+
shimmer.wrap(app, 'activity', activityHandler)
20+
21+
return df
22+
})
23+
24+
function entityWrapper (method) {
25+
return function (entityName, arg) {
26+
// because this method is overloaded, the second argument can either be an object
27+
// with the handler or the handler itself, so first we figure which type it is
28+
if (typeof arg === 'function') {
29+
// if a function, this is the handler we want to wrap and trace
30+
arguments[1] = shimmer.wrapFunction(arg, handler => entityHandler(handler, entityName))
31+
} else {
32+
// if an object, access the handler then trace it
33+
shimmer.wrap(arg, 'handler', handler => entityHandler(handler, entityName))
34+
}
35+
36+
return method.apply(this, arguments)
37+
}
38+
}
39+
40+
function entityHandler (handler, entityName) {
41+
return function () {
42+
if (!azureDurableFunctionsChannel.hasSubscribers) return handler.apply(this, arguments)
43+
44+
const entityContext = arguments[0]
45+
return azureDurableFunctionsChannel.traceSync(
46+
handler,
47+
{ trigger: 'Entity', functionName: entityName, operationName: entityContext?.df?.operationName },
48+
this, ...arguments)
49+
}
50+
}
51+
52+
function activityHandler (method) {
53+
return function (activityName, activityOptions) {
54+
shimmer.wrap(activityOptions, 'handler', handler => {
55+
const isAsync =
56+
handler && handler.constructor && handler.constructor.name === 'AsyncFunction'
57+
58+
return function () {
59+
if (!azureDurableFunctionsChannel.hasSubscribers) return handler.apply(this, arguments)
60+
61+
// use tracePromise if this is an async handler. otherwise, use traceSync
62+
return isAsync
63+
? azureDurableFunctionsChannel.tracePromise(
64+
handler,
65+
{ trigger: 'Activity', functionName: activityName },
66+
this, ...arguments)
67+
: azureDurableFunctionsChannel.traceSync(
68+
handler,
69+
{ trigger: 'Activity', functionName: activityName },
70+
this, ...arguments)
71+
}
72+
})
73+
return method.apply(this, arguments)
74+
}
75+
}

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

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@ module.exports = {
88
'@aws-sdk/smithy-client': () => require('../aws-sdk'),
99
'@azure/event-hubs': () => require('../azure-event-hubs'),
1010
'@azure/functions': () => require('../azure-functions'),
11+
'durable-functions': () => require('../azure-durable-functions'),
1112
'@azure/service-bus': () => require('../azure-service-bus'),
1213
'@cucumber/cucumber': () => require('../cucumber'),
1314
'@playwright/test': () => require('../playwright'),
Lines changed: 49 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,49 @@
1+
'use strict'
2+
3+
const TracingPlugin = require('../../dd-trace/src/plugins/tracing')
4+
5+
class AzureDurableFunctionsPlugin extends TracingPlugin {
6+
static get id () { return 'azure-durable-functions' }
7+
static get operation () { return 'invoke' }
8+
static get prefix () { return 'tracing:datadog:azure:durable-functions:invoke' }
9+
static get type () { return 'serverless' }
10+
static get kind () { return 'server' }
11+
12+
bindStart (ctx) {
13+
const span = this.startSpan(this.operationName(), {
14+
kind: 'internal',
15+
type: 'serverless',
16+
17+
meta: {
18+
component: 'azure-functions',
19+
'aas.function.name': ctx.functionName,
20+
'aas.function.trigger': ctx.trigger,
21+
'resource.name': `${ctx.trigger} ${ctx.functionName}`,
22+
},
23+
}, ctx)
24+
25+
// in the case of entity functions, operationName should be available
26+
if (ctx.operationName) {
27+
span.setTag('aas.function.operation', ctx.operationName)
28+
span.setTag('resource.name', `${ctx.trigger} ${ctx.functionName} ${ctx.operationName}`
29+
)
30+
}
31+
32+
ctx.span = span
33+
return ctx.currentStore
34+
}
35+
36+
end (ctx) {
37+
// We only want to run finish here if this is a synchronous operation
38+
// Only synchronous operations would have `result` or `error` on `end`
39+
// So we skip operations that dont
40+
if (!ctx.hasOwnProperty('result') && !ctx.hasOwnProperty('error')) return
41+
super.finish(ctx)
42+
}
43+
44+
asyncStart (ctx) {
45+
super.finish(ctx)
46+
}
47+
}
48+
49+
module.exports = AzureDurableFunctionsPlugin
Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,14 @@
1+
{
2+
"version": "2.0",
3+
"extensionBundle": {
4+
"id": "Microsoft.Azure.Functions.ExtensionBundle",
5+
"version": "[4.0.0, 4.28.0)"
6+
},
7+
"extensions": {
8+
"durableTask": {
9+
"storageProvider": {
10+
"type": "AzureStorage"
11+
}
12+
}
13+
}
14+
}
Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,8 @@
1+
{
2+
"IsEncrypted": false,
3+
"Values": {
4+
"FUNCTIONS_WORKER_RUNTIME": "node",
5+
"AzureWebJobsFeatureFlags": "EnableWorkerIndexing",
6+
"AzureWebJobsStorage": "UseDevelopmentStorage=true"
7+
}
8+
}
Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,9 @@
1+
{
2+
"name": "azure-durable-functions-tests",
3+
"version": "1.0.0",
4+
"description": "",
5+
"main": "./server.mjs",
6+
"scripts": {
7+
"start": "func start"
8+
}
9+
}
Lines changed: 145 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,145 @@
1+
'use strict'
2+
3+
const assert = require('node:assert/strict')
4+
5+
const { spawn } = require('child_process')
6+
const { describe, it } = require('mocha')
7+
const {
8+
FakeAgent,
9+
hookFile,
10+
sandboxCwd,
11+
useSandbox,
12+
curlAndAssertMessage,
13+
assertObjectContains,
14+
} = require('../../../../integration-tests/helpers')
15+
const { withVersions } = require('../../../dd-trace/test/setup/mocha')
16+
17+
describe('esm', () => {
18+
let agent
19+
let proc
20+
21+
withVersions('azure-durable-functions', 'durable-functions', version => {
22+
useSandbox([
23+
`durable-functions@${version}`,
24+
'@azure/functions',
25+
'azure-functions-core-tools@4',
26+
],
27+
false,
28+
['./packages/datadog-plugin-azure-durable-functions/test/integration-test/*',
29+
'./packages/datadog-plugin-azure-durable-functions/test/fixtures/*',
30+
])
31+
32+
beforeEach(async () => {
33+
agent = await new FakeAgent().start()
34+
})
35+
36+
afterEach(async () => {
37+
// after each test, kill process and wait for exit before continuing
38+
if (proc) {
39+
proc.kill('SIGINT')
40+
await new Promise(resolve => proc.on('exit', resolve))
41+
}
42+
await agent.stop()
43+
})
44+
45+
it('is instrumented', async () => {
46+
proc = await spawnPluginIntegrationTestProc(agent.port)
47+
return await curlAndAssertMessage(agent, 'http://127.0.0.1:7071/api/httptest', ({ headers, payload }) => {
48+
assert.strictEqual(headers.host, `127.0.0.1:${agent.port}`)
49+
assert.ok(Array.isArray(payload))
50+
51+
// should expect spans for http.request, activity.hola, entity.counter.add_n, entity.counter.get_count
52+
assert.strictEqual(payload.length, 4)
53+
54+
for (const maybeArray of payload) {
55+
assert.ok(Array.isArray(maybeArray))
56+
}
57+
58+
const [maybeHttpSpan, maybeHolaActivity, maybeAddNEntity, maybeGetCountEntity] = payload
59+
60+
assert.strictEqual(maybeHttpSpan.length, 2)
61+
assert.strictEqual(maybeHttpSpan[0].resource, 'GET /api/httptest')
62+
63+
assert.strictEqual(maybeHolaActivity.length, 1)
64+
assertObjectContains(maybeHolaActivity[0], {
65+
name: 'azure.functions.invoke',
66+
resource: 'Activity hola',
67+
meta: {
68+
'aas.function.trigger': 'Activity',
69+
'aas.function.name': 'hola',
70+
},
71+
})
72+
73+
assert.strictEqual(maybeAddNEntity.length, 1)
74+
assertObjectContains(maybeAddNEntity[0], {
75+
name: 'azure.functions.invoke',
76+
resource: 'Entity counter add_n',
77+
meta: {
78+
'aas.function.trigger': 'Entity',
79+
'aas.function.name': 'counter',
80+
'aas.function.operation': 'add_n',
81+
},
82+
})
83+
84+
assert.strictEqual(maybeGetCountEntity.length, 1)
85+
assertObjectContains(maybeGetCountEntity[0], {
86+
name: 'azure.functions.invoke',
87+
resource: 'Entity counter get_count',
88+
meta: {
89+
'aas.function.trigger': 'Entity',
90+
'aas.function.name': 'counter',
91+
'aas.function.operation': 'get_count',
92+
},
93+
})
94+
})
95+
}).timeout(60_000)
96+
})
97+
})
98+
99+
/**
100+
* - spawns process for azure func start commands
101+
* - connects to azurite (running in container)
102+
* then runs the durable function locally
103+
*/
104+
async function spawnPluginIntegrationTestProc (agentPort) {
105+
const cwd = sandboxCwd()
106+
const env = {
107+
NODE_OPTIONS: `--loader=${hookFile}`,
108+
DD_TRACE_AGENT_PORT: agentPort,
109+
DD_TRACE_DISABLED_PLUGINS: 'amqplib,amqp10,rhea,net',
110+
PATH: `${cwd}/node_modules/azure-functions-core-tools/bin:${process.env.PATH}`,
111+
}
112+
113+
const options = { cwd, env }
114+
115+
const proc = await spawnProc('func', ['start'], options)
116+
return proc
117+
}
118+
119+
function spawnProc (command, args, options = {}) {
120+
const proc = spawn(command, args, { ...options, stdio: 'pipe' })
121+
return new Promise((resolve, reject) => {
122+
proc
123+
.on('error', reject)
124+
.on('exit', code => {
125+
if (code !== 0) {
126+
reject(new Error(`Process exited with status code ${code}.`))
127+
}
128+
resolve()
129+
})
130+
131+
proc.stdout.on('data', data => {
132+
// eslint-disable-next-line no-console
133+
if (!options.silent) console.log(data.toString())
134+
135+
if (data.toString().includes('Host lock lease acquired by instance')) {
136+
resolve(proc)
137+
}
138+
})
139+
140+
proc.stderr.on('data', data => {
141+
// eslint-disable-next-line no-console
142+
if (!options.silent) console.error(data.toString())
143+
})
144+
})
145+
}

0 commit comments

Comments
 (0)