feat(flows): add telemetry support

src/classes/flow-producer.ts

@@ -8,11 +8,14 @@ import {
   IoredisListener,
   QueueBaseOptions,
   RedisClient,
+  Tracer,
+  ContextManager,
 } from '../interfaces';
-import { getParentKey, isRedisInstance } from '../utils';
+import { getParentKey, isRedisInstance, trace } from '../utils';
 import { Job } from './job';
 import { KeysMap, QueueKeys } from './queue-keys';
 import { RedisConnection } from './redis-connection';
+import { SpanKind, TelemetryAttributes } from '../enums';
 
 export interface AddNodeOpts {
   multi: ChainableCommander;
@@ -95,6 +98,10 @@ export class FlowProducer extends EventEmitter {
   queueKeys: QueueKeys;
 
   protected connection: RedisConnection;
+  protected telemetry: {
+    tracer: Tracer | undefined;
+    contextManager: ContextManager | undefined;
+  };
 
   constructor(
     public opts: QueueBaseOptions = { connection: {} },
@@ -122,6 +129,10 @@ export class FlowProducer extends EventEmitter {
     });
 
     this.queueKeys = new QueueKeys(opts.prefix);
+
+    if (opts?.telemetry) {
+      this.telemetry = opts.telemetry;
+    }
   }
 
   emit<U extends keyof FlowProducerListener>(
@@ -196,19 +207,32 @@ export class FlowProducer extends EventEmitter {
       ? `${parentKey}:dependencies`
       : undefined;
 
-    const jobsTree = this.addNode({
-      multi,
-      node: flow,
-      queuesOpts: opts?.queuesOptions,
-      parent: {
-        parentOpts,
-        parentDependenciesKey,
-      },
-    });
+    return trace<Promise<JobNode>>(
+      this.telemetry,
+      SpanKind.PRODUCER,
+      flow.queueName,
+      'addFlow',
+      flow.queueName,
+      async span => {
+        span?.setAttributes({
+          [TelemetryAttributes.FlowName]: flow.name,
+        });
+
+        const jobsTree = await this.addNode({
+          multi,
+          node: flow,
+          queuesOpts: opts?.queuesOptions,
+          parent: {
+            parentOpts,
+            parentDependenciesKey,
+          },
+        });
 
-    await multi.exec();
+        await multi.exec();
 
-    return jobsTree;
+        return jobsTree;
+      },
+    );
   }
 
   /**
@@ -255,11 +279,27 @@ export class FlowProducer extends EventEmitter {
     const client = await this.connection.client;
     const multi = client.multi();
 
-    const jobsTrees = this.addNodes(multi, flows);
-
-    await multi.exec();
-
-    return jobsTrees;
+    return trace<Promise<JobNode[]>>(
+      this.telemetry,
+      SpanKind.PRODUCER,
+      '',
+      'addBulkFlows',
+      '',
+      async span => {
+        span?.setAttributes({
+          [TelemetryAttributes.BulkCount]: flows.length,
+          [TelemetryAttributes.BulkNames]: flows
+            .map(flow => flow.name)
+            .join(','),
+        });
+
+        const jobsTrees = await this.addNodes(multi, flows);
+
+        await multi.exec();
+
+        return jobsTrees;
+      },
+    );
   }
 
   /**
@@ -273,70 +313,92 @@ export class FlowProducer extends EventEmitter {
    * @param parent - parent data sent to children to create the "links" to their parent
    * @returns
    */
-  protected addNode({ multi, node, parent, queuesOpts }: AddNodeOpts): JobNode {
+  protected async addNode({
+    multi,
+    node,
+    parent,
+    queuesOpts,
+  }: AddNodeOpts): Promise<JobNode> {
     const prefix = node.prefix || this.opts.prefix;
     const queue = this.queueFromNode(node, new QueueKeys(prefix), prefix);
     const queueOpts = queuesOpts && queuesOpts[node.queueName];
 
     const jobsOpts = queueOpts?.defaultJobOptions ?? {};
     const jobId = node.opts?.jobId || v4();
 
-    const job = new this.Job(
-      queue,
+    return trace<Promise<JobNode>>(
+      this.telemetry,
+      SpanKind.PRODUCER,
       node.name,
-      node.data,
-      {
-        ...jobsOpts,
-        ...node.opts,
-        parent: parent?.parentOpts,
-      },
-      jobId,
-    );
-
-    const parentKey = getParentKey(parent?.parentOpts);
-
-    if (node.children && node.children.length > 0) {
-      // Create parent job, will be a job in status "waiting-children".
-      const parentId = jobId;
-      const queueKeysParent = new QueueKeys(node.prefix || this.opts.prefix);
-      const waitChildrenKey = queueKeysParent.toKey(
-        node.queueName,
-        'waiting-children',
-      );
-
-      job.addJob(<Redis>(multi as unknown), {
-        parentDependenciesKey: parent?.parentDependenciesKey,
-        waitChildrenKey,
-        parentKey,
-      });
-
-      const parentDependenciesKey = `${queueKeysParent.toKey(
-        node.queueName,
-        parentId,
-      )}:dependencies`;
-
-      const children = this.addChildren({
-        multi,
-        nodes: node.children,
-        parent: {
-          parentOpts: {
-            id: parentId,
-            queue: queueKeysParent.getQueueQualifiedName(node.queueName),
+      'addNode',
+      node.queueName,
+      async (span, dstPropagationMetadata) => {
+        span?.setAttributes({
+          [TelemetryAttributes.JobName]: node.name,
+          [TelemetryAttributes.JobId]: jobId,
+        });
+
+        const job = new this.Job(
+          queue,
+          node.name,
+          node.data,
+          {
+            ...jobsOpts,
+            ...node.opts,
+            parent: parent?.parentOpts,
+            telemetryMetadata: dstPropagationMetadata,
           },
-          parentDependenciesKey,
-        },
-        queuesOpts,
-      });
-
-      return { job, children };
-    } else {
-      job.addJob(<Redis>(multi as unknown), {
-        parentDependenciesKey: parent?.parentDependenciesKey,
-        parentKey,
-      });
+          jobId,
+        );
 
-      return { job };
-    }
+        const parentKey = getParentKey(parent?.parentOpts);
+
+        if (node.children && node.children.length > 0) {
+          // Create the parent job, it will be a job in status "waiting-children".
+          const parentId = jobId;
+          const queueKeysParent = new QueueKeys(
+            node.prefix || this.opts.prefix,
+          );
+          const waitChildrenKey = queueKeysParent.toKey(
+            node.queueName,
+            'waiting-children',
+          );
+
+          await job.addJob(<Redis>(multi as unknown), {
+            parentDependenciesKey: parent?.parentDependenciesKey,
+            waitChildrenKey,
+            parentKey,
+          });
+
+          const parentDependenciesKey = `${queueKeysParent.toKey(
+            node.queueName,
+            parentId,
+          )}:dependencies`;
+
+          const children = await this.addChildren({
+            multi,
+            nodes: node.children,
+            parent: {
+              parentOpts: {
+                id: parentId,
+                queue: queueKeysParent.getQueueQualifiedName(node.queueName),
+              },
+              parentDependenciesKey,
+            },
+            queuesOpts,
+          });
+
+          return { job, children };
+        } else {
+          await job.addJob(<Redis>(multi as unknown), {
+            parentDependenciesKey: parent?.parentDependenciesKey,
+            parentKey,
+          });
+
+          return { job };
+        }
+      },
+    );
   }
 
   /**
@@ -349,23 +411,28 @@ export class FlowProducer extends EventEmitter {
    * @param nodes - the nodes representing jobs to be added to some queue
    * @returns
    */
-  protected addNodes(multi: ChainableCommander, nodes: FlowJob[]): JobNode[] {
-    return nodes.map(node => {
-      const parentOpts = node?.opts?.parent;
-      const parentKey = getParentKey(parentOpts);
-      const parentDependenciesKey = parentKey
-        ? `${parentKey}:dependencies`
-        : undefined;
-
-      return this.addNode({
-        multi,
-        node,
-        parent: {
-          parentOpts,
-          parentDependenciesKey,
-        },
-      });
-    });
+  protected addNodes(
+    multi: ChainableCommander,
+    nodes: FlowJob[],
+  ): Promise<JobNode[]> {
+    return Promise.all(
+      nodes.map(node => {
+        const parentOpts = node?.opts?.parent;
+        const parentKey = getParentKey(parentOpts);
+        const parentDependenciesKey = parentKey
+          ? `${parentKey}:dependencies`
+          : undefined;
+
+        return this.addNode({
+          multi,
+          node,
+          parent: {
+            parentOpts,
+            parentDependenciesKey,
+          },
+        });
+      }),
+    );
   }
 
   private async getNode(client: RedisClient, node: NodeOpts): Promise<JobNode> {
@@ -406,7 +473,9 @@ export class FlowProducer extends EventEmitter {
   }
 
   private addChildren({ multi, nodes, parent, queuesOpts }: AddChildrenOpts) {
-    return nodes.map(node => this.addNode({ multi, node, parent, queuesOpts }));
+    return Promise.all(
+      nodes.map(node => this.addNode({ multi, node, parent, queuesOpts })),
+    );
   }
 
   private getChildren(