Adding Span Link support for distributed tracing header extractions with invalid traces (#4874)

mhlidd · web-flow · commit f0df061a4bf9 · 2024-11-18T10:21:33.000-05:00
* initial commit

* updating _links and when links are created

* logging

* add link to instrumentation

* updating integrations to include span links

* fixing syntax error

* fixing ci tests

* updating unit test

* fix ci

* fixing moleculer tests

* safe checking all contexts before getting links
diff --git a/packages/datadog-plugin-amqplib/src/consumer.js b/packages/datadog-plugin-amqplib/src/consumer.js
@@ -26,7 +26,8 @@ class AmqplibConsumerPlugin extends ConsumerPlugin {
         'amqp.consumerTag': fields.consumerTag,
         'amqp.source': fields.source,
         'amqp.destination': fields.destination
-      }
+      },
+      extractedLinks: childOf?._links
     })
 
     if (
diff --git a/packages/datadog-plugin-aws-sdk/src/services/kinesis.js b/packages/datadog-plugin-aws-sdk/src/services/kinesis.js
@@ -42,7 +42,8 @@ class Kinesis extends BaseAwsSdkPlugin {
               {},
               this.requestTags.get(request) || {},
               { 'span.kind': 'server' }
-            )
+            ),
+            extractedLinks: responseExtraction.maybeChildOf._links
           }
           span = plugin.tracer.startSpan('aws.response', options)
           this.enter(span, store)
diff --git a/packages/datadog-plugin-aws-sdk/src/services/sqs.js b/packages/datadog-plugin-aws-sdk/src/services/sqs.js
@@ -33,7 +33,8 @@ class Sqs extends BaseAwsSdkPlugin {
             {},
             this.requestTags.get(request) || {},
             { 'span.kind': 'server' }
-          )
+          ),
+          extractedLinks: contextExtraction.datadogContext._links
         }
         parsedMessageAttributes = contextExtraction.parsedAttributes
         span = plugin.tracer.startSpan('aws.response', options)
diff --git a/packages/datadog-plugin-child_process/test/index.spec.js b/packages/datadog-plugin-child_process/test/index.spec.js
@@ -62,7 +62,8 @@ describe('Child process plugin', () => {
               'span.type': 'system',
               'cmd.exec': JSON.stringify(['ls', '-l'])
             },
-            integrationName: 'system'
+            integrationName: 'system',
+            links: undefined
           }
         )
       })
@@ -84,7 +85,8 @@ describe('Child process plugin', () => {
               'span.type': 'system',
               'cmd.shell': 'ls -l'
             },
-            integrationName: 'system'
+            integrationName: 'system',
+            links: undefined
           }
         )
       })
@@ -109,7 +111,8 @@ describe('Child process plugin', () => {
               'cmd.exec': JSON.stringify(['echo', arg, '']),
               'cmd.truncated': 'true'
             },
-            integrationName: 'system'
+            integrationName: 'system',
+            links: undefined
           }
         )
       })
@@ -134,7 +137,8 @@ describe('Child process plugin', () => {
               'cmd.shell': 'ls -l /h ',
               'cmd.truncated': 'true'
             },
-            integrationName: 'system'
+            integrationName: 'system',
+            links: undefined
           }
         )
       })
@@ -160,7 +164,8 @@ describe('Child process plugin', () => {
               'cmd.exec': JSON.stringify(['ls', '-l', '', '']),
               'cmd.truncated': 'true'
             },
-            integrationName: 'system'
+            integrationName: 'system',
+            links: undefined
           }
         )
       })
@@ -186,7 +191,8 @@ describe('Child process plugin', () => {
               'cmd.shell': 'ls -l /home -t',
               'cmd.truncated': 'true'
             },
-            integrationName: 'system'
+            integrationName: 'system',
+            links: undefined
           }
         )
       })
