Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

feat(queue): observability for queue #2721

Draft
wants to merge 22 commits into
base: master
Choose a base branch
from
Draft

feat(queue): observability for queue #2721

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
22 commits
Select commit Hold shift + click to select a range
722fb5d
feat(queue): observability for queue
fgozdz Aug 22, 2024
1b2fd0b
feat(worker): observability for worker
fgozdz Aug 23, 2024
e6dfb85
feat(queue, worker): fix telemetry setup
fgozdz Aug 26, 2024
a75a8dd
feat(queue, worker): describe features
fgozdz Aug 27, 2024
1b176c6
Merge branch 'master' of github.com:taskforcesh/bullmq into feat/open…
fgozdz Aug 28, 2024
66b85c5
feat(queue, worker): add exception handler
fgozdz Aug 29, 2024
54a2c18
feat(queue, worker): remove redundancy in code by creating helper met…
fgozdz Aug 29, 2024
2e3fe37
feat(queue, worker): telemetry attributes change, and fix promises
fgozdz Sep 4, 2024
c4cb517
feat(queue, worker): resolve conflict
fgozdz Sep 4, 2024
14b2f2e
Merge branch 'master' of github.com:taskforcesh/bullmq into feat/open…
fgozdz Sep 5, 2024
7f6c674
Revert "feat(queue, worker): resolve conflict"
fgozdz Sep 5, 2024
059d50b
fix(documentation): fix documentation formatting
fgozdz Sep 5, 2024
f70cc8a
feat(queue-base): add context manager for telemetry
fgozdz Sep 5, 2024
8f38087
feat(worker, queue): add spankind and distributed context propagation
fgozdz Sep 10, 2024
379abf5
fix(worker, queue): remove unused import
fgozdz Sep 10, 2024
3e8e0d8
Merge branch 'master' into feat/opentelemetry
roggervalf Sep 10, 2024
6169ea3
feat(queue, worker): documentation and changes to propagation, basic …
fgozdz Sep 16, 2024
908ec13
feat(test_telemetry_interface): remove unused imports
fgozdz Sep 16, 2024
2dab35f
feat(queue, worker): correct tests and spankind, use property mapping…
fgozdz Sep 22, 2024
846e3ef
feat(worker): minor changes
fgozdz Sep 23, 2024
fc733e9
feat(queue, worker): distributed tracing
fgozdz Oct 9, 2024
4c199f8
feat(queue, worker): resolve conflicts
fgozdz Oct 9, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions src/classes/queue-base.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,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
Expand Down Expand Up @@ -191,6 +192,10 @@ export class QueueBase extends EventEmitter implements MinimalQueue {

const span = this.tracer.startSpan(getSpanName());
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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,
  ) {


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

try {
return callback(span);
} catch (err) {
Expand Down
95 changes: 26 additions & 69 deletions src/classes/queue.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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>>(
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

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

if (opts && opts.repeat) {
if (opts.repeat.endDate) {
if (+new Date(opts.repeat.endDate) < Date.now()) {
Expand Down Expand Up @@ -283,11 +279,10 @@ export class Queue<
async addBulk(
jobs: { name: NameType; data: DataType; opts?: BulkJobOptions }[],
): Promise<Job<DataType, ResultType, NameType>[]> {
return this.trace(
return await this.trace<Job<DataType, ResultType, NameType>[]>(
() => `${this.name} Queue.addBulk`,
async span => {
span?.setAttributes({
[TelemetryAttributes.QueueName]: this.name,
[TelemetryAttributes.BulkNames]: jobs.map(job => job.name),
[TelemetryAttributes.BulkCount]: jobs.length,
});
Expand Down Expand Up @@ -320,13 +315,9 @@ export class Queue<
* and in that case it will add it there instead of the wait list.
*/
async pause(): Promise<void> {
this.trace(
await this.trace<void>(
() => `${this.name} Queue.pause`,
async span => {
span?.setAttributes({
[TelemetryAttributes.QueueName]: this.name,
});

async () => {
await this.scripts.pause(true);

this.emit('paused');
Expand All @@ -339,13 +330,9 @@ export class Queue<
*
*/
async close(): Promise<void> {
this.trace(
await this.trace<void>(
() => `${this.name} Queue.close`,
async span => {
span?.setAttributes({
[TelemetryAttributes.QueueName]: this.name,
});

async () => {
if (!this.closing) {
if (this._repeat) {
await this._repeat.close();
Expand All @@ -363,13 +350,9 @@ export class Queue<
* queue.
*/
async resume(): Promise<void> {
this.trace(
await this.trace<void>(
() => `${this.name} Queue.resume`,
async span => {
span?.setAttributes({
[TelemetryAttributes.QueueName]: this.name,
});

async () => {
await this.scripts.pause(false);

this.emit('resumed');
Expand Down Expand Up @@ -427,13 +410,9 @@ export class Queue<
repeatOpts: RepeatOptions,
jobId?: string,
): Promise<boolean> {
return this.trace(
return await this.trace<boolean>(
() => `${this.name} ${name} Queue.removeRepeatable`,
async span => {
span?.setAttributes({
[TelemetryAttributes.QueueName]: this.name,
});

async () => {
const repeat = await this.repeat;
const removed = await repeat.removeRepeatable(name, repeatOpts, jobId);

Expand All @@ -448,13 +427,9 @@ export class Queue<
* @param id - identifier
*/
async removeDebounceKey(id: string): Promise<number> {
return this.trace(
return await this.trace<number>(
() => `${this.name} ${id} Queue.removeDebounceKey`,
async span => {
span?.setAttributes({
[TelemetryAttributes.QueueName]: this.name,
});

async () => {
const client = await this.client;

return await client.del(`${this.keys.de}:${id}`);
Expand All @@ -473,11 +448,10 @@ export class Queue<
* @returns
*/
async removeRepeatableByKey(key: string): Promise<boolean> {
return this.trace(
return await this.trace<boolean>(
() => `${this.name} ${key} Queue.removeRepeatableByKey`,
async span => {
span?.setAttributes({
[TelemetryAttributes.QueueName]: this.name,
[TelemetryAttributes.JobKey]: key,
});

Expand All @@ -499,11 +473,10 @@ export class Queue<
* any of its dependencies were locked.
*/
async remove(jobId: string, { removeChildren = true } = {}): Promise<number> {
return this.trace(
return await this.trace<number>(
() => `${this.name} ${jobId} Queue.remove`,
async span => {
span?.setAttributes({
[TelemetryAttributes.QueueName]: this.name,
[TelemetryAttributes.JobId]: jobId,
[TelemetryAttributes.JobOptions]: JSON.stringify({
removeChildren,
Expand All @@ -525,11 +498,10 @@ export class Queue<
jobId: string,
progress: number | object,
): Promise<void> {
this.trace(
await this.trace<void>(
() => `${this.name} Queue.updateJobProgress`,
async span => {
span?.setAttributes({
[TelemetryAttributes.QueueName]: this.name,
[TelemetryAttributes.JobId]: jobId,
[TelemetryAttributes.JobProgress]: JSON.stringify(progress),
});
Expand Down Expand Up @@ -564,11 +536,10 @@ export class Queue<
* delayed jobs.
*/
async drain(delayed = false): Promise<void> {
this.trace(
await this.trace<void>(
() => `${this.name} Queue.drain`,
async span => {
span?.setAttributes({
[TelemetryAttributes.QueueName]: this.name,
[TelemetryAttributes.QueueDrainDelay]: delayed,
});

Expand Down Expand Up @@ -599,26 +570,15 @@ export class Queue<
| 'delayed'
| 'failed' = 'completed',
): Promise<string[]> {
return this.trace(
return await this.trace<string[]>(
() => `${this.name} Queue.clean`,
async span => {
span?.setAttributes({
[TelemetryAttributes.QueueName]: this.name,
[TelemetryAttributes.QueueGrace]: grace,
[TelemetryAttributes.JobType]: type,
});

const maxCount = limit || Infinity;
const maxCountPerCall = Math.min(10000, maxCount);
const timestamp = Date.now() - grace;
let deletedCount = 0;
const deletedJobsIds: string[] = [];

span?.setAttributes({
[TelemetryAttributes.QueueCleanLimit]: maxCount,
[TelemetryAttributes.JobTimestamp]: timestamp,
});

while (deletedCount < maxCount) {
const jobsIds = await this.scripts.cleanJobsInSet(
type,
Expand All @@ -636,6 +596,10 @@ export class Queue<
}

span?.setAttributes({
[TelemetryAttributes.QueueGrace]: grace,
[TelemetryAttributes.JobType]: type,
[TelemetryAttributes.QueueCleanLimit]: maxCount,
[TelemetryAttributes.JobTimestamp]: timestamp,
[TelemetryAttributes.JobId]: deletedJobsIds,
});

Expand All @@ -656,13 +620,9 @@ export class Queue<
* @param opts - Obliterate options.
*/
async obliterate(opts?: ObliterateOpts): Promise<void> {
this.trace(
await this.trace<void>(
() => `${this.name} Queue.obliterate`,
async span => {
span?.setAttributes({
[TelemetryAttributes.QueueName]: this.name,
});

async () => {
await this.pause();

let cursor = 0;
Expand Down Expand Up @@ -690,11 +650,10 @@ export class Queue<
async retryJobs(
opts: { count?: number; state?: FinishedStatus; timestamp?: number } = {},
): Promise<void> {
this.trace(
await this.trace<void>(
() => `${this.name} Queue.retryJobs`,
async span => {
span?.setAttributes({
[TelemetryAttributes.QueueName]: this.name,
[TelemetryAttributes.QueueOptions]: JSON.stringify(opts),
});

Expand All @@ -719,11 +678,10 @@ export class Queue<
* @returns
*/
async promoteJobs(opts: { count?: number } = {}): Promise<void> {
this.trace(
await this.trace<void>(
() => `${this.name} Queue.promoteJobs`,
async span => {
span?.setAttributes({
[TelemetryAttributes.QueueName]: this.name,
[TelemetryAttributes.QueueOptions]: JSON.stringify(opts),
});

Expand All @@ -741,11 +699,10 @@ export class Queue<
* @param maxLength -
*/
async trimEvents(maxLength: number): Promise<number> {
return this.trace(
return await this.trace<number>(
() => `${this.name} Queue.trimEvents`,
async span => {
span?.setAttributes({
[TelemetryAttributes.QueueName]: this.name,
[TelemetryAttributes.QueueEventMaxLength]: maxLength,
});

Expand Down
Loading