feat(queue): observability for queue

src/classes/worker.ts

@@ -35,6 +35,8 @@ import {
   RATE_LIMIT_ERROR,
   WaitingChildrenError,
 } from './errors';
+import { Tracer } from '../interfaces';
+import { TelemetryAttributes } from '../enums';
 
 // 10 seconds is the maximum time a BRPOPLPUSH can block.
 const maximumBlockTimeout = 10;
@@ -190,6 +192,13 @@ export class Worker<
   protected processFn: Processor<DataType, ResultType, NameType>;
   protected running = false;
 
+  /**
+   * Instance of a telemetry client
+   * To use it create if statement in a method to observe with start and end of a span
+   * It will check if tracer is provided and if not it will continue as is
+   */
+  private tracer: Tracer | undefined;
+
   static RateLimitError(): Error {
     return new RateLimitError();
   }
@@ -314,6 +323,8 @@ export class Worker<
     this.blockingConnection.on('ready', () =>
       setTimeout(() => this.emit('ready'), 0),
     );
+
+    this.tracer = opts?.telemetry?.tracer;
   }
 
   emit<U extends keyof WorkerListener<DataType, ResultType, NameType>>(
@@ -402,11 +413,30 @@ export class Worker<
   }
 
   async run() {
+    let span;
+    if (this.tracer) {
+      const spanName = `${this.name} ${this.id} Worker.run`;
+      span = this.tracer.startSpan(spanName);
+      span.setAttributes({
+        [TelemetryAttributes.WorkerName]: this.name,
+        [TelemetryAttributes.WorkerId]: this.id,
+        [TelemetryAttributes.WorkerOptions]: JSON.stringify(this.opts),
+      });
+    }
+
     if (!this.processFn) {
+      if (this.tracer) {
+        span.end();
+      }
+
       throw new Error('No process function is defined.');
     }
 
     if (this.running) {
+      if (this.tracer) {
+        span.end();
+      }
+
       throw new Error('Worker is already running.');
     }
 
@@ -507,10 +537,14 @@ export class Worker<
       }
 
       this.running = false;
-      return asyncFifoQueue.waitAll();
+      return await asyncFifoQueue.waitAll();
     } catch (error) {
       this.running = false;
       throw error;
+    } finally {
+      if (this.tracer) {
+        span.end();
+      }
     }
   }
 
