V4.13.1 proposal

.github/workflows/release-3.yml

@@ -28,3 +28,26 @@ jobs:
       - run: |
           git tag v${{ fromJson(steps.pkg.outputs.json).version }}
           git push origin v${{ fromJson(steps.pkg.outputs.json).version }}
+
+  injection-image-publish:
+    runs-on: ubuntu-latest
+    needs: ['publish']
+    steps:
+      - uses: actions/checkout@v3
+      - uses: actions/setup-node@v3
+      - name: Log in to the Container registry
+        uses: docker/login-action@49ed152c8eca782a232dede0303416e8f356c37b
+        with:
+          registry: ghcr.io
+          username: ${{ github.actor }}
+          password: ${{ secrets.GITHUB_TOKEN }}
+      - id: pkg
+        run: |
+          content=`cat ./package.json | tr '\n' ' '`
+          echo "::set-output name=json::$content"
+      - name: npm pack for injection image
+        run: |
+          npm pack dd-trace@${{ fromJson(steps.pkg.outputs.json).version }}
+      - uses: ./.github/actions/injection
+        with:
+          init-image-version: v${{ fromJson(steps.pkg.outputs.json).version }}

integration-tests/helpers.js

@@ -205,6 +205,7 @@ async function createSandbox (dependencies = [], isGitRepo = false, integrationT
 
   integrationTestsPaths.forEach(async (path) => {
     await exec(`cp -R ${path} ${folder}`)
+    await exec(`sync ${folder}`)
   })
 
   if (isGitRepo) {

integration-tests/telemetry.spec.js

@@ -0,0 +1,64 @@
+'use strict'
+
+const { createSandbox, FakeAgent, spawnProc } = require('./helpers')
+const path = require('path')
+
+describe('telemetry', () => {
+  describe('dependencies', () => {
+    let sandbox
+    let cwd
+    let startupTestFile
+    let agent
+    let proc
+
+    before(async () => {
+      sandbox = await createSandbox()
+      cwd = sandbox.folder
+      startupTestFile = path.join(cwd, 'startup/index.js')
+    })
+
+    after(async () => {
+      await sandbox.remove()
+    })
+
+    beforeEach(async () => {
+      agent = await new FakeAgent().start()
+      proc = await spawnProc(startupTestFile, {
+        cwd,
+        env: {
+          AGENT_PORT: agent.port
+        }
+      })
+    })
+
+    afterEach(async () => {
+      proc.kill()
+      await agent.stop()
+    })
+
+    it('Test that tracer and iitm are sent as dependencies', (done) => {
+      let ddTraceFound = false
+      let importInTheMiddleFound = false
+
+      agent.assertTelemetryReceived(msg => {
+        const { payload } = msg
+
+        if (payload.request_type === 'app-dependencies-loaded') {
+          if (payload.payload.dependencies) {
+            payload.payload.dependencies.forEach(dependency => {
+              if (dependency.name === 'dd-trace') {
+                ddTraceFound = true
+              }
+              if (dependency.name === 'import-in-the-middle') {
+                importInTheMiddleFound = true
+              }
+            })
+            if (ddTraceFound && importInTheMiddleFound) {
+              done()
+            }
+          }
+        }
+      }, null, 'app-dependencies-loaded', 1)
+    })
+  })
+})

package.json

@@ -1,6 +1,6 @@
 {
   "name": "dd-trace",
-  "version": "4.13.0",
+  "version": "4.13.1",
   "description": "Datadog APM tracing client for JavaScript",
   "main": "index.js",
   "typings": "index.d.ts",

packages/datadog-instrumentations/src/kafkajs.js

@@ -15,15 +15,14 @@ const consumerStartCh = channel('apm:kafkajs:consume:start')
 const consumerFinishCh = channel('apm:kafkajs:consume:finish')
 const consumerErrorCh = channel('apm:kafkajs:consume:error')
 
-addHook({ name: 'kafkajs', versions: ['>=1.4'] }, (obj) => {
-  class Kafka extends obj.Kafka {
+addHook({ name: 'kafkajs', file: 'src/index.js', versions: ['>=1.4'] }, (BaseKafka) => {
+  class Kafka extends BaseKafka {
     constructor (options) {
       super(options)
       this._brokers = (options.brokers && typeof options.brokers !== 'function')
         ? options.brokers.join(',') : undefined
     }
   }
-  obj.Kafka = Kafka
 
   shimmer.wrap(Kafka.prototype, 'producer', createProducer => function () {
     const producer = createProducer.apply(this, arguments)
@@ -117,5 +116,5 @@ addHook({ name: 'kafkajs', versions: ['>=1.4'] }, (obj) => {
     }
     return consumer
   })
-  return obj
+  return Kafka
 })

packages/datadog-plugin-amqp10/test/integration-test/server.mjs

@@ -2,15 +2,14 @@ import 'dd-trace/init.js'
 import amqp from 'amqp10'
 
 const client = new amqp.Client()
-await client.connect('amqp://admin:admin@localhost:5673')
+await client.connect('amqp://admin:admin@127.0.0.1:5673')
 const handlers = await Promise.all([client.createSender('amq.topic')])
 const sender = handlers[0]
-
 sender.send({ key: 'value' })
 
 if (sender) {
   await sender.detach()
 }
 if (client) {
   await client.disconnect()
-}
+}

packages/datadog-plugin-kafkajs/test/integration-test/client.spec.js

@@ -0,0 +1,48 @@
+'use strict'
+
+const {
+  FakeAgent,
+  createSandbox,
+  checkSpansForServiceName,
+  spawnPluginIntegrationTestProc
+} = require('../../../../integration-tests/helpers')
+const { assert } = require('chai')
+
+describe('esm', () => {
+  let agent
+  let proc
+  let sandbox
+
+  before(async function () {
+    this.timeout(20000)
+    sandbox = await createSandbox(['kafkajs@>=1.4.0'], false, [
+      `./packages/datadog-plugin-kafkajs/test/integration-test/*`])
+  })
+
+  after(async () => {
+    await sandbox.remove()
+  })
+
+  beforeEach(async () => {
+    agent = await new FakeAgent().start()
+  })
+
+  afterEach(async () => {
+    proc && proc.kill()
+    await agent.stop()
+  })
+
+  context('kafkajs', () => {
+    it('is instrumented', async () => {
+      const res = agent.assertMessageReceived(({ headers, payload }) => {
+        assert.propertyVal(headers, 'host', `127.0.0.1:${agent.port}`)
+        assert.isArray(payload)
+        assert.strictEqual(checkSpansForServiceName(payload, 'kafka.produce'), true)
+      })
+
+      proc = await spawnPluginIntegrationTestProc(sandbox.folder, 'server.mjs', agent.port)
+
+      await res
+    }).timeout(20000)
+  })
+})

packages/datadog-plugin-kafkajs/test/integration-test/server.mjs

@@ -0,0 +1,19 @@
+import 'dd-trace/init.js'
+import { Kafka } from 'kafkajs'
+
+const kafka = new Kafka({
+  clientId: 'my-app',
+  brokers: ['127.0.0.1:9092']
+})
+
+const sendMessage = async (topic, messages) => {
+  const producer = kafka.producer()
+  await producer.connect()
+  await producer.send({
+    topic,
+    messages
+  })
+  await producer.disconnect()
+}
+
+await sendMessage('test-topic', [{ key: 'key1', value: 'test2' }])

packages/datadog-plugin-mongodb-core/src/index.js

@@ -36,10 +36,14 @@ class MongodbCorePlugin extends DatabasePlugin {
   }
 }
 
+function sanitizeBigInt (data) {
+  return JSON.stringify(data, (_key, value) => typeof value === 'bigint' ? value.toString() : value)
+}
+
 function getQuery (cmd) {
   if (!cmd || typeof cmd !== 'object' || Array.isArray(cmd)) return
-  if (cmd.query) return JSON.stringify(limitDepth(cmd.query))
-  if (cmd.filter) return JSON.stringify(limitDepth(cmd.filter))
+  if (cmd.query) return sanitizeBigInt(limitDepth(cmd.query))
+  if (cmd.filter) return sanitizeBigInt(limitDepth(cmd.filter))
 }
 
 function getResource (plugin, ns, query, operationName) {

packages/datadog-plugin-mongodb-core/test/core.spec.js

@@ -133,6 +133,39 @@ describe('Plugin', () => {
             }, () => {})
           })
 
+          it('should serialize BigInt without erroring', done => {
+            agent
+              .use(traces => {
+                const span = traces[0][0]
+                const resource = `find test.${collection}`
+                const query = `{"_id":"9999999999999999999999"}`
+
+                expect(span).to.have.property('resource', resource)
+                expect(span.meta).to.have.property('mongodb.query', query)
+              })
+              .then(done)
+              .catch(done)
+
+            try {
+              server.command(`test.${collection}`, {
+                find: `test.${collection}`,
+                query: {
+                  _id: 9999999999999999999999n
+                }
+              }, () => {})
+            } catch (err) {
+              // It appears that most versions of MongodDB are happy to use a BigInt instance.
+              // For example, 2.0.0, 3.2.0, 3.1.10, etc.
+              // However, version 3.1.9 throws a synchronous error that it wants a Decimal128 instead.
+              if (err.message.includes('Decimal128')) {
+                // eslint-disable-next-line no-console
+                console.log('This version of mongodb-core does not accept BigInt instances')
+                return done()
+              }
+              done(err)
+            }
+          })
+
           it('should stringify BSON objects', done => {
             const BSON = require(`../../../versions/bson@4.0.0`).get()
             const id = '123456781234567812345678'

packages/dd-trace/src/appsec/iast/vulnerabilities-formatter/evidence-redaction/sensitive-handler.js

@@ -76,7 +76,18 @@ class SensitiveHandler {
         while (nextSensitive != null && contains(nextTainted, nextSensitive)) {
           const redactionStart = nextSensitive.start - nextTainted.start
           const redactionEnd = nextSensitive.end - nextTainted.start
-          this.redactSource(sources, redactedSources, redactedSourcesContext, sourceIndex, redactionStart, redactionEnd)
+          if (redactionStart === redactionEnd) {
+            this.writeRedactedValuePart(valueParts, 0)
+          } else {
+            this.redactSource(
+              sources,
+              redactedSources,
+              redactedSourcesContext,
+              sourceIndex,
+              redactionStart,
+              redactionEnd
+            )
+          }
           nextSensitive = sensitive.shift()
         }
 

packages/dd-trace/src/telemetry/dependencies.js

@@ -15,6 +15,7 @@ const FILE_URI_START = `file://`
 const moduleLoadStartChannel = dc.channel('dd-trace:moduleLoadStart')
 
 let immediate, config, application, host
+let isFirstModule = true
 
 function waitAndSend (config, application, host) {
   if (!immediate) {
@@ -36,7 +37,21 @@ function waitAndSend (config, application, host) {
   }
 }
 
+function loadAllTheLoadedModules () {
+  if (require.cache) {
+    const filenames = Object.keys(require.cache)
+    filenames.forEach(filename => {
+      onModuleLoad({ filename })
+    })
+  }
+}
+
 function onModuleLoad (data) {
+  if (isFirstModule) {
+    isFirstModule = false
+    loadAllTheLoadedModules()
+  }
+
   if (data) {
     let filename = data.filename
     if (filename && filename.startsWith(FILE_URI_START)) {

packages/dd-trace/src/telemetry/index.js

@@ -17,6 +17,7 @@ let pluginManager
 let application
 let host
 let interval
+let heartbeatTimeout
 let heartbeatInterval
 const sentIntegrations = new Set()
 
@@ -110,6 +111,14 @@ function getTelemetryData () {
   return { config, application, host, heartbeatInterval }
 }
 
+function heartbeat (config, application, host) {
+  heartbeatTimeout = setTimeout(() => {
+    sendData(config, application, host, 'app-heartbeat')
+    heartbeat(config, application, host)
+  }, heartbeatInterval).unref()
+  return heartbeatTimeout
+}
+
 function start (aConfig, thePluginManager) {
   if (!aConfig.telemetry.enabled) {
     return
@@ -122,9 +131,9 @@ function start (aConfig, thePluginManager) {
 
   dependencies.start(config, application, host)
   sendData(config, application, host, 'app-started', appStarted())
+  heartbeat(config, application, host)
   interval = setInterval(() => {
     metricsManager.send(config, application, host)
-    sendData(config, application, host, 'app-heartbeat')
   }, heartbeatInterval)
   interval.unref()
   process.on('beforeExit', onBeforeExit)
@@ -137,6 +146,7 @@ function stop () {
     return
   }
   clearInterval(interval)
+  clearTimeout(heartbeatTimeout)
   process.removeListener('beforeExit', onBeforeExit)
 
   telemetryStopChannel.publish(getTelemetryData())