fix(datastreams): preserve extracted schema field descriptions (#7838)

Populate extracted schema property descriptions from Avro field docs and
protobufjs field comments so shipped schema metadata matches the source
definitions. Add regression tests and clean up related test/helper
issues surfaced by stricter linting.

Author: watson

benchmark/e2e-test-optimization/benchmark-run.js

@@ -36,7 +36,8 @@ const parseGitHubJsonResponse = ({ body, endpoint, res }) => {
     return JSON.parse(body)
   } catch (e) {
     throw new Error(
-      `GitHub API ${endpoint} returned invalid JSON. Body preview: ${getResponsePreview(body)}`
+      `GitHub API ${endpoint} returned invalid JSON. Body preview: ${getResponsePreview(body)}`,
+      { cause: e }
     )
   }
 }

integration-tests/ci-visibility-intake.js

@@ -351,7 +351,7 @@ class FakeCiVisIntake extends FakeAgent {
       this.server = http.createServer(app)
       this.server.on('error', reject)
       this.server.listen(this.port, () => {
-        this.port = this.server.address().port
+        this.port = (/** @type {import('net').AddressInfo} */ (this.server.address())).port
         clearTimeout(timeoutObj)
         resolve(this)
       })
@@ -473,7 +473,7 @@ class FakeCiVisIntake extends FakeAgent {
   // always be faster or as fast as gatherPayloads
   gatherPayloadsMaxTimeout (payloadMatch, onPayload, maxGatheringTime = 15000) {
     const payloads = []
-    return new Promise((resolve, reject) => {
+    return /** @type {Promise<void>} */ (new Promise((resolve, reject) => {
       const timeoutId = setTimeout(() => {
         try {
           onPayload(payloads)
@@ -499,7 +499,7 @@ class FakeCiVisIntake extends FakeAgent {
         }
       }
       this.on('message', messageHandler)
-    })
+    }))
   }
 
   gatherPayloads (payloadMatch, gatheringTime = 15000) {
@@ -540,37 +540,7 @@ class FakeCiVisIntake extends FakeAgent {
   }
 
   assertPayloadReceived (fn, messageMatch, timeout) {
-    let resultResolve
-    let resultReject
-    let error
-
-    const timeoutObj = setTimeout(() => {
-      resultReject([error, new Error('timeout')])
-    }, timeout || 15000)
-
-    const messageHandler = (message) => {
-      if (!messageMatch || messageMatch(message)) {
-        try {
-          fn(message)
-          resultResolve()
-        } catch (e) {
-          resultReject(e)
-        }
-        this.off('message', messageHandler)
-      }
-    }
-    this.on('message', messageHandler)
-
-    return new Promise((resolve, reject) => {
-      resultResolve = () => {
-        clearTimeout(timeoutObj)
-        resolve()
-      }
-      resultReject = (e) => {
-        clearTimeout(timeoutObj)
-        reject(e)
-      }
-    })
+    return this.payloadReceived(messageMatch, timeout).then(fn)
   }
 }
 

integration-tests/ci-visibility/intake.spec.js

@@ -0,0 +1,36 @@
+'use strict'
+
+const assert = require('node:assert/strict')
+
+const { describe, it } = require('mocha')
+
+const { FakeCiVisIntake } = require('../ci-visibility-intake')
+
+describe('FakeCiVisIntake', () => {
+  it('should reject with a timeout error and remove the listener', async () => {
+    const intake = new FakeCiVisIntake()
+
+    await assert.rejects(
+      intake.assertPayloadReceived(() => {}, undefined, 5),
+      {
+        message: 'Timeout',
+      }
+    )
+    assert.strictEqual(intake.listenerCount('message'), 0)
+  })
+
+  it('should resolve when a matching payload satisfies the assertion', async () => {
+    const intake = new FakeCiVisIntake()
+    const payloadPromise = intake.assertPayloadReceived(({ payload }) => {
+      assert.deepStrictEqual(payload, { ok: true })
+    }, ({ url }) => url === '/expected')
+
+    intake.emit('message', {
+      payload: { ok: true },
+      url: '/expected',
+    })
+
+    await payloadPromise
+    assert.strictEqual(intake.listenerCount('message'), 0)
+  })
+})

packages/datadog-plugin-avsc/src/schema_iterator.js

@@ -43,9 +43,9 @@ class SchemaExtractor {
     let type
     let format
     let enumValues
-    let description
     let ref
 
+    const description = field.doc
     const fieldType = field.type?.types ?? field.type?.typeName ?? field.type
 
     if (Array.isArray(fieldType)) {
@@ -129,7 +129,7 @@ class SchemaExtractor {
   }
 
   iterateOverSchema (builder) {
-    this.constructor.extractSchema(this.schema, builder, 0)
+    SchemaExtractor.extractSchema(this.schema, builder, 0)
   }
 
   static attachSchemaOnSpan (args, span, operation, tracer) {

packages/datadog-plugin-avsc/test/index.spec.js

@@ -27,7 +27,7 @@ const ADVANCED_USER_SCHEMA_DEF = JSON.parse(
   fs.readFileSync(path.join(__dirname, 'schemas/expected_advanced_user_schema.json'), 'utf8')
 )
 
-const BASIC_USER_SCHEMA_ID = '1605040621379664412'
+const BASIC_USER_SCHEMA_ID = '15683462889181473629'
 const ADVANCED_USER_SCHEMA_ID = '919692610494986520'
 function compareJson (expected, span) {
   const actual = JSON.parse(span.context().getTag(SCHEMA_DEFINITION))

packages/datadog-plugin-avsc/test/schema-iterator.spec.js

@@ -0,0 +1,35 @@
+'use strict'
+
+const assert = require('node:assert/strict')
+
+const { describe, it } = require('mocha')
+
+const { SchemaBuilder } = require('../../dd-trace/src/datastreams/schemas/schema_builder')
+const SchemaExtractor = require('../src/schema_iterator')
+
+describe('SchemaExtractor', () => {
+  describe('avsc', () => {
+    it('should include field docs in the extracted schema', () => {
+      const schema = {
+        name: 'UserWithDocs',
+        fields: [
+          {
+            name: 'name',
+            type: 'string',
+            doc: 'The user name',
+          },
+        ],
+      }
+
+      const schemaData = SchemaBuilder.getSchemaDefinition(
+        new SchemaBuilder(new SchemaExtractor(schema)).build()
+      )
+      const property = JSON.parse(schemaData.definition).components.schemas.UserWithDocs.properties.name
+
+      assert.deepStrictEqual(property, {
+        description: 'The user name',
+        type: 'string',
+      })
+    })
+  })
+})

packages/datadog-plugin-avsc/test/schemas/expected_user_schema.json

@@ -6,7 +6,8 @@
               "type": "object",
               "properties": {
                   "name": {
-                      "type": "string"
+                      "type": "string",
+                      "description": "The user name"
                   },
                   "favorite_number": {
                       "type": "union[integer,null]"

packages/datadog-plugin-avsc/test/schemas/user.avsc

@@ -5,7 +5,8 @@
     "fields": [
         {
             "name": "name",
-            "type": "string"
+            "type": "string",
+            "doc": "The user name"
         },
         {
             "name": "favorite_number",

packages/datadog-plugin-confluentinc-kafka-javascript/test/index.spec.js

@@ -173,27 +173,30 @@ describe('Plugin', () => {
               ])
             })
 
-            it('should run the consumer in the context of the consumer span', done => {
+            it('should run the consumer in the context of the consumer span', async () => {
               const firstSpan = tracer.scope().active()
-              let consumerReceiveMessagePromise
-              let eachMessage = async ({ topic, partition, message }) => {
-                const currentSpan = tracer.scope().active()
+              let hasReceivedMessage = false
+              const consumerReceiveMessagePromise = /** @type {Promise<void>} */(new Promise((resolve, reject) => {
+                const eachMessage = async () => {
+                  if (hasReceivedMessage) return
 
-                try {
-                  assert.notStrictEqual(currentSpan, firstSpan)
-                  assert.strictEqual(currentSpan.context()._name, expectedSchema.receive.opName)
-                  eachMessage = async () => {} // avoid being called for each message
-                  done()
-                } catch (e) {
-                  eachMessage = async () => {}
-                  done(e)
+                  hasReceivedMessage = true
+                  const currentSpan = tracer.scope().active()
+
+                  try {
+                    assert.notStrictEqual(currentSpan, firstSpan)
+                    assert.strictEqual(currentSpan.context()._name, expectedSchema.receive.opName)
+                    resolve()
+                  } catch (e) {
+                    reject(e)
+                  }
                 }
-              }
 
-              consumer.run({ eachMessage: (...args) => eachMessage(...args) })
-                .then(() => sendMessages(kafka, testTopic, messages))
-                .then(() => consumerReceiveMessagePromise)
-                .catch(done)
+                consumer.run({ eachMessage }).catch(reject)
+              }))
+
+              await sendMessages(kafka, testTopic, messages)
+              await consumerReceiveMessagePromise
             })
 
             it('should propagate context', async () => {

packages/datadog-plugin-express-mongo-sanitize/test/integration-test/client.spec.js

@@ -34,7 +34,7 @@ withVersions('express-mongo-sanitize', 'express-mongo-sanitize', version => {
 
     for (const variant of varySandbox.VARIANTS) {
       it(`is instrumented loaded with ${variant}`, async () => {
-        const proc = await spawnPluginIntegrationTestProc(sandboxCwd(), variants[variant], agent.port)
+        proc = await spawnPluginIntegrationTestProc(sandboxCwd(), variants[variant], agent.port)
         const response = await axios.get(`${proc.url}/?param=paramvalue`)
         assert.equal(response.headers['x-counter'], '1')
       })