fix(test-optimization): disable telemetry in test workers (#9687)

Author: juan-fernandez

ci/init.js

@@ -1,6 +1,7 @@
 'use strict'
 
 /* eslint-disable no-console */
+const exporters = require('../ext/exporters')
 const log = require('../packages/dd-trace/src/log')
 const { getEnvironmentVariable, getValueFromEnvSources } = require('../packages/dd-trace/src/config/helper')
 const { isFalse, isTrue } = require('../packages/dd-trace/src/util')
@@ -13,11 +14,11 @@ const VALIDATION_MODE_ENV = '_DD_TEST_OPTIMIZATION_VALIDATION_MODE'
 const VALIDATION_MANIFEST_ENV = '_DD_TEST_OPTIMIZATION_VALIDATION_MANIFEST_FILE'
 const VALIDATION_OUTPUT_ENV = '_DD_TEST_OPTIMIZATION_VALIDATION_OUTPUT_DIR'
 const EXPORTER_MAP = {
-  jest: 'jest_worker',
-  cucumber: 'cucumber_worker',
-  mocha: 'mocha_worker',
-  playwright: 'playwright_worker',
-  vitest: 'vitest_worker',
+  jest: exporters.JEST_WORKER,
+  cucumber: exporters.CUCUMBER_WORKER,
+  mocha: exporters.MOCHA_WORKER,
+  playwright: exporters.PLAYWRIGHT_WORKER,
+  vitest: exporters.VITEST_WORKER,
 }
 
 function isPackageManager () {
@@ -70,7 +71,6 @@ if (!isTestWorker && isPackageManager()) {
 }
 
 if (isTestWorker) {
-  baseOptions.telemetry = { enabled: false }
   baseOptions.experimental = {
     exporter: EXPORTER_MAP[testWorkerType],
   }

integration-tests/ci-visibility/test-optimization-startup.spec.js

@@ -147,4 +147,34 @@ describe('test optimization startup', () => {
     assert.match(processOutput, /hello!/)
     assert.doesNotMatch(processOutput, /dd-trace will not be initialized/)
   })
+
+  it('does not log an unknown telemetry option in a Vitest worker', async () => {
+    childProcess = exec('node -e "console.log(\'hello!\')"',
+      {
+        cwd,
+        env: {
+          ...process.env,
+          NODE_OPTIONS: '-r dd-trace/ci/init',
+          DD_TRACE_DEBUG: '1',
+          TINYPOOL_WORKER_ID: '1',
+        },
+      }
+    )
+
+    childProcess.stdout?.on('data', (chunk) => {
+      processOutput += chunk.toString()
+    })
+    childProcess.stderr?.on('data', (chunk) => {
+      processOutput += chunk.toString()
+    })
+
+    await Promise.all([
+      once(childProcess, 'exit'),
+      once(childProcess.stdout, 'end'),
+      once(childProcess.stderr, 'end'),
+    ])
+
+    assert.match(processOutput, /hello!/)
+    assert.doesNotMatch(processOutput, /Unknown option telemetry/)
+  })
 })

integration-tests/cucumber/cucumber.spec.js

@@ -295,6 +295,49 @@ describe(`cucumber@${version} commonJS`, () => {
     ])
   })
 
+  it('forwards telemetry from parallel workers', async () => {
+    receiver.setInfoResponse({ endpoints: ['/evp_proxy/v4'] })
+
+    const eventsPromise = receiver
+      .gatherPayloadsMaxTimeout(({ url }) => url.endsWith('/api/v2/citestcycle'), payloads => {
+        const events = payloads.flatMap(({ payload }) => payload.events)
+        const testSession = events.find(event => event.type === 'test_session_end').content
+
+        assert.strictEqual(testSession.meta[TEST_STATUS], 'pass')
+      })
+    const telemetryPromise = receiver
+      .gatherPayloadsMaxTimeout(({ url }) => url.endsWith('/api/v2/apmtelemetry'), (payloads) => {
+        const telemetryMetrics = payloads.flatMap(({ payload }) => payload.payload.series)
+        const testFinishedMetric = telemetryMetrics.find(({ metric, tags }) =>
+          metric === 'event_finished' && tags.includes('event_type:test')
+        )
+
+        assert.ok(testFinishedMetric, 'test event telemetry from a worker should be sent')
+      })
+
+    childProcess = exec(
+      './node_modules/.bin/cucumber-js ci-visibility/features/farewell.feature --parallel 2',
+      {
+        cwd,
+        env: {
+          ...getCiVisEvpProxyConfig(receiver.port),
+          DD_TRACE_AGENT_PORT: String(receiver.port),
+          DD_INSTRUMENTATION_TELEMETRY_ENABLED: 'true',
+        },
+      }
+    )
+    childProcess.stdout?.on('data', chunk => { testOutput += chunk.toString() })
+    childProcess.stderr?.on('data', chunk => { testOutput += chunk.toString() })
+
+    const [[exitCode]] = await Promise.all([
+      once(childProcess, 'exit'),
+      eventsPromise,
+      telemetryPromise,
+    ])
+
+    assert.strictEqual(exitCode, 0, testOutput)
+  })
+
   onlyLatestIt('waits for the final payload before the programmatic run resolves', async () => {
     const completionOrder = []
     const completedMessage = 'programmatic Cucumber run completed'
@@ -620,11 +663,7 @@ describe(`cucumber@${version} commonJS`, () => {
           })
       })
 
