feat: span compression

CHANGELOG.asciidoc

@@ -28,6 +28,19 @@ Notes:
 [[release-notes-3.x]]
 === Node.js Agent version 3.x
 
+==== Unreleased
+
+[float]
+===== Breaking changes
+
+[float]
+===== Features
+
+* Adds initial implementaion of
+  https://github.com/elastic/apm/blob/main/specs/agents/handling-huge-traces/tracing-spans-compress.md[span compression]. To enable, set the `spanCompressionEnabled` configuration field to `true`.
+
+[float]
+===== Bug fixes
 
 [[release-notes-3.31.0]]
 ==== 3.31.0 2022/03/23

docs/configuration.asciidoc

@@ -1181,3 +1181,57 @@ require('elastic-apm-node').start({
   ]
 })
 ----
+
+[[span-compression-enabled]]
+==== `spanCompressionEnabled`
+* *Type:* Boolean
+* *Default:* `false`
+* *Env:* `ELASTIC_SPAN_COMPRESSION_ENABLED`
+
+Setting this option to true will enable span compression feature. Span compression reduces the collection, processing, and storage overhead, and removes clutter from the UI. The tradeoff is that some information such as DB statements of all the compressed spans will not be collected.
+
+Example usage:
+
+[source,js]
+----
+require('elastic-apm-node').start({
+  spanCompressionEnabled: true
+})
+----
+
+[[span-compression-exact-match-max-duration]]
+==== `spanCompressionExactMatchMaxDuration`
+* *Type:* String
+* *Default:* `50ms`
+* *Env:* `ELASTIC_SPAN_COMPRESSION_EXACT_MATCH_MAX_DURATION`
+
+Consecutive spans that are exact match and that are under this threshold will be compressed into a single composite span. This option does not apply to composite spans. This reduces the collection, processing, and storage overhead, and removes clutter from the UI. The tradeoff is that the DB statements of all the compressed spans will not be collected.
+
+Supports the duration suffixes ms (milliseconds), s (seconds) and m (minutes).
+
+Example usage:
+
+[source,js]
+----
+require('elastic-apm-node').start({
+  spanCompressionExactMatchMaxDuration:'100ms'
+})
+----
+
+[[span-compression-same-kind-max-duration]]
+==== `spanCompressionSameKindMaxDuration`
+* *Type:* String
+* *Default:* `0ms`
+* *Env:* `ELASTIC_SPAN_COMPRESSION_SAME_KIND_MAX_DURATION`
+
+Consecutive spans to the same destination that are under this threshold will be compressed into a single composite span. This option does not apply to composite spans. This reduces the collection, processing, and storage overhead, and removes clutter from the UI. The tradeoff is that the DB statements of all the compressed spans will not be collected.
+
+Example usage:
+
+[source,js]
+----
+require('elastic-apm-node').start({
+  spanCompressionSameKindMaxDuration:'0ms'
+})
+----
+

index.d.ts

