Use errorMonitor symbol instead of listening to the error event (#5682)

BridgeAR · web-flow · commit 22d38aaf5d20 · 2025-05-19T10:46:50.000+02:00
This is important to guarantee user code behaves as it did without
the instrumentation. So user crashes should happen as before.
diff --git a/packages/datadog-instrumentations/src/child_process.js b/packages/datadog-instrumentations/src/child_process.js
@@ -1,5 +1,6 @@
 'use strict'
 
+const { errorMonitor } = require('events')
 const util = require('util')
 
 const {
@@ -227,7 +228,7 @@ function wrapChildProcessAsyncMethod (ChildProcess, shell = false) {
         if (childProcess) {
           let errorExecuted = false
 
-          childProcess.on('error', (e) => {
+          childProcess.on(errorMonitor, (e) => {
             errorExecuted = true
             childProcessChannel.error.publish(e)
           })
diff --git a/packages/datadog-instrumentations/src/couchbase.js b/packages/datadog-instrumentations/src/couchbase.js
@@ -1,5 +1,6 @@
 'use strict'
 
+const { errorMonitor } = require('events')
 const {
   channel,
   addHook,
@@ -186,7 +187,7 @@ addHook({ name: 'couchbase', file: 'lib/bucket.js', versions: ['^2.6.12'] }, Buc
         finishCh.publish(undefined)
       }))
 
-      emitter.once('error', asyncResource.bind((error) => {
+      emitter.once(errorMonitor, asyncResource.bind((error) => {
         errorCh.publish(error)
         finishCh.publish(undefined)
       }))
diff --git a/packages/datadog-instrumentations/src/fs.js b/packages/datadog-instrumentations/src/fs.js
@@ -1,5 +1,6 @@
 'use strict'
 
+const { errorMonitor } = require('events')
 const { channel, addHook } = require('./helpers/instrument')
 const shimmer = require('../../datadog-shimmer')
 
@@ -199,16 +200,16 @@ function wrapCreateStream (original) {
         }
         const onFinish = () => {
           finishChannel.runStores(ctx, () => {})
-          stream.off('close', onFinish)
-          stream.off('end', onFinish)
-          stream.off('finish', onFinish)
-          stream.off('error', onError)
+          stream.removeListener('close', onFinish)
+          stream.removeListener('end', onFinish)
+          stream.removeListener('finish', onFinish)
+          stream.removeListener(errorMonitor, onError)
         }
 
         stream.once('close', onFinish)
         stream.once('end', onFinish)
         stream.once('finish', onFinish)
-        stream.once('error', onError)
+        stream.once(errorMonitor, onError)
 
         return stream
       } catch (error) {
diff --git a/packages/datadog-instrumentations/src/http/client.js b/packages/datadog-instrumentations/src/http/client.js
@@ -3,6 +3,7 @@
 /* eslint-disable no-fallthrough */
 
 const url = require('url')
+const { errorMonitor } = require('events')
 const { channel, addHook } = require('../helpers/instrument')
 const shimmer = require('../../../datadog-shimmer')
 
@@ -88,7 +89,7 @@ function patch (http, methodName) {
                 const res = arg
                 ctx.res = res
                 res.on('end', finish)
-                res.on('error', finish)
+                res.on(errorMonitor, finish)
                 break
               }
               case 'connect':
diff --git a/packages/datadog-instrumentations/src/mysql2.js b/packages/datadog-instrumentations/src/mysql2.js
@@ -1,5 +1,7 @@
 'use strict'
 
+const { errorMonitor } = require('node:events')
+
 const {
   channel,
   addHook,
@@ -138,7 +140,7 @@ function wrapConnection (Connection, version) {
           onResult.apply(this, arguments)
         }, 'bound-anonymous-fn', this))
       } else {
-        this.on('error', asyncResource.bind(error => errorCh.publish(error)))
+        this.on(errorMonitor, asyncResource.bind(error => errorCh.publish(error)))
         this.on('end', asyncResource.bind(() => finishCh.publish(undefined)))
       }
 
diff --git a/packages/datadog-instrumentations/src/net.js b/packages/datadog-instrumentations/src/net.js
@@ -1,5 +1,7 @@
 'use strict'
 
+const { errorMonitor } = require('events')
+
 const { channel, addHook } = require('./helpers/instrument')
 const shimmer = require('../../datadog-shimmer')
 
@@ -99,7 +101,7 @@ function getOptions (args) {
 }
 
 function setupListeners (socket, protocol, ctx, finishCh, errorCh) {
-  const events = ['connect', 'error', 'close', 'timeout']
+  const events = ['connect', errorMonitor, 'close', 'timeout']
 
   const wrapListener = function (error) {
     if (error) {
diff --git a/packages/datadog-instrumentations/src/pg.js b/packages/datadog-instrumentations/src/pg.js
@@ -129,21 +129,15 @@ function wrapQuery (query) {
         }
       } else if (newQuery.once) {
         newQuery
-          .once('error', finish)
+          .once(errorMonitor, finish)
           .once('end', (res) => finish(null, res))
       } else {
+        // TODO: This code is never reached in our tests.
+        // Internally, pg always uses callbacks or streams, even for promise based queries.
+        // Investigate if this code should just be removed.
         newQuery.then((res) => finish(null, res), finish)
       }
 
-      if (stream) {
-        newQuery.on('end', () => {
-          finish(null, [])
-        })
-        newQuery.on(errorMonitor, (err) => {
-          finish(err)
-        })
-      }
-
       try {
         return retval
       } catch (err) {
diff --git a/packages/datadog-instrumentations/test/child_process.spec.js b/packages/datadog-instrumentations/test/child_process.spec.js
@@ -102,6 +102,10 @@ describe('child process', () => {
               it('should execute error callback', (done) => {
                 const childEmitter = childProcess[methodName]('invalid_command_test')
 
+                expect(childEmitter.listenerCount('error')).to.equal(methodName.includes('spawn') ? 0 : 1)
+
+                childEmitter.once('error', () => {})
+
                 childEmitter.once('close', () => {
                   expect(start).to.have.been.calledOnceWith({
                     command: 'invalid_command_test',
diff --git a/packages/datadog-instrumentations/test/mysql2.spec.js b/packages/datadog-instrumentations/test/mysql2.spec.js
@@ -4,6 +4,7 @@ const { channel } = require('../src/helpers/instrument')
 const agent = require('../../dd-trace/test/plugins/agent')
 const { assert } = require('chai')
 const semver = require('semver')
+const { once } = require('events')
 
 describe('mysql2 instrumentation', () => {
   withVersions('mysql2', 'mysql2', version => {
@@ -435,16 +436,17 @@ describe('mysql2 instrumentation', () => {
               query.on('end', () => done())
             })
 
-            it('should work without abortController.abort()', (done) => {
+            it('should work without abortController.abort()', async () => {
               startCh.subscribe(noop)
               const query = pool.query(sql)
 
-              query.on('error', err => done(err))
-              query.on('end', () => {
-                sinon.assert.called(apmQueryStart)
+              expect(query.listenerCount('error')).to.equal(0)
 
-                done()
-              })
+              await once(query, 'end')
+
+              expect(query.listenerCount('error')).to.equal(0)
+
+              sinon.assert.called(apmQueryStart)
             })
 
             it('should work without subscriptions', (done) => {
diff --git a/packages/datadog-plugin-fs/test/index.spec.js b/packages/datadog-plugin-fs/test/index.spec.js
@@ -675,7 +675,10 @@ describe('Plugin', () => {
               'file.flag': 'r'
             }
           })
-          fs.createReadStream(__filename).on('error', done).resume()
+          const stream = fs.createReadStream(__filename)
+          expect(stream.listenerCount('error')).to.equal(0)
+
+          stream.on('error', done).resume()
         })
 
         it('should be instrumented when closed', (done) => {
diff --git a/packages/datadog-plugin-http/test/client.spec.js b/packages/datadog-plugin-http/test/client.spec.js
@@ -414,6 +414,7 @@ describe('Plugin', () => {
           appListener = server(app, port => {
             const req = http.request(`${protocol}://localhost:${port}/user`, res => {
               setTimeout(() => {
+                expect(res.listenerCount('error')).to.equal(0)
                 res.on('data', () => done())
               })
             })
diff --git a/packages/datadog-plugin-net/test/index.spec.js b/packages/datadog-plugin-net/test/index.spec.js
@@ -69,7 +69,8 @@ describe('Plugin', () => {
         }).then(done).catch(done)
 
         tracer.scope().activate(parent, () => {
-          net.connect('/tmp/dd-trace.sock')
+          const socket = net.connect('/tmp/dd-trace.sock')
+          expect(socket.listenerCount('error')).to.equal(0)
         })
       })
 
@@ -84,6 +85,7 @@ describe('Plugin', () => {
               resource: 'localhost'
             }, 2000).then(done).catch(done)
           })
+          expect(socket.listenerCount('error')).to.equal(0)
         })
       })
 
@@ -123,6 +125,7 @@ describe('Plugin', () => {
               parent_id: BigInt(parent.context()._spanId.toString(10))
             }, 2000).then(done).catch(done)
           })
+          expect(socket.listenerCount('error')).to.equal(0)
         })
       })
 
diff --git a/packages/datadog-plugin-pg/test/index.spec.js b/packages/datadog-plugin-pg/test/index.spec.js
@@ -1,6 +1,7 @@
 'use strict'
 
 const { expect } = require('chai')
+const assert = require('assert')
 const semver = require('semver')
 const agent = require('../../dd-trace/test/plugins/agent')
 const { ERROR_MESSAGE, ERROR_TYPE, ERROR_STACK } = require('../../dd-trace/src/constants')
@@ -263,6 +264,7 @@ describe('Plugin', () => {
                     expect(traces[0][0]).to.have.property('service', expectedSchema.outbound.serviceName)
                     expect(traces[0][0]).to.have.property('resource', 'SELECT * FROM generate_series(0, 1) num')
                     expect(traces[0][0]).to.have.property('type', 'sql')
+                    expect(traces[0][0]).to.have.property('error', 0)
                     expect(traces[0][0].meta).to.have.property('span.kind', 'client')
                     expect(traces[0][0].meta).to.have.property('db.name', 'postgres')
                     expect(traces[0][0].meta).to.have.property('db.type', 'postgres')
@@ -274,11 +276,48 @@ describe('Plugin', () => {
                   const query = new QueryStream('SELECT * FROM generate_series(0, 1) num', [])
                   const stream = client.query(query)
 
-                  for await (const row of stream) {
-                    expect(row).to.have.property('num')
-                  }
+                  expect(stream.listenerCount('error')).to.equal(0)
 
-                  await agentPromise
+                  const readPromise = (async () => {
+                    for await (const row of stream) {
+                      expect(row).to.have.property('num')
+                    }
+                  })()
+
+                  await Promise.all([readPromise, agentPromise])
+                })
+
+                it('should instrument stream-based queries with pg-query-stream and catch errors', async () => {
+                  const agentPromise = agent.use(traces => {
+                    expect(traces[0][0]).to.have.property('name', expectedSchema.outbound.opName)
+                    expect(traces[0][0]).to.have.property('service', expectedSchema.outbound.serviceName)
+                    expect(traces[0][0]).to.have.property('resource', 'SELECT * FROM generate_series(0, 1) num')
+                    expect(traces[0][0]).to.have.property('type', 'sql')
+                    expect(traces[0][0]).to.have.property('error', 1)
+                    expect(traces[0][0].meta).to.have.property('span.kind', 'client')
+                    expect(traces[0][0].meta).to.have.property('db.name', 'postgres')
+                    expect(traces[0][0].meta).to.have.property('db.type', 'postgres')
+                    expect(traces[0][0].meta).to.have.property('component', 'pg')
+                    expect(traces[0][0].metrics).to.have.property('db.stream', 1)
+                    expect(traces[0][0].metrics).to.have.property('network.destination.port', 5432)
+                  })
+
+                  const query = new QueryStream('SELECT * FROM generate_series(0, 1) num', [])
+                  const stream = client.query(query)
+
+                  expect(stream.listenerCount('error')).to.equal(0)
+
+                  const rejectedRead = assert.rejects(async () => {
+                    // eslint-disable-next-line no-unreachable-loop
+                    for await (const row of stream) {
+                      expect(row).to.have.property('num')
+                      throw new Error('Test error')
+                    }
+                  }, {
+                    message: 'Test error'
+                  })
+
+                  await Promise.all([rejectedRead, agentPromise])
                 })
               })
             })
