Skip to content

Commit edde8b5

Browse files
rochdevrochdev
authored
remove async resource usage from dns/fs/net integrations (#5673)
* remove async resource usage from net integration * remove async resource usage from dns integration * code cleanup * fix fs and profiling
1 parent 56b04d0 commit edde8b5

File tree

17 files changed

+169
-139
lines changed

17 files changed

+169
-139
lines changed

‎packages/datadog-instrumentations/src/dns.js

+16-14
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
'use strict'
22

3-
const { channel, addHook, AsyncResource } = require('./helpers/instrument')
3+
const { channel, addHook } = require('./helpers/instrument')
44
const shimmer = require('../../datadog-shimmer')
55

66
const rrtypes = {
@@ -53,7 +53,7 @@ function wrap (prefix, fn, expectedArgs, rrtype) {
5353
const errorCh = channel(prefix + ':error')
5454

5555
const wrapped = function () {
56-
const cb = AsyncResource.bind(arguments[arguments.length - 1])
56+
const cb = arguments[arguments.length - 1]
5757
if (
5858
!startCh.hasSubscribers ||
5959
arguments.length < expectedArgs ||
@@ -62,30 +62,32 @@ function wrap (prefix, fn, expectedArgs, rrtype) {
6262
return fn.apply(this, arguments)
6363
}
6464

65-
const startArgs = Array.from(arguments)
66-
startArgs.pop() // gets rid of the callback
65+
const args = Array.from(arguments)
66+
args.pop() // gets rid of the callback
6767
if (rrtype) {
68-
startArgs.push(rrtype)
68+
args.push(rrtype)
6969
}
7070

71-
const asyncResource = new AsyncResource('bound-anonymous-fn')
72-
return asyncResource.runInAsyncScope(() => {
73-
startCh.publish(startArgs)
71+
const ctx = { args }
7472

75-
arguments[arguments.length - 1] = shimmer.wrapFunction(cb, cb => asyncResource.bind(function (error, result) {
73+
return startCh.runStores(ctx, () => {
74+
arguments[arguments.length - 1] = shimmer.wrapFunction(cb, cb => function (error, result, ...args) {
7675
if (error) {
77-
errorCh.publish(error)
76+
ctx.error = error
77+
errorCh.publish(ctx)
7878
}
79-
finishCh.publish(result)
80-
cb.apply(this, arguments)
81-
}))
79+
80+
ctx.result = result
81+
finishCh.runStores(ctx, cb, this, error, result, ...args)
82+
})
8283

8384
try {
8485
return fn.apply(this, arguments)
8586
// TODO deal with promise versions when we support `dns/promises`
8687
} catch (error) {
8788
error.stack // trigger getting the stack at the original throwing point
88-
errorCh.publish(error)
89+
ctx.error = error
90+
errorCh.publish(ctx)
8991

9092
throw error
9193
}

‎packages/datadog-instrumentations/src/fs.js

+37-46
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,6 @@
11
'use strict'
22

3-
const {
4-
channel,
5-
addHook,
6-
AsyncResource
7-
} = require('./helpers/instrument')
3+
const { channel, addHook } = require('./helpers/instrument')
84
const shimmer = require('../../datadog-shimmer')
95

106
const startChannel = channel('apm:fs:operation:start')
@@ -191,25 +187,23 @@ function wrapCreateStream (original) {
191187
return function (path, options) {
192188
if (!startChannel.hasSubscribers) return original.apply(this, arguments)
193189

194-
const innerResource = new AsyncResource('bound-anonymous-fn')
195-
const message = getMessage(name, ['path', 'options'], arguments)
196-
197-
return innerResource.runInAsyncScope(() => {
198-
startChannel.publish(message)
190+
const ctx = getMessage(name, ['path', 'options'], arguments)
199191

192+
return startChannel.runStores(ctx, () => {
200193
try {
201194
const stream = original.apply(this, arguments)
202-
const onError = innerResource.bind(error => {
203-
errorChannel.publish(error)
195+
const onError = error => {
196+
ctx.error = error
197+
errorChannel.publish(ctx)
204198
onFinish()
205-
})
206-
const onFinish = innerResource.bind(() => {
207-
finishChannel.publish()
199+
}
200+
const onFinish = () => {
201+
finishChannel.runStores(ctx, () => {})
208202
stream.off('close', onFinish)
209203
stream.off('end', onFinish)
210204
stream.off('finish', onFinish)
211205
stream.off('error', onError)
212-
})
206+
}
213207

214208
stream.once('close', onFinish)
215209
stream.once('end', onFinish)
@@ -218,8 +212,9 @@ function wrapCreateStream (original) {
218212

219213
return stream
220214
} catch (error) {
221-
errorChannel.publish(error)
222-
finishChannel.publish()
215+
ctx.error = error
216+
errorChannel.publish(ctx)
217+
finishChannel.runStores(ctx, () => {})
223218
}
224219
})
225220
}
@@ -239,17 +234,16 @@ function createWatchWrapFunction (override = '') {
239234
const operation = name
240235
return function () {
241236
if (!startChannel.hasSubscribers) return original.apply(this, arguments)
242-
const message = getMessage(method, watchMethods[operation], arguments, this)
243-
const innerResource = new AsyncResource('bound-anonymous-fn')
244-
return innerResource.runInAsyncScope(() => {
245-
startChannel.publish(message)
237+
const ctx = getMessage(method, watchMethods[operation], arguments, this)
238+
return startChannel.runStores(ctx, () => {
246239
try {
247240
const result = original.apply(this, arguments)
248-
finishChannel.publish()
241+
finishChannel.runStores(ctx, () => {})
249242
return result
250243
} catch (error) {
251-
errorChannel.publish(error)
252-
finishChannel.publish()
244+
ctx.error = error
245+
errorChannel.publish(ctx)
246+
finishChannel.runStores(ctx, () => {})
253247
throw error
254248
}
255249
})
@@ -268,30 +262,25 @@ function createWrapFunction (prefix = '', override = '') {
268262

269263
const lastIndex = arguments.length - 1
270264
const cb = typeof arguments[lastIndex] === 'function' && arguments[lastIndex]
271-
const innerResource = new AsyncResource('bound-anonymous-fn')
272265
const params = getMethodParamsRelationByPrefix(prefix)[operation]
273266
const abortController = new AbortController()
274-
const message = { ...getMessage(method, params, arguments, this), abortController }
267+
const ctx = { ...getMessage(method, params, arguments, this), abortController }
275268

276-
const finish = innerResource.bind(function (error) {
269+
const finish = function (error, cb = () => {}) {
277270
if (error !== null && typeof error === 'object') { // fs.exists receives a boolean
278-
errorChannel.publish(error)
271+
ctx.error = error
272+
errorChannel.publish(ctx)
279273
}
280-
finishChannel.publish()
281-
})
274+
return finishChannel.runStores(ctx, cb)
275+
}
282276

283277
if (cb) {
284-
const outerResource = new AsyncResource('bound-anonymous-fn')
285-
286-
arguments[lastIndex] = shimmer.wrapFunction(cb, cb => innerResource.bind(function (e) {
287-
finish(e)
288-
return outerResource.runInAsyncScope(() => cb.apply(this, arguments))
289-
}))
278+
arguments[lastIndex] = shimmer.wrapFunction(cb, cb => function (e) {
279+
return finish(e, () => cb.apply(this, arguments))
280+
})
290281
}
291282

292-
return innerResource.runInAsyncScope(() => {
293-
startChannel.publish(message)
294-
283+
return startChannel.runStores(ctx, () => {
295284
if (abortController.signal.aborted) {
296285
const error = abortController.signal.reason || new Error('Aborted')
297286

@@ -318,23 +307,25 @@ function createWrapFunction (prefix = '', override = '') {
318307
if (isFirstMethodReturningFileHandle(original)) {
319308
wrapFileHandle(value)
320309
}
321-
finishChannel.publish()
310+
finishChannel.runStores(ctx, () => {})
322311
return value
323312
},
324313
error => {
325-
errorChannel.publish(error)
326-
finishChannel.publish()
314+
ctx.error = error
315+
errorChannel.publish(ctx)
316+
finishChannel.runStores(ctx, () => {})
327317
throw error
328318
}
329319
)
330320
}
331321

332-
finishChannel.publish()
322+
finishChannel.runStores(ctx, () => {})
333323

334324
return result
335325
} catch (error) {
336-
errorChannel.publish(error)
337-
finishChannel.publish()
326+
ctx.error = error
327+
errorChannel.publish(ctx)
328+
finishChannel.runStores(ctx, () => {})
338329
throw error
339330
}
340331
})

‎packages/datadog-instrumentations/src/net.js

+24-28
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,6 @@
11
'use strict'
22

3-
const {
4-
channel,
5-
addHook,
6-
AsyncResource
7-
} = require('./helpers/instrument')
3+
const { channel, addHook } = require('./helpers/instrument')
84
const shimmer = require('../../datadog-shimmer')
95

106
const startICPCh = channel('apm:net:ipc:start')
@@ -15,6 +11,7 @@ const startTCPCh = channel('apm:net:tcp:start')
1511
const finishTCPCh = channel('apm:net:tcp:finish')
1612
const errorTCPCh = channel('apm:net:tcp:error')
1713

14+
const readyCh = channel('apm:net:tcp:ready')
1815
const connectionCh = channel('apm:net:tcp:connection')
1916

2017
const names = ['net', 'node:net']
@@ -39,30 +36,27 @@ addHook({ name: names }, (net, version, name) => {
3936

4037
if (!options) return connect.apply(this, arguments)
4138

42-
const callbackResource = new AsyncResource('bound-anonymous-fn')
43-
const asyncResource = new AsyncResource('bound-anonymous-fn')
39+
const protocol = options.path ? 'ipc' : 'tcp'
40+
const startCh = protocol === 'ipc' ? startICPCh : startTCPCh
41+
const finishCh = protocol === 'ipc' ? finishICPCh : finishTCPCh
42+
const errorCh = protocol === 'ipc' ? errorICPCh : errorTCPCh
43+
const ctx = { options }
4444

4545
if (typeof callback === 'function') {
46-
arguments[lastIndex] = callbackResource.bind(callback)
46+
arguments[lastIndex] = function (...args) {
47+
return finishCh.runStores(ctx, callback, this, ...args)
48+
}
4749
}
4850

49-
const protocol = options.path ? 'ipc' : 'tcp'
50-
51-
return asyncResource.runInAsyncScope(() => {
52-
if (protocol === 'ipc') {
53-
startICPCh.publish({ options })
54-
setupListeners(this, 'ipc', asyncResource)
55-
} else {
56-
startTCPCh.publish({ options })
57-
setupListeners(this, 'tcp', asyncResource)
58-
}
51+
return startCh.runStores(ctx, () => {
52+
setupListeners(this, protocol, ctx, finishCh, errorCh)
5953

6054
const emit = this.emit
6155
this.emit = shimmer.wrapFunction(emit, emit => function (eventName) {
6256
switch (eventName) {
6357
case 'ready':
6458
case 'connect':
65-
return callbackResource.runInAsyncScope(() => {
59+
return readyCh.runStores(ctx, () => {
6660
return emit.apply(this, arguments)
6761
})
6862
default:
@@ -73,7 +67,7 @@ addHook({ name: names }, (net, version, name) => {
7367
try {
7468
return connect.apply(this, arguments)
7569
} catch (err) {
76-
protocol === 'ipc' ? errorICPCh.publish(err) : errorTCPCh.publish(err)
70+
errorCh.publish(err)
7771

7872
throw err
7973
}
@@ -104,19 +98,21 @@ function getOptions (args) {
10498
}
10599
}
106100

107-
function setupListeners (socket, protocol, asyncResource) {
101+
function setupListeners (socket, protocol, ctx, finishCh, errorCh) {
108102
const events = ['connect', 'error', 'close', 'timeout']
109103

110-
const wrapListener = asyncResource.bind(function (error) {
104+
const wrapListener = function (error) {
111105
if (error) {
112-
protocol === 'ipc' ? errorICPCh.publish(error) : errorTCPCh.publish(error)
106+
ctx.error = error
107+
errorCh.publish(ctx)
113108
}
114-
protocol === 'ipc' ? finishICPCh.publish(undefined) : finishTCPCh.publish(undefined)
115-
})
109+
finishCh.runStores(ctx, () => {})
110+
}
116111

117-
const localListener = asyncResource.bind(function () {
118-
connectionCh.publish({ socket })
119-
})
112+
const localListener = function () {
113+
ctx.socket = socket
114+
connectionCh.publish(ctx)
115+
}
120116

121117
const cleanupListener = function () {
122118
socket.removeListener('connect', localListener)

‎packages/datadog-plugin-dns/src/lookup.js

+10-5
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,9 @@ class DNSLookupPlugin extends ClientPlugin {
66
static get id () { return 'dns' }
77
static get operation () { return 'lookup' }
88

9-
start ([hostname]) {
9+
bindStart (ctx) {
10+
const [hostname] = ctx.args
11+
1012
this.startSpan('dns.lookup', {
1113
service: this.config.service,
1214
resource: hostname,
@@ -16,11 +18,14 @@ class DNSLookupPlugin extends ClientPlugin {
1618
'dns.address': '',
1719
'dns.addresses': ''
1820
}
19-
})
21+
}, ctx)
22+
23+
return ctx.currentStore
2024
}
2125

22-
finish (result) {
23-
const span = this.activeSpan
26+
bindFinish (ctx) {
27+
const span = ctx.currentStore.span
28+
const result = ctx.result
2429

2530
if (Array.isArray(result)) {
2631
const addresses = Array.isArray(result)
@@ -33,7 +38,7 @@ class DNSLookupPlugin extends ClientPlugin {
3338
span.setTag('dns.address', result)
3439
}
3540

36-
super.finish()
41+
return ctx.parentStore
3742
}
3843
}
3944

‎packages/datadog-plugin-dns/src/lookup_service.js

+6-2
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,9 @@ class DNSLookupServicePlugin extends ClientPlugin {
66
static get id () { return 'dns' }
77
static get operation () { return 'lookup_service' }
88

9-
start ([address, port]) {
9+
bindStart (ctx) {
10+
const [address, port] = ctx.args
11+
1012
this.startSpan('dns.lookup_service', {
1113
service: this.config.service,
1214
resource: `${address}:${port}`,
@@ -17,7 +19,9 @@ class DNSLookupServicePlugin extends ClientPlugin {
1719
metrics: {
1820
'dns.port': port
1921
}
20-
})
22+
}, ctx)
23+
24+
return ctx.currentStore
2125
}
2226
}
2327

‎packages/datadog-plugin-dns/src/resolve.js

+5-2
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,8 @@ class DNSResolvePlugin extends ClientPlugin {
66
static get id () { return 'dns' }
77
static get operation () { return 'resolve' }
88

9-
start ([hostname, maybeType]) {
9+
bindStart (ctx) {
10+
const [hostname, maybeType] = ctx.args
1011
const rrtype = typeof maybeType === 'string' ? maybeType : 'A'
1112

1213
this.startSpan('dns.resolve', {
@@ -17,7 +18,9 @@ class DNSResolvePlugin extends ClientPlugin {
1718
'dns.hostname': hostname,
1819
'dns.rrtype': rrtype
1920
}
20-
})
21+
}, ctx)
22+
23+
return ctx.currentStore
2124
}
2225
}
2326

0 commit comments

Comments
 (0)