fix: add queue attribute to jobs waiting gauge metric (#21324)

## Summary
- The `twenty_queue_jobs_waiting_total` gauge was summing all queues
into a single value without a `queue` label, making the Grafana "Jobs
Waiting by Queue" panel show a single aggregated line instead of
per-queue breakdown.
- Uses `getMeter()` directly to call `observableResult.observe(count, {
queue: queueName })` per queue, matching the `by (queue)` grouping the
dashboard already expects.

## Test plan
- [x] Deploy and verify the Grafana "Jobs Waiting by Queue" panel
displays separate series per queue
- [x] Confirm Prometheus scrape returns
`twenty_queue_jobs_waiting_total{queue="..."}` with distinct queue
labels
This commit is contained in:
Thomas Trompette
2026-06-08 17:46:04 +02:00
committed by GitHub
parent 77d1e8ced6
commit 16db47fb70
2 changed files with 62 additions and 8 deletions
@@ -54,17 +54,23 @@ export class BullMQDriver
) {}
onModuleInit() {
this.metricsService.createObservableGauge({
this.metricsService.createMultiObservableGauge({
metricName: 'twenty_queue_jobs_waiting_total',
options: { description: 'Current number of jobs waiting in queue' },
callback: async () => {
let totalWaiting = 0;
const observations: Array<{
value: number;
attributes: { queue: string };
}> = [];
for (const [queueName, queue] of Object.entries(this.queueMap)) {
try {
const waitingCount = await queue.count();
totalWaiting += waitingCount;
observations.push({
value: waitingCount,
attributes: { queue: queueName },
});
} catch (error) {
this.logger.error(
`Failed to collect waiting jobs metrics for queue ${queueName}`,
@@ -73,7 +79,7 @@ export class BullMQDriver
}
}
return totalWaiting;
return observations;
},
});
}
@@ -6,6 +6,7 @@ import {
type Meter,
type MetricOptions,
type ObservableGauge,
type ObservableResult,
} from '@opentelemetry/api';
import { isDefined } from 'twenty-shared/utils';
@@ -42,16 +43,63 @@ export class MetricsService {
options: MetricOptions;
callback: () => number | Promise<number>;
cacheValue?: boolean;
}): ObservableGauge {
return this.createObservableGaugeInternal({
metricName,
options,
callback,
cacheValue,
observeResult: (observableResult, result) => {
observableResult.observe(result);
},
});
}
createMultiObservableGauge({
metricName,
options,
callback,
cacheValue = false,
}: {
metricName: string;
options: MetricOptions;
callback: () => Promise<Array<{ value: number; attributes: Attributes }>>;
cacheValue?: boolean;
}): ObservableGauge {
return this.createObservableGaugeInternal({
metricName,
options,
callback,
cacheValue,
observeResult: (observableResult, observations) => {
for (const observation of observations) {
observableResult.observe(observation.value, observation.attributes);
}
},
});
}
private createObservableGaugeInternal<T>({
metricName,
options,
callback,
cacheValue,
observeResult,
}: {
metricName: string;
options: MetricOptions;
callback: () => T | Promise<T>;
cacheValue: boolean;
observeResult: (observableResult: ObservableResult, result: T) => void;
}): ObservableGauge {
const gauge = this.getMeter().createObservableGauge(metricName, options);
gauge.addCallback(async (observableResult) => {
if (cacheValue) {
const cachedResult =
await this.healthCacheStorage.get<number>(metricName);
const cachedResult = await this.healthCacheStorage.get<T>(metricName);
if (isDefined(cachedResult)) {
observableResult.observe(cachedResult);
observeResult(observableResult, cachedResult);
return;
}
@@ -60,7 +108,7 @@ export class MetricsService {
try {
const result = await callback();
observableResult.observe(result);
observeResult(observableResult, result);
if (cacheValue) {
await this.healthCacheStorage.set(