-      const runModes = ['serial']
-
-      if (version !== '7.0.0') { // only on latest or 9 if node is old
-        runModes.push('parallel')
-      }
+      const runModes = ['serial', 'parallel']
 
       runModes.forEach((runMode) => {
         it(`(${runMode}) can run and report tests`, (done) => {

integration-tests/mocha/mocha.spec.js

@@ -6,6 +6,9 @@ const assert = require('node:assert/strict')
 const { once } = require('node:events')
 const path = require('path')
 const { inspect } = require('node:util')
+
+const satisfies = require('semifies')
+
 const { assertObjectContains } = require('../helpers')
 
 const {
@@ -119,6 +122,8 @@ const MOCHA_VERSION = requestedMochaVersion === 'oldest' ? oldestMochaVersion :
 const mochaMajor = MOCHA_VERSION === 'latest' ? Infinity : Number.parseInt(MOCHA_VERSION, 10)
 const supportsMochaRetryEvents = mochaMajor >= 6
 const onlyLatestIt = MOCHA_VERSION === 'latest' ? it : it.skip
+// Mocha 8.0 through 8.2 use workerpool 6.0.x, which cannot start process workers on supported Node versions.
+const parallelIt = MOCHA_VERSION === 'latest' || satisfies(MOCHA_VERSION, '>=8.3.0') ? it : it.skip
 
 describe('mocha failed test replay helpers', () => {
   describe('finishDeferredHookEnd', () => {
@@ -313,6 +318,38 @@ describe(`mocha@${MOCHA_VERSION}`, function () {
     ])
   })
 
+  parallelIt('forwards telemetry from parallel workers', async () => {
+    receiver.setInfoResponse({ endpoints: ['/evp_proxy/v4'] })
+
+    const telemetryPromise = receiver
+      .gatherPayloadsMaxTimeout(({ url }) => url.endsWith('/api/v2/apmtelemetry'), (payloads) => {
+        const telemetryMetrics = payloads.flatMap(({ payload }) => payload.payload.series)
+        const testFinishedMetric = telemetryMetrics.find(({ metric, tags }) =>
+          metric === 'event_finished' && tags.includes('event_type:test')
+        )
+
+        assert.ok(testFinishedMetric, 'test event telemetry from a worker should be sent')
+      })
+
+    childProcess = exec(
+      runTestsCommand,
+      {
+        cwd,
+        env: {
+          ...getCiVisEvpProxyConfig(receiver.port),
+          DD_TRACE_AGENT_PORT: String(receiver.port),
+          DD_INSTRUMENTATION_TELEMETRY_ENABLED: 'true',
+          RUN_IN_PARALLEL: '1',
+        },
+      }
+    )
+
+    await Promise.all([
+      once(childProcess, 'exit'),
+      telemetryPromise,
+    ])
+  })
+
   /**
    * @param {boolean} runInParallel
    * @returns {Promise<void>}

integration-tests/playwright/playwright-active-test-span.spec.js

@@ -257,6 +257,7 @@ versions.forEach((version) => {
             const eventFinishedTestEvents = telemetryEvents
               .filter(({ metric, tags }) => metric === 'event_finished' && tags.includes('event_type:test'))
 
+            assert.ok(eventFinishedTestEvents.length > 0, 'test event telemetry from a worker should be sent')
             eventFinishedTestEvents.forEach(({ tags }) => {
               assert.ok(tags.includes('is_rum'), `Got: ${inspect(tags)}`)
               assert.ok(tags.includes('test_framework:playwright'), `Got: ${inspect(tags)}`)

packages/datadog-instrumentations/src/cucumber.js

@@ -21,6 +21,7 @@ const {
   getTestSuitePath,
   getRelativeCoverageFiles,
   CUCUMBER_WORKER_TRACE_PAYLOAD_CODE,
+  CUCUMBER_WORKER_TELEMETRY_PAYLOAD_CODE,
   getIsFaultyEarlyFlakeDetection,
   applySkippedCoverageToCoverage,
   getTestCoverageLinesPercentage,
@@ -62,6 +63,7 @@ const modifiedFilesCh = channel('ci:cucumber:modified-files')
 const isModifiedCh = channel('ci:cucumber:is-modified-test')
 
 const workerReportTraceCh = channel('ci:cucumber:worker-report:trace')
+const workerReportTelemetryCh = channel('ci:cucumber:worker-report:telemetry')
 
 const itrSkippedSuitesCh = channel('ci:cucumber:itr:skipped-suites')
 
@@ -303,6 +305,10 @@ function handleDdWorkerMessage (message) {
       workerReportTraceCh.publish(payload)
       return true
     }
+    if (messageCode === CUCUMBER_WORKER_TELEMETRY_PAYLOAD_CODE) {
+      workerReportTelemetryCh.publish(payload)
+      return true
+    }
   }
 
   if (message?.[DD_EFD_RETRY_COUNT_MESSAGE]) {
@@ -351,8 +357,10 @@ function maybeStartParallelSuite (pickle) {
   })
 }
 
-function handleParallelTestCaseFinished (pickle, worstTestStepResult) {
-  const { status } = getStatusFromResultLatest(worstTestStepResult)
+function handleParallelTestCaseFinished (pickle, worstTestStepResult, usesNumericStatus = false) {
+  const { status } = usesNumericStatus
+    ? getStatusFromResult(worstTestStepResult)
+    : getStatusFromResultLatest(worstTestStepResult)
   let isNew = false
 
   if (isKnownTestsEnabled) {
@@ -1488,7 +1496,7 @@ function patchCucumberWorkerRunTestCase (runtimeExecutorPackage, isWorker) {
   )
 }
 
-function getWrappedParseWorkerMessage (parseWorkerMessageFunction, isNewVersion) {
+function getWrappedParseWorkerMessage (parseWorkerMessageFunction, isNewVersion, usesNumericStatus = false) {
   return function (worker, message) {
     if (!testSuiteFinishCh.hasSubscribers) {
       return parseWorkerMessageFunction.apply(this, arguments)
@@ -1539,13 +1547,31 @@ function getWrappedParseWorkerMessage (parseWorkerMessageFunction, isNewVersion)
         pickle = testCase.pickle
       }
 
-      handleParallelTestCaseFinished(pickle, worstTestStepResult)
+      handleParallelTestCaseFinished(pickle, worstTestStepResult, usesNumericStatus)
     }
 
     return parseWorkerResponse
   }
 }
 
+/**
+ * Adapts Cucumber 7's callback-based parallel coordinator to the Promise contract used by getWrappedStart.
+ *
+ * @param {Function} run
+ * @param {string} frameworkVersion
+ * @returns {Function}
+ */
+function getWrappedCoordinatorRun (run, frameworkVersion) {
+  const runAsPromise = function (numberOfWorkers) {
+    return new Promise(resolve => run.call(this, numberOfWorkers, resolve))
+  }
+  const wrappedStart = getWrappedStart(runAsPromise, frameworkVersion, true)
+
+  return function (numberOfWorkers, done) {
+    return wrappedStart.call(this, numberOfWorkers).then(done)
+  }
+}
+
 module.exports.patchCucumberWorkerRunTestCase = patchCucumberWorkerRunTestCase
 
 // Test start / finish for older versions. The only hook executed in workers when in parallel mode
@@ -1593,19 +1619,33 @@ addHook({
   return runtimePackage
 })
 
-// Only executed in parallel mode.
-// `getWrappedStart` generates session start and finish events
+// Only executed in parallel mode in Cucumber 7 through 10.
+// `getWrappedCoordinatorRun` or `getWrappedStart` generates session start and finish events
 // `getWrappedParseWorkerMessage` generates suite start and finish events
+// Shimmer is required because the coordinator must be changed before it starts workers and exposes no lifecycle hook.
 addHook({
   name: '@cucumber/cucumber',
-  versions: ['>=8.0.0 <11.0.0'],
+  versions: ['>=7.0.0 <11.0.0'],
   file: 'lib/runtime/parallel/coordinator.js',
 }, (coordinatorPackage, frameworkVersion) => {
-  shimmer.wrap(coordinatorPackage.default.prototype, 'start', start => getWrappedStart(start, frameworkVersion, true))
+  const isCucumber7 = satisfies(frameworkVersion, '<8.0.0')
+  if (isCucumber7) {
+    shimmer.wrap(
+      coordinatorPackage.default.prototype,
+      'run',
+      run => getWrappedCoordinatorRun(run, frameworkVersion)
+    )
+  } else {
+    shimmer.wrap(
+      coordinatorPackage.default.prototype,
+      'start',
+      start => getWrappedStart(start, frameworkVersion, true)
+    )
+  }
   shimmer.wrap(
     coordinatorPackage.default.prototype,
     'parseWorkerMessage',
-    parseWorkerMessage => getWrappedParseWorkerMessage(parseWorkerMessage)
+    parseWorkerMessage => getWrappedParseWorkerMessage(parseWorkerMessage, false, isCucumber7)
   )
   return coordinatorPackage
 })

packages/datadog-instrumentations/src/mocha/main.js

@@ -17,6 +17,7 @@ const log = require('../../../dd-trace/src/log')
 const { getEnvironmentVariable } = require('../../../dd-trace/src/config/helper')
 const {
   getTestSuitePath,
+  MOCHA_WORKER_TELEMETRY_PAYLOAD_CODE,
   MOCHA_WORKER_TRACE_PAYLOAD_CODE,
   fromCoverageMapToCoverage,
   getCoveredFilesFromCoverage,
@@ -100,6 +101,7 @@ const mochaGlobalRunCh = channel('ci:mocha:global:run')
 const testManagementTestsCh = channel('ci:mocha:test-management-tests')
 const modifiedFilesCh = channel('ci:mocha:modified-files')
 const workerReportTraceCh = channel('ci:mocha:worker-report:trace')
+const workerReportTelemetryCh = channel('ci:mocha:worker-report:telemetry')
 const testSessionStartCh = channel('ci:mocha:session:start')
 const testSessionFinishCh = channel('ci:mocha:session:finish')
 const itrSkippedSuitesCh = channel('ci:mocha:itr:skipped-suites')
@@ -994,6 +996,8 @@ function onMessage (message) {
         attemptToFixExecutions,
       })
       workerReportTraceCh.publish(payload)
+    } else if (messageCode === MOCHA_WORKER_TELEMETRY_PAYLOAD_CODE) {
+      workerReportTelemetryCh.publish(payload)
     }
   }
 }
@@ -1127,9 +1131,34 @@ addHook({
   name: 'mocha',
   versions: ['>=8.0.0'],
   file: 'lib/nodejs/buffered-worker-pool.js',
-}, (BufferedWorkerPoolPackage) => {
+}, (BufferedWorkerPoolPackage, frameworkVersion) => {
   const { BufferedWorkerPool } = BufferedWorkerPoolPackage
 
+  if (satisfies(frameworkVersion, '<9.2.0')) {
+    // Shimmer is required because the worker environment must be changed before workerpool forks,
+    // before any test lifecycle hook can run. Mocha added this worker ID itself in 9.2.0.
+    shimmer.wrap(BufferedWorkerPool, 'create', create => function () {
+      const pool = create.apply(this, arguments)
+
+      if (!testFinishCh.hasSubscribers) return pool
+
+      let workerId = 0
+      shimmer.wrap(pool._pool, '_createWorkerHandler', createWorkerHandler => function () {
+        this.forkOpts = {
+          ...this.forkOpts,
+          env: {
+            // eslint-disable-next-line eslint-rules/eslint-process-env
+            ...(this.forkOpts.env || process.env),
+            MOCHA_WORKER_ID: String(workerId++),
+          },
+        }
+        return createWorkerHandler.apply(this, arguments)
+      })
+
+      return pool
+    })
+  }
+
   shimmer.wrap(BufferedWorkerPool.prototype, 'run', run => async function (testSuiteAbsolutePath, workerArgs) {
     if (!testFinishCh.hasSubscribers ||
         (!config.isKnownTestsEnabled &&