Add support for endpoint_counts (#4980)

Also:

* Extract event.json creation in profile exporters so it can be shared between all exporters.
File exporter will be writing it so we can more easily write integration tests.

* Extract web span handling in profiler

Author: szegedi

integration-tests/profiler/profiler.spec.js

@@ -75,6 +75,12 @@ function processExitPromise (proc, timeout, expectBadExit = false) {
 }
 
 async function getLatestProfile (cwd, pattern) {
+  const pprofGzipped = await readLatestFile(cwd, pattern)
+  const pprofUnzipped = zlib.gunzipSync(pprofGzipped)
+  return { profile: Profile.decode(pprofUnzipped), encoded: pprofGzipped.toString('base64') }
+}
+
+async function readLatestFile (cwd, pattern) {
   const dirEntries = await fs.readdir(cwd)
   // Get the latest file matching the pattern
   const pprofEntries = dirEntries.filter(name => pattern.test(name))
@@ -83,9 +89,7 @@ async function getLatestProfile (cwd, pattern) {
     .map(name => ({ name, modified: fsync.statSync(path.join(cwd, name), { bigint: true }).mtimeNs }))
     .reduce((a, b) => a.modified > b.modified ? a : b)
     .name
-  const pprofGzipped = await fs.readFile(path.join(cwd, pprofEntry))
-  const pprofUnzipped = zlib.gunzipSync(pprofGzipped)
-  return { profile: Profile.decode(pprofUnzipped), encoded: pprofGzipped.toString('base64') }
+  return await fs.readFile(path.join(cwd, pprofEntry))
 }
 
 function expectTimeout (messagePromise, allowErrors = false) {
@@ -212,6 +216,10 @@ describe('profiler', () => {
       await processExitPromise(proc, 30000)
       const procEnd = BigInt(Date.now() * 1000000)
 
+      // Must've counted the number of times each endpoint was hit
+      const event = JSON.parse((await readLatestFile(cwd, /^event_.+\.json$/)).toString())
+      assert.deepEqual(event.endpoint_counts, { 'endpoint-0': 1, 'endpoint-1': 1, 'endpoint-2': 1 })
+
       const { profile, encoded } = await getLatestProfile(cwd, /^wall_.+\.pprof$/)
 
       // We check the profile for following invariants:

packages/dd-trace/src/profiling/exporters/agent.js

@@ -3,13 +3,13 @@
 const retry = require('retry')
 const { request: httpRequest } = require('http')
 const { request: httpsRequest } = require('https')
+const { EventSerializer } = require('./event_serializer')
 
 // TODO: avoid using dd-trace internals. Make this a separate module?
 const docker = require('../../exporters/common/docker')
 const FormData = require('../../exporters/common/form-data')
 const { storage } = require('../../../../datadog-core')
 const version = require('../../../../../package.json').version
-const os = require('os')
 const { urlToHttpOptions } = require('url')
 const perf = require('perf_hooks').performance
 
@@ -89,83 +89,24 @@ function computeRetries (uploadTimeout) {
   return [tries, Math.floor(uploadTimeout)]
 }
 
-class AgentExporter {
-  constructor ({ url, logger, uploadTimeout, env, host, service, version, libraryInjected, activation } = {}) {
+class AgentExporter extends EventSerializer {
+  constructor (config = {}) {
+    super(config)
+    const { url, logger, uploadTimeout } = config
     this._url = url
     this._logger = logger
 
     const [backoffTries, backoffTime] = computeRetries(uploadTimeout)
 
     this._backoffTime = backoffTime
     this._backoffTries = backoffTries
-    this._env = env
-    this._host = host
-    this._service = service
-    this._appVersion = version
-    this._libraryInjected = !!libraryInjected
-    this._activation = activation || 'unknown'
   }
 
-  export ({ profiles, start, end, tags }) {
+  export (exportSpec) {
+    const { profiles } = exportSpec
     const fields = []
 
-    function typeToFile (type) {
-      return `${type}.pprof`
-    }
-
-    const event = JSON.stringify({
-      attachments: Object.keys(profiles).map(typeToFile),
-      start: start.toISOString(),
-      end: end.toISOString(),
-      family: 'node',
-      version: '4',
-      tags_profiler: [
-        'language:javascript',
-        'runtime:nodejs',
-        `runtime_arch:${process.arch}`,
-        `runtime_os:${process.platform}`,
-        `runtime_version:${process.version}`,
-        `process_id:${process.pid}`,
-        `profiler_version:${version}`,
-        'format:pprof',
-        ...Object.entries(tags).map(([key, value]) => `${key}:${value}`)
-      ].join(','),
-      info: {
-        application: {
-          env: this._env,
-          service: this._service,
-          start_time: new Date(perf.nodeTiming.nodeStart + perf.timeOrigin).toISOString(),
-          version: this._appVersion
-        },
-        platform: {
-          hostname: this._host,
-          kernel_name: os.type(),
-          kernel_release: os.release(),
-          kernel_version: os.version()
-        },
-        profiler: {
-          activation: this._activation,
-          ssi: {
-            mechanism: this._libraryInjected ? 'injected_agent' : 'none'
-          },
-          version
-        },
-        runtime: {
-          // Using `nodejs` for consistency with the existing `runtime` tag.
-          // Note that the event `family` property uses `node`, as that's what's
-          // proscribed by the Intake API, but that's an internal enum and is
-          // not customer visible.
-          engine: 'nodejs',
-          // strip off leading 'v'. This makes the format consistent with other
-          // runtimes (e.g. Ruby) but not with the existing `runtime_version` tag.
-          // We'll keep it like this as we want cross-engine consistency. We
-          // also aren't changing the format of the existing tag as we don't want
-          // to break it.
-          version: process.version.substring(1)
-        }
-      }
-    })
-
+    const event = this.getEventJSON(exportSpec)
     fields.push(['event', event, {
       filename: 'event.json',
       contentType: 'application/json'
@@ -181,7 +122,7 @@ class AgentExporter {
         return `Adding ${type} profile to agent export: ` + bytes
       })
 
-      const filename = typeToFile(type)
+      const filename = this.typeToFile(type)
       fields.push([filename, buffer, {
         filename,
         contentType: 'application/octet-stream'

packages/dd-trace/src/profiling/exporters/event_serializer.js

@@ -0,0 +1,76 @@
+const os = require('os')
+const perf = require('perf_hooks').performance
+const version = require('../../../../../package.json').version
+
+class EventSerializer {
+  constructor ({ env, host, service, version, libraryInjected, activation } = {}) {
+    this._env = env
+    this._host = host
+    this._service = service
+    this._appVersion = version
+    this._libraryInjected = !!libraryInjected
+    this._activation = activation || 'unknown'
+  }
+
+  typeToFile (type) {
+    return `${type}.pprof`
+  }
+
+  getEventJSON ({ profiles, start, end, tags = {}, endpointCounts }) {
+    return JSON.stringify({
+      attachments: Object.keys(profiles).map(t => this.typeToFile(t)),
+      start: start.toISOString(),
+      end: end.toISOString(),
+      family: 'node',
+      version: '4',
+      tags_profiler: [
+        'language:javascript',
+        'runtime:nodejs',
+        `runtime_arch:${process.arch}`,
+        `runtime_os:${process.platform}`,
+        `runtime_version:${process.version}`,
+        `process_id:${process.pid}`,
+        `profiler_version:${version}`,
+        'format:pprof',
+        ...Object.entries(tags).map(([key, value]) => `${key}:${value}`)
+      ].join(','),
+      endpoint_counts: endpointCounts,
+      info: {
+        application: {
+          env: this._env,
+          service: this._service,
+          start_time: new Date(perf.nodeTiming.nodeStart + perf.timeOrigin).toISOString(),
+          version: this._appVersion
+        },
+        platform: {
+          hostname: this._host,
+          kernel_name: os.type(),
+          kernel_release: os.release(),
+          kernel_version: os.version()
+        },
+        profiler: {
+          activation: this._activation,
+          ssi: {
+            mechanism: this._libraryInjected ? 'injected_agent' : 'none'
+          },
+          version
+        },
+        runtime: {
+          // Using `nodejs` for consistency with the existing `runtime` tag.
+          // Note that the event `family` property uses `node`, as that's what's
+          // proscribed by the Intake API, but that's an internal enum and is
+          // not customer visible.
+          engine: 'nodejs',
+          // strip off leading 'v'. This makes the format consistent with other
+          // runtimes (e.g. Ruby) but not with the existing `runtime_version` tag.
+          // We'll keep it like this as we want cross-engine consistency. We
+          // also aren't changing the format of the existing tag as we don't want
+          // to break it.
+          version: process.version.substring(1)
+        }
+      }
+    })
+  }
+}
+
+module.exports = { EventSerializer }

packages/dd-trace/src/profiling/exporters/file.js

@@ -4,25 +4,29 @@ const fs = require('fs')
 const { promisify } = require('util')
 const { threadId } = require('worker_threads')
 const writeFile = promisify(fs.writeFile)
+const { EventSerializer } = require('./event_serializer')
 
 function formatDateTime (t) {
   const pad = (n) => String(n).padStart(2, '0')
   return `${t.getUTCFullYear()}${pad(t.getUTCMonth() + 1)}${pad(t.getUTCDate())}` +
          `T${pad(t.getUTCHours())}${pad(t.getUTCMinutes())}${pad(t.getUTCSeconds())}Z`
 }
 
-class FileExporter {
-  constructor ({ pprofPrefix } = {}) {
+class FileExporter extends EventSerializer {
+  constructor (config = {}) {
+    super(config)
+    const { pprofPrefix } = config
     this._pprofPrefix = pprofPrefix || ''
   }
 
-  export ({ profiles, end }) {
+  export (exportSpec) {
+    const { profiles, end } = exportSpec
     const types = Object.keys(profiles)
     const dateStr = formatDateTime(end)
     const tasks = types.map(type => {
       return writeFile(`${this._pprofPrefix}${type}_worker_${threadId}_${dateStr}.pprof`, profiles[type])
     })
-
+    tasks.push(writeFile(`event_worker_${threadId}_${dateStr}.json`, this.getEventJSON(exportSpec)))
     return Promise.all(tasks)
   }
 }

packages/dd-trace/src/profiling/profiler.js

@@ -4,9 +4,11 @@ const { EventEmitter } = require('events')
 const { Config } = require('./config')
 const { snapshotKinds } = require('./constants')
 const { threadNamePrefix } = require('./profilers/shared')
+const { isWebServerSpan, endpointNameFromTags, getStartedSpans } = require('./webspan-utils')
 const dc = require('dc-polyfill')
 
 const profileSubmittedChannel = dc.channel('datadog:profiling:profile-submitted')
+const spanFinishedChannel = dc.channel('dd-trace:span:finish')
 
 function maybeSourceMap (sourceMap, SourceMapper, debug) {
   if (!sourceMap) return
@@ -21,6 +23,20 @@ function logError (logger, err) {
   }
 }
 
+function findWebSpan (startedSpans, spanId) {
+  for (let i = startedSpans.length; --i >= 0;) {
+    const ispan = startedSpans[i]
+    const context = ispan.context()
+    if (context._spanId === spanId) {
+      if (isWebServerSpan(context._tags)) {
+        return true
+      }
+      spanId = context._parentId
+    }
+  }
+  return false
+}
+
 class Profiler extends EventEmitter {
   constructor () {
     super()
@@ -30,6 +46,7 @@ class Profiler extends EventEmitter {
     this._timer = undefined
     this._lastStart = undefined
     this._timeoutInterval = undefined
+    this.endpointCounts = new Map()
   }
 
   start (options) {
@@ -82,6 +99,11 @@ class Profiler extends EventEmitter {
         this._logger.debug(`Started ${profiler.type} profiler in ${threadNamePrefix} thread`)
       }
 
+      if (config.endpointCollectionEnabled) {
+        this._spanFinishListener = this._onSpanFinish.bind(this)
+        spanFinishedChannel.subscribe(this._spanFinishListener)
+      }
+
       this._capture(this._timeoutInterval, start)
       return true
     } catch (e) {
@@ -117,6 +139,11 @@ class Profiler extends EventEmitter {
 
     this._enabled = false
 
+    if (this._spanFinishListener !== undefined) {
+      spanFinishedChannel.unsubscribe(this._spanFinishListener)
+      this._spanFinishListener = undefined
+    }
+
     for (const profiler of this._config.profilers) {
       profiler.stop()
       this._logger.debug(`Stopped ${profiler.type} profiler in ${threadNamePrefix} thread`)
@@ -137,6 +164,26 @@ class Profiler extends EventEmitter {
     }
   }
 
+  _onSpanFinish (span) {
+    const context = span.context()
+    const tags = context._tags
+    if (!isWebServerSpan(tags)) return
+
+    const endpointName = endpointNameFromTags(tags)
+    if (!endpointName) return
+
+    // Make sure this is the outermost web span, just in case so we don't overcount
+    if (findWebSpan(getStartedSpans(context), context._parentId)) return
+
+    let counter = this.endpointCounts.get(endpointName)
+    if (counter === undefined) {
+      counter = { count: 1 }
+      this.endpointCounts.set(endpointName, counter)
+    } else {
+      counter.count++
+    }
+  }
+
   async _collect (snapshotKind, restart = true) {
     if (!this._enabled) return
 
@@ -194,18 +241,23 @@ class Profiler extends EventEmitter {
 
   _submit (profiles, start, end, snapshotKind) {
     const { tags } = this._config
-    const tasks = []
 
-    tags.snapshot = snapshotKind
-    for (const exporter of this._config.exporters) {
-      const task = exporter.export({ profiles, start, end, tags })
-        .catch(err => {
-          if (this._logger) {
-            this._logger.warn(err)
-          }
-        })
-      tasks.push(task)
+    // Flatten endpoint counts
+    const endpointCounts = {}
+    for (const [endpoint, { count }] of this.endpointCounts) {
+      endpointCounts[endpoint] = count
     }
+    this.endpointCounts.clear()
+
+    tags.snapshot = snapshotKind
+    const exportSpec = { profiles, start, end, tags, endpointCounts }
+    const tasks = this._config.exporters.map(exporter =>
+      exporter.export(exportSpec).catch(err => {
+        if (this._logger) {
+          this._logger.warn(err)
+        }
+      })
+    )
 
     return Promise.all(tasks)
   }