@@ -281,6 +281,9 @@ declare namespace apm {
     sourceLinesErrorLibraryFrames?: number;
     sourceLinesSpanAppFrames?: number;
     sourceLinesSpanLibraryFrames?: number;
+    spanCompressionEnabled?: boolean;
+    spanCompressionExactMatchMaxDuration?: string;
+    spanCompressionSameKindMaxDuration?: string;
     /**
      * @deprecated Use `spanStackTraceMinDuration`.
      */

lib/config.js

@@ -70,6 +70,9 @@ var DEFAULTS = {
   sourceLinesErrorLibraryFrames: 5,
   sourceLinesSpanAppFrames: 0,
   sourceLinesSpanLibraryFrames: 0,
+  spanCompressionEnabled: false,
+  spanCompressionExactMatchMaxDuration: '50ms',
+  spanCompressionSameKindMaxDuration: '0ms',
   // 'spanStackTraceMinDuration' is explicitly *not* included in DEFAULTS
   // because normalizeSpanStackTraceMinDuration() needs to know if a value
   // was provided by the user.
@@ -136,6 +139,9 @@ var ENV_TABLE = {
   sourceLinesErrorLibraryFrames: 'ELASTIC_APM_SOURCE_LINES_ERROR_LIBRARY_FRAMES',
   sourceLinesSpanAppFrames: 'ELASTIC_APM_SOURCE_LINES_SPAN_APP_FRAMES',
   sourceLinesSpanLibraryFrames: 'ELASTIC_APM_SOURCE_LINES_SPAN_LIBRARY_FRAMES',
+  spanCompressionEnabled: 'ELASTIC_SPAN_COMPRESSION_ENABLED',
+  spanCompressionExactMatchMaxDuration: 'ELASTIC_SPAN_COMPRESSION_EXACT_MATCH_MAX_DURATION',
+  spanCompressionSameKindMaxDuration: 'ELASTIC_SPAN_COMPRESSION_SAME_KIND_MAX_DURATION',
   spanStackTraceMinDuration: 'ELASTIC_APM_SPAN_STACK_TRACE_MIN_DURATION',
   spanFramesMinDuration: 'ELASTIC_APM_SPAN_FRAMES_MIN_DURATION',
   stackTraceLimit: 'ELASTIC_APM_STACK_TRACE_LIMIT',
@@ -173,6 +179,7 @@ var BOOL_OPTS = [
   'instrument',
   'instrumentIncomingHTTPRequests',
   'logUncaughtExceptions',
+  'spanCompressionEnabled',
   'usePathAsTransactionName',
   'verifyServerCert'
 ]
@@ -215,7 +222,18 @@ var DURATION_OPTS = [
     allowedUnits: ['ms', 's', 'm'],
     allowNegative: false
   },
-
+  {
+    name: 'spanCompressionExactMatchMaxDuration',
+    defaultUnit: 'ms',
+    allowedUnits: ['ms', 's', 'm'],
+    allowNegative: false
+  },
+  {
+    name: 'spanCompressionSameKindMaxDuration',
+    defaultUnit: 'ms',
+    allowedUnits: ['ms', 's', 'm'],
+    allowNegative: false
+  },
   {
     // Deprecated: use `spanStackTraceMinDuration`.
     name: 'spanFramesMinDuration',

lib/instrumentation/generic-span.js

@@ -6,6 +6,8 @@ const config = require('../config')
 const constants = require('../constants')
 const Timer = require('./timer')
 const TraceContext = require('../tracecontext')
+const { SpanCompression } = require('./span-compression')
+
 module.exports = GenericSpan
 
 function GenericSpan (agent, ...args) {
@@ -17,13 +19,21 @@ function GenericSpan (agent, ...args) {
 
   this._context = TraceContext.startOrResume(opts.childOf, agent._conf, opts.tracestate)
 
+  this._parentSpan = null
+  if (opts.childOf instanceof GenericSpan) {
+    this.setParentSpan(opts.childOf)
+  }
+  this._compression = new SpanCompression(agent)
+  this._compression.setBufferedSpan(null)
+
   this._agent = agent
   this._labels = null
   this._ids = null // Populated by sub-types of GenericSpan
 
   this.timestamp = this._timer.start
   this.ended = false
   this._duration = null // Duration in milliseconds. Set on `.end()`.
+  this._endTimestamp = null
 
   this.outcome = constants.OUTCOME_UNKNOWN
 
@@ -141,3 +151,48 @@ GenericSpan.prototype._isValidOutcome = function (outcome) {
     outcome === constants.OUTCOME_SUCCESS ||
     outcome === constants.OUTCOME_UNKNOWN
 }
+
+GenericSpan.prototype.isCompressionEligible = function () {
+  if (!this.getParentSpan()) {
+    return false
+  }
+
+  // TODO: Once we have every span type marked as an exit span and
+  //       a trace context api that can detect whether the trace context
+  //       has been propagated to a span or not this hard coded list
+  //       will be replaced with detection routines looks for exit spans
+  //       whose trace context has _not_ been propagated.
+
+  const subtypes = ['elasticsearch', 'redis', 'memcached', 'mongodb', 'mysql',
+    'postgresql']
+
+  return subtypes.indexOf(this.subtype) !== -1
+}
+
+GenericSpan.prototype.tryToCompress = function (spanToCompress) {
+  return this._compression.tryToCompress(this, spanToCompress)
+}
+
+GenericSpan.prototype.setParentSpan = function (span) {
+  this._parentSpan = span
+}
+
+GenericSpan.prototype.getParentSpan = function (span) {
+  return this._parentSpan
+}
+
+GenericSpan.prototype.getBufferedSpan = function () {
+  return this._compression.getBufferedSpan()
+}
+
+GenericSpan.prototype.setBufferedSpan = function (span) {
+  return this._compression.setBufferedSpan(span)
+}
+
+GenericSpan.prototype.isCompositeSameKind = function () {
+  return this._compression.isCompositeSameKind()
+}
+
+GenericSpan.prototype.isComposite = function () {
+  return this._compression.isComposite()
+}

lib/instrumentation/index.js

@@ -312,6 +312,11 @@ Instrumentation.prototype.addEndedTransaction = function (transaction) {
     return
   }
 
+  // if I have ended and I have something buffered, send that buffered thing
+  if (transaction.getBufferedSpan()) {
+    this.encodeAndSendSpan(transaction.getBufferedSpan())
+  }
+
   var payload = agent._transactionFilters.process(transaction._encode())
   if (!payload) {
     agent.logger.debug('transaction ignored by filter %o', { trans: transaction.id, trace: transaction.traceId })
@@ -345,6 +350,33 @@ Instrumentation.prototype.addEndedSpan = function (span) {
     return
   }
 
+  // TODO: branching code for "is compression enabled or not"
+  if (!this._agent._conf.spanCompressionEnabled) {
+    this.encodeAndSendSpan(span)
+  } else {
+    // if I have ended and I have something buffered, send that buffered thing
+    if (span.getBufferedSpan()) {
+      this.encodeAndSendSpan(span.getBufferedSpan())
+    }
+
+    if (!span.isCompressionEligible()) {
+      // span is not compressible -- send
+      this.encodeAndSendSpan(span)
+    } else if (!span.getParentSpan().getBufferedSpan()) {
+      // span is compressible and there's nothing buffered
+      // add to buffer, move on
+      span.getParentSpan().setBufferedSpan(span)
+    } else if (!span.getParentSpan().getBufferedSpan().tryToCompress(span)) {
+      // we could not compress span so SEND bufferend span
+      // and buffer the span we could not compress
+      this.encodeAndSendSpan(span.getParentSpan().getBufferedSpan())
+      span.getParentSpan().setBufferedSpan(span)
+    }
+  }
+}
+
+Instrumentation.prototype.encodeAndSendSpan = function (span) {
+  const agent = this._agent
   // Note this error as an "inflight" event. See Agent#flush().
   const inflightEvents = agent._inflightEvents
   inflightEvents.add(span.id)