Respect and propagate X-Datadog-Sampling-Priority.

benchmark/core.js

@@ -16,6 +16,7 @@ const Writer = proxyquire('../src/writer', {
   './platform': { request: () => Promise.resolve() }
 })
 const Sampler = require('../src/sampler')
+const PrioritySampler = require('../src/priority_sampler')
 const format = require('../src/format')
 const encode = require('../src/encode')
 const config = new Config({ service: 'benchmark' })
@@ -30,6 +31,7 @@ let writer
 let sampler
 let queue
 
+const prioritySampler = new PrioritySampler('bench')
 const traceStub = require('./stubs/trace')
 const spanStub = require('./stubs/span')
 
@@ -44,7 +46,7 @@ suite
   })
   .add('TextMapPropagator#inject', {
     onStart () {
-      propagator = new TextMapPropagator()
+      propagator = new TextMapPropagator(prioritySampler)
       carrier = {}
       spanContext = new DatadogSpanContext({
         traceId: new Uint64BE(0x12345678, 0x12345678),
@@ -58,7 +60,7 @@ suite
   })
   .add('TextMapPropagator#extract', {
     onStart () {
-      propagator = new TextMapPropagator()
+      propagator = new TextMapPropagator(prioritySampler)
       carrier = {
         'x-datadog-trace-id': '1234567891234567',
         'x-datadog-parent-id': '1234567891234567',

benchmark/platform/node.js

@@ -34,7 +34,7 @@ suite
           protocol: 'http:',
           hostname: 'test',
           port: '8080',
-          path: '/v0.3/traces',
+          path: '/v0.4/traces',
           method: 'PUT',
           headers: {
             'Content-Type': 'application/msgpack'

benchmark/stubs/span.js

@@ -18,17 +18,19 @@ const span = {
     trace: {
       started: [span, span],
       finished: [span, span]
-    }
+    },
+    tags: {
+      resource: '/resource',
+      type: 'web',
+      error: true
+    },
+    metrics: {
+      [SAMPLE_RATE_METRIC_KEY]: 1
+    },
+    sampled: true,
+    sampling: {}
   }),
   _operationName: 'operation',
-  _tags: {
-    resource: '/resource',
-    type: 'web',
-    error: true
-  },
-  _metrics: {
-    [SAMPLE_RATE_METRIC_KEY]: 1
-  },
   _startTime: 1500000000000.123456,
   _duration: 100
 }

ext/index.js

@@ -0,0 +1,9 @@
+'use strict'
+
+const priority = require('./priority')
+const tags = require('./tags')
+
+module.exports = {
+  priority,
+  tags
+}

ext/priority.js

@@ -0,0 +1,8 @@
+'use strict'
+
+module.exports = {
+  USER_REJECT: -1,
+  AUTO_REJECT: 0,
+  AUTO_KEEP: 1,
+  USER_KEEP: 2
+}

ext/tags.js

@@ -0,0 +1,8 @@
+'use strict'
+
+module.exports = {
+  SERVICE_NAME: 'service.name',
+  RESOURCE_NAME: 'resource.name',
+  SPAN_TYPE: 'span.type',
+  SAMPLING_PRIORITY: 'sampling.priority'
+}

src/format.js

@@ -11,9 +11,9 @@ const map = {
 function format (span) {
   const formatted = formatSpan(span)
 
-  extractError(formatted, span._error)
-  extractTags(formatted, span._tags)
-  extractMetrics(formatted, span._metrics)
+  extractError(formatted, span)
+  extractTags(formatted, span)
+  extractMetrics(formatted, span)
 
   return formatted
 }
@@ -37,7 +37,9 @@ function formatSpan (span) {
   }
 }
 
-function extractTags (trace, tags) {
+function extractTags (trace, span) {
+  const tags = span.context().tags
+
   Object.keys(tags).forEach(tag => {
     switch (tag) {
       case 'service.name':
@@ -62,20 +64,28 @@ function extractTags (trace, tags) {
   })
 }
 
-function extractMetrics (trace, metrics) {
-  Object.keys(metrics).forEach(metric => {
-    if (typeof metrics[metric] === 'number') {
-      trace.metrics[metric] = metrics[metric]
-    }
-  })
-}
+function extractError (trace, span) {
+  const error = span._error
 
-function extractError (trace, error) {
   if (error) {
     trace.meta['error.msg'] = error.message
     trace.meta['error.type'] = error.name
     trace.meta['error.stack'] = error.stack
   }
 }
 
+function extractMetrics (trace, span) {
+  const spanContext = span.context()
+
+  Object.keys(spanContext.metrics).forEach(metric => {
+    if (typeof spanContext.metrics[metric] === 'number') {
+      trace.metrics[metric] = spanContext.metrics[metric]
+    }
+  })
+
+  if (spanContext.sampling.priority !== undefined) {
+    trace.metrics['_sampling_priority_v1'] = spanContext.sampling.priority
+  }
+}
+
 module.exports = format

src/opentracing/propagation/text_map.js

@@ -6,39 +6,67 @@ const DatadogSpanContext = require('../span_context')
 
 const traceKey = 'x-datadog-trace-id'
 const spanKey = 'x-datadog-parent-id'
+const samplingKey = 'x-datadog-sampling-priority'
 const baggagePrefix = 'ot-baggage-'
 const baggageExpr = new RegExp(`^${baggagePrefix}(.+)$`)
 
 class TextMapPropagator {
+  constructor (prioritySampler) {
+    this._prioritySampler = prioritySampler
+  }
+
   inject (spanContext, carrier) {
     carrier[traceKey] = new Int64BE(spanContext.traceId.toBuffer()).toString()
     carrier[spanKey] = new Int64BE(spanContext.spanId.toBuffer()).toString()
 
-    spanContext.baggageItems && Object.keys(spanContext.baggageItems).forEach(key => {
-      carrier[baggagePrefix + key] = String(spanContext.baggageItems[key])
-    })
+    this._injectSamplingPriority(spanContext, carrier)
+    this._injectBaggageItems(spanContext, carrier)
   }
 
   extract (carrier) {
     if (!carrier[traceKey] || !carrier[spanKey]) {
       return null
     }
 
-    const baggageItems = {}
+    const spanContext = new DatadogSpanContext({
+      traceId: new Uint64BE(carrier[traceKey], 10),
+      spanId: new Uint64BE(carrier[spanKey], 10)
+    })
+
+    this._extractBaggageItems(carrier, spanContext)
+    this._extractSamplingPriority(carrier, spanContext)
+
+    return spanContext
+  }
+
+  _injectSamplingPriority (spanContext, carrier) {
+    this._prioritySampler.sample(spanContext)
 
+    carrier[samplingKey] = spanContext.sampling.priority.toString()
+  }
+
+  _injectBaggageItems (spanContext, carrier) {
+    spanContext.baggageItems && Object.keys(spanContext.baggageItems).forEach(key => {
+      carrier[baggagePrefix + key] = String(spanContext.baggageItems[key])
+    })
+  }
+
+  _extractBaggageItems (carrier, spanContext) {
     Object.keys(carrier).forEach(key => {
       const match = key.match(baggageExpr)
 
       if (match) {
-        baggageItems[match[1]] = carrier[key]
+        spanContext.baggageItems[match[1]] = carrier[key]
       }
     })
+  }
 
-    return new DatadogSpanContext({
-      traceId: new Uint64BE(carrier[traceKey], 10),
-      spanId: new Uint64BE(carrier[spanKey], 10),
-      baggageItems
-    })
+  _extractSamplingPriority (carrier, spanContext) {
+    const priority = parseInt(carrier[samplingKey], 10)
+
+    if (this._prioritySampler.validate(priority)) {
+      spanContext.sampling.priority = priority
+    }
   }
 }
 

src/opentracing/span.js

@@ -10,28 +10,30 @@ const constants = require('../constants')
 const SAMPLE_RATE_METRIC_KEY = constants.SAMPLE_RATE_METRIC_KEY
 
 class DatadogSpan extends Span {
-  constructor (tracer, sampler, fields) {
+  constructor (tracer, recorder, sampler, prioritySampler, fields) {
     super()
 
     const startTime = fields.startTime || platform.now()
     const operationName = fields.operationName
     const parent = fields.parent || null
-    const tags = fields.tags || {}
+    const tags = Object.assign({}, fields.tags)
     const metrics = {
       [SAMPLE_RATE_METRIC_KEY]: sampler.rate()
     }
 
     this._parentTracer = tracer
     this._sampler = sampler
+    this._recorder = recorder
+    this._prioritySampler = prioritySampler
     this._operationName = operationName
-    this._tags = Object.assign({}, tags)
-    this._metrics = metrics
     this._startTime = startTime
 
     this._spanContext = this._createContext(parent)
+    this._spanContext.tags = tags
+    this._spanContext.metrics = metrics
   }
 
-  _createContext (parent) {
+  _createContext (parent, tags) {
     let spanContext
 
     if (parent) {
@@ -40,6 +42,7 @@ class DatadogSpan extends Span {
         spanId: platform.id(),
         parentId: parent.spanId,
         sampled: parent.sampled,
+        sampling: parent.sampling,
         baggageItems: Object.assign({}, parent.baggageItems),
         trace: parent.trace.started.length !== parent.trace.finished.length ? parent.trace : null
       })
@@ -80,7 +83,7 @@ class DatadogSpan extends Span {
   _addTags (keyValuePairs) {
     try {
       Object.keys(keyValuePairs).forEach(key => {
-        this._tags[key] = String(keyValuePairs[key])
+        this._spanContext.tags[key] = String(keyValuePairs[key])
       })
     } catch (e) {
       log.error(e)
@@ -96,9 +99,10 @@ class DatadogSpan extends Span {
 
     this._duration = finishTime - this._startTime
     this._spanContext.trace.finished.push(this)
+    this._prioritySampler.sample(this)
 
     if (this._spanContext.sampled) {
-      this._parentTracer._record(this)
+      this._recorder.record(this)
     }
   }
 }

src/opentracing/span_context.js

@@ -9,7 +9,10 @@ class DatadogSpanContext extends SpanContext {
     this.traceId = props.traceId
     this.spanId = props.spanId
     this.parentId = props.parentId || null
+    this.tags = props.tags || {}
+    this.metrics = props.metrics || {}
     this.sampled = props.sampled === undefined || props.sampled
+    this.sampling = props.sampling || {}
     this.baggageItems = props.baggageItems || {}
     this.trace = props.trace || {
       started: [],

src/opentracing/tracer.js

@@ -5,8 +5,10 @@ const Tracer = opentracing.Tracer
 const Reference = opentracing.Reference
 const Span = require('./span')
 const SpanContext = require('./span_context')
+const Writer = require('../writer')
 const Recorder = require('../recorder')
 const Sampler = require('../sampler')
+const PrioritySampler = require('../priority_sampler')
 const TextMapPropagator = require('./propagation/text_map')
 const HttpPropagator = require('./propagation/http')
 const BinaryPropagator = require('./propagation/binary')
@@ -23,13 +25,15 @@ class DatadogTracer extends Tracer {
     this._url = config.url
     this._env = config.env
     this._tags = config.tags
-    this._recorder = new Recorder(config.url, config.flushInterval, config.bufferSize)
+    this._prioritySampler = new PrioritySampler(config.env)
+    this._writer = new Writer(this._prioritySampler, config.url, config.bufferSize)
+    this._recorder = new Recorder(this._writer, config.flushInterval)
     this._recorder.init()
     this._sampler = new Sampler(config.sampleRate)
     this._propagators = {
-      [opentracing.FORMAT_TEXT_MAP]: new TextMapPropagator(),
-      [opentracing.FORMAT_HTTP_HEADERS]: new HttpPropagator(),
-      [opentracing.FORMAT_BINARY]: new BinaryPropagator()
+      [opentracing.FORMAT_TEXT_MAP]: new TextMapPropagator(this._prioritySampler),
+      [opentracing.FORMAT_HTTP_HEADERS]: new HttpPropagator(this._prioritySampler),
+      [opentracing.FORMAT_BINARY]: new BinaryPropagator(this._prioritySampler)
     }
   }
 
@@ -38,26 +42,20 @@ class DatadogTracer extends Tracer {
       'resource.name': name
     }
 
-    if (this._service) {
-      tags['service.name'] = this._service
-    }
+    tags['service.name'] = this._service
 
     if (this._env) {
       tags.env = this._env
     }
 
-    return new Span(this, this._sampler, {
+    return new Span(this, this._recorder, this._sampler, this._prioritySampler, {
       operationName: fields.operationName || name,
       parent: getParent(fields.references),
       tags: Object.assign(tags, this._tags, fields.tags),
       startTime: fields.startTime
     })
   }
 
-  _record (span) {
-    this._recorder.record(span)
-  }
-
   _inject (spanContext, format, carrier) {
     try {
       this._propagators[format].inject(spanContext, carrier)

src/plugins/http.js

@@ -15,7 +15,7 @@ function patch (http, tracer, config) {
       const uri = extractUrl(options)
       const method = (options.method || 'GET').toUpperCase()
 
-      if (uri === `${tracer._url.href}/v0.3/traces`) {
+      if (uri === `${tracer._url.href}/v0.4/traces`) {
         return request.apply(this, [options, callback])
       }