feat(queue): observability for queue

src/classes/queue-base.ts

@@ -1,5 +1,14 @@
 import { EventEmitter } from 'events';
-import { QueueBaseOptions, RedisClient } from '../interfaces';
+import {
+  QueueBaseOptions,
+  RedisClient,
+  Span,
+  Tracer,
+  SetSpan,
+  ContextManager,
+  SpanKind,
+  Propagation,
+} from '../interfaces';
 import { MinimalQueue } from '../types';
 import {
   delay,
@@ -11,6 +20,7 @@ import { RedisConnection } from './redis-connection';
 import { Job } from './job';
 import { KeysMap, QueueKeys } from './queue-keys';
 import { Scripts } from './scripts';
+import { TelemetryAttributes } from '../enums';
 
 /**
  * @class QueueBase
@@ -30,6 +40,16 @@ export class QueueBase extends EventEmitter implements MinimalQueue {
   protected connection: RedisConnection;
   public readonly qualifiedName: string;
 
+  /**
+   * 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;
+  private setSpan: SetSpan | undefined;
+  protected contextManager: ContextManager | undefined;
+  protected propagation: Propagation | undefined;
+
   /**
    *
    * @param name - The name of the queue.
@@ -76,6 +96,13 @@ export class QueueBase extends EventEmitter implements MinimalQueue {
     this.keys = queueKeys.getKeys(name);
     this.toKey = (type: string) => queueKeys.toKey(name, type);
     this.setScripts();
+
+    if (opts?.telemetry) {
+      this.tracer = opts.telemetry.trace.getTracer(opts.telemetry.tracerName);
+      this.setSpan = opts.telemetry.trace.setSpan;
+      this.contextManager = opts.telemetry.contextManager;
+      this.propagation = opts.telemetry.propagation;
+    }
   }
 
   /**
@@ -175,4 +202,59 @@ export class QueueBase extends EventEmitter implements MinimalQueue {
       }
     }
   }
+
+  /**
+   * Wraps the code with telemetry and provides span for configuration.
+   *
+   * @param spanType - type of the span: Producer, Consumer, Internal
+   * @param getSpanName - name of the span
+   * @param callback - code to wrap with telemetry
+   * @returns
+   */
+  protected trace<T>(
+    getSpanType: () => SpanKind,
+    getSpanName: () => string,
+    callback: (
+      span?: Span,
+      telemetryHeaders?: Record<string, string>,
+    ) => Promise<T> | T,
+    activeTelemetryHeaders?: Record<string, string>,
+  ) {
+    if (!this.tracer) {
+      return callback();
+    }
+
+    const span = this.tracer.startSpan(getSpanName(), {
+      kind: getSpanType(),
+    });
+
+    span.setAttributes({
+      [TelemetryAttributes.QueueName]: this.name,
+    });
+
+    try {
+      if (activeTelemetryHeaders) {
+        const activeContext = this.propagation.extract(
+          this.contextManager.active(),
+          activeTelemetryHeaders,
+        );
+
+        return this.contextManager.with(activeContext, () => callback(span));
+      }
+
+      const telemetryHeaders: Record<string, string> = {};
+
+      this.propagation.inject(this.contextManager.active(), telemetryHeaders);
+
+      return this.contextManager.with(
+        this.setSpan(this.contextManager.active(), span),
+        () => callback(span, telemetryHeaders),
+      );
+    } catch (err) {
+      span.recordException(err as Error);
+      throw err;
+    } finally {
+      span.end();
+    }
+  }
 }