fix(worker): propagate process <job_name> span's metadata into processor function so child spans are correctly nested

src/classes/job.ts

@@ -578,6 +578,26 @@ export class Job<
     return options as RedisJobOptions;
   }
 
+  /**
+   * Creates a clone of the job with new options merged with the existing ones.
+   *
+   * @param opts - job options to be shallowly merged with the existing ones.
+   * @returns New Job instance.
+   */
+  cloneWithOpts(opts: JobsOptions): Job<DataType, ReturnType, NameType> {
+    return new Job<DataType, ReturnType, NameType>(
+      this.queue,
+      this.name,
+      this.data,
+      {
+        ...this.opts,
+        timestamp: this.timestamp,
+        ...opts,
+      },
+      this.id,
+    );
+  }
+
   /**
    * Prepares a job to be passed to Sandbox.
    * @returns

src/classes/worker.ts

@@ -436,7 +436,27 @@ export class Worker<
     job: Job<DataType, ResultType, NameType>,
     token: string,
     signal?: AbortSignal,
+    srcPropagationMetadata?: string,
   ): Promise<ResultType> {
+    // Need to overwrite telemetry.metadata to be able to restore
+    // propagation context in a worker thread/fork.
+    // This is only necessary for sandboxed workers because OTEL doesn't
+    // propagate context cross-threads/cross-processes.
+    if (this.childPool && job.opts.telemetry) {
+      return this.processFn(
+        job.cloneWithOpts({
+          telemetry: {
+            ...job.opts.telemetry,
+            metadata: job.opts.telemetry.omitContext
+              ? undefined
+              : srcPropagationMetadata,
+          },
+        }),
+        token,
+        signal,
+      );
+    }
+
     return this.processFn(job, token, signal);
   }
 
@@ -978,7 +998,7 @@ will never work with more accuracy than 1ms. */
       SpanKind.CONSUMER,
       'process',
       this.name,
-      async span => {
+      async (span, srcPropagationMetadata) => {
         span?.setAttributes({
           [TelemetryAttributes.WorkerId]: this.id,
           [TelemetryAttributes.WorkerName]: this.opts.name,
@@ -1023,6 +1043,7 @@ will never work with more accuracy than 1ms. */
             abortController
               ? (abortController.signal as AbortSignal)
               : undefined,
+            srcPropagationMetadata,
           );
           return await this.retryIfFailed<void | Job<
             DataType,

tests/fixtures/fixture_processor_custom_span.js

@@ -0,0 +1,9 @@
+/**
+ * A processor file to be used in tests.
+ *
+ */
+'use strict';
+
+module.exports = function (job) {
+  return job.opts.tm;
+};

tests/job.test.ts

@@ -1987,4 +1987,77 @@ describe('Job', () => {
       await worker.close();
     });
   });
+
+  describe('.cloneWithOpts', () => {
+    it('should clone one-off debounced and delayed job', async () => {
+      const sourceJob = await Job.create(
+        queue,
+        'test_job',
+        { foo: 'bar' },
+        {
+          jobId: 'custom_job_id',
+          delay: 1000,
+          deduplication: { id: 'dedup-id' },
+          debounce: { id: 'debounce-id' },
+          backoff: { type: 'fixed', jitter: 0 },
+          telemetry: {
+            metadata: 'source_metadata',
+          },
+        },
+      );
+
+      const clonedJob = sourceJob.cloneWithOpts({
+        telemetry: {
+          metadata: 'target_metadata',
+        },
+      });
+
+      expect(clonedJob.asJSON()).toEqual({
+        ...sourceJob.asJSON(),
+        opts: {
+          ...sourceJob.asJSON().opts,
+          // timestamp for cloned job is passed as opts,
+          // whereas for sourceJob it's autogenerated hence the difference
+          timestamp: sourceJob.timestamp,
+          tm: clonedJob.asJSON().opts.tm,
+        },
+      });
+    });
+
+    it('should clone repeatable job with parent', async () => {
+      const parentJob = await Job.create(queue, 'parent', {});
+
+      const sourceJob = await Job.create(
+        queue,
+        'test_job',
+        { foo: 'bar' },
+        {
+          jobId: 'custom_job_id',
+          parent: { id: parentJob.id!, queue: `${prefix}:${queueName}` },
+          repeat: { every: 200 },
+          removeDependencyOnFailure: true,
+          telemetry: {
+            metadata: 'source_metadata',
+          },
+        },
+      );
+
+      const clonedJob = sourceJob.cloneWithOpts({
+        telemetry: {
+          metadata: 'target_metadata',
+        },
+      });
+
+      expect(clonedJob.asJSON()).toEqual({
+        ...sourceJob.asJSON(),
+        opts: {
+          ...sourceJob.asJSON().opts,
+          // timestamp for cloned job is passed as opts,
+          // whereas for sourceJob it's autogenerated hence the difference
+          timestamp: sourceJob.timestamp,
+          tm: clonedJob.asJSON().opts.tm,
+        },
+      });
+    });
+  });
 });

tests/telemetry_interface.test.ts

@@ -10,7 +10,7 @@ import {
 } from 'vitest';
 
 import { v4 } from 'uuid';
-import { FlowProducer, JobScheduler, Queue, Worker } from '../src/classes';
+import { FlowProducer, Job, JobScheduler, Queue, Worker } from '../src/classes';
 import { removeAllQueueData } from '../src/utils';
 import {
   Telemetry,
@@ -29,6 +29,7 @@ import {
 } from '../src/interfaces';
 import * as sinon from 'sinon';
 import { SpanKind, TelemetryAttributes, MetricNames } from '../src/enums';
+import { pathToFileURL } from 'url';
 
 describe('Telemetry', () => {
   type ExtendedException = Exception & {
@@ -394,6 +395,65 @@ describe('Telemetry', () => {
       await worker.close();
     });
 
+    it('should correctly interact with telemetry when processing a sandboxed job', async () => {
+      const processFile =
+        __dirname + '/fixtures/fixture_processor_custom_span.js';
+      const processUrl = pathToFileURL(processFile);
+
+      const worker = new Worker(queueName, processUrl, {
+        connection,
+        telemetry: telemetryClient,
+        name: 'testWorker',
+        prefix,
+        useWorkerThreads: true,
+      });
+
+      const completing = new Promise<string>((resolve, reject) => {
+        worker.on('completed', async (_, value: string) => {
+          try {
+            resolve(value);
+          } catch (err) {
+            reject(err);
+          }
+        });
+      });
+
+      const startSpanSpy = sinon.spy(telemetryClient.tracer, 'startSpan');
+
+      const job = await queue.add('testJob', { foo: 'bar' });
+
+      const processContextMetadata = JSON.parse(await completing);
+
+      const addSpan = startSpanSpy.returnValues[0] as MockSpan;
+      expect(addSpan).toBeInstanceOf(MockSpan);
+      expect(addSpan.name).toBe(`add ${queueName}.${job.name}`);
+      expect(addSpan.options?.kind).toBe(SpanKind.PRODUCER);
+      expect(addSpan.attributes[TelemetryAttributes.QueueName]).toBe(
+        queue.name,
+      );
+      expect(addSpan.attributes[TelemetryAttributes.QueueOperation]).toBe(
+        'add',
+      );
+      expect(addSpan.attributes[TelemetryAttributes.JobName]).toBe(job.name);
+      expect(addSpan.attributes[TelemetryAttributes.JobId]).toBe(job.id);
+
+      const processSpan = startSpanSpy.returnValues[1] as MockSpan;
+      expect(processSpan).toBeInstanceOf(MockSpan);
+      expect(processSpan.name).toBe(`process ${queueName}`);
+      expect(processSpan.options?.kind).toBe(SpanKind.CONSUMER);
+      expect(processSpan.attributes[TelemetryAttributes.WorkerId]).toBe(
+        worker.id,
+      );
+      expect(processSpan.attributes[TelemetryAttributes.WorkerName]).toBe(
+        'testWorker',
+      );
+      expect(processSpan.attributes[TelemetryAttributes.JobId]).toBe(job.id);
+
+      expect(processContextMetadata['getMetadata_span']).toBe(processSpan.name);
+
+      await worker.close();
+    });
+
     it('should propagate context correctly between queue and worker using telemetry', async () => {
       const job = await queue.add('testJob', { foo: 'bar' });