BB-764: All Kafka consumers use span Links, never parent-child

Convention: every Kafka consumer read starts a new root trace and adds
an OTEL Link to the upstream span from the message headers. No Kafka
hop uses parent-child. Rationale:

- Async work (replication, lifecycle, GC, notifications) can fire
  minutes to hours after the original S3 request. Parent-child puts
  both in one trace → huge wall-clock gap in the waterfall, traces with
  millions of spans for high-volume buckets.
- Links keep each trace small and focused. Jaeger renders them as
  clickable cross-trace jumps.
- OTEL messaging semantic conventions recommend links for async hops
  with significant time gaps.

Changes:
- BackbeatConsumer._processTask always calls startLinkedSpanFromKafkaEntry.
  Removed entryHasLinkHeaders auto-detect and the parent-child code path.
- kafkaTraceContext.js simplified: removed linkHeadersFromCurrentContext()
  and linkContextFromKafkaHeaders(). startLinkedSpanFromKafkaEntry now
  reads standard traceparent/tracestate. startSpanFromKafkaEntry kept
  as backward-compat alias (same function ref).
- Producer-side: LifecycleTask.js and ReplicationAPI.js switched from
  linkHeadersFromCurrentContext() to traceHeadersFromCurrentContext()
  (same output, just standard traceparent headers).

Issue: BB-764

Author: delthas

extensions/lifecycle/tasks/LifecycleTask.js

@@ -24,10 +24,7 @@ const ReplicationAPI = require('../../replication/ReplicationAPI');
 const { LifecycleMetrics, LIFECYCLE_MARKER_METRICS_LOCATION } = require('../LifecycleMetrics');
 const locationsConfig = require('../../../conf/locationConfig.json') || {};
 const { rulesSupportTransition } = require('../util/rules');
-const {
-    linkHeadersFromCurrentContext,
-    traceHeadersFromCurrentContext,
-} = require('../../../lib/tracing/kafkaTraceContext');
+const { traceHeadersFromCurrentContext } = require('../../../lib/tracing/kafkaTraceContext');
 const { decode } = versioning.VersionID;
 
 const errorTransitionInProgress = errors.InternalError.
@@ -194,11 +191,10 @@ class LifecycleTask extends BackbeatTask {
             location,
             Date.now() - entry.getAttribute('transitionTime'));
 
-        // Attach link-headers (NOT traceparent). The object-processor
-        // consumer starts a NEW root trace that links back to this
-        // bucket-processor span — prevents 1M-object traces from breaking
-        // the Jaeger UI.
-        const headers = linkHeadersFromCurrentContext();
+        // Attach traceparent so the downstream consumer can link back to
+        // this bucket-processor span. The consumer always creates a new
+        // root trace with a Link (never parent-child across Kafka).
+        const headers = traceHeadersFromCurrentContext();
         const kafkaEntry = { message: entry.toKafkaMessage() };
         if (headers) kafkaEntry.headers = headers;
         const entries = [kafkaEntry];

extensions/replication/ReplicationAPI.js

@@ -3,7 +3,7 @@ const locations = require('../../conf/locationConfig.json') || {};
 
 const ActionQueueEntry = require('../../lib/models/ActionQueueEntry');
 const ReplicationMetrics = require('./ReplicationMetrics');
-const { linkHeadersFromCurrentContext } = require('../../lib/tracing/kafkaTraceContext');
+const { traceHeadersFromCurrentContext } = require('../../lib/tracing/kafkaTraceContext');
 
 let { dataMoverTopic } = config.extensions.replication;
 const { coldStorageArchiveTopicPrefix } = config.extensions.lifecycle;
@@ -79,13 +79,10 @@ class ReplicationAPI {
             key: `${bucket}/${key}`,
             message: action.toKafkaMessage(),
         };
-        // Fan-out transitions can produce many entries per bucket-processor
-        // span. Use link-headers so the data-mover consumer starts a new
-        // root trace rather than a child of this one. If no active OTEL
-        // span (OTEL disabled), linkHeadersFromCurrentContext returns
-        // undefined and we ship the entry without headers.
-        const linkHeaders = linkHeadersFromCurrentContext();
-        if (linkHeaders) kafkaEntry.headers = linkHeaders;
+        // Attach traceparent so the downstream consumer can link back.
+        // All Kafka consumers create linked spans (never parent-child).
+        const traceHeaders = traceHeadersFromCurrentContext();
+        if (traceHeaders) kafkaEntry.headers = traceHeaders;
         let topic = dataMoverTopic;
         const toLocation = action.getAttribute('toLocation');
         const locationConfig = locations[toLocation];

lib/BackbeatConsumer.js

@@ -24,24 +24,7 @@ const {
 } = require('./constants');
 
 const { context: otelContext, SpanStatusCode } = require('@opentelemetry/api');
-const {
-    startSpanFromKafkaEntry,
-    startLinkedSpanFromKafkaEntry,
-} = require('./tracing/kafkaTraceContext');
-
-/**
- * Detect whether a Kafka entry carries link-traceparent (fan-out break)
- * rather than standard traceparent. Link semantics mean the consumer
- * starts a NEW root trace that references the upstream via OTEL Link,
- * instead of becoming a child of it.
- */
-function entryHasLinkHeaders(entry) {
-    if (!entry || !Array.isArray(entry.headers)) return false;
-    for (const h of entry.headers) {
-        if (h['link-traceparent']) return true;
-    }
-    return false;
-}
+const { startLinkedSpanFromKafkaEntry } = require('./tracing/kafkaTraceContext');
 
 const CLIENT_ID = 'BackbeatConsumer';
 const { withTopicPrefix } = require('./util/topic');
@@ -534,9 +517,7 @@ class BackbeatConsumer extends EventEmitter {
         const { topic, partition } = entry;
         KafkaBacklogMetrics.onTaskStarted(topic, partition, this._groupId);
 
-        const { ctx, span } = entryHasLinkHeaders(entry)
-            ? startLinkedSpanFromKafkaEntry(entry, `${topic}.process`)
-            : startSpanFromKafkaEntry(entry, `${topic}.process`);
+        const { ctx, span } = startLinkedSpanFromKafkaEntry(entry, `${topic}.process`);
         span.setAttribute('messaging.kafka.topic', topic);
         span.setAttribute('messaging.kafka.partition', partition);
 

lib/tracing/kafkaTraceContext.js

@@ -33,47 +33,15 @@ function contextFromKafkaHeaders(kafkaHeaders) {
     return propagation.extract(context.active(), carrier);
 }
 
-/**
- * Extract a link-trace-context from Kafka message headers.
- * Reads `link-traceparent` / `link-tracestate` (the link-variant that
- * marks the upstream as a related trace, not a parent).
- * @param {Array|undefined} kafkaHeaders - array of single-key objects from node-rdkafka
- * @returns {Object|null} OTEL context carrying the linked SpanContext, or null
- */
-function linkContextFromKafkaHeaders(kafkaHeaders) {
-    if (!kafkaHeaders) return null;
-    const carrier = {};
-    for (const h of kafkaHeaders) {
-        if (h['link-traceparent']) carrier.traceparent = h['link-traceparent'].toString();
-        if (h['link-tracestate']) carrier.tracestate = h['link-tracestate'].toString();
-    }
-    if (!carrier.traceparent) return null;
-    return propagation.extract(context.active(), carrier);
-}
-
-/**
- * Start a span linked to a Kafka entry's trace context.
- * Returns { ctx, span } — caller must call span.end() when done.
- * @param {Object} kafkaEntry - Kafka consumer entry with .headers
- * @param {string} operationName - name for the span
- * @returns {{ ctx: Object, span: Object }}
- */
-function startSpanFromKafkaEntry(kafkaEntry, operationName) {
-    const parentCtx = contextFromKafkaHeaders(kafkaEntry.headers);
-    const tracer = trace.getTracer('backbeat');
-    const span = tracer.startSpan(operationName, {
-        kind: SpanKind.CONSUMER,
-    }, parentCtx);
-    return { ctx: trace.setSpan(parentCtx, span), span };
-}
-
 /**
  * Start a NEW root span that LINKS to the span referenced by the Kafka
- * entry's `link-traceparent` header — as opposed to being a child of it.
+ * entry's `traceparent` header — as opposed to being a child of it.
  *
- * Use this for high-volume fan-out consumers (e.g. the lifecycle
- * object-processor). Traces stay small and clickable from Jaeger's Link
- * UI, instead of producing 1M-span super-traces that break the UI.
+ * Convention: every Kafka consumer read starts a new trace (new trace ID)
+ * and adds an OTEL Link to the upstream span. Async work (replication,
+ * lifecycle, GC, notifications) can fire minutes to hours after the
+ * original S3 request — links keep traces small and navigable via
+ * Jaeger's link UI.
  *
  * Returns { ctx, span } — caller must call span.end() when done.
  *
@@ -85,17 +53,13 @@ function startLinkedSpanFromKafkaEntry(kafkaEntry, operationName) {
     const tracer = trace.getTracer('backbeat');
     const links = [];
 
-    const linkedCtx = linkContextFromKafkaHeaders(kafkaEntry.headers);
-    if (linkedCtx) {
-        const linkedSpan = trace.getSpan(linkedCtx);
-        const linkedSpanCtx = linkedSpan && linkedSpan.spanContext();
-        if (linkedSpanCtx && linkedSpanCtx.traceId) {
-            links.push({ context: linkedSpanCtx });
-        }
+    const parentCtx = contextFromKafkaHeaders(kafkaEntry.headers);
+    const remoteSpan = trace.getSpan(parentCtx);
+    const remoteSpanCtx = remoteSpan && remoteSpan.spanContext();
+    if (remoteSpanCtx && remoteSpanCtx.traceId) {
+        links.push({ context: remoteSpanCtx });
     }
 
-    // NEW root span — do not pass an active parent. The link is the only
-    // reference to the upstream trace.
     const span = tracer.startSpan(operationName, {
         kind: SpanKind.CONSUMER,
         links,
@@ -104,10 +68,16 @@ function startLinkedSpanFromKafkaEntry(kafkaEntry, operationName) {
     return { ctx: trace.setSpan(context.active(), span), span };
 }
 
+// Alias kept for backward-compatibility with any call sites that still
+// reference the old name. Behaviour is identical: always creates a linked
+// root span, never a parent-child.
+const startSpanFromKafkaEntry = startLinkedSpanFromKafkaEntry;
+
 /**
  * Serialize the currently active OTEL context into standard
- * `traceparent` / `tracestate` Kafka headers — the consumer will
- * become a CHILD of the current span.
+ * `traceparent` / `tracestate` Kafka headers. The downstream consumer
+ * will create a LINK to this span (not a child — all Kafka hops use
+ * links).
  *
  * Returns an array of single-key header objects compatible with
  * node-rdkafka's header format, or undefined if no active span.
@@ -128,40 +98,10 @@ function traceHeadersFromCurrentContext() {
     return headers.length > 0 ? headers : undefined;
 }
 
-/**
- * Serialize the currently active OTEL context into `link-traceparent`
- * and `link-tracestate` Kafka headers — the link-variant, NOT the
- * standard `traceparent`. Producers use this when the downstream
- * consumer should LINK to the current span rather than become its
- * child.
- *
- * Returns an array of single-key header objects compatible with
- * node-rdkafka's header format, or undefined if no active span.
- *
- * @returns {Array|undefined}
- */
-function linkHeadersFromCurrentContext() {
-    const span = trace.getSpan(context.active());
-    if (!span) return undefined;
-    const spanCtx = span.spanContext();
-    if (!spanCtx || !spanCtx.traceId || !spanCtx.spanId) return undefined;
-
-    const carrier = {};
-    propagation.inject(context.active(), carrier);
-    // carrier now has { traceparent, tracestate? } — rename them to the
-    // link-variant so the consumer knows to LINK, not parent.
-    const headers = [];
-    if (carrier.traceparent) headers.push({ 'link-traceparent': carrier.traceparent });
-    if (carrier.tracestate) headers.push({ 'link-tracestate': carrier.tracestate });
-    return headers.length > 0 ? headers : undefined;
-}
-
 module.exports = {
     traceHeadersFromEntry,
     traceHeadersFromCurrentContext,
     contextFromKafkaHeaders,
-    linkContextFromKafkaHeaders,
     startSpanFromKafkaEntry,
     startLinkedSpanFromKafkaEntry,
-    linkHeadersFromCurrentContext,
 };

tests/unit/lib/tracing/kafkaTraceContext.spec.js

@@ -7,10 +7,8 @@ const {
     traceHeadersFromEntry,
     traceHeadersFromCurrentContext,
     contextFromKafkaHeaders,
-    linkContextFromKafkaHeaders,
     startSpanFromKafkaEntry,
     startLinkedSpanFromKafkaEntry,
-    linkHeadersFromCurrentContext,
 } = require('../../../../lib/tracing/kafkaTraceContext');
 
 describe('kafkaTraceContext', () => {
@@ -74,9 +72,6 @@ describe('kafkaTraceContext', () => {
             assert.strictEqual(result, context.active());
         });
 
-        // NOTE: With a full OTEL SDK (propagator registered), extract()
-        // returns a new context. Without SDK, the no-op propagator returns
-        // the active context unchanged. Full extraction is tested on-cluster.
         it('should call propagation.extract with correct carrier', () => {
             const tp = '00-abcdef1234567890abcdef1234567890-1234567890abcdef-01';
             const headers = [
@@ -88,17 +83,17 @@ describe('kafkaTraceContext', () => {
         });
     });
 
-    describe('startSpanFromKafkaEntry', () => {
-        it('should return ctx and span with no headers', () => {
+    describe('startLinkedSpanFromKafkaEntry', () => {
+        it('returns a span with no headers (no link)', () => {
             const entry = { topic: 'test-topic', partition: 0 };
-            const { ctx, span } = startSpanFromKafkaEntry(entry, 'test-op');
+            const { ctx, span } = startLinkedSpanFromKafkaEntry(entry, 'test-op');
             assert(ctx);
             assert(span);
             assert.strictEqual(typeof span.end, 'function');
             span.end();
         });
 
-        it('should return ctx and span with traceparent header', () => {
+        it('returns a span with traceparent header (creates link)', () => {
             const tp = '00-abcdef1234567890abcdef1234567890-1234567890abcdef-01';
             const entry = {
                 topic: 'test-topic',
@@ -107,7 +102,7 @@ describe('kafkaTraceContext', () => {
                     { traceparent: Buffer.from(tp) },
                 ],
             };
-            const { ctx, span } = startSpanFromKafkaEntry(entry, 'test-op');
+            const { ctx, span } = startLinkedSpanFromKafkaEntry(entry, 'test-op');
             assert(ctx);
             assert(span);
             assert.strictEqual(typeof span.setAttribute, 'function');
@@ -116,57 +111,14 @@ describe('kafkaTraceContext', () => {
         });
     });
 
-    describe('linkContextFromKafkaHeaders', () => {
-        it('returns null when headers are missing', () => {
-            assert.strictEqual(linkContextFromKafkaHeaders(undefined), null);
-            assert.strictEqual(linkContextFromKafkaHeaders(null), null);
-        });
-
-        it('returns null when no link-traceparent header', () => {
-            const result = linkContextFromKafkaHeaders([
-                { unrelated: Buffer.from('x') },
-            ]);
-            assert.strictEqual(result, null);
-        });
-
-        it('returns a context when link-traceparent is present', () => {
-            const tp = '00-abcdef1234567890abcdef1234567890-1234567890abcdef-01';
-            const result = linkContextFromKafkaHeaders([
-                { 'link-traceparent': Buffer.from(tp) },
-            ]);
-            // Without a propagator registered this may be the active
-            // context; the important bit is the call doesn't throw.
-            assert(result !== undefined);
-        });
-    });
-
-    describe('startLinkedSpanFromKafkaEntry', () => {
-        it('returns a span even with no link headers', () => {
-            const entry = { topic: 'lifecycle.object-tasks', partition: 0 };
-            const { ctx, span } = startLinkedSpanFromKafkaEntry(entry, 'linked-op');
-            assert(ctx);
-            assert(span);
-            span.end();
-        });
-
-        it('returns a span with link headers present', () => {
-            const tp = '00-abcdef1234567890abcdef1234567890-1234567890abcdef-01';
-            const entry = {
-                topic: 'lifecycle.object-tasks',
-                partition: 0,
-                headers: [{ 'link-traceparent': Buffer.from(tp) }],
-            };
-            const { ctx, span } = startLinkedSpanFromKafkaEntry(entry, 'linked-op');
-            assert(ctx);
-            assert(span);
-            span.end();
+    describe('startSpanFromKafkaEntry (backward-compat alias)', () => {
+        it('is the same function as startLinkedSpanFromKafkaEntry', () => {
+            assert.strictEqual(startSpanFromKafkaEntry, startLinkedSpanFromKafkaEntry);
         });
     });
 
-    describe('linkHeadersFromCurrentContext / traceHeadersFromCurrentContext', () => {
+    describe('traceHeadersFromCurrentContext', () => {
         it('returns undefined when no active span', () => {
-            // In a fresh test context, no span is active.
-            assert.strictEqual(linkHeadersFromCurrentContext(), undefined);
             assert.strictEqual(traceHeadersFromCurrentContext(), undefined);
         });
     });