[test visibility] Add dynamic instrumentation logs writer for test vi…

…sibility (#4821)

* Apply suggestions from code review

Co-authored-by: Nikita Tkachenko <121111529+nikita-tkachenko-datadog@users.noreply.github.com>

packages/dd-trace/src/ci-visibility/exporters/agent-proxy/index.js

@@ -7,6 +7,7 @@ const CiVisibilityExporter = require('../ci-visibility-exporter')
 
 const AGENT_EVP_PROXY_PATH_PREFIX = '/evp_proxy/v'
 const AGENT_EVP_PROXY_PATH_REGEX = /\/evp_proxy\/v(\d+)\/?/
+const AGENT_DEBUGGER_INPUT = '/debugger/v1/input'
 
 function getLatestEvpProxyVersion (err, agentInfo) {
   if (err) {
@@ -24,6 +25,10 @@ function getLatestEvpProxyVersion (err, agentInfo) {
   }, 0)
 }
 
+function getCanForwardDebuggerLogs (err, agentInfo) {
+  return !err && agentInfo.endpoints.some(endpoint => endpoint === AGENT_DEBUGGER_INPUT)
+}
+
 class AgentProxyCiVisibilityExporter extends CiVisibilityExporter {
   constructor (config) {
     super(config)
@@ -33,7 +38,8 @@ class AgentProxyCiVisibilityExporter extends CiVisibilityExporter {
       prioritySampler,
       lookup,
       protocolVersion,
-      headers
+      headers,
+      isTestDynamicInstrumentationEnabled
     } = config
 
     this.getAgentInfo((err, agentInfo) => {
@@ -60,6 +66,18 @@ class AgentProxyCiVisibilityExporter extends CiVisibilityExporter {
           url: this._url,
           evpProxyPrefix
         })
+        if (isTestDynamicInstrumentationEnabled) {
+          const canFowardLogs = getCanForwardDebuggerLogs(err, agentInfo)
+          if (canFowardLogs) {
+            const DynamicInstrumentationLogsWriter = require('../agentless/di-logs-writer')
+            this._logsWriter = new DynamicInstrumentationLogsWriter({
+              url: this._url,
+              tags,
+              isAgentProxy: true
+            })
+            this._canForwardLogs = true
+          }
+        }
       } else {
         this._writer = new AgentWriter({
           url: this._url,

packages/dd-trace/src/ci-visibility/exporters/agentless/di-logs-writer.js

@@ -0,0 +1,53 @@
+'use strict'
+const request = require('../../../exporters/common/request')
+const log = require('../../../log')
+const { safeJSONStringify } = require('../../../exporters/common/util')
+const { JSONEncoder } = require('../../encode/json-encoder')
+
+const BaseWriter = require('../../../exporters/common/writer')
+
+// Writer used by the integration between Dynamic Instrumentation and Test Visibility
+// It is used to encode and send logs to both the logs intake directly and the
+// `/debugger/v1/input` endpoint in the agent, which is a proxy to the logs intake.
+class DynamicInstrumentationLogsWriter extends BaseWriter {
+  constructor ({ url, timeout, isAgentProxy = false }) {
+    super(...arguments)
+    this._url = url
+    this._encoder = new JSONEncoder()
+    this._isAgentProxy = isAgentProxy
+    this.timeout = timeout
+  }
+
+  _sendPayload (data, _, done) {
+    const options = {
+      path: '/api/v2/logs',
+      method: 'POST',
+      headers: {
+        'dd-api-key': process.env.DATADOG_API_KEY || process.env.DD_API_KEY,
+        'Content-Type': 'application/json'
+      },
+      // TODO: what's a good value for timeout for the logs intake?
+      timeout: this.timeout || 15000,
+      url: this._url
+    }
+
+    if (this._isAgentProxy) {
+      delete options.headers['dd-api-key']
+      options.path = '/debugger/v1/input'
+    }
+
+    log.debug(() => `Request to the logs intake: ${safeJSONStringify(options)}`)
+
+    request(data, options, (err, res) => {
+      if (err) {
+        log.error(err)
+        done()
+        return
+      }
+      log.debug(`Response from the logs intake: ${res}`)
+      done()
+    })
+  }
+}
+
+module.exports = DynamicInstrumentationLogsWriter

packages/dd-trace/src/ci-visibility/exporters/agentless/index.js

@@ -9,17 +9,24 @@ const log = require('../../../log')
 class AgentlessCiVisibilityExporter extends CiVisibilityExporter {
   constructor (config) {
     super(config)
-    const { tags, site, url } = config
+    const { tags, site, url, isTestDynamicInstrumentationEnabled } = config
     // we don't need to request /info because we are using agentless by configuration
     this._isInitialized = true
     this._resolveCanUseCiVisProtocol(true)
+    this._canForwardLogs = true
 
     this._url = url || new URL(`https://citestcycle-intake.${site}`)
     this._writer = new Writer({ url: this._url, tags })
 
     this._coverageUrl = url || new URL(`https://citestcov-intake.${site}`)
     this._coverageWriter = new CoverageWriter({ url: this._coverageUrl })
 
+    if (isTestDynamicInstrumentationEnabled) {
+      const DynamicInstrumentationLogsWriter = require('./di-logs-writer')
+      this._logsUrl = url || new URL(`https://http-intake.logs.${site}`)
+      this._logsWriter = new DynamicInstrumentationLogsWriter({ url: this._logsUrl, tags })
+    }
+
     this._apiUrl = url || new URL(`https://api.${site}`)
     // Agentless is always gzip compatible
     this._isGzipCompatible = true

packages/dd-trace/src/ci-visibility/exporters/ci-visibility-exporter.js

@@ -8,6 +8,7 @@ const { getSkippableSuites: getSkippableSuitesRequest } = require('../intelligen
 const { getKnownTests: getKnownTestsRequest } = require('../early-flake-detection/get-known-tests')
 const log = require('../../log')
 const AgentInfoExporter = require('../../exporters/common/agent-info-exporter')
+const { GIT_REPOSITORY_URL, GIT_COMMIT_SHA } = require('../../plugins/util/tags')
 
 function getTestConfigurationTags (tags) {
   if (!tags) {
@@ -36,6 +37,7 @@ class CiVisibilityExporter extends AgentInfoExporter {
     super(config)
     this._timer = undefined
     this._coverageTimer = undefined
+    this._logsTimer = undefined
     this._coverageBuffer = []
     // The library can use new features like ITR and test suite level visibility
     // AKA CI Vis Protocol
@@ -255,6 +257,47 @@ class CiVisibilityExporter extends AgentInfoExporter {
     this._export(formattedCoverage, this._coverageWriter, '_coverageTimer')
   }
 
+  formatLogMessage (testConfiguration, logMessage) {
+    const {
+      [GIT_REPOSITORY_URL]: gitRepositoryUrl,
+      [GIT_COMMIT_SHA]: gitCommitSha
+    } = testConfiguration
+
+    const { service, env, version } = this._config
+
+    return {
+      ddtags: [
+        ...(logMessage.ddtags || []),
+        `${GIT_REPOSITORY_URL}:${gitRepositoryUrl}`,
+        `${GIT_COMMIT_SHA}:${gitCommitSha}`
+      ].join(','),
+      level: 'error',
+      service,
+      dd: {
+        ...(logMessage.dd || []),
+        service,
+        env,
+        version
+      },
+      ddsource: 'dd_debugger',
+      ...logMessage
+    }
+  }
+
+  // DI logs
+  exportDiLogs (testConfiguration, logMessage) {
+    // TODO: could we lose logs if it's not initialized?
+    if (!this._config.isTestDynamicInstrumentationEnabled || !this._isInitialized || !this._canForwardLogs) {
+      return
+    }
+
+    this._export(
+      this.formatLogMessage(testConfiguration, logMessage),
+      this._logsWriter,
+      '_logsTimer'
+    )
+  }
+
   flush (done = () => {}) {
     if (!this._isInitialized) {
       return done()

packages/dd-trace/src/config.js

@@ -503,6 +503,7 @@ class Config {
     this._setValue(defaults, 'isManualApiEnabled', false)
     this._setValue(defaults, 'ciVisibilityTestSessionName', '')
     this._setValue(defaults, 'ciVisAgentlessLogSubmissionEnabled', false)
+    this._setValue(defaults, 'isTestDynamicInstrumentationEnabled', false)
     this._setValue(defaults, 'logInjection', false)
     this._setValue(defaults, 'lookup', undefined)
     this._setValue(defaults, 'memcachedCommandEnabled', false)
@@ -1054,7 +1055,8 @@ class Config {
       DD_CIVISIBILITY_FLAKY_RETRY_ENABLED,
       DD_CIVISIBILITY_FLAKY_RETRY_COUNT,
       DD_TEST_SESSION_NAME,
-      DD_AGENTLESS_LOG_SUBMISSION_ENABLED
+      DD_AGENTLESS_LOG_SUBMISSION_ENABLED,
+      DD_TEST_DYNAMIC_INSTRUMENTATION_ENABLED
     } = process.env
 
     if (DD_CIVISIBILITY_AGENTLESS_URL) {
@@ -1072,6 +1074,7 @@ class Config {
       this._setBoolean(calc, 'isManualApiEnabled', !isFalse(this._isCiVisibilityManualApiEnabled()))
       this._setString(calc, 'ciVisibilityTestSessionName', DD_TEST_SESSION_NAME)
       this._setBoolean(calc, 'ciVisAgentlessLogSubmissionEnabled', isTrue(DD_AGENTLESS_LOG_SUBMISSION_ENABLED))
+      this._setBoolean(calc, 'isTestDynamicInstrumentationEnabled', isTrue(DD_TEST_DYNAMIC_INSTRUMENTATION_ENABLED))
     }
     this._setString(calc, 'dogstatsd.hostname', this._getHostname())
     this._setBoolean(calc, 'isGitUploadEnabled',

packages/dd-trace/test/ci-visibility/exporters/agent-proxy/agent-proxy.spec.js

@@ -6,6 +6,7 @@ const nock = require('nock')
 
 const AgentProxyCiVisibilityExporter = require('../../../../src/ci-visibility/exporters/agent-proxy')
 const AgentlessWriter = require('../../../../src/ci-visibility/exporters/agentless/writer')
+const DynamicInstrumentationLogsWriter = require('../../../../src/ci-visibility/exporters/agentless/di-logs-writer')
 const CoverageWriter = require('../../../../src/ci-visibility/exporters/agentless/coverage-writer')
 const AgentWriter = require('../../../../src/exporters/agent/writer')
 
@@ -68,7 +69,10 @@ describe('AgentProxyCiVisibilityExporter', () => {
         .get('/info')
         .delay(queryDelay)
         .reply(200, JSON.stringify({
-          endpoints: ['/evp_proxy/v2/']
+          endpoints: [
+            '/evp_proxy/v2/',
+            '/debugger/v1/input'
+          ]
         }))
     })
 
@@ -112,6 +116,35 @@ describe('AgentProxyCiVisibilityExporter', () => {
       agentProxyCiVisibilityExporter.exportCoverage(coverage)
       expect(mockWriter.append).to.have.been.calledWith({ spanId: '1', traceId: '1', files: [] })
     })
+
+    context('if isTestDynamicInstrumentationEnabled is set', () => {
+      it('should initialise DynamicInstrumentationLogsWriter', async () => {
+        const agentProxyCiVisibilityExporter = new AgentProxyCiVisibilityExporter({
+          port,
+          tags,
+          isTestDynamicInstrumentationEnabled: true
+        })
+        await agentProxyCiVisibilityExporter._canUseCiVisProtocolPromise
+        expect(agentProxyCiVisibilityExporter._logsWriter).to.be.instanceOf(DynamicInstrumentationLogsWriter)
+      })
+
+      it('should process logs', async () => {
+        const mockWriter = {
+          append: sinon.spy(),
+          flush: sinon.spy()
+        }
+        const agentProxyCiVisibilityExporter = new AgentProxyCiVisibilityExporter({
+          port,
+          tags,
+          isTestDynamicInstrumentationEnabled: true
+        })
+        await agentProxyCiVisibilityExporter._canUseCiVisProtocolPromise
+        agentProxyCiVisibilityExporter._logsWriter = mockWriter
+        const log = { message: 'hello' }
+        agentProxyCiVisibilityExporter.exportDiLogs({}, log)
+        expect(mockWriter.append).to.have.been.calledWith(sinon.match(log))
+      })
+    })
   })
 
   describe('agent is not evp compatible', () => {
@@ -166,6 +199,35 @@ describe('AgentProxyCiVisibilityExporter', () => {
       })
       expect(mockWriter.append).not.to.have.been.called
     })
+
+    context('if isTestDynamicInstrumentationEnabled is set', () => {
+      it('should not initialise DynamicInstrumentationLogsWriter', async () => {
+        const agentProxyCiVisibilityExporter = new AgentProxyCiVisibilityExporter({
+          port,
+          tags,
+          isTestDynamicInstrumentationEnabled: true
+        })
+        await agentProxyCiVisibilityExporter._canUseCiVisProtocolPromise
+        expect(agentProxyCiVisibilityExporter._logsWriter).to.be.undefined
+      })
+
+      it('should not process logs', async () => {
+        const mockWriter = {
+          append: sinon.spy(),
+          flush: sinon.spy()
+        }
+        const agentProxyCiVisibilityExporter = new AgentProxyCiVisibilityExporter({
+          port,
+          tags,
+          isTestDynamicInstrumentationEnabled: true
+        })
+        await agentProxyCiVisibilityExporter._canUseCiVisProtocolPromise
+        agentProxyCiVisibilityExporter._logsWriter = mockWriter
+        const log = { message: 'hello' }
+        agentProxyCiVisibilityExporter.exportDiLogs({}, log)
+        expect(mockWriter.append).not.to.have.been.called
+      })
+    })
   })
 
   describe('export', () => {