feat(queue): observability for queue #2721

fgozdz · 2024-08-22T19:07:31Z

For the record, we are trying to adhere to the following guidelines: https://opentelemetry.io/docs/specs/semconv/messaging/messaging-spans/#conventions

manast · 2024-08-22T22:49:43Z

src/classes/queue.ts

@@ -250,6 +270,11 @@ export class Queue<
        },
      );
      this.emit('waiting', job);
+
+      if (this.tracer) {


wouldn't be useful to store the job.id in the span? (I have no idea, just wondering if this is something that maybe is needed in order to match this trace when the trace continues to the worker or something.

id won't be available as traces are happening before addition

@roggervalf yes, but it is an attribute that can be added to the span using setAttributes

manast · 2024-08-22T22:52:07Z

src/classes/queue.ts

+      });
+    }
+
+    const bulk = this.Job.createBulk<DataType, ResultType, NameType>(


I think we need to await here, as createBulk is an asynchronous operation, otherwise the span will actually end before the jobs are actually added to Redis.

manast · 2024-08-22T22:56:52Z

src/classes/queue.ts

    if (!this.closing) {
      if (this._repeat) {
        await this._repeat.close();
      }
    }
-    return super.close();
+
+    super.close();


we need to await here as well as super.close() is asynchronous.

manast · 2024-08-22T23:03:31Z

src/enums/opentelemetry-attributes.ts

@@ -0,0 +1,18 @@
+export enum OpenTelemetryAttributes {


As this interface is generic, it should be called just TelemetryAttributes

manast · 2024-08-22T23:03:54Z

src/interfaces/opentelemetry.ts

@@ -0,0 +1,25 @@
+export interface Telemetry {


opentelemetry.ts -> telemetry.ts

manast · 2024-08-22T23:05:12Z

src/interfaces/opentelemetry.ts

+}
+
+export interface Tracer {
+  startSpan(name: string): Span;


I wonder if this method startSpan wouldn't need to be asynchronous?

(same with end() method)

looks like it's synchronous so no need to use async

roggervalf

There are few other package/repos where you can base like:
https://github.com/appsignal/opentelemetry-instrumentation-bullmq
https://github.com/jenniferplusplus/opentelemetry-instrumentation-bullmq
Where authors were trying to follow openTel conventions

roggervalf · 2024-08-23T04:29:31Z

src/classes/queue.ts

+      const spanName = `${this.name}.${name} Queue.add`;
+      span = this.tracer.startSpan(spanName);
+      span.setAttributes({
+        [OpenTelemetryAttributes.QueueName]: this.name,


there are other attributes that can be added like:

span.setAttributes({ [SemanticAttributes.MESSAGING_SYSTEM]: 'BullMQ', // it could be a constant [SemanticAttributes.MESSAGING_DESTINATION]: this.name, [OpenTelemetryAttributes.JOB_NAME]: name });

You can use this package https://www.npmjs.com/package/@opentelemetry/semantic-conventions for semantic conventions

roggervalf · 2024-08-23T04:33:36Z

src/classes/queue.ts

@@ -250,6 +270,11 @@ export class Queue<
        },
      );
      this.emit('waiting', job);
+
+      if (this.tracer) {


id won't be available as traces are happening before addition

roggervalf · 2024-08-23T04:37:52Z

src/classes/queue.ts

-    return this.Job.createBulk<DataType, ResultType, NameType>(
+    let span;
+    if (this.tracer) {
+      const jobsInBulk = jobs.map(job => job.name);


it seems like there is a way to span different values when doing bulk operation https://github.com/appsignal/opentelemetry-instrumentation-bullmq/blob/main/src/instrumentation.ts#L285-L325

roggervalf · 2024-08-23T04:39:47Z

src/classes/queue.ts

@@ -299,12 +356,26 @@ export class Queue<
   *
   */
  async close(): Promise<void> {
+    let span;
+    if (this.tracer) {
+      const spanName = `${this.name} Queue.close`;


no sure if we need this span for now 🤔. Maybe we can start with add operations and then expand with other values while receiving more requests.

I will leave it for now i think and delete if it were to cause some problems

still I think to expose few metrics for now as needed, I don't see that the other packages are exposing so many. If we expose them now and then we want to remove, we will wait until a next breaking change. could you consider to spans same methods as in these packages #2721 (review) that looks like people are using now

My two cents: maybe close in particular is not so useful, however I do not think we need to implement only the same methods as in the existing packages, we can make it more powerful right in the first version. In the future we may even be able to make it configurable, which methods and which not to add spans to.

roggervalf · 2024-08-23T04:47:47Z

src/interfaces/opentelemetry.ts

+}
+
+export interface Tracer {
+  startSpan(name: string): Span;


looks like it's synchronous so no need to use async

manast · 2024-08-23T09:27:49Z

One thing that I forgot to mention, the interface should be properly documented using typedoc syntax, similarly as we document the rest of the API: https://typedoc.org/example/

…telemetry

manast · 2024-08-28T08:57:15Z

src/classes/worker.ts

    } catch (error) {
      this.running = false;
      throw error;
+    } finally {


This made me think, why do we need to make sure the span ends even in the case of an exception, but not in all other places? As most functions perform asynchronous calls, they could also potentially raise an exception. Have you thought about that?

Also, in the case of error, shouldn't we store this information in the span? just wondering here... how does the current open telemetry for BullMQ package handle exceptions?

manast · 2024-08-28T09:10:02Z

src/classes/worker.ts

+      span.setAttributes({
+        [TelemetryAttributes.WorkerName]: this.name,
+        [TelemetryAttributes.WorkerId]: this.id,
+        [TelemetryAttributes.WorkerJobsInvolved]: JSON.stringify(jobs),


wouldn't we also want to store all the IDs of the jobs whose locks we are extending?

manast · 2024-08-28T09:21:26Z

src/classes/worker.ts

      await this.client,
      await this.blockingConnection.client,
      token,
      { block },
    );
+
+    if (this.tracer) {


In order to match the job that was added to the queue we also need to specify the job.id here in the span attributes.

manast · 2024-08-28T09:23:01Z

src/classes/worker.ts

@@ -1065,6 +1239,10 @@ will never work with more accuracy than 1ms. */
    }

    this.notifyFailedJobs(await Promise.all(jobPromises));
+


I think we need to store the ids of the failed/stalled jobs here because they need to be mapped to the jobs added to the queue so that we can trace all the way of a job added to the queue to the place where they complete/fail/stall

…hod in queue base

manast · 2024-08-29T16:33:49Z

src/classes/queue.ts

+          [TelemetryAttributes.BulkCount]: jobs.length,
+        });
+
+        return await this.Job.createBulk<DataType, ResultType, NameType>(


In the case of addBulk, we are adding a bunch of jobs, so createBulk will return an array of jobs including their ids. However, later on, in the worker, when these jobs are processed, they are not processed in a bulk, so the question here is how do we handle it? should the span created by the worker when processing a job from this bulk span be mapped to it somehow? can a span have children?

Indeed, the spans seems to form a parent-child hierarchy in Otel at least: https://opentelemetry.io/docs/concepts/signals/traces/
Does it make sense that a span created for adding a job to the queue (or a bunch of jobs in a bulk) be the parent span of spans created later when the jobs are being processed?

The question is how would a worker when processing a job, find the parent span for that job, unless we could use the job Id for the span id...

manast · 2024-08-29T16:43:08Z

src/classes/queue.ts

+      () => `${this.name}.${name} Queue.add`,
+      async span => {
+        span?.setAttributes({
+          [TelemetryAttributes.QueueName]: this.name,


It looks like this line:

span?.setAttributes({ [TelemetryAttributes.QueueName]: this.name, });

is repeated on every span, couldn't we just move it to the "trace" method in QueueBase?

manast · 2024-08-29T16:46:31Z

src/classes/queue.ts

+        let deletedCount = 0;
+        const deletedJobsIds: string[] = [];
+
+        span?.setAttributes({


Better to move all attributes here instead of splitting in 2 calls to setAttributes

manast · 2024-08-29T16:53:40Z

src/classes/queue.ts

-      if (jobId == '0' || jobId?.startsWith('0:')) {
-        throw new Error("JobId cannot be '0' or start with 0:");
-      }
+    return this.trace(


for all the calls to trace we need to specify the correct return type, for example in this method it would look like this:

return this.trace<Job<DataType, ResultType, NameType>>(

This will help in reducing issues derived from returning the wrong type in the trace callback.

manast · 2024-08-29T16:56:17Z

src/classes/worker.ts

+          token,
+          { block },
+        );
+


it is possible that nextJob is undefined here, so we need to check before trying to set the attribute.

Actually I am not sure we need to create a span at all if there is no next job :thinking_face:

manast · 2024-08-29T17:00:52Z

src/classes/worker.ts

+        span?.setAttributes({
+          [TelemetryAttributes.WorkerName]: this.name,
+          [TelemetryAttributes.WorkerId]: this.id,
+          [TelemetryAttributes.WorkerJobsInvolved]: JSON.stringify(jobs),


Not sure we want to store the whole jobs here, job ids should be enough, otherwise the amount of memory required can be huge.

Another thing, why would this attribute be called WorkerJobsIdsInvolved? what's its purpose?

…telemetry

This reverts commit c4cb517, reversing changes made to 69c1296.

manast · 2024-09-04T19:46:14Z

src/classes/queue.ts

@@ -221,13 +221,9 @@ export class Queue<
    data: DataType,
    opts?: JobsOptions,
  ): Promise<Job<DataType, ResultType, NameType>> {
-    return this.trace(
+    return await this.trace<Job<DataType, ResultType, NameType>>(


There is no need to await if we are only returning and not waiting for the result. Same for all the places where there is a return followed by an await.

manast · 2024-09-04T19:52:51Z

src/classes/queue-base.ts

@@ -191,6 +192,10 @@ export class QueueBase extends EventEmitter implements MinimalQueue {

    const span = this.tracer.startSpan(getSpanName());


I think we should also set the span type, either producer (queue) or consumer (worker), as this is useful later on specially when implementing Otel. Maybe a new argument to trace like spanType:

protected trace<T>( spanType: SpanType // Producer, Consumer or Internal. getSpanName: () => string, callback: (span?: Span) => Promise<T> | T, ) {

manast · 2024-09-10T09:14:29Z

src/classes/queue-base.ts

+   * @returns
+   */
+  protected trace<T>(
+    getSpanType: () => SpanKind,


We do not need to have a function for getting the span type here, the reason for having it for the span name is because as the span name usually requires some computation (like concatenating strings, etc), by having it as a callback it does not perform those computations if telemetry is not enabled. For the spanType we are just passing a const, so this will not impact performance in any way.

manast · 2024-09-10T09:16:25Z

src/classes/queue-base.ts

+          activeTelemetryHeaders,
+        );
+
+        return this.contextManager.with(activeContext, () => callback(span));


I assume contextManager.with will (for the Otel case), create a parent-child relationship for the spans?

manast · 2024-09-10T09:18:01Z

src/classes/queue-base.ts

+          activeTelemetryHeaders,
+        );
+
+        return this.contextManager.with(activeContext, () => callback(span));


Also, in this case we need to call it like awaiting like this:

return await this.contextManager.with(activeContext, () => callback(span));

The reason being that we want to catch potential exceptions thrown by the callback, without this await the exception will just bubble up and we loose the ability to record the exception in the tracer.

manast · 2024-09-10T09:20:37Z

src/classes/queue-base.ts

+
+      this.propagation.inject(this.contextManager.active(), telemetryHeaders);
+
+      return this.contextManager.with(


same issue here we need to await before returning.

manast · 2024-09-10T09:29:16Z

src/classes/queue.ts

-        if (+new Date(opts.repeat.endDate) < Date.now()) {
-          throw new Error('End date must be greater than current timestamp');
+    return this.trace<Job<DataType, ResultType, NameType>>(
+      () => 3,


We should always use enums or constants instead of magic numbers.

In this case SpanType.PRODUCER (I think)

manast · 2024-09-10T09:37:26Z

src/classes/queue-base.ts

+
+      const telemetryHeaders: Record<string, string> = {};
+
+      this.propagation.inject(this.contextManager.active(), telemetryHeaders);


Ok, so we are going to always inject the headers, hmm, we may need to give it a bit more of thought, not sure this is 100% correct.

manast · 2024-09-10T09:40:02Z

src/classes/queue.ts

+      () => `${this.name}.${name} Queue.add`,
+      async (span, telemetryHeaders) => {
+        if (telemetryHeaders) {
+          data = { ...data, telemetryHeaders };


The telemetry data should be stored in the options, as the data is user data which we should not tamper with.

we probably need to define a "telemetryData" or similar optional option in JobOpts with type "any" so that we can store the telemetry data there.

manast · 2024-09-10T09:41:27Z

src/classes/queue.ts

+            name: job.name,
+            data: {
+              ...job.data,
+              ...(span && telemetryHeaders),


Should go on the options object.

src/classes/queue-base.ts

manast · 2024-09-10T13:16:15Z

src/interfaces/telemetry.ts

+}
+
+export interface ContextManager {
+  with<A extends (...args: any[]) => any>(


The documentation for "with":

Creates a new context and sets it as active for the fn passed as last argument

manast · 2024-09-10T13:17:25Z

src/interfaces/telemetry.ts

+type HighResolutionTime = [number, number];
+
+export interface Propagation {
+  inject<T>(context: Context, carrier: T, setter?: TextMapSetter): void;


As we are using the propagation in a very controlled environment I am not sure we need the setter, if so better remove it to make integrations easier.

manast · 2024-09-10T13:19:14Z

src/interfaces/worker-options.ts

@@ -144,6 +145,11 @@ export interface WorkerOptions extends QueueBaseOptions {
   * @default false
   */
  useWorkerThreads?: boolean;
+
+  /**
+   * Telemetry client


Rename comment to "Telemetry Addon"

manast · 2024-09-10T13:23:00Z

src/interfaces/telemetry.ts

+  get<T>(carrier: T, key: string): undefined | string | string[];
+}
+
+export interface JobDataWithHeaders {


We can move telemetryHeaders to BaseJobOptions: https://github.com/taskforcesh/bullmq/blob/master/src/interfaces/base-job-options.ts#L77
I would rename it to maybe "telemetryMetadata", later when serialising the options before storing in Redis we will optimise it to a shorter name.

or maybe a better name: telemetryPropagation not sure.

manast · 2024-09-10T13:39:22Z

src/classes/queue.ts

-        if (+new Date(opts.repeat.endDate) < Date.now()) {
-          throw new Error('End date must be greater than current timestamp');
+    return this.trace<Job<DataType, ResultType, NameType>>(
+      () => 3,


In this case SpanType.PRODUCER (I think)

…tests for telemetry interface

manast

Looks good! But please address the new and previous comments.

manast · 2024-09-17T09:46:00Z

src/classes/queue-base.ts

@@ -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


I do not think this comment is correct anymore as we are using the trace helper.

manast · 2024-09-17T09:48:26Z

src/classes/queue-base.ts

+      this.propagation = opts.telemetry.propagation;
+
+      this.contextManager.getMetadata = (context: Context) => {
+        const metadata = {};


getMetadata and fromMetadata should be implemented in the integration. For Otel for example they will use the propagation module, but we do not need to expose the propagation module in our generic telemetry interface.

manast · 2024-09-17T09:49:01Z

src/classes/queue-base.ts

@@ -175,4 +216,61 @@ export class QueueBase extends EventEmitter implements MinimalQueue {
      }
    }
  }
+
+  /**
+   * Wraps the code with telemetry and provides span for configuration.


"... provides a span for configuration."

manast · 2024-09-17T10:05:52Z

src/classes/queue.ts

-    await this.scripts.pause(true);
-    this.emit('paused');
+    await this.trace<void>(
+      SpanKind.PRODUCER,


In this case I do not think the kind is PRODUCER, as we are just pausing the queue, not producing any jobs. Probably INTERNAL is the correct one, not sure.

manast · 2024-09-17T10:06:04Z

src/classes/queue.ts

-    }
-    return super.close();
+    await this.trace<void>(
+      SpanKind.PRODUCER,


Same here, should not be PRODUCER

manast · 2024-09-17T10:09:55Z

src/classes/queue.ts

-    const client = await this.client;
-    return client.xtrim(this.keys.events, 'MAXLEN', '~', maxLength);
+    return this.trace<number>(
+      SpanKind.PRODUCER,


I am doubting if it is correct or necessary to have trace on all these methods actually. Because the most important duty of the telemetry is to be able to trace a job from its creation to its completion. But sure there may be situations where a job gets removed manually and things like that, probably we want to trace them too. But "pausing" a queue, or "trimEvents", it does not look like this is something we want to have spans for.

trimEvents I guess is not needed for tracing at all.

manast · 2024-09-17T10:12:17Z

src/classes/queue.ts

@@ -424,7 +510,18 @@ export class Queue<
    jobId: string,
    progress: number | object,
  ): Promise<void> {
-    return this.scripts.updateProgress(jobId, progress);
+    await this.trace<void>(
+      SpanKind.PRODUCER,


Updating a job progress should somehow also map to the span that created the job. Also the SpanKind cannot be PRODUCER.

manast · 2024-09-17T10:14:37Z

tests/test_telemetry_interface.ts

+    end(): void {}
+  }
+
+  class MockPropagation implements Propagation {


Should not be needed as Propagation is OTel implementation specific

manast · 2024-09-17T10:18:27Z

tests/test_telemetry_interface.ts

+
+        expect.fail('Expected an error to be thrown');
+      } catch (error) {
+        span.recordException(error);


We should not need to call recordException here as this is supposed to be called inside queue.add when an exception occurs.

I see, the issue you have here is that you are throwing an exception as soon as queue.add is called, but by doing so we are not really testing the try/catch inside queue.add So you could try to throw an error when calling Job.create instead which is called inside queue.add.

manast · 2024-09-17T10:37:18Z

tests/test_telemetry_interface.ts

+      const span = telemetryClient.trace
+        .getTracer('testtracer')
+        .startSpan('Queue.addBulk.error') as MockSpan;
+      const recordExceptionSpy = sinon.spy(span, 'recordException');


Same comment as above, in order for this test to be meaningful it should throw an exception from inside addBulk.

manast · 2024-09-20T08:32:15Z

src/classes/queue.ts

-      cursor = await this.scripts.promoteJobs(opts.count);
-    } while (cursor);
+    await this.trace<void>(
+      3,


Make sure to change the 3 to KindSpan enum.

… for shorter names

manast · 2024-09-23T09:32:26Z

src/classes/worker.ts

-      const client = await this.client;
-      const bclient = await this.blockingConnection.client;
+    await this.trace<void>(
+      SpanKind.CONSUMER,


I do not think this is the correct type, as this span is not consuming but starting the main worker loop.

manast · 2024-09-23T09:37:53Z

src/classes/worker.ts

-      token,
-      { block },
+    return this.trace<Job<DataType, ResultType, NameType>>(
+      SpanKind.CONSUMER,


I think this should still be kind INTERNAL, because I do not think we should have more than one kind CONSUMER per job, and in this case getNextJob would lead to processJob which also has kind CONSUMER

manast · 2024-09-23T09:40:17Z

src/classes/worker.ts

-    if (!job || this.closing || this.paused) {
-      return;
-    }
+    const { telemetryMetadata: dstPropagationMedatada } = job.opts;


Shouldn't this actually be the "source propagation metadata" ?

manast · 2024-09-23T09:40:42Z

src/classes/worker.ts

-    }
+    const { telemetryMetadata: dstPropagationMedatada } = job.opts;
+
+    return this.trace<void | Job<DataType, ResultType, NameType>>(


Shouldn't we need to pass the source propagation metadata as last parameter here so that it maps the job added by the PRODUCER?

As the job can be completed or failed in this case, shouldn't we mark the span with this information somehow? Could you please check the old Otel implementation to see how completed and failed jobs are handled?

Please check the above comment I think you missed it before.

manast

I wrote some comments, please do not forget to address the older comments as well.

manast · 2024-10-09T12:13:02Z

src/classes/queue-base.ts

+      {
+        kind: spanKind,
+      },
+      currentContext,


If srcPropagationMetadata is not defined, nor will currentContext, is this correct?

yes, startSpan method will be called without existing context, later it is bond to a new active context with setSpan method

But setSpan is only called if spanKind is of type PRODUCER.

I addressed it in newest commit

manast · 2024-10-09T12:24:00Z

src/classes/worker.ts

-        this.emit('completed', job, result, 'active');
-        const [jobData, jobId, limitUntil, delayUntil] = completed || [];
-        this.updateDelays(limitUntil, delayUntil);
+        if (!job || this.closing || this.paused) {


I think this is wrong, if there is no job, we should not create a span here. It is common that there is no job to process when this function is called, so this should be left as it was, i.e. returning before creating a new span.

manast · 2024-10-09T12:24:22Z

src/classes/worker.ts

-    }
+    const { telemetryMetadata: dstPropagationMedatada } = job.opts;
+
+    return this.trace<void | Job<DataType, ResultType, NameType>>(


Please check the above comment I think you missed it before.

manast · 2024-10-09T12:25:58Z

src/classes/worker.ts

+          [TelemetryAttributes.WorkerId]: this.id,
+        });
+
+        if (this.resumeWorker) {


Same here, we should not create a span if we are not resuming (as this check if used to see if the worker is paused to begin with).

manast · 2024-10-09T12:26:46Z

src/classes/worker.ts

-        } catch (err) {
-          this.emit('error', <Error>err);
+  async close(force = false): Promise<void> {
+    await this.trace<void>(


We should not need to create a span if we are already closing (as somebody already called this method before and is wrongly calling it again).

manast · 2024-10-09T12:27:48Z

src/classes/worker.ts

@@ -958,20 +1044,32 @@ will never work with more accuracy than 1ms. */
   * @see {@link https://docs.bullmq.io/patterns/manually-fetching-jobs}
   */
  async startStalledCheckTimer(): Promise<void> {
-    if (!this.opts.skipStalledCheck) {


same here, do not create a span if we are skipping the stalled checker. The spans are not just for wrapping methods, but if there is actually something useful to do in them! :)

…there is a point to

…or every SpanKind

…eat/opentelemetry

fgozdz

LGTM

…eat/opentelemetry

# [5.22.0](v5.21.2...v5.22.0) (2024-10-31) ### Bug Fixes * **commands:** add missing build statement when releasing [python] ([#2869](#2869)) fixes [#2868](#2868) ([ff2a47b](ff2a47b)) ### Features * **job:** add getChildrenValues method [python] ([#2853](#2853)) ([0f25213](0f25213)) * **queue:** add a telemetry interface ([#2721](#2721)) ([273b574](273b574))

This PR contains the following updates: | Package | Type | Update | Change | |---|---|---|---| | [bullmq](https://bullmq.io/) ([source](https://github.com/taskforcesh/bullmq)) | dependencies | minor | [`5.21.2` -> `5.23.0`](https://renovatebot.com/diffs/npm/bullmq/5.21.2/5.23.0) | --- ### Release Notes <details> <summary>taskforcesh/bullmq (bullmq)</summary> ### [`v5.23.0`](https://github.com/taskforcesh/bullmq/releases/tag/v5.23.0) [Compare Source](taskforcesh/bullmq@v5.22.0...v5.23.0) ##### Features - **scheduler:** add getJobScheduler method ([#2877](taskforcesh/bullmq#2877)) ref [#2875](taskforcesh/bullmq#2875) ([956d98c](taskforcesh/bullmq@956d98c)) ### [`v5.22.0`](https://github.com/taskforcesh/bullmq/releases/tag/v5.22.0) [Compare Source](taskforcesh/bullmq@v5.21.2...v5.22.0) ##### Bug Fixes - **commands:** add missing build statement when releasing \[python] ([#2869](taskforcesh/bullmq#2869)) fixes [#2868](taskforcesh/bullmq#2868) ([ff2a47b](taskforcesh/bullmq@ff2a47b)) ##### Features - **job:** add getChildrenValues method \[python] ([#2853](taskforcesh/bullmq#2853)) ([0f25213](taskforcesh/bullmq@0f25213)) - **queue:** add a telemetry interface ([#2721](taskforcesh/bullmq#2721)) ([273b574](taskforcesh/bullmq@273b574)) </details> --- ### Configuration 📅 **Schedule**: Branch creation - At any time (no schedule defined), Automerge - At any time (no schedule defined). 🚦 **Automerge**: Disabled by config. Please merge this manually once you are satisfied. ♻ **Rebasing**: Whenever PR becomes conflicted, or you tick the rebase/retry checkbox. 🔕 **Ignore**: Close this PR and you won't be reminded about this update again. --- - [ ] If you want to rebase/retry this PR, check this box --- This PR has been generated by [Renovate Bot](https://github.com/renovatebot/renovate).  Reviewed-on: https://git.tristess.app/alexandresoro/ouca/pulls/290 Reviewed-by: Alexandre Soro <code@soro.dev> Co-authored-by: renovate <renovate@git.tristess.app> Co-committed-by: renovate <renovate@git.tristess.app>

feat(queue): observability for queue

722fb5d

fgozdz requested a review from manast August 22, 2024 19:07

manast reviewed Aug 22, 2024

View reviewed changes

roggervalf reviewed Aug 23, 2024

View reviewed changes

feat(worker): observability for worker

1b2fd0b

fgozdz requested review from manast and roggervalf August 23, 2024 15:17

fgozdz added 3 commits August 26, 2024 18:59

feat(queue, worker): fix telemetry setup

e6dfb85

feat(queue, worker): describe features

a75a8dd

Merge branch 'master' of github.com:taskforcesh/bullmq into feat/open…

1b176c6

…telemetry

fgozdz marked this pull request as ready for review August 28, 2024 09:14

manast reviewed Aug 28, 2024

View reviewed changes

fgozdz added 2 commits August 29, 2024 09:44

feat(queue, worker): add exception handler

66b85c5

feat(queue, worker): remove redundancy in code by creating helper met…

54a2c18

…hod in queue base

fgozdz marked this pull request as draft August 29, 2024 15:14

manast reviewed Aug 29, 2024

View reviewed changes

fgozdz added 5 commits September 4, 2024 16:28

feat(queue, worker): telemetry attributes change, and fix promises

2e3fe37

feat(queue, worker): resolve conflict

c4cb517

Merge branch 'master' of github.com:taskforcesh/bullmq into feat/open…

14b2f2e

…telemetry

Revert "feat(queue, worker): resolve conflict"

7f6c674

This reverts commit c4cb517, reversing changes made to 69c1296.

fix(documentation): fix documentation formatting

059d50b

manast reviewed Sep 5, 2024

View reviewed changes

fgozdz added 3 commits September 6, 2024 01:29

feat(queue-base): add context manager for telemetry

f70cc8a

feat(worker, queue): add spankind and distributed context propagation

8f38087

fix(worker, queue): remove unused import

379abf5

manast reviewed Sep 10, 2024

View reviewed changes

Merge branch 'master' into feat/opentelemetry

3e8e0d8

manast reviewed Sep 10, 2024

View reviewed changes

feat(queue, worker): documentation and changes to propagation, basic …

6169ea3

…tests for telemetry interface

feat(test_telemetry_interface): remove unused imports

908ec13

manast reviewed Sep 17, 2024

View reviewed changes

manast reviewed Sep 20, 2024

View reviewed changes

feat(queue, worker): correct tests and spankind, use property mapping…

2dab35f

… for shorter names

manast reviewed Sep 23, 2024

View reviewed changes

This was referenced Sep 23, 2024

Built-in Observability interface in BullMQ jenniferplusplus/opentelemetry-instrumentation-bullmq#26

Open

Built-in Observability interface in BullMQ appsignal/opentelemetry-instrumentation-bullmq#9

Open

fgozdz added 3 commits September 24, 2024 00:07

feat(worker): minor changes

846e3ef

feat(queue, worker): distributed tracing

fc733e9

feat(queue, worker): resolve conflicts

4c199f8

manast reviewed Oct 9, 2024

View reviewed changes

fgozdz and others added 7 commits October 10, 2024 12:48

feat(worker): job error handling, do not trace certain methods until …

7261e40

…there is a point to

feat(queue-base): add propagation for internal spanKind and setSpan f…

677fcc6

…or every SpanKind

refactor(telemetry): several improvements and small changes

51a353e

fix(telemetry): fix return type of setSpanOnContext

6742a03

fix(job): return result of trace

167a7dd

test(telemetry): fix broken tests

210d36e

Merge branch 'master' of https://github.com/taskforcesh/bullmq into f…

0f172ff

…eat/opentelemetry

manast self-assigned this Oct 29, 2024

manast marked this pull request as ready for review October 29, 2024 10:45

manast added 3 commits October 29, 2024 12:03

chore(job): rename srcPropagationMetadata to dstPropagationMetadata

23924d3

docs(telemetry): initial interface documentation

49b2c36

chore: small improvements to naming of telemetry spans

9b0b3fa

fgozdz commented Oct 31, 2024

View reviewed changes

Merge branch 'master' of https://github.com/taskforcesh/bullmq into f…

e35233b

…eat/opentelemetry

manast merged commit 273b574 into master Oct 31, 2024
11 checks passed

manast deleted the feat/opentelemetry branch October 31, 2024 15:26

		@@ -1065,6 +1239,10 @@ will never work with more accuracy than 1ms. */
		}

		this.notifyFailedJobs(await Promise.all(jobPromises));

		@@ -191,6 +192,10 @@ export class QueueBase extends EventEmitter implements MinimalQueue {

		const span = this.tracer.startSpan(getSpanName());


		this.propagation.inject(this.contextManager.active(), telemetryHeaders);

		return this.contextManager.with(


		const telemetryHeaders: Record<string, string> = {};

		this.propagation.inject(this.contextManager.active(), telemetryHeaders);

feat(queue): observability for queue #2721

feat(queue): observability for queue #2721

Conversation

fgozdz commented Aug 22, 2024 • edited by manast Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

roggervalf left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

manast commented Aug 23, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

manast left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

fgozdz commented Aug 22, 2024 •

edited by manast

Loading