@@ -520,12 +554,30 @@ export class Worker<
    * @returns a Job or undefined if no job was available in the queue.
    */
   async getNextJob(token: string, { block = true }: GetNextJobOptions = {}) {
-    return this._getNextJob(
+    let span;
+    if (this.tracer) {
+      const spanName = `${this.name} ${this.id} Worker.getNextJob`;
+      span = this.tracer.startSpan(spanName);
+      span.setAttributes({
+        [TelemetryAttributes.WorkerName]: this.name,
+        [TelemetryAttributes.WorkerId]: this.id,
+        [TelemetryAttributes.WorkerToken]: token,
+        [TelemetryAttributes.WorkerOptions]: JSON.stringify({ block }),
+      });
+    }
+
+    const nextJob = await this._getNextJob(
       await this.client,
       await this.blockingConnection.client,
       token,
       { block },
     );
+
+    if (this.tracer) {
+      span.end();
+    }
+
+    return nextJob;
   }
 
   private async _getNextJob(
@@ -585,6 +637,17 @@ export class Worker<
    * @param expireTimeMs - expire time in ms of this rate limit.
    */
   async rateLimit(expireTimeMs: number): Promise<void> {
+    let span;
+    if (this.tracer) {
+      const spanName = `${this.name} ${this.id} Worker.rateLimit`;
+      span = this.tracer.startSpan(spanName);
+      span.setAttributes({
+        [TelemetryAttributes.WorkerName]: this.name,
+        [TelemetryAttributes.WorkerId]: this.id,
+        [TelemetryAttributes.WorkerRateLimit]: expireTimeMs,
+      });
+    }
+
     await this.client.then(client =>
       client.set(
         this.keys.limiter,
@@ -593,6 +656,10 @@ export class Worker<
         expireTimeMs,
       ),
     );
+
+    if (this.tracer) {
+      span.end();
+    }
   }
 
   get minimumBlockTimeout(): number {
@@ -743,7 +810,23 @@ will never work with more accuracy than 1ms. */
     fetchNextCallback = () => true,
     jobsInProgress: Set<{ job: Job; ts: number }>,
   ): Promise<void | Job<DataType, ResultType, NameType>> {
+    let span;
+    if (this.tracer) {
+      const spanName = `${this.name} ${this.id} Worker.processJob`;
+      span = this.tracer.startSpan(spanName);
+      span.setAttributes({
+        [TelemetryAttributes.WorkerName]: this.name,
+        [TelemetryAttributes.WorkerId]: this.id,
+        [TelemetryAttributes.WorkerToken]: token,
+        [TelemetryAttributes.JobId]: job.id,
+      });
+    }
+
     if (!job || this.closing || this.paused) {
+      if (this.tracer) {
+        span.end();
+      }
+
       return;
     }
 
@@ -808,6 +891,10 @@ will never work with more accuracy than 1ms. */
       return handleFailed(<Error>err);
     } finally {
       jobsInProgress.delete(inProgressItem);
+
+      if (this.tracer) {
+        span.end();
+      }
     }
   }
 
@@ -816,6 +903,17 @@ will never work with more accuracy than 1ms. */
    * Pauses the processing of this queue only for this worker.
    */
   async pause(doNotWaitActive?: boolean): Promise<void> {
+    let span;
+    if (this.tracer) {
+      const spanName = `${this.name} ${this.id} Worker.pause`;
+      span = this.tracer.startSpan(spanName);
+      span.setAttributes({
+        [TelemetryAttributes.WorkerName]: this.name,
+        [TelemetryAttributes.WorkerId]: this.id,
+        [TelemetryAttributes.WorkerDoNotWaitActive]: doNotWaitActive,
+      });
+    }
+
     if (!this.paused) {
       this.paused = new Promise(resolve => {
         this.resumeWorker = function () {
@@ -827,17 +925,35 @@ will never work with more accuracy than 1ms. */
       await (!doNotWaitActive && this.whenCurrentJobsFinished());
       this.emit('paused');
     }
+
+    if (this.tracer) {
+      span.end();
+    }
   }
 
   /**
    *
    * Resumes processing of this worker (if paused).
    */
   resume(): void {
+    let span;
+    if (this.tracer) {
+      const spanName = `${this.name} ${this.id} Worker.resume`;
+      span = this.tracer.startSpan(spanName);
+      span.setAttributes({
+        [TelemetryAttributes.WorkerName]: this.name,
+        [TelemetryAttributes.WorkerId]: this.id,
+      });
+    }
+
     if (this.resumeWorker) {
       this.resumeWorker();
       this.emit('resumed');
     }
+
+    if (this.tracer) {
+      span.end();
+    }
   }
 
   /**
@@ -872,7 +988,22 @@ will never work with more accuracy than 1ms. */
    * @returns Promise that resolves when the worker has been closed.
    */
   close(force = false): Promise<void> {
+    let span;
+    if (this.tracer) {
+      const spanName = `${this.name} ${this.id} Worker.close`;
+      span = this.tracer.startSpan(spanName);
+      span.setAttributes({
+        [TelemetryAttributes.WorkerName]: this.name,
+        [TelemetryAttributes.WorkerId]: this.id,
+        [TelemetryAttributes.WorkerForceClose]: force,
+      });
+    }
+
     if (this.closing) {
+      if (this.tracer) {
+        span.end();
+      }
+
       return this.closing;
     }
     this.closing = (async () => {
@@ -905,6 +1036,10 @@ will never work with more accuracy than 1ms. */
 
       this.closed = true;
       this.emit('closed');
+
+      if (this.tracer) {
+        span.end();
+      }
     })();
     return this.closing;
   }
@@ -922,6 +1057,16 @@ will never work with more accuracy than 1ms. */
    * @see {@link https://docs.bullmq.io/patterns/manually-fetching-jobs}
    */
   async startStalledCheckTimer(): Promise<void> {
+    let span;
+    if (this.tracer) {
+      const spanName = `${this.name} ${this.id} Worker.startStalledCheckTimer`;
+      span = this.tracer.startSpan(spanName);
+      span.setAttributes({
+        [TelemetryAttributes.WorkerName]: this.name,
+        [TelemetryAttributes.WorkerId]: this.id,
+      });
+    }
+
     if (!this.opts.skipStalledCheck) {
       clearTimeout(this.stalledCheckTimer);
 
@@ -936,6 +1081,10 @@ will never work with more accuracy than 1ms. */
         }
       }
     }
+
+    if (this.tracer) {
+      span.end();
+    }
   }
 
   private startLockExtenderTimer(
@@ -1017,6 +1166,17 @@ will never work with more accuracy than 1ms. */
   }
 
   protected async extendLocks(jobs: Job[]) {
+    let span;
+    if (this.tracer) {
+      const spanName = `${this.name} ${this.id} Worker.extendLocks`;
+      span = this.tracer.startSpan(spanName);
+      span.setAttributes({
+        [TelemetryAttributes.WorkerName]: this.name,
+        [TelemetryAttributes.WorkerId]: this.id,
+        [TelemetryAttributes.WorkerJobsInvolved]: JSON.stringify(jobs),
+      });
+    }
+
     try {
       const pipeline = (await this.client).pipeline();
       for (const job of jobs) {
@@ -1040,10 +1200,24 @@ will never work with more accuracy than 1ms. */
       }
     } catch (err) {
       this.emit('error', <Error>err);
+    } finally {
+      if (this.tracer) {
+        span.end();
+      }
     }
   }
 
   private async moveStalledJobsToWait() {
+    let span;
+    if (this.tracer) {
+      const spanName = `${this.name} ${this.id} Worker.moveStalledJobsToWait`;
+      span = this.tracer.startSpan(spanName);
+      span.setAttributes({
+        [TelemetryAttributes.WorkerName]: this.name,
+        [TelemetryAttributes.WorkerId]: this.id,
+      });
+    }
+
     const chunkSize = 50;
     const [failed, stalled] = await this.scripts.moveStalledJobsToWait();
 
@@ -1065,6 +1239,10 @@ will never work with more accuracy than 1ms. */
     }
 
     this.notifyFailedJobs(await Promise.all(jobPromises));
+
+    if (this.tracer) {
+      span.end();
+    }
   }
 
   private notifyFailedJobs(failedJobs: Job<DataType, ResultType, NameType>[]) {

src/enums/index.ts

@@ -2,3 +2,4 @@ export * from './child-command';
 export * from './error-code';
 export * from './parent-command';
 export * from './metrics-time';
+export * from './telemetry-attributes';

src/enums/telemetry-attributes.ts

@@ -0,0 +1,25 @@
+export enum TelemetryAttributes {
+  QueueName = 'bullmq.queue.name',
+  WorkerName = 'bullmq.worker.name',
+  BulkCount = 'bullmq.job.bulk.count',
+  BulkNames = 'bullmq.job.bulk.names',
+  JobName = 'bullmq.job.name',
+  JobId = 'bullmq.job.id',
+  JobKey = 'bullmq.job.key',
+  JobOptions = 'bullmq.job.options',
+  JobProgress = 'bullmq.job.progress',
+  QueueDrainDelay = 'bullmq.queue.drain.delay',
+  QueueGrace = 'bullmq.queue.grace',
+  QueueCleanLimit = 'bullmq.queue.clean.limit',
+  JobType = 'bullmq.job.type',
+  JobTimestamp = 'bullmq.job.timestamp',
+  QueueOptions = 'bullmq.queue.options',
+  QueueEventMaxLength = 'bullmq.queue.event.max.length',
+  WorkerOptions = 'bullmq.worker.options',
+  WorkerToken = 'bullmq.worker.token',
+  WorkerId = 'bullmq.worker.id',
+  WorkerRateLimit = 'bullmq.worker.rate.limit',
+  WorkerDoNotWaitActive = 'bullmq.worker.do.not.wait.active',
+  WorkerForceClose = 'bullmq.worker.force.close',
+  WorkerJobsInvolved = 'bullmq.worker.jobs.involved',
+}

src/interfaces/index.ts

@@ -23,3 +23,4 @@ export * from './repeat-options';
 export * from './sandboxed-job-processor';
 export * from './sandboxed-job';
 export * from './worker-options';
+export * from './telemetry';