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,
+  Propagation,
+  Context,
+} 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, SpanKind } 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,27 @@ 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;
+
+      this.contextManager.getMetadata = (context: Context) => {
+        const metadata = {};
+        this.propagation.inject(context, metadata);
+        return metadata;
+      };
+
+      this.contextManager.fromMetadata = (
+        activeContext: Context,
+        metadata: Record<string, string>,
+      ) => {
+        const context = this.propagation.extract(activeContext, metadata);
+        return context;
+      };
+    }
   }
 
   /**
@@ -175,4 +216,61 @@ export class QueueBase extends EventEmitter implements MinimalQueue {
       }
     }
   }
+
+  /**
+   * Wraps the code with telemetry and provides span for configuration.
+   *
+   * @param spanKind - kind of the span: Producer, Consumer, Internal
+   * @param getSpanName - name of the span
+   * @param callback - code to wrap with telemetry
+   * @param srcPropagationMedatada -
+   * @returns
+   */
+  protected async trace<T>(
+    spanKind: SpanKind,
+    getSpanName: () => string,
+    callback: (
+      span?: Span,
+      dstPropagationMetadata?: Record<string, string>,
+    ) => Promise<T> | T,
+    srcPropagationMetadata?: Record<string, string>,
+  ) {
+    if (!this.tracer) {
+      return callback();
+    }
+
+    const span = this.tracer.startSpan(getSpanName(), {
+      kind: spanKind,
+    });
+
+    try {
+      span.setAttributes({
+        [TelemetryAttributes.QueueName]: this.name,
+      });
+
+      let activeContext = this.contextManager.active();
+      if (srcPropagationMetadata) {
+        activeContext = this.contextManager.fromMetadata(
+          activeContext,
+          srcPropagationMetadata,
+        );
+      }
+
+      let dstPropagationMetadata: undefined | Record<string, string>;
+      if (spanKind === SpanKind.PRODUCER) {
+        dstPropagationMetadata = this.contextManager.getMetadata(activeContext);
+      }
+
+      const messageContext = this.setSpan(activeContext, span);
+
+      return await this.contextManager.with(messageContext, () =>
+        callback(span, dstPropagationMetadata),
+      );
+    } catch (err) {
+      span.recordException(err as Error);
+      throw err;
+    } finally {
+      span.end();
+    }
+  }
 }