feat(core): Export the stored schedule of durable system tasks as metrics (#40009)

This commit is contained in:
Lorent Lempereur 2026-10-01 11:46:12 +00:00 committed by GitHub
parent a4b87fccbe
commit 3faf54fea3
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 449 additions and 10 deletions

View File

@ -132,6 +132,35 @@ describe('ScheduledJobRepository', () => {
});
});
describe('findScheduleStatesByOwnerType', () => {
it('reads the owner id and schedule state of every job owners of one kind hold', async () => {
entityManager.find.mockResolvedValueOnce([
{ ownerId: 'a', enabled: true, nextRunAt: CLOCK, orphanedAt: null } as ScheduledJob,
]);
const result = await repository.findScheduleStatesByOwnerType('system-task');
expect(entityManager.find).toHaveBeenCalledWith(ScheduledJob, {
where: { ownerType: 'system-task' },
select: ['ownerId', 'enabled', 'nextRunAt', 'orphanedAt'],
});
expect(result).toEqual([{ ownerId: 'a', runnable: true, nextRunAt: CLOCK }]);
});
it.each([
['disabled', { enabled: false, orphanedAt: null }],
['quarantined', { enabled: true, orphanedAt: CLOCK }],
])('marks a %s job not runnable', async (_, state) => {
entityManager.find.mockResolvedValueOnce([
{ ownerId: 'a', nextRunAt: CLOCK, ...state } as ScheduledJob,
]);
const [result] = await repository.findScheduleStatesByOwnerType('system-task');
expect(result.runnable).toBe(false);
});
});
describe('deleteIfPayloadUnchanged', () => {
const OBSERVED = { n8nVersion: '1.0.0' };

View File

@ -156,6 +156,24 @@ export class ScheduledJobRepository extends Repository<ScheduledJob> {
return await this.find({ where: { ownerType }, select: ['id', 'ownerId', 'payload'] });
}
/**
* The owner id and schedule state of every job owners of one kind hold.
* A job is runnable when the scheduler will claim it: enabled and not quarantined.
*/
async findScheduleStatesByOwnerType(
ownerType: string,
): Promise<Array<Pick<ScheduledJob, 'ownerId' | 'nextRunAt'> & { runnable: boolean }>> {
const rows = await this.find({
where: { ownerType },
select: ['ownerId', 'enabled', 'nextRunAt', 'orphanedAt'],
});
return rows.map(({ ownerId, enabled, nextRunAt, orphanedAt }) => ({
ownerId,
nextRunAt,
runnable: enabled && orphanedAt === null,
}));
}
/**
* The member ids under which an owner holds jobs of one task type, each once.
* A caller uses them to tell which of its parts still provision a job.

View File

@ -4,7 +4,7 @@ import '@/zod-alias-support';
import { Logger } from '@n8n/backend-common';
import { mockInstance, mockLogger } from '@n8n/backend-test-utils';
import { GlobalConfig } from '@n8n/config';
import { DbConnection, SettingsRepository } from '@n8n/db';
import { DbConnection, ScheduledJobRepository, SettingsRepository } from '@n8n/db';
import { SystemTaskMetadata } from '@n8n/decorators';
import { Container } from '@n8n/di';
import type { NextFunction, Request, Response } from 'express';
@ -91,6 +91,7 @@ describe('Start system task metrics', () => {
mockInstance(SystemTaskJobRegistrar);
mockInstance(SystemTaskScheduledJobOwner);
mockInstance(SettingsRepository, { findBy: async () => [] });
mockInstance(ScheduledJobRepository);
activeWorkflowManager = mockInstance(ActiveWorkflowManager);
// Keep the real system task collector. Other collectors are outside this test.

View File

@ -24,7 +24,14 @@ export type SystemTaskRunResult = 'success' | 'failure' | 'aborted';
* after the run signal had already aborted during stepdown, or the process
* slept through it and the timer coalesced it into one late fire.
*/
export type SystemTaskSkipReason = 'overlap' | 'provisioned_elsewhere' | 'aborted' | 'coalesced';
export const SYSTEM_TASK_SKIP_REASONS = [
'overlap',
'provisioned_elsewhere',
'aborted',
'coalesced',
] as const;
export type SystemTaskSkipReason = (typeof SYSTEM_TASK_SKIP_REASONS)[number];
export type SystemTaskMetricsEventMap = {
'system-task-routed': {

View File

@ -1,6 +1,7 @@
import type { EventService } from '@n8n/backend-services';
import type { CacheService, EventService } from '@n8n/backend-services';
import { mockInstance } from '@n8n/backend-test-utils';
import { PrometheusMetricsConfig } from '@n8n/config';
import type { ScheduledJob, ScheduledJobRepository } from '@n8n/db';
import promClient from 'prom-client';
import type { Mock } from 'vitest';
import { mock } from 'vitest-mock-extended';
@ -18,6 +19,8 @@ describe('PrometheusSystemTaskMetricsService', () => {
includeSystemTaskMetrics: true,
});
const eventService = mock<EventService>();
const cacheService = mock<CacheService>();
const scheduledJobRepository = mock<ScheduledJobRepository>();
let service: PrometheusSystemTaskMetricsService;
const ctorByType = { Counter: vi.fn(), Gauge: vi.fn(), Histogram: vi.fn() };
@ -36,8 +39,11 @@ describe('PrometheusSystemTaskMetricsService', () => {
this.valuesByLabels.set(JSON.stringify(labels), value),
);
observe = vi.fn();
zero = vi.fn((labels: object) => this.valuesByLabels.set(JSON.stringify(labels), 0));
remove = vi.fn((labels: object) => this.valuesByLabels.delete(JSON.stringify(labels)));
collect?: () => Promise<void>;
value(labels: object) {
return this.valuesByLabels.get(JSON.stringify(labels));
}
@ -50,8 +56,9 @@ describe('PrometheusSystemTaskMetricsService', () => {
function fake(type: keyof typeof ctorByType) {
return class extends FakeMetric {
constructor(opts: { name: string }) {
constructor(opts: { name: string; collect?: () => Promise<void> }) {
super();
this.collect = opts.collect;
ctorByType[type](opts);
instancesByName.set(opts.name, this);
}
@ -72,7 +79,13 @@ describe('PrometheusSystemTaskMetricsService', () => {
Histogram: fake('Histogram'),
});
service = new PrometheusSystemTaskMetricsService(config, eventService);
cacheService.get.mockResolvedValue(undefined);
service = new PrometheusSystemTaskMetricsService(
config,
eventService,
cacheService,
scheduledJobRepository,
);
});
afterEach(() => {
@ -126,6 +139,34 @@ describe('PrometheusSystemTaskMetricsService', () => {
]);
});
it('keeps the label names of every series', () => {
service.init();
const labelsByName = Object.fromEntries(
[
...ctorByType.Histogram.mock.calls,
...ctorByType.Counter.mock.calls,
...ctorByType.Gauge.mock.calls,
].map(([opts]) => {
const { name, labelNames } = opts as { name: string; labelNames: string[] };
return [name, labelNames];
}),
);
expect(labelsByName).toEqual({
n8n_system_task_run_duration_seconds: ['task', 'mode', 'result'],
n8n_system_task_fire_lag_seconds: ['task'],
n8n_system_task_runs_skipped_total: ['task', 'reason'],
n8n_system_task_provision_check_failures_total: ['task'],
n8n_system_task_retries_total: ['task'],
n8n_system_task_last_success_timestamp_seconds: ['task', 'mode'],
n8n_system_task_runs_in_flight: ['task', 'mode'],
n8n_system_task_info: ['task', 'mode'],
n8n_system_task_interval_seconds: ['task'],
n8n_system_task_next_run_timestamp_seconds: ['task'],
n8n_system_task_scheduled: ['task', 'mode'],
});
});
it('gives the fire lag histogram buckets that reach a day', () => {
service.init();
@ -287,6 +328,166 @@ describe('PrometheusSystemTaskMetricsService', () => {
});
});
describe('zeroed counts', () => {
it('starts the success and failure series of a routed task at zero', () => {
service.init();
handler('system-task-routed')({ name: 'prune', mode: 'durable' });
const runDuration = metric('system_task_run_duration_seconds');
expect(runDuration.value({ task: 'prune', mode: 'durable', result: 'success' })).toBe(0);
expect(runDuration.value({ task: 'prune', mode: 'durable', result: 'failure' })).toBe(0);
});
it('starts the retry and skip counters of a timer task at zero', () => {
service.init();
handler('system-task-routed')({ name: 'sweep', mode: 'instance_timer' });
expect(metric('system_task_retries_total').value({ task: 'sweep' })).toBe(0);
for (const reason of ['overlap', 'provisioned_elsewhere', 'aborted', 'coalesced']) {
expect(metric('system_task_runs_skipped_total').value({ task: 'sweep', reason })).toBe(0);
}
});
it('does not start the retry and skip counters of a durable task', () => {
service.init();
handler('system-task-routed')({ name: 'prune', mode: 'durable' });
expect(metric('system_task_retries_total').inc).not.toHaveBeenCalled();
expect(metric('system_task_runs_skipped_total').inc).not.toHaveBeenCalled();
});
it('zeroes the run series of a leader timer task only on the first takeover', () => {
service.init();
handler('system-task-routed')({ name: 'prune', mode: 'leader_timer' });
handler('system-task-timers-started')({});
handler('system-task-timers-stopped')({});
handler('system-task-timers-started')({});
expect(metric('system_task_run_duration_seconds').zero).toHaveBeenCalledTimes(2);
});
});
describe('durable job state', () => {
const job = (
overrides: Partial<Pick<ScheduledJob, 'ownerId' | 'nextRunAt'> & { runnable: boolean }>,
) => ({
ownerId: 'prune',
runnable: true,
nextRunAt: new Date(NOW.getTime() + 60_000),
...overrides,
});
async function scrape() {
await metric('system_task_scheduled').collect!();
await metric('system_task_next_run_timestamp_seconds').collect!();
}
it('reads nothing while no task runs durably', async () => {
service.init();
handler('system-task-routed')({ name: 'prune', mode: 'leader_timer' });
await scrape();
expect(scheduledJobRepository.findScheduleStatesByOwnerType).not.toHaveBeenCalled();
});
it('exports the stored next run of a durable task and marks it scheduled', async () => {
scheduledJobRepository.findScheduleStatesByOwnerType.mockResolvedValue([job({})]);
service.init();
handler('system-task-routed')({ name: 'prune', mode: 'durable' });
await scrape();
expect(scheduledJobRepository.findScheduleStatesByOwnerType).toHaveBeenCalledWith(
'system-task',
);
expect(metric('system_task_scheduled').value({ task: 'prune', mode: 'durable' })).toBe(1);
expect(metric('system_task_next_run_timestamp_seconds').value({ task: 'prune' })).toBe(
NOW.getTime() / 1000 + 60,
);
});
it.each([
['is missing', []],
['is not runnable', [job({ runnable: false })]],
])('marks a durable task unscheduled when its job %s', async (_, jobs) => {
scheduledJobRepository.findScheduleStatesByOwnerType.mockResolvedValue(jobs);
service.init();
handler('system-task-routed')({ name: 'prune', mode: 'durable' });
await scrape();
expect(metric('system_task_scheduled').value({ task: 'prune', mode: 'durable' })).toBe(0);
expect(metric('system_task_next_run_timestamp_seconds').remove).toHaveBeenCalledWith({
task: 'prune',
});
});
it('ignores the jobs of tasks this instance does not run durably', async () => {
scheduledJobRepository.findScheduleStatesByOwnerType.mockResolvedValue([
job({}),
job({ ownerId: 'other' }),
]);
service.init();
handler('system-task-routed')({ name: 'prune', mode: 'durable' });
await scrape();
expect(metric('system_task_scheduled').value({ task: 'other', mode: 'durable' })).toBe(
undefined,
);
});
it('reads the jobs once for both gauges and caches them for the metrics interval', async () => {
const cache = new Map<string, unknown>();
cacheService.get.mockImplementation(async (key: string) => cache.get(key));
cacheService.set.mockImplementation(async (key: string, value: unknown) => {
cache.set(key, value);
});
scheduledJobRepository.findScheduleStatesByOwnerType.mockResolvedValue([job({})]);
config.schedulerMetricsInterval = 20;
service.init();
handler('system-task-routed')({ name: 'prune', mode: 'durable' });
await scrape();
expect(scheduledJobRepository.findScheduleStatesByOwnerType).toHaveBeenCalledTimes(1);
expect(cacheService.set).toHaveBeenCalledExactlyOnceWith(
'metrics:system-tasks:durable-jobs:v1',
[{ task: 'prune', runnable: true, nextRunAtSeconds: NOW.getTime() / 1000 + 60 }],
20_000,
);
});
it('serves both gauges from the cache without reading the jobs', async () => {
cacheService.get.mockResolvedValue([
{ task: 'prune', runnable: true, nextRunAtSeconds: 1234 },
]);
service.init();
handler('system-task-routed')({ name: 'prune', mode: 'durable' });
await scrape();
expect(scheduledJobRepository.findScheduleStatesByOwnerType).not.toHaveBeenCalled();
expect(metric('system_task_scheduled').value({ task: 'prune', mode: 'durable' })).toBe(1);
expect(metric('system_task_next_run_timestamp_seconds').value({ task: 'prune' })).toBe(1234);
});
it('keeps the last values when the job read fails', async () => {
scheduledJobRepository.findScheduleStatesByOwnerType.mockRejectedValue(new Error('db down'));
service.init();
handler('system-task-routed')({ name: 'prune', mode: 'durable' });
await expect(scrape()).resolves.toBeUndefined();
expect(metric('system_task_scheduled').value({ task: 'prune', mode: 'durable' })).toBe(1);
});
});
describe('system-task-run-started and -settled', () => {
it('raises the in-flight gauge on start and lowers it on settle', () => {
service.init();

View File

@ -1,14 +1,29 @@
import { EventService } from '@n8n/backend-services';
import { CacheService, EventService } from '@n8n/backend-services';
import { PrometheusMetricsConfig } from '@n8n/config';
import { Time } from '@n8n/constants';
import { ScheduledJobOwnerType, Time } from '@n8n/constants';
import { ScheduledJobRepository } from '@n8n/db';
import { Service } from '@n8n/di';
import promClient from 'prom-client';
import type { SystemTaskMode } from '@/events/maps/system-task-metrics.event-map';
import {
SYSTEM_TASK_SKIP_REASONS,
type SystemTaskMode,
} from '@/events/maps/system-task-metrics.event-map';
import type { PrometheusMetricsCollector } from './base';
import { CachedMetricQuery } from './cached-metric-query';
import { DURATION_BUCKETS_SECONDS, LAG_BUCKETS_SECONDS } from './constant';
const DURABLE_JOBS_CACHE_KEY = 'metrics:system-tasks:durable-jobs:v1';
/** The stored schedule of one durable system task, as JSON so the cache can hold it. */
type DurableJobState = {
task: string;
/** Enabled and not quarantined, so the scheduler will claim it. */
runnable: boolean;
nextRunAtSeconds: number | null;
};
/**
* Collects Prometheus metrics for system tasks, on all of their paths: the
* in-memory timers of the leader, the durable scheduler and the per-instance
@ -22,6 +37,13 @@ import { DURATION_BUCKETS_SECONDS, LAG_BUCKETS_SECONDS } from './constant';
* instance leads: seeded when the timers start, removed when they stop, so a
* former leader does not export frozen series for runs it no longer makes.
*
* The run outcome and skip series start at zero when a task is seeded, so
* `increase()` counts the first failure after a restart.
*
* The scheduled and next-run series of a durable task come from its stored job,
* read on scrape: the job is shared by every main, so another main can remove it
* without this instance seeing an event.
*
* Labels are bounded (task name, mode, result, reason): no instance label,
* Prometheus adds one per scrape target.
*/
@ -30,6 +52,8 @@ export class PrometheusSystemTaskMetricsService implements PrometheusMetricsColl
constructor(
private readonly config: PrometheusMetricsConfig,
private readonly eventService: EventService,
private readonly cacheService: CacheService,
private readonly scheduledJobRepository: ScheduledJobRepository,
) {}
get enabled(): boolean {
@ -76,16 +100,31 @@ export class PrometheusSystemTaskMetricsService implements PrometheusMetricsColl
labelNames: ['task'],
});
const durableTasks = new Set<string>();
const durableJobs = new CachedMetricQuery<DurableJobState[]>({
cacheService: this.cacheService,
cacheKey: DURABLE_JOBS_CACHE_KEY,
ttlMs: this.config.schedulerMetricsInterval * Time.seconds.toMilliseconds,
query: async () => await this.findDurableJobStates(),
});
const collectDurableJobs = async () => {
if (durableTasks.size > 0) {
await this.applyDurableJobStates(durableJobs, durableTasks, scheduled, nextRun);
}
};
const nextRun = new promClient.Gauge({
name: `${prefix}system_task_next_run_timestamp_seconds`,
help: 'Unix timestamp in seconds of the next occurrence a timer-driven system task is armed for on this instance, by task.',
help: 'Unix timestamp in seconds of the next occurrence of a system task, by task: the occurrence a timer is armed for on this instance, or the first occurrence not yet planned for a durable task, identical on every main. Planned occurrences of a durable task can fire before this timestamp.',
labelNames: ['task'],
collect: collectDurableJobs,
});
const scheduled = new promClient.Gauge({
name: `${prefix}system_task_scheduled`,
help: '1 while a system task is scheduled to run on this instance, 0 once it stopped being scheduled, by task and mode: the schedule of a leader_timer or instance_timer task could not be planned, or the job of a durable task could not be provisioned.',
help: '1 while a system task is scheduled to run, 0 once it stopped being scheduled, by task and mode: the schedule of a leader_timer or instance_timer task could not be planned on this instance, or no runnable job is stored for a durable task.',
labelNames: ['task', 'mode'],
collect: collectDurableJobs,
});
const provisionCheckFailures = new promClient.Counter({
@ -108,8 +147,23 @@ export class PrometheusSystemTaskMetricsService implements PrometheusMetricsColl
});
const leaderTimerTasks = new Set<string>();
const zeroedRunOutcomes = new Set<string>();
let timersRunning = false;
const zeroCounts = (task: string, mode: SystemTaskMode) => {
// `zero()` resets a histogram, so a later takeover must not zero it again.
const key = `${task}:${mode}`;
if (!zeroedRunOutcomes.has(key)) {
zeroedRunOutcomes.add(key);
runDuration.zero({ task, mode, result: 'success' });
runDuration.zero({ task, mode, result: 'failure' });
}
if (mode !== 'durable') {
retries.inc({ task }, 0);
SYSTEM_TASK_SKIP_REASONS.forEach((reason) => runsSkipped.inc({ task, reason }, 0));
}
};
const seed = (task: string, mode: SystemTaskMode) => {
info.set({ task, mode }, 1);
scheduled.set({ task, mode }, 1);
@ -117,6 +171,7 @@ export class PrometheusSystemTaskMetricsService implements PrometheusMetricsColl
// a takeover has its own decrement to come, and an absolute 0 here would
// turn that decrement into a permanent -1.
runsInFlight.inc({ task, mode }, 0);
zeroCounts(task, mode);
};
const leaderTimerGauges = [info, scheduled, runsInFlight, lastSuccess];
@ -132,6 +187,9 @@ export class PrometheusSystemTaskMetricsService implements PrometheusMetricsColl
if (mode === 'leader_timer') {
leaderTimerTasks.add(name);
}
if (mode === 'durable') {
durableTasks.add(name);
}
if (mode !== 'leader_timer' || timersRunning) {
seed(name, mode);
}
@ -185,4 +243,45 @@ export class PrometheusSystemTaskMetricsService implements PrometheusMetricsColl
fireLag.observe({ task: name }, lagMs / Time.seconds.toMilliseconds);
});
}
private async findDurableJobStates(): Promise<DurableJobState[]> {
const jobs = await this.scheduledJobRepository.findScheduleStatesByOwnerType(
ScheduledJobOwnerType.SystemTask,
);
return jobs.map(({ ownerId, runnable, nextRunAt }) => ({
task: ownerId,
runnable,
nextRunAtSeconds:
nextRunAt === null ? null : nextRunAt.getTime() / Time.seconds.toMilliseconds,
}));
}
/**
* Set the scheduled and next-run series of every durable task from its stored
* job. A failed read keeps the last values, so one database error does not
* fail the whole scrape.
*/
private async applyDurableJobStates(
durableJobs: CachedMetricQuery<DurableJobState[]>,
durableTasks: Set<string>,
scheduled: promClient.Gauge<'task' | 'mode'>,
nextRun: promClient.Gauge<'task'>,
): Promise<void> {
let states: DurableJobState[];
try {
states = await durableJobs.get();
} catch {
return;
}
const stateByTask = new Map(states.map((state) => [state.task, state]));
for (const task of durableTasks) {
const state = stateByTask.get(task);
scheduled.set({ task, mode: 'durable' }, state?.runnable ? 1 : 0);
if (state?.runnable && state.nextRunAtSeconds !== null) {
nextRun.set({ task }, state.nextRunAtSeconds);
} else {
nextRun.remove({ task });
}
}
}
}

View File

@ -0,0 +1,84 @@
import { testDb } from '@n8n/backend-test-utils';
import { CacheService, EventService } from '@n8n/backend-services';
import { GlobalConfig } from '@n8n/config';
import { ScheduledJobRepository } from '@n8n/db';
import { Container } from '@n8n/di';
import promClient from 'prom-client';
import { PrometheusSystemTaskMetricsService } from '@/metrics/prometheus/system-task-metrics.service';
import { emitSystemTaskMetric } from '@/scheduling/system-tasks/emit-system-task-metric';
import { createDueJobFactory } from './shared/job-factory';
/**
* The scheduled and next-run series of a durable system task, read from its
* stored job on scrape, against the real database.
*/
describe('system task durable job metrics', () => {
let jobRepo: ScheduledJobRepository;
let cacheService: CacheService;
let createJob: ReturnType<typeof createDueJobFactory>;
beforeAll(async () => {
await testDb.init();
jobRepo = Container.get(ScheduledJobRepository);
cacheService = Container.get(CacheService);
await cacheService.init();
createJob = createDueJobFactory(jobRepo, 'system:prune', 'metrics');
const { metrics } = Container.get(GlobalConfig).endpoints;
metrics.prefix = 'n8n_';
metrics.includeSystemTaskMetrics = true;
});
beforeEach(async () => {
await testDb.truncate(['ScheduledTask', 'ScheduledJob']);
await cacheService.reset();
promClient.register.clear();
Container.get(EventService).removeAllListeners();
Container.get(PrometheusSystemTaskMetricsService).init();
emitSystemTaskMetric(Container.get(EventService), 'system-task-routed', {
name: 'prune',
mode: 'durable',
});
emitSystemTaskMetric(Container.get(EventService), 'system-task-next-run-planned', {
name: 'prune',
nextRunAtMs: Date.now(),
});
});
afterAll(async () => {
promClient.register.clear();
await testDb.terminate();
});
async function valueOf(metricName: string, labels: Record<string, string>) {
const { values } = await promClient.register.getSingleMetric(metricName)!.get();
return values.find(({ labels: actual }) =>
Object.entries(labels).every(([key, value]) => actual[key] === value),
)?.value;
}
it('exports the stored next run of a runnable job', async () => {
const nextRunAt = new Date('2030-01-01T00:00:00.000Z');
await createJob({ ownerId: 'prune', nextRunAt });
expect(await valueOf('n8n_system_task_scheduled', { task: 'prune', mode: 'durable' })).toBe(1);
expect(await valueOf('n8n_system_task_next_run_timestamp_seconds', { task: 'prune' })).toBe(
nextRunAt.getTime() / 1000,
);
});
it('marks the task unscheduled once its job is quarantined', async () => {
await createJob({ ownerId: 'prune', orphanedAt: new Date() });
expect(await valueOf('n8n_system_task_scheduled', { task: 'prune', mode: 'durable' })).toBe(0);
expect(
await valueOf('n8n_system_task_next_run_timestamp_seconds', { task: 'prune' }),
).toBeUndefined();
});
it('marks the task unscheduled when no job is stored', async () => {
expect(await valueOf('n8n_system_task_scheduled', { task: 'prune', mode: 'durable' })).toBe(0);
});
});