diff --git a/packages/datadog-plugin-google-cloud-pubsub/src/consumer.js b/packages/datadog-plugin-google-cloud-pubsub/src/consumer.js
@@ -22,7 +22,8 @@ class GoogleCloudPubsubConsumerPlugin extends ConsumerPlugin {
       },
       metrics: {
         'pubsub.ack': 0
-      }
+      },
+      extractedLinks: childOf?._links
     })
     if (this.config.dsmEnabled && message?.attributes) {
       const payloadSize = getMessageSize(message)
diff --git a/packages/datadog-plugin-grpc/src/server.js b/packages/datadog-plugin-grpc/src/server.js
@@ -48,7 +48,8 @@ class GrpcServerPlugin extends ServerPlugin {
       },
       metrics: {
         'grpc.status.code': 0
-      }
+      },
+      extractedLinks: childOf?._links
     })
 
     addMetadataTags(span, metadata, metadataFilter, 'request')
diff --git a/packages/datadog-plugin-jest/src/index.js b/packages/datadog-plugin-jest/src/index.js
@@ -219,7 +219,8 @@ class JestPlugin extends CiPlugin {
           [COMPONENT]: this.constructor.id,
           ...this.testEnvironmentMetadata,
           ...testSuiteMetadata
-        }
+        },
+        extractedLinks: testSessionSpanContext?._links
       })
       this.telemetry.ciVisEvent(TELEMETRY_EVENT_CREATED, 'suite')
       if (_ddTestCodeCoverageEnabled) {
diff --git a/packages/datadog-plugin-kafkajs/src/consumer.js b/packages/datadog-plugin-kafkajs/src/consumer.js
@@ -76,7 +76,8 @@ class KafkajsConsumerPlugin extends ConsumerPlugin {
       },
       metrics: {
         'kafka.partition': partition
-      }
+      },
+      extractedLinks: childOf?._links
     })
     if (this.config.dsmEnabled && message?.headers) {
       const payloadSize = getMessageSize(message)
diff --git a/packages/datadog-plugin-moleculer/src/server.js b/packages/datadog-plugin-moleculer/src/server.js
@@ -9,7 +9,6 @@ class MoleculerServerPlugin extends ServerPlugin {
 
   start ({ action, ctx, broker }) {
     const followsFrom = this.tracer.extract('text_map', ctx.meta)
-
     this.startSpan(this.operationName(), {
       childOf: followsFrom || this.activeSpan,
       service: this.config.service || this.serviceName(),
@@ -19,7 +18,8 @@ class MoleculerServerPlugin extends ServerPlugin {
       meta: {
         'resource.name': action.name,
         ...moleculerTags(broker, ctx, this.config)
-      }
+      },
+      extractedLinks: followsFrom?._links
     })
   }
 }
diff --git a/packages/datadog-plugin-rhea/src/consumer.js b/packages/datadog-plugin-rhea/src/consumer.js
@@ -28,7 +28,8 @@ class RheaConsumerPlugin extends ConsumerPlugin {
         component: 'rhea',
         'amqp.link.source.address': name,
         'amqp.link.role': 'receiver'
-      }
+      },
+      extractedLinks: childOf?._links
     })
 
     if (
diff --git a/packages/datadog-plugin-vitest/src/index.js b/packages/datadog-plugin-vitest/src/index.js
@@ -191,7 +191,8 @@ class VitestPlugin extends CiPlugin {
           [COMPONENT]: this.constructor.id,
           ...this.testEnvironmentMetadata,
           ...testSuiteMetadata
-        }
+        },
+        extractedLinks: testSessionSpanContext?._links
       })
       this.telemetry.ciVisEvent(TELEMETRY_EVENT_CREATED, 'suite')
       const store = storage.getStore()
diff --git a/packages/dd-trace/src/opentracing/propagation/text_map.js b/packages/dd-trace/src/opentracing/propagation/text_map.js
@@ -290,50 +290,63 @@ class TextMapPropagator {
   }
 
   _extractSpanContext (carrier) {
-    let spanContext = null
+    let context = null
     for (const extractor of this._config.tracePropagationStyle.extract) {
-      // add logic to ensure tracecontext headers takes precedence over other extracted headers
-      if (spanContext !== null) {
-        if (this._config.tracePropagationExtractFirst) {
-          return spanContext
-        }
-        if (extractor !== 'tracecontext') {
-          continue
-        }
-        spanContext = this._resolveTraceContextConflicts(
-          this._extractTraceparentContext(carrier), spanContext, carrier)
-        break
-      }
-
+      let extractedContext = null
       switch (extractor) {
         case 'datadog':
-          spanContext = this._extractDatadogContext(carrier)
+          extractedContext = this._extractDatadogContext(carrier)
           break
         case 'tracecontext':
-          spanContext = this._extractTraceparentContext(carrier)
+          extractedContext = this._extractTraceparentContext(carrier)
           break
         case 'b3' && this
           ._config
           .tracePropagationStyle
           .otelPropagators: // TODO: should match "b3 single header" in next major
         case 'b3 single header': // TODO: delete in major after singular "b3"
-          spanContext = this._extractB3SingleContext(carrier)
+          extractedContext = this._extractB3SingleContext(carrier)
           break
         case 'b3':
         case 'b3multi':
-          spanContext = this._extractB3MultiContext(carrier)
+          extractedContext = this._extractB3MultiContext(carrier)
           break
         default:
           if (extractor !== 'baggage') log.warn(`Unknown propagation style: ${extractor}`)
       }
 
+      if (extractedContext === null) { // If the current extractor was invalid, continue to the next extractor
+        continue
+      }
+
+      if (context === null) {
+        context = extractedContext
+        if (this._config.tracePropagationExtractFirst) {
+          return context
+        }
+      } else {
+        // If extractor is tracecontext, add tracecontext specific information to the context
+        if (extractor === 'tracecontext') {
+          context = this._resolveTraceContextConflicts(
+            this._extractTraceparentContext(carrier), context, carrier)
+        }
+        if (extractedContext._traceId && extractedContext._spanId &&
+           extractedContext.toTraceId(true) !== context.toTraceId(true)) {
+          const link = {
+            context: extractedContext,
+            attributes: { reason: 'terminated_context', context_headers: extractor }
+          }
+          context._links.push(link)
+        }
+      }
+
       if (this._config.tracePropagationStyle.extract.includes('baggage') && carrier.baggage) {
-        spanContext = spanContext || new DatadogSpanContext()
-        this._extractBaggageItems(carrier, spanContext)
+        context = context || new DatadogSpanContext()
+        this._extractBaggageItems(carrier, context)
       }
     }
 
-    return spanContext || this._extractSqsdContext(carrier)
+    return context || this._extractSqsdContext(carrier)
   }
 
   _extractDatadogContext (carrier) {
diff --git a/packages/dd-trace/src/opentracing/span_context.js b/packages/dd-trace/src/opentracing/span_context.js
@@ -18,6 +18,7 @@ class DatadogSpanContext {
     this._tags = props.tags || {}
     this._sampling = props.sampling || {}
     this._spanSampling = undefined
+    this._links = props.links || []
     this._baggageItems = props.baggageItems || {}
     this._traceparent = props.traceparent
     this._tracestate = props.tracestate
diff --git a/packages/dd-trace/src/plugins/tracing.js b/packages/dd-trace/src/plugins/tracing.js
@@ -101,9 +101,8 @@ class TracingPlugin extends Plugin {
     }
   }
 
-  startSpan (name, { childOf, kind, meta, metrics, service, resource, type } = {}, enter = true) {
+  startSpan (name, { childOf, kind, meta, metrics, service, resource, type, extractedLinks } = {}, enter = true) {
     const store = storage.getStore()
-
     if (store && childOf === undefined) {
       childOf = store.span
     }
@@ -119,7 +118,8 @@ class TracingPlugin extends Plugin {
         ...meta,
         ...metrics
       },
-      integrationName: type
+      integrationName: type,
+      links: extractedLinks
     })
 
     analyticsSampler.sample(span, this.config.measured)
diff --git a/packages/dd-trace/src/plugins/util/web.js b/packages/dd-trace/src/plugins/util/web.js
@@ -267,7 +267,7 @@ const web = {
       }
     }
 
-    const span = tracer.startSpan(name, { childOf })
+    const span = tracer.startSpan(name, { childOf, extractedLinks: childOf?.links })
 
     return span
   },
diff --git a/packages/dd-trace/test/opentracing/propagation/text_map.spec.js b/packages/dd-trace/test/opentracing/propagation/text_map.spec.js
@@ -692,6 +692,23 @@ describe('TextMapPropagator', () => {
       }
     })
 
+    it('should create span links when traces have inconsistent traceids', () => {
+      // Add a traceparent header and it will prioritize it
+      const traceId = '1111aaaa2222bbbb3333cccc4444dddd'
+      const spanId = '5555eeee6666ffff'
+      textMap.traceparent = `00-${traceId}-${spanId}-01`
+
+      config.tracePropagationStyle.extract = ['tracecontext', 'datadog']
+
+      const first = propagator.extract(textMap)
+
+      expect(first._links.length).to.equal(1)
+      expect(first._links[0].context.toTraceId()).to.equal(textMap['x-datadog-trace-id'])
+      expect(first._links[0].context.toSpanId()).to.equal(textMap['x-datadog-parent-id'])
+      expect(first._links[0].attributes.reason).to.equal('terminated_context')
+      expect(first._links[0].attributes.context_headers).to.equal('datadog')
+    })
+
     describe('with B3 propagation as multiple headers', () => {
       beforeEach(() => {
         config.tracePropagationStyle.extract = ['b3multi']
diff --git a/packages/dd-trace/test/opentracing/span_context.spec.js b/packages/dd-trace/test/opentracing/span_context.spec.js
@@ -48,6 +48,7 @@ describe('SpanContext', () => {
       _tags: {},
       _sampling: { priority: 2 },
       _spanSampling: undefined,
+      _links: [],
       _baggageItems: { foo: 'bar' },
       _noop: noop,
       _trace: {
@@ -77,6 +78,7 @@ describe('SpanContext', () => {
       _tags: {},
       _sampling: {},
       _spanSampling: undefined,
+      _links: [],
       _baggageItems: {},
       _noop: null,
       _trace: {

Original file line number	Diff line number	Diff line change
`@@ -62,7 +62,8 @@ describe('Child process plugin', () => {`
`62`	`62`	`'span.type': 'system',`
`63`	`63`	`'cmd.exec': JSON.stringify(['ls', '-l'])`
`64`	`64`	`},`
`65`		`- integrationName: 'system'`
	`65`	`+ integrationName: 'system',`
	`66`	`+ links: undefined`
`66`	`67`	`}`
`67`	`68`	`)`
`68`	`69`	`})`
`@@ -84,7 +85,8 @@ describe('Child process plugin', () => {`
`84`	`85`	`'span.type': 'system',`
`85`	`86`	`'cmd.shell': 'ls -l'`
`86`	`87`	`},`
`87`		`- integrationName: 'system'`
	`88`	`+ integrationName: 'system',`
	`89`	`+ links: undefined`
`88`	`90`	`}`
`89`	`91`	`)`
`90`	`92`	`})`
`@@ -109,7 +111,8 @@ describe('Child process plugin', () => {`
`109`	`111`	`'cmd.exec': JSON.stringify(['echo', arg, '']),`
`110`	`112`	`'cmd.truncated': 'true'`
`111`	`113`	`},`
`112`		`- integrationName: 'system'`
	`114`	`+ integrationName: 'system',`
	`115`	`+ links: undefined`
`113`	`116`	`}`
`114`	`117`	`)`
`115`	`118`	`})`
`@@ -134,7 +137,8 @@ describe('Child process plugin', () => {`
`134`	`137`	`'cmd.shell': 'ls -l /h ',`
`135`	`138`	`'cmd.truncated': 'true'`
`136`	`139`	`},`
`137`		`- integrationName: 'system'`
	`140`	`+ integrationName: 'system',`
	`141`	`+ links: undefined`
`138`	`142`	`}`
`139`	`143`	`)`
`140`	`144`	`})`
`@@ -160,7 +164,8 @@ describe('Child process plugin', () => {`
`160`	`164`	`'cmd.exec': JSON.stringify(['ls', '-l', '', '']),`
`161`	`165`	`'cmd.truncated': 'true'`
`162`	`166`	`},`
`163`		`- integrationName: 'system'`
	`167`	`+ integrationName: 'system',`
	`168`	`+ links: undefined`
`164`	`169`	`}`
`165`	`170`	`)`
`166`	`171`	`})`
`@@ -186,7 +191,8 @@ describe('Child process plugin', () => {`
`186`	`191`	`'cmd.shell': 'ls -l /home -t',`
`187`	`192`	`'cmd.truncated': 'true'`
`188`	`193`	`},`
`189`		`- integrationName: 'system'`
	`194`	`+ integrationName: 'system',`
	`195`	`+ links: undefined`
`190`	`196`	`}`
`191`	`197`	`)`
`192`	`198`	`})`