feat(telemetry): add option to omit context propagation

src/classes/flow-producer.ts

@@ -346,7 +346,10 @@ export class FlowProducer extends EventEmitter {
             ...jobsOpts,
             ...node.opts,
             parent: parent?.parentOpts,
-            telemetryMetadata: dstPropagationMetadata,
+            telemetryMetadata:
+              span &&
+              !node.opts?.telemetry?.omitContext &&
+              dstPropagationMetadata,
           },
           jobId,
         );

src/classes/job-scheduler.ts

@@ -132,7 +132,8 @@ export class JobScheduler extends QueueBase {
             {
               ...opts,
               repeat: filteredRepeatOpts,
-              telemetryMetadata: srcPropagationMedatada,
+              telemetryMetadata:
+                span && !opts?.telemetry?.omitContext && srcPropagationMedatada,
             },
             jobData,
             iterationCount,

src/classes/job.ts

@@ -46,6 +46,7 @@ const optsDecodeMap = {
   kl: 'keepLogs',
   rdof: 'removeDependencyOnFailure',
   tm: 'telemetryMetadata',
+  omc: 'omitContext',
 };
 
 const optsEncodeMap = invertObject(optsDecodeMap);
@@ -747,7 +748,7 @@ export class Job<
       this.getSpanOperation(command),
       this.queue.name,
       async (span, dstPropagationMedatadata) => {
-        if (dstPropagationMedatadata) {
+        if (!this.opts?.telemetry?.omitContext && dstPropagationMedatadata) {
           this.scripts.execCommand(multi, 'updateJobOption', [
             this.toKey(this.id),
             'tm',

src/classes/queue-base.ts

@@ -1,6 +1,6 @@
 import { EventEmitter } from 'events';
 import { QueueBaseOptions, RedisClient, Span, Tracer } from '../interfaces';
-import { MinimalQueue } from '../types';
+import { JobsOptions, MinimalQueue } from '../types';
 import {
   delay,
   DELAY_TIME_5,

src/classes/queue.ts

@@ -307,7 +307,7 @@ export class Queue<
       'add',
       `${this.name}.${name}`,
       async (span, srcPropagationMedatada) => {
-        if (srcPropagationMedatada) {
+        if (!opts?.telemetry?.omitContext && srcPropagationMedatada) {
           opts = { ...opts, telemetryMetadata: srcPropagationMedatada };
         }
 
@@ -404,7 +404,10 @@ export class Queue<
               ...this.jobsOpts,
               ...job.opts,
               jobId: job.opts?.jobId,
-              tm: span && srcPropagationMedatada,
+              tm:
+                span &&
+                !job.opts?.telemetry?.omitContext &&
+                srcPropagationMedatada,
             },
           })),
         );

src/interfaces/base-job-options.ts

@@ -117,4 +117,10 @@ export interface BaseJobOptions extends DefaultJobOptions {
    * TelemetryMetadata, provide for context propagation.
    */
   telemetryMetadata?: string;
+
+  /**
+   * If `true` telemetry will omit the context propagation
+   * @default false
+   */
+  omitContext?: boolean;
 }

src/types/job-options.ts

@@ -26,6 +26,22 @@ export type JobsOptions = BaseJobOptions & {
    * If true, removes the job from its parent dependencies when it fails after all attempts.
    */
   removeDependencyOnFailure?: boolean;
+
+  /**
+   * Telemetry options
+   */
+  telemetry?: {
+    /**
+     * TelemetryMetadata, provide for context propagation.
+     */
+    metadata?: string;
+
+    /**
+     * If `true` telemetry will omit the context propagation
+     * @default false
+     */
+    omitContext?: boolean;
+  };
 };
 
 /**
@@ -61,4 +77,9 @@ export type RedisJobOptions = BaseJobOptions & {
    * TelemetryMetadata, provide for context propagation.
    */
   tm?: string;
+
+  /**
+   * Omit Context Propagation
+   */
+  omc?: boolean;
 };

tests/test_telemetry_interface.ts

@@ -2,7 +2,7 @@ import { expect, assert } from 'chai';
 import { default as IORedis } from 'ioredis';
 import { after, beforeEach, describe, it, before } from 'mocha';
 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,
@@ -93,7 +93,7 @@ describe('Telemetry', () => {
       this.options = options;
     }
 
-    setSpanOnContext(ctx: any): any {
+    setSpanOnContext(ctx: any, omitContext?: boolean): any {
       context['getSpan'] = () => this;
       return { ...context, getMetadata_span: this['name'] };
     }
@@ -260,6 +260,7 @@ describe('Telemetry', () => {
     });
 
     it('should correctly handle errors and record them in telemetry for upsertJobScheduler', async () => {
+      const originalCreateNextJob = JobScheduler.prototype.createNextJob;
       const recordExceptionSpy = sinon.spy(
         MockSpan.prototype,
         'recordException',
@@ -283,6 +284,7 @@ describe('Telemetry', () => {
         const recordedError = recordExceptionSpy.firstCall.args[0];
         assert.equal(recordedError.message, errMessage);
       } finally {
+        JobScheduler.prototype.createNextJob = originalCreateNextJob;
         recordExceptionSpy.restore();
       }
     });
@@ -511,4 +513,131 @@ describe('Telemetry', () => {
       }
     });
   });
+
+  describe('Omit Propagation', () => {
+    let fromMetadataSpy;
+
+    beforeEach(() => {
+      fromMetadataSpy = sinon.spy(
+        telemetryClient.contextManager,
+        'fromMetadata',
+      );
+    });
+
+    afterEach(() => fromMetadataSpy.restore());
+
+    it('should omit propagation on queue add', async () => {
+      const worker = new Worker(queueName, async () => 'some result', {
+        connection,
+        telemetry: telemetryClient,
+      });
+      await worker.waitUntilReady();
+
+      const job = await queue.add(
+        'testJob',
+        { foo: 'bar' },
+        { telemetry: { omitContext: true } },
+      );
+
+      await worker.processJob(job, 'some-token', () => false, new Set());
+
+      expect(fromMetadataSpy.callCount).to.equal(0);
+      await worker.close();
+    });
+
+    it('should omit propagation on queue addBulk', async () => {
+      const worker = new Worker(queueName, async () => 'some result', {
+        connection,
+        telemetry: telemetryClient,
+      });
+      await worker.waitUntilReady();
+
+      const jobs = [
+        {
+          name: 'job1',
+          data: { foo: 'bar' },
+          opts: { telemetry: { omitContext: true } },
+        },
+      ];
+      const jobArray = await queue.addBulk(jobs);
+
+      await Promise.all(
+        jobArray.map(job =>
+          worker.processJob(job, 'some-token', () => false, new Set()),
+        ),
+      );
+
+      expect(fromMetadataSpy.callCount).to.equal(0);
+      await worker.close();
+    });
+
+    it('should omit propagation on job scheduler', async () => {
+      const worker = new Worker(queueName, async () => 'some result', {
+        connection,
+        telemetry: telemetryClient,
+      });
+      await worker.waitUntilReady();
+
+      const jobSchedulerId = 'testJobScheduler';
+      const data = { foo: 'bar' };
+
+      const job = await queue.upsertJobScheduler(
+        jobSchedulerId,
+        { every: 1000, endDate: Date.now() + 1000, limit: 1 },
+        {
+          name: 'repeatable-job',
+          data,
+          opts: { telemetry: { omitContext: true } },
+        },
+      );
+
+      await worker.processJob(job, 'some-token', () => false, new Set());
+
+      expect(fromMetadataSpy.callCount).to.equal(0);
+      await worker.close();
+    });
+
+    it('should omit propagation on flow producer', async () => {
+      const worker = new Worker(queueName, async () => 'some result', {
+        connection,
+        telemetry: telemetryClient,
+      });
+      await worker.waitUntilReady();
+
+      const flowProducer = new FlowProducer({
+        connection,
+        telemetry: telemetryClient,
+      });
+
+      const testFlow = {
+        name: 'parentJob',
+        queueName,
+        data: { foo: 'bar' },
+        children: [
+          {
+            name: 'childJob',
+            queueName,
+            data: { baz: 'qux' },
+            opts: { telemetry: { omitContext: true } },
+          },
+        ],
+        opts: { telemetry: { omitContext: true } },
+      };
+
+      const jobNode = await flowProducer.add(testFlow);
+      const jobs = jobNode.children
+        ? [jobNode.job, ...jobNode.children.map(c => c.job)]
+        : [jobNode.job];
+
+      await Promise.all(
+        jobs.map(job =>
+          worker.processJob(job, 'some-token', () => false, new Set()),
+        ),
+      );
+
+      expect(fromMetadataSpy.callCount).to.equal(0);
+      await flowProducer.close();
+      await worker.close();
+    });
+  });
 });