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

feat(queue): observability for queue #2721

Merged
merged 33 commits into from
Oct 31, 2024
Merged

feat(queue): observability for queue #2721

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
33 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
7261e40
feat(worker): job error handling, do not trace certain methods until …
fgozdz Oct 10, 2024
677fcc6
feat(queue-base): add propagation for internal spanKind and setSpan f…
fgozdz Oct 11, 2024
51a353e
refactor(telemetry): several improvements and small changes
manast Oct 28, 2024
6742a03
fix(telemetry): fix return type of setSpanOnContext
manast Oct 29, 2024
167a7dd
fix(job): return result of trace
manast Oct 29, 2024
210d36e
test(telemetry): fix broken tests
manast Oct 29, 2024
0f172ff
Merge branch 'master' of https://github.com/taskforcesh/bullmq into f…
manast Oct 29, 2024
23924d3
chore(job): rename srcPropagationMetadata to dstPropagationMetadata
manast Oct 29, 2024
49b2c36
docs(telemetry): initial interface documentation
manast Oct 30, 2024
9b0b3fa
chore: small improvements to naming of telemetry spans
manast Oct 31, 2024
e35233b
Merge branch 'master' of https://github.com/taskforcesh/bullmq into f…
manast Oct 31, 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