From 3faf54fea360d8bebb2c598d6a730c9e034dee90 Mon Sep 17 00:00:00 2001 From: Lorent Lempereur Date: Thu, 1 Oct 2026 11:46:12 +0000 Subject: [PATCH] feat(core): Export the stored schedule of durable system tasks as metrics (#40009) --- .../scheduled-job.repository.test.ts | 29 +++ .../repositories/scheduled-job.repository.ts | 18 ++ .../start-system-task-metrics.test.ts | 3 +- .../maps/system-task-metrics.event-map.ts | 9 +- .../system-task-metrics.service.test.ts | 207 +++++++++++++++++- .../prometheus/system-task-metrics.service.ts | 109 ++++++++- .../system-task-durable-metrics.test.ts | 84 +++++++ 7 files changed, 449 insertions(+), 10 deletions(-) create mode 100644 packages/cli/test/integration/scheduling/system-task-durable-metrics.test.ts diff --git a/packages/@n8n/db/src/repositories/__tests__/scheduled-job.repository.test.ts b/packages/@n8n/db/src/repositories/__tests__/scheduled-job.repository.test.ts index d62ae848599..decd72ccefb 100644 --- a/packages/@n8n/db/src/repositories/__tests__/scheduled-job.repository.test.ts +++ b/packages/@n8n/db/src/repositories/__tests__/scheduled-job.repository.test.ts @@ -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' }; diff --git a/packages/@n8n/db/src/repositories/scheduled-job.repository.ts b/packages/@n8n/db/src/repositories/scheduled-job.repository.ts index 795bab6c618..b05a842e7b4 100644 --- a/packages/@n8n/db/src/repositories/scheduled-job.repository.ts +++ b/packages/@n8n/db/src/repositories/scheduled-job.repository.ts @@ -156,6 +156,24 @@ export class ScheduledJobRepository extends Repository { 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 & { 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. diff --git a/packages/cli/src/commands/__tests__/start-system-task-metrics.test.ts b/packages/cli/src/commands/__tests__/start-system-task-metrics.test.ts index 554814985c8..9383be4b38f 100644 --- a/packages/cli/src/commands/__tests__/start-system-task-metrics.test.ts +++ b/packages/cli/src/commands/__tests__/start-system-task-metrics.test.ts @@ -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. diff --git a/packages/cli/src/events/maps/system-task-metrics.event-map.ts b/packages/cli/src/events/maps/system-task-metrics.event-map.ts index 24391e56048..56f2d23b36a 100644 --- a/packages/cli/src/events/maps/system-task-metrics.event-map.ts +++ b/packages/cli/src/events/maps/system-task-metrics.event-map.ts @@ -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': { diff --git a/packages/cli/src/metrics/prometheus/__tests__/system-task-metrics.service.test.ts b/packages/cli/src/metrics/prometheus/__tests__/system-task-metrics.service.test.ts index a1346a7d6db..1c97b52f9c0 100644 --- a/packages/cli/src/metrics/prometheus/__tests__/system-task-metrics.service.test.ts +++ b/packages/cli/src/metrics/prometheus/__tests__/system-task-metrics.service.test.ts @@ -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(); + const cacheService = mock(); + const scheduledJobRepository = mock(); 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; + 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 }) { 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 & { 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(); + 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(); diff --git a/packages/cli/src/metrics/prometheus/system-task-metrics.service.ts b/packages/cli/src/metrics/prometheus/system-task-metrics.service.ts index 165e1b51293..710efb65485 100644 --- a/packages/cli/src/metrics/prometheus/system-task-metrics.service.ts +++ b/packages/cli/src/metrics/prometheus/system-task-metrics.service.ts @@ -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(); + const durableJobs = new CachedMetricQuery({ + 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(); + const zeroedRunOutcomes = new Set(); 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 { + 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, + durableTasks: Set, + scheduled: promClient.Gauge<'task' | 'mode'>, + nextRun: promClient.Gauge<'task'>, + ): Promise { + 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 }); + } + } + } } diff --git a/packages/cli/test/integration/scheduling/system-task-durable-metrics.test.ts b/packages/cli/test/integration/scheduling/system-task-durable-metrics.test.ts new file mode 100644 index 00000000000..aec34b0c3b4 --- /dev/null +++ b/packages/cli/test/integration/scheduling/system-task-durable-metrics.test.ts @@ -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; + + 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) { + 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); + }); +});