BB-764: Add OpenTelemetry tracing to backbeat replication pipeline

bin/queuePopulator.js

@@ -1,3 +1,5 @@
+require('../lib/otel');
+
 const async = require('async');
 const schedule = require('node-schedule');
 

extensions/gc/service.js

@@ -1,4 +1,5 @@
 'use strict';
+require('../../lib/otel');
 
 const { errors } = require('arsenal');
 const werelogs = require('werelogs');

extensions/lifecycle/bucketProcessor/task.js

@@ -1,4 +1,5 @@
 'use strict';
+require('../../../lib/otel');
 
 const async = require('async');
 const werelogs = require('werelogs');

extensions/lifecycle/conductor/LifecycleConductor.js

@@ -29,6 +29,10 @@
     startCircuitBreakerMetricsExport,
     updateCircuitBreakerConfigForImplicitOutputQueue,
 } = require('../../../lib/CircuitBreaker');
+const { context: otelContext, trace, SpanKind, SpanStatusCode } =
+    require('@opentelemetry/api');
+const { traceHeadersFromCurrentContext } =
+    require('../../../lib/tracing/kafkaTraceContext');
 
 const DEFAULT_CRON_RULE = '* * * * *';
 const DEFAULT_CONCURRENCY = 10;
@@ -309,7 +313,11 @@
     }
 
     _taskToMessage(task, taskVersion, log) {
-        return {
+        // Attach traceparent so the bucket-processor consumer becomes a
+        // child of the conductor.scan root span. Returns undefined when
+        // OTEL is disabled (no active span).
+        const headers = traceHeadersFromCurrentContext();
+        const kafkaEntry = {
             message: JSON.stringify({
                 action: 'processObjects',
                 contextInfo: {
@@ -324,6 +332,8 @@
                 details: {},
             }),
         };
+        if (headers) kafkaEntry.headers = headers;
+        return kafkaEntry;
     }
 
     _getAccountIds(unknownCanonicalIds, log, cb) {
@@ -371,6 +381,27 @@
     }
 
     processBuckets(cb) {
+        // Wrap every cron firing in its own root trace so bucket-listing
+        // and downstream bucket-task produces appear as children. The
+        // trace is INTERNAL because there's no upstream HTTP/Kafka parent.
+        const tracer = trace.getTracer('backbeat');
+        const span = tracer.startSpan('lifecycle.conductor.scan', {
+            kind: SpanKind.INTERNAL,
+        });
+        const ctx = trace.setSpan(otelContext.active(), span);
+        otelContext.with(ctx, () => {
+            this._processBucketsInternal((err, res) => {
+                if (err) {
+                    span.recordException(err);
+                    span.setStatus({ code: SpanStatusCode.ERROR });
+                }
+                span.end();
+                if (cb) cb(err, res);
+            });
+        });
+    }
+
+    _processBucketsInternal(cb) {
         const log = this.logger.newRequestLogger();
         const start = new Date();
         let nBucketsQueued = 0;

extensions/lifecycle/conductor/service.js

@@ -1,4 +1,5 @@
 'use strict';
+require('../../../lib/otel');
 
 const async = require('async');
 const werelogs = require('werelogs');

extensions/lifecycle/objectProcessor/task.js

@@ -1,4 +1,5 @@
 'use strict';
+require('../../../lib/otel');
 
 const async = require('async');
 const werelogs = require('werelogs');

extensions/lifecycle/tasks/LifecycleTask.js

@@ -24,6 +24,7 @@
 const { LifecycleMetrics, LIFECYCLE_MARKER_METRICS_LOCATION } = require('../LifecycleMetrics');
 const locationsConfig = require('../../../conf/locationConfig.json') || {};
 const { rulesSupportTransition } = require('../util/rules');
+const { traceHeadersFromCurrentContext } = require('../../../lib/tracing/kafkaTraceContext');
 const { decode } = versioning.VersionID;
 
 const errorTransitionInProgress = errors.InternalError.
@@ -121,7 +122,14 @@
      * @return {undefined}
      */
     _sendBucketEntry(entry, cb) {
-        const entries = [{ message: JSON.stringify(entry) }];
+        // Bucket-task re-queue: keep normal parent-child semantics so the
+        // continuation stays in the same trace as the originating conductor
+        // scan. Re-queues happen rarely (only when a bucket is too big to
+        // finish in one batch) so the trace stays small.
+        const headers = traceHeadersFromCurrentContext();
+        const kafkaEntry = { message: JSON.stringify(entry) };
+        if (headers) kafkaEntry.headers = headers;
+        const entries = [kafkaEntry];
         this.producer.sendToTopic(this.bucketTasksTopic, entries, err => {
             LifecycleMetrics.onKafkaPublish(null, 'BucketTopic', 'bucket', err, 1);
             return cb(err);
@@ -183,7 +191,13 @@
             location,
             Date.now() - entry.getAttribute('transitionTime'));
 
-        const entries = [{ message: entry.toKafkaMessage() }];
+        // 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];
         this.producer.sendToTopic(this.objectTasksTopic, entries,  err => {
             LifecycleMetrics.onKafkaPublish(null, 'ObjectTopic', 'bucket', err, 1);
             return cb(err);

extensions/notification/NotificationQueuePopulator.js

@@ -10,6 +10,7 @@ const messageUtil = require('./utils/message');
 const notifConstants = require('./constants');
 const QueuePopulatorExtension =
     require('../../lib/queuePopulator/QueuePopulatorExtension');
+const { traceHeadersFromEntry } = require('../../lib/tracing/kafkaTraceContext');
 
 class NotificationQueuePopulator extends QueuePopulatorExtension {
     /**
@@ -290,13 +291,20 @@ class NotificationQueuePopulator extends QueuePopulatorExtension {
                             eventTime: message.dateTime,
                             matchingConfig,
                         });
+                        // Propagate the oplog entry's trace context as
+                        // Kafka headers so the notification-processor
+                        // consumer becomes a child of the original S3
+                        // trace (same pattern as ReplicationQueuePopulator).
+                        const traceHeaders = traceHeadersFromEntry(value);
                         this.publish(topic,
                             // keeping all messages for same object
                             // in the same partition to keep the order.
                             // here we use the object name and not the
                             // "_id" which also includes the versionId
                             `${bucket}/${message.key}`,
-                            JSON.stringify(message));
+                            JSON.stringify(message),
+                            undefined,
+                            traceHeaders);
                         // keep track of internal topics we have pushed to
                         pushedToTopic[topic] = true;
                     }

extensions/notification/destination/KafkaNotificationDestination.js

@@ -100,7 +100,22 @@ class KafkaNotificationDestination extends NotificationDestination {
      */
     send(messages, done) {
         const starTime = Date.now();
-        this._notificationProducer.send(messages, error => {
+        // Trust boundary: strip any trace headers before producing to the
+        // external customer Kafka destination. There's no ingress-layer
+        // strip on the Kafka protocol (unlike HTTP via nginx), so the only
+        // place to drop headers is at the producer. Keeps everything else
+        // on the message intact.
+        const safeMessages = Array.isArray(messages)
+            ? messages.map(m => {
+                if (m && m.headers) {
+                    // eslint-disable-next-line no-unused-vars
+                    const { headers, ...rest } = m;
+                    return rest;
+                }
+                return m;
+            })
+            : messages;
+        this._notificationProducer.send(safeMessages, error => {
             if (error) {
                 const { host, topic } = this._destinationConfig;
                 this._log.error('error in message delivery to external Kafka destination', {

extensions/notification/queueProcessor/task.js

@@ -1,4 +1,6 @@
 'use strict';
+require('../../../lib/otel');
+
 const assert = require('assert');
 const { errors } = require('arsenal');
 const async = require('async');

extensions/replication/ReplicationAPI.js

@@ -3,6 +3,7 @@
 
 const ActionQueueEntry = require('../../lib/models/ActionQueueEntry');
 const ReplicationMetrics = require('./ReplicationMetrics');
+const { traceHeadersFromCurrentContext } = require('../../lib/tracing/kafkaTraceContext');
 
 let { dataMoverTopic } = config.extensions.replication;
 const { coldStorageArchiveTopicPrefix } = config.extensions.lifecycle;
@@ -78,6 +79,10 @@
             key: `${bucket}/${key}`,
             message: action.toKafkaMessage(),
         };
+        // 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];

extensions/replication/ReplicationQueuePopulator.js

@@ -5,6 +5,7 @@ const QueuePopulatorExtension =
           require('../../lib/queuePopulator/QueuePopulatorExtension');
 const ObjectQueueEntry = require('../../lib/models/ObjectQueueEntry');
 const locationsConfig = require('../../conf/locationConfig.json') || {};
+const { traceHeadersFromEntry } = require('../../lib/tracing/kafkaTraceContext');
 
 class ReplicationQueuePopulator extends QueuePopulatorExtension {
     constructor(params) {
@@ -95,11 +96,15 @@ class ReplicationQueuePopulator extends QueuePopulatorExtension {
         const publishedEntry = Object.assign({}, entry);
         delete publishedEntry.logReader;
 
+        const traceHeaders = traceHeadersFromEntry(value);
+
         this.log.trace('publishing object replication entry',
                        { entry: queueEntry.getLogInfo() });
         this.publish(this.repConfig.topic,
                      `${queueEntry.getBucket()}/${queueEntry.getObjectKey()}`,
-                     JSON.stringify(publishedEntry));
+                     JSON.stringify(publishedEntry),
+                     undefined,
+                     traceHeaders);
     }
 
     /**

extensions/replication/queueProcessor/task.js

@@ -1,4 +1,6 @@
 'use strict';
+require('../../../lib/otel');
+
 const async = require('async');
 const assert = require('assert');
 const werelogs = require('werelogs');

extensions/replication/replicationStatusProcessor/task.js

@@ -1,4 +1,5 @@
 'use strict';
+require('../../../lib/otel');
 
 const werelogs = require('werelogs');
 

lib/BackbeatConsumer.js

@@ -23,6 +23,9 @@ const {
     }
 } = require('./constants');
 
+const { context: otelContext, SpanStatusCode } = require('@opentelemetry/api');
+const { startLinkedSpanFromKafkaEntry } = require('./tracing/kafkaTraceContext');
+
 const CLIENT_ID = 'BackbeatConsumer';
 const { withTopicPrefix } = require('./util/topic');
 
@@ -514,7 +517,20 @@ class BackbeatConsumer extends EventEmitter {
         const { topic, partition } = entry;
         KafkaBacklogMetrics.onTaskStarted(topic, partition, this._groupId);
 
-        this._queueProcessor(entry, (err, completionArgs) => done(err, completionArgs, finishProcessingTask));
+        const { ctx, span } = startLinkedSpanFromKafkaEntry(entry, `${topic}.process`);
+        span.setAttribute('messaging.kafka.topic', topic);
+        span.setAttribute('messaging.kafka.partition', partition);
+
+        otelContext.with(ctx, () => {
+            this._queueProcessor(entry, (err, completionArgs) => {
+                if (err) {
+                    span.recordException(err);
+                    span.setStatus({ code: SpanStatusCode.ERROR });
+                }
+                span.end();
+                done(err, completionArgs, finishProcessingTask);
+            });
+        });
     }
 
     /**

lib/BackbeatProducer.js

@@ -288,7 +288,8 @@ class BackbeatProducer extends EventEmitter {
                     Buffer.from(item.message), // value
                     item.key, // key (for keyed partitioning)
                     Date.now(), // timestamp
-                    sendCtx // opaque
+                    sendCtx, // opaque
+                    item.headers || undefined // Kafka message headers
                 );
             });
         } catch (err) {