diff --git a/packages/@n8n/config/src/configs/endpoints.config.ts b/packages/@n8n/config/src/configs/endpoints.config.ts index 7b713ceb92b..288f97c60ea 100644 --- a/packages/@n8n/config/src/configs/endpoints.config.ts +++ b/packages/@n8n/config/src/configs/endpoints.config.ts @@ -79,6 +79,10 @@ class PrometheusMetricsConfig { /** How often (in seconds) to update workflow statistics metrics. */ @Env('N8N_METRICS_WORKFLOW_STATISTICS_INTERVAL') workflowStatisticsInterval: number = 300; + + /** Whether to include metrics for execution data reads and writes. */ + @Env('N8N_METRICS_INCLUDE_EXECUTION_DATA_METRICS') + includeExecutionDataMetrics: boolean = false; } @Config diff --git a/packages/@n8n/config/test/config.test.ts b/packages/@n8n/config/test/config.test.ts index 934c0e78da4..13245fb08d0 100644 --- a/packages/@n8n/config/test/config.test.ts +++ b/packages/@n8n/config/test/config.test.ts @@ -235,6 +235,7 @@ describe('GlobalConfig', () => { activeWorkflowCountInterval: 60, includeWorkflowStatistics: false, workflowStatisticsInterval: 300, + includeExecutionDataMetrics: false, }, additionalNonUIRoutes: '', disableProductionWebhooksOnMainProcess: false, diff --git a/packages/cli/src/events/__tests__/telemetry-event-relay.test.ts b/packages/cli/src/events/__tests__/telemetry-event-relay.test.ts index 3c2b66efa97..4a6ab29d8db 100644 --- a/packages/cli/src/events/__tests__/telemetry-event-relay.test.ts +++ b/packages/cli/src/events/__tests__/telemetry-event-relay.test.ts @@ -62,6 +62,7 @@ describe('TelemetryEventRelay', () => { includeCacheMetrics: false, includeMessageEventBusMetrics: false, includeQueueMetrics: false, + includeExecutionDataMetrics: false, }, }, logging: { @@ -2105,6 +2106,7 @@ describe('TelemetryEventRelay', () => { metrics_category_logs: false, metrics_category_queue: false, metrics_category_routes: false, + metrics_category_execution_data: false, metrics_enabled: true, }, n8n_binary_data_mode: 'default', @@ -2142,6 +2144,7 @@ describe('TelemetryEventRelay', () => { metrics_category_cache: false, metrics_category_logs: false, metrics_category_queue: false, + metrics_category_execution_data: false, }, otel: { enabled: false, diff --git a/packages/cli/src/events/event.service.ts b/packages/cli/src/events/event.service.ts index 29a198a0d6b..d7d99f680bd 100644 --- a/packages/cli/src/events/event.service.ts +++ b/packages/cli/src/events/event.service.ts @@ -3,10 +3,11 @@ import { Service } from '@n8n/di'; import { TypedEmitter } from '@/typed-emitter'; import type { AiEventMap } from './maps/ai.event-map'; +import type { ExecutionDataEventMap } from './maps/execution-data.event-map'; import type { QueueMetricsEventMap } from './maps/queue-metrics.event-map'; import type { RelayEventMap } from './maps/relay.event-map'; -type EventMap = RelayEventMap & QueueMetricsEventMap & AiEventMap; +type EventMap = RelayEventMap & QueueMetricsEventMap & AiEventMap & ExecutionDataEventMap; @Service() export class EventService extends TypedEmitter {} diff --git a/packages/cli/src/events/maps/execution-data.event-map.ts b/packages/cli/src/events/maps/execution-data.event-map.ts new file mode 100644 index 00000000000..3db210b66ef --- /dev/null +++ b/packages/cli/src/events/maps/execution-data.event-map.ts @@ -0,0 +1,25 @@ +import type { ExecutionDataStorageLocation } from '@n8n/db'; + +/** + * A "read" spans fetching the bundle from its store AND deserializing it into usable data, so a + * deserialization failure counts as a failed read and its duration includes deserialize time. + * A "write" is symmetric: it spans serializing the data AND writing it to its store. + * + * `unreadableBundles` counts bundles that were missing (the store had none) or corrupt (fetched + * but failed to deserialize) — the data-loss signal. It excludes infrastructure failures (e.g. the + * store being unreachable), where the bundles' integrity is unknown; those are failed reads with + * `unreadableBundles: 0`. + */ +export type ExecutionDataEventMap = { + 'execution-data-read': { + mode: ExecutionDataStorageLocation; + durationMs: number; + success: boolean; + unreadableBundles: number; + }; + 'execution-data-write': { + mode: ExecutionDataStorageLocation; + durationMs: number; + success: boolean; + }; +}; diff --git a/packages/cli/src/events/relays/telemetry.event-relay.ts b/packages/cli/src/events/relays/telemetry.event-relay.ts index 442d417527c..1ca793293b8 100644 --- a/packages/cli/src/events/relays/telemetry.event-relay.ts +++ b/packages/cli/src/events/relays/telemetry.event-relay.ts @@ -1296,6 +1296,8 @@ export class TelemetryEventRelay extends EventRelay { metrics_category_cache: this.globalConfig.endpoints.metrics.includeCacheMetrics, metrics_category_logs: this.globalConfig.endpoints.metrics.includeMessageEventBusMetrics, metrics_category_queue: this.globalConfig.endpoints.metrics.includeQueueMetrics, + metrics_category_execution_data: + this.globalConfig.endpoints.metrics.includeExecutionDataMetrics, }, }; diff --git a/packages/cli/src/executions/__tests__/execution-persistence.test.ts b/packages/cli/src/executions/__tests__/execution-persistence.test.ts index 7c90b6cd993..cd8e13e3cec 100644 --- a/packages/cli/src/executions/__tests__/execution-persistence.test.ts +++ b/packages/cli/src/executions/__tests__/execution-persistence.test.ts @@ -16,8 +16,10 @@ import type { IWorkflowBase } from 'n8n-workflow'; import { createEmptyRunExecutionData, UnexpectedError } from 'n8n-workflow'; import { DuplicateExecutionError } from '@/errors/duplicate-execution.error'; +import type { EventService } from '@/events/event.service'; import type { DbStore } from '@/executions/execution-data/db-store'; import type { FsStore } from '@/executions/execution-data/fs-store'; +import { CorruptedExecutionDataError } from '@/executions/execution-data/corrupted-execution-data.error'; import { MissingExecutionDataError } from '@/executions/execution-data/missing-execution-data.error'; import { ExecutionPersistence } from '@/executions/execution-persistence'; @@ -27,6 +29,7 @@ describe('ExecutionPersistence', () => { const fsStore = mock(); const dbStore = mock(); const errorReporter = mock(); + const eventService = mock(); const executionsConfig = mock({ pruneData: true, pruneDataHardDeleteBuffer: 1, @@ -73,6 +76,7 @@ describe('ExecutionPersistence', () => { executionsConfig, mock({ type: dbType }), errorReporter, + eventService, ); describe('create', () => { @@ -1146,7 +1150,8 @@ describe('ExecutionPersistence', () => { const result = await executionPersistence.findMultipleExecutions({}, { includeData: true }); - expect(executionRepository.reportInvalidExecutions).toHaveBeenCalledWith([dbB, fsC]); + expect(executionRepository.reportInvalidExecutions).toHaveBeenCalledWith([dbB]); + expect(executionRepository.reportInvalidExecutions).toHaveBeenCalledWith([fsC]); expect(result.map((e) => e.id)).toEqual(['a']); }); @@ -1356,4 +1361,277 @@ describe('ExecutionPersistence', () => { expect(executionRepository.update).not.toHaveBeenCalled(); }); }); + + describe('execution data metrics events', () => { + const createPayload: CreateExecutionPayload = { + data: runData, + workflowData, + mode: 'manual', + finished: false, + status: 'new', + workflowId: 'workflow-123', + }; + + const readBundle = { + data: '[{"resultData":"1"},{}]', + workflowData: { id: 'wf-1', name: 's', nodes: [], connections: {}, settings: undefined }, + workflowVersionId: 'v-1', + version: 1 as const, + }; + + const entity = (storedAt: 'db' | 'fs', id = 'exec-1') => + ({ + id, + workflowId: 'wf-1', + storedAt, + metadata: [], + annotation: undefined, + status: 'success', + }) as unknown as ExecutionEntity; + + it('emits a successful write event on create', async () => { + const executionPersistence = createPersistenceService('db'); + executionRepository.manager.transaction = createMockTx(createMockTransaction()); + + await executionPersistence.create(createPayload); + + expect(eventService.emit).toHaveBeenCalledWith( + 'execution-data-write', + expect.objectContaining({ mode: 'db', success: true, durationMs: expect.any(Number) }), + ); + }); + + it('emits a failed write event when the store write throws', async () => { + const executionPersistence = createPersistenceService('db'); + executionRepository.manager.transaction = createMockTx(createMockTransaction()); + dbStore.write.mockRejectedValueOnce(new Error('disk full')); + + await expect(executionPersistence.create(createPayload)).rejects.toThrow('disk full'); + + expect(eventService.emit).toHaveBeenCalledWith( + 'execution-data-write', + expect.objectContaining({ mode: 'db', success: false }), + ); + }); + + it('emits a successful read event when a single bundle is found', async () => { + const executionPersistence = createPersistenceService('db'); + executionRepository.findOne.mockResolvedValue(entity('db')); + dbStore.read.mockResolvedValue(readBundle); + + await executionPersistence.findSingleExecution('exec-1', { includeData: true }); + + expect(eventService.emit).toHaveBeenCalledWith( + 'execution-data-read', + expect.objectContaining({ mode: 'db', success: true, unreadableBundles: 0 }), + ); + }); + + it('emits a failed read with one unreadable bundle when a single bundle is missing', async () => { + const executionPersistence = createPersistenceService('db'); + executionRepository.findOne.mockResolvedValue(entity('db')); + dbStore.read.mockResolvedValue(null); + + await executionPersistence.findSingleExecution('exec-1', { includeData: true }); + + expect(eventService.emit).toHaveBeenCalledWith( + 'execution-data-read', + expect.objectContaining({ mode: 'db', success: false, unreadableBundles: 1 }), + ); + }); + + it('emits a read failure and no write when the read-merge bundle is missing', async () => { + const executionPersistence = createPersistenceService('db'); + executionRepository.findOne.mockResolvedValue(entity('db')); + executionRepository.manager.transaction = createMockTx(createMockTransaction()); + dbStore.read.mockResolvedValue(null); + + await expect( + executionPersistence.updateExistingExecution('exec-1', { data: runData }), + ).rejects.toBeInstanceOf(MissingExecutionDataError); + + expect(eventService.emit).toHaveBeenCalledWith( + 'execution-data-read', + expect.objectContaining({ mode: 'db', success: false, unreadableBundles: 1 }), + ); + expect(eventService.emit).not.toHaveBeenCalledWith('execution-data-write', expect.anything()); + }); + + it('counts dropped bundles from a partial batch read as unreadable', async () => { + const executionPersistence = createPersistenceService('db'); + executionRepository.find.mockResolvedValue([entity('db', 'exec-1'), entity('db', 'exec-2')]); + dbStore.readMany.mockResolvedValue(new Map([['exec-1', readBundle]])); + + await executionPersistence.findMultipleExecutions({}, { includeData: true }); + + expect(eventService.emit).toHaveBeenCalledWith( + 'execution-data-read', + expect.objectContaining({ mode: 'db', success: true, unreadableBundles: 1 }), + ); + }); + + it('emits a failed read with zero unreadable bundles when the store read throws', async () => { + const executionPersistence = createPersistenceService('db'); + executionRepository.findOne.mockResolvedValue(entity('db')); + dbStore.read.mockRejectedValueOnce(new Error('db connection lost')); + + await expect( + executionPersistence.findSingleExecution('exec-1', { includeData: true }), + ).rejects.toThrow('db connection lost'); + + expect(eventService.emit).toHaveBeenCalledWith( + 'execution-data-read', + expect.objectContaining({ mode: 'db', success: false, unreadableBundles: 0 }), + ); + }); + + it('emits a failed read with one unreadable bundle when a fetched bundle fails to deserialize', async () => { + const executionPersistence = createPersistenceService('db'); + executionRepository.findOne.mockResolvedValue(entity('db')); + dbStore.read.mockResolvedValue({ ...readBundle, data: 'not-valid-flatted' }); + + await expect( + executionPersistence.findSingleExecution('exec-1', { + includeData: true, + unflattenData: true, + }), + ).rejects.toBeInstanceOf(CorruptedExecutionDataError); + + expect(eventService.emit).toHaveBeenCalledWith( + 'execution-data-read', + expect.objectContaining({ mode: 'db', success: false, unreadableBundles: 1 }), + ); + }); + + it('emits a failed read with one unreadable bundle when the store read reports corruption', async () => { + const executionPersistence = createPersistenceService('fs'); + executionRepository.findOne.mockResolvedValue(entity('fs')); + fsStore.read.mockRejectedValueOnce( + new CorruptedExecutionDataError( + { workflowId: 'wf-1', executionId: 'exec-1' }, + new Error('x'), + ), + ); + + await expect( + executionPersistence.findSingleExecution('exec-1', { includeData: true }), + ).rejects.toBeInstanceOf(CorruptedExecutionDataError); + + expect(eventService.emit).toHaveBeenCalledWith( + 'execution-data-read', + expect.objectContaining({ mode: 'fs', success: false, unreadableBundles: 1 }), + ); + }); + + it('emits a read failure with one unreadable bundle when the read-merge read reports corruption', async () => { + const executionPersistence = createPersistenceService('db'); + executionRepository.findOne.mockResolvedValue(entity('db')); + executionRepository.manager.transaction = createMockTx(createMockTransaction()); + dbStore.read.mockRejectedValueOnce( + new CorruptedExecutionDataError( + { workflowId: 'wf-1', executionId: 'exec-1' }, + new Error('x'), + ), + ); + + await expect( + executionPersistence.updateExistingExecution('exec-1', { data: runData }), + ).rejects.toBeInstanceOf(CorruptedExecutionDataError); + + expect(eventService.emit).toHaveBeenCalledWith( + 'execution-data-read', + expect.objectContaining({ mode: 'db', success: false, unreadableBundles: 1 }), + ); + }); + + it('counts a corrupt bundle in a batch as unreadable, reports it, and returns the rest', async () => { + const executionPersistence = createPersistenceService('db'); + const good = entity('db', 'exec-1'); + const corrupt = entity('db', 'exec-2'); + executionRepository.find.mockResolvedValue([good, corrupt]); + dbStore.readMany.mockResolvedValue( + new Map([ + ['exec-1', readBundle], + ['exec-2', { ...readBundle, data: 'not-valid-flatted' }], + ]), + ); + + const result = await executionPersistence.findMultipleExecutions( + {}, + { includeData: true, unflattenData: true }, + ); + + expect(result.map((e) => e.id)).toEqual(['exec-1']); + expect(executionRepository.reportInvalidExecutions).toHaveBeenCalledWith([corrupt]); + expect(eventService.emit).toHaveBeenCalledWith( + 'execution-data-read', + expect.objectContaining({ mode: 'db', success: true, unreadableBundles: 1 }), + ); + }); + + it('fails the batch when an assembly error is not a corrupt bundle', async () => { + const executionPersistence = createPersistenceService('db'); + // `metadata: undefined` makes assembly throw a plain `TypeError`, not a corruption error. + const broken = { + id: 'exec-1', + workflowId: 'wf-1', + storedAt: 'db', + metadata: undefined, + annotation: undefined, + status: 'success', + } as unknown as ExecutionEntity; + executionRepository.find.mockResolvedValue([broken]); + dbStore.readMany.mockResolvedValue(new Map([['exec-1', readBundle]])); + + await expect( + executionPersistence.findMultipleExecutions({}, { includeData: true }), + ).rejects.toThrow(); + + expect(executionRepository.reportInvalidExecutions).not.toHaveBeenCalled(); + }); + + it('treats an unsupported execution-data version as a corrupt bundle', async () => { + const executionPersistence = createPersistenceService('db'); + executionRepository.findOne.mockResolvedValue(entity('db')); + // Parses fine, but carries a version the migration step does not support. + dbStore.read.mockResolvedValue({ ...readBundle, data: '[{"version":99}]' }); + + await expect( + executionPersistence.findSingleExecution('exec-1', { + includeData: true, + unflattenData: true, + }), + ).rejects.toBeInstanceOf(CorruptedExecutionDataError); + + expect(eventService.emit).toHaveBeenCalledWith( + 'execution-data-read', + expect.objectContaining({ mode: 'db', success: false, unreadableBundles: 1 }), + ); + }); + + it('keeps the missing-bundle count when an unexpected error fails the batch', async () => { + const executionPersistence = createPersistenceService('db'); + const missing = entity('db', 'exec-1'); + // `metadata: undefined` makes assembly throw a plain `TypeError`, not a corruption error. + const broken = { + id: 'exec-2', + workflowId: 'wf-1', + storedAt: 'db', + metadata: undefined, + annotation: undefined, + status: 'success', + } as unknown as ExecutionEntity; + executionRepository.find.mockResolvedValue([missing, broken]); + dbStore.readMany.mockResolvedValue(new Map([['exec-2', readBundle]])); + + await expect( + executionPersistence.findMultipleExecutions({}, { includeData: true }), + ).rejects.toThrow(); + + expect(eventService.emit).toHaveBeenCalledWith( + 'execution-data-read', + expect.objectContaining({ mode: 'db', success: false, unreadableBundles: 1 }), + ); + }); + }); }); diff --git a/packages/cli/src/executions/execution-persistence.ts b/packages/cli/src/executions/execution-persistence.ts index 518ceadd129..4e6985e183b 100644 --- a/packages/cli/src/executions/execution-persistence.ts +++ b/packages/cli/src/executions/execution-persistence.ts @@ -19,6 +19,7 @@ import { BinaryDataService, ErrorReporter, StorageConfig } from 'n8n-core'; import type { IRunExecutionData, IRunExecutionDataAll } from 'n8n-workflow'; import { migrateRunExecutionData, UnexpectedError } from 'n8n-workflow'; +import { CorruptedExecutionDataError } from './execution-data/corrupted-execution-data.error'; import { DbStore } from './execution-data/db-store'; import { FsStore } from './execution-data/fs-store'; import { MissingExecutionDataError } from './execution-data/missing-execution-data.error'; @@ -29,9 +30,12 @@ import type { WorkflowSnapshot, } from './execution-data/types'; import { DuplicateExecutionError } from '../errors/duplicate-execution.error'; +import { EventService } from '../events/event.service'; type DeletionTarget = ExecutionRef & { storedAt: ExecutionDataStorageLocation }; +type FoundExecution = IExecutionFlattedDb | IExecutionResponse | IExecutionBase; + type UpdatableEntityColumns = Omit< Partial, | 'id' @@ -59,6 +63,7 @@ export class ExecutionPersistence { private readonly executionsConfig: ExecutionsConfig, private readonly databaseConfig: DatabaseConfig, private readonly errorReporter: ErrorReporter, + private readonly eventService: EventService, ) {} /** @@ -72,7 +77,6 @@ export class ExecutionPersistence { const workflowSnapshot: WorkflowSnapshot = { connections, nodes, name, settings, id }; const storedAt = this.storageConfig.modeTag; const executionEntity = { ...rest, createdAt: new Date(), storedAt }; - const data = stringify(rawData); const workflowVersionId = workflowData.versionId ?? null; try { @@ -80,9 +84,16 @@ export class ExecutionPersistence { const { identifiers } = await tx.insert(ExecutionEntity, executionEntity); const executionId = String(identifiers[0].id); const ref = { workflowId: id, executionId }; - const bundle = { data, workflowData: workflowSnapshot, workflowVersionId }; + const store = this.getStoreFor(storedAt); - await this.getStoreFor(storedAt).write(ref, bundle, tx); + await this.trackWrite(storedAt, async () => { + const bundle = { + data: stringify(rawData), + workflowData: workflowSnapshot, + workflowVersionId, + }; + await store.write(ref, bundle, tx); + }); return executionId; }); @@ -120,7 +131,7 @@ export class ExecutionPersistence { const ref = { workflowId: entity.workflowId, executionId }; const store = this.getStoreFor(entity.storedAt); - return await this.applyDataUpdate(ref, store, execution, conditions); + return await this.applyDataUpdate(ref, store, entity.storedAt, execution, conditions); } /** @@ -169,7 +180,7 @@ export class ExecutionPersistence { unflattenData?: boolean; where?: FindOptionsWhere; }, - ): Promise { + ): Promise { if (!options?.includeData) { return await this.executionRepository.findSingleExecution(id, options); } @@ -185,23 +196,35 @@ export class ExecutionPersistence { if (!entity) return undefined; const store = this.getStoreFor(entity.storedAt); - const bundle = await store.read({ workflowId: entity.workflowId, executionId: entity.id }); + const ref = { workflowId: entity.workflowId, executionId: entity.id }; - if (!bundle) { - if (entity.storedAt === 'db') { - this.executionRepository.reportInvalidExecutions([entity]); - return undefined; + const start = Date.now(); + let success = false; + let unreadableBundles = 0; + try { + const bundle = await store.read(ref); + if (!bundle) { + unreadableBundles = 1; + if (entity.storedAt === 'db') { + this.executionRepository.reportInvalidExecutions([entity]); + return undefined; + } + throw new MissingExecutionDataError(ref); } - throw new MissingExecutionDataError({ - workflowId: entity.workflowId, - executionId: entity.id, + const assembled = await this.assembleExecution(entity, bundle, options); + success = true; + return assembled as FoundExecution; + } catch (error) { + if (error instanceof CorruptedExecutionDataError) unreadableBundles = 1; + throw error; + } finally { + this.eventService.emit('execution-data-read', { + mode: entity.storedAt, + durationMs: Date.now() - start, + success, + unreadableBundles, }); } - - return (await this.assembleExecution(entity, bundle, options)) as - | IExecutionFlattedDb - | IExecutionResponse - | IExecutionBase; } /** @@ -275,30 +298,60 @@ export class ExecutionPersistence { entitiesByLocation.set(entity.storedAt, group); } - const bundlesById = new Map(); + const assembledById = new Map>>(); await Promise.all( [...entitiesByLocation].map(async ([location, group]) => { const refs = group.map((e) => ({ workflowId: e.workflowId, executionId: e.id })); - const bundles = await this.getStoreFor(location).readMany(refs); - for (const [id, bundle] of bundles) bundlesById.set(id, bundle); + const store = this.getStoreFor(location); + const start = Date.now(); + let success = false; + let unreadableBundles = 0; + try { + const bundles = await store.readMany(refs); + const missing = group.filter((e) => !bundles.has(e.id)); + if (missing.length > 0) this.executionRepository.reportInvalidExecutions(missing); + unreadableBundles = missing.length; + + const settled = await Promise.allSettled( + group.map(async (entity) => { + const bundle = bundles.get(entity.id); + if (!bundle) return; + assembledById.set(entity.id, await this.assembleExecution(entity, bundle, options)); + }), + ); + const corrupt = group.filter((_, i) => { + const outcome = settled[i]; + return ( + outcome.status === 'rejected' && outcome.reason instanceof CorruptedExecutionDataError + ); + }); + unreadableBundles += corrupt.length; + if (corrupt.length > 0) this.executionRepository.reportInvalidExecutions(corrupt); + + for (const outcome of settled) { + if ( + outcome.status === 'rejected' && + !(outcome.reason instanceof CorruptedExecutionDataError) + ) { + throw outcome.reason; + } + } + + success = true; + } finally { + this.eventService.emit('execution-data-read', { + mode: location, + durationMs: Date.now() - start, + success, + unreadableBundles, + }); + } }), ); - // Report invalid entities when they are found to be missing from the stores. - const invalidEntities = entities.filter((e) => !bundlesById.has(e.id)); - if (invalidEntities.length > 0) { - this.executionRepository.reportInvalidExecutions(invalidEntities); - } - - const assembled = await Promise.all( - entities.map(async (entity) => { - const bundle = bundlesById.get(entity.id); - if (!bundle) return null; - return await this.assembleExecution(entity, bundle, options); - }), - ); - - return assembled.filter((e): e is NonNullable => e !== null) as + return entities + .map((e) => assembledById.get(e.id)) + .filter((e): e is NonNullable => e !== undefined) as | IExecutionFlattedDb[] | IExecutionResponse[] | IExecutionBase[]; @@ -358,6 +411,7 @@ export class ExecutionPersistence { private async applyDataUpdate( ref: ExecutionRef, store: ExecutionDataStore, + mode: ExecutionDataStorageLocation, execution: Partial, conditions?: UpdateExecutionConditions, ): Promise { @@ -385,28 +439,34 @@ export class ExecutionPersistence { } if (data !== undefined && workflowData !== undefined && store === this.dbStore) { - const result = await tx.update( - ExecutionData, - { executionId: ref.executionId }, - { data: stringify(data), workflowData: this.toWorkflowSnapshot(workflowData) }, - ); - if ((result.affected ?? 0) === 0) throw new MissingExecutionDataError(ref); + await this.trackWrite(mode, async () => { + const result = await tx.update( + ExecutionData, + { executionId: ref.executionId }, + { data: stringify(data), workflowData: this.toWorkflowSnapshot(workflowData) }, + ); + if ((result.affected ?? 0) === 0) throw new MissingExecutionDataError(ref); + }); return true; } - const existing = await store.read(ref, tx); + const existing = await this.trackRead(mode, async () => await store.read(ref, tx)); if (!existing) throw new MissingExecutionDataError(ref); - await store.write( - ref, - { - data: data !== undefined ? stringify(data) : existing.data, - workflowData: workflowData - ? this.toWorkflowSnapshot(workflowData) - : existing.workflowData, - workflowVersionId: existing.workflowVersionId, - }, - tx, + await this.trackWrite( + mode, + async () => + await store.write( + ref, + { + data: data !== undefined ? stringify(data) : existing.data, + workflowData: workflowData + ? this.toWorkflowSnapshot(workflowData) + : existing.workflowData, + workflowVersionId: existing.workflowVersionId, + }, + tx, + ), ); return true; @@ -461,6 +521,46 @@ export class ExecutionPersistence { return where; } + private async trackRead(mode: ExecutionDataStorageLocation, op: () => Promise): Promise { + const start = Date.now(); + let success = false; + let unreadableBundles = 0; + try { + const result = await op(); + success = result !== null && result !== undefined; + if (!success) unreadableBundles = 1; + return result; + } catch (error) { + if (error instanceof CorruptedExecutionDataError) unreadableBundles = 1; + throw error; + } finally { + this.eventService.emit('execution-data-read', { + mode, + durationMs: Date.now() - start, + success, + unreadableBundles, + }); + } + } + + private async trackWrite( + mode: ExecutionDataStorageLocation, + op: () => Promise, + ): Promise { + const start = Date.now(); + let success = false; + try { + await op(); + success = true; + } finally { + this.eventService.emit('execution-data-write', { + mode, + durationMs: Date.now() - start, + success, + }); + } + } + private getStoreFor(location: ExecutionDataStorageLocation): ExecutionDataStore { switch (location) { case 'db': @@ -485,7 +585,8 @@ export class ExecutionPersistence { options: { unflattenData?: boolean; includeAnnotation?: boolean }, ) { const { metadata, annotation, ...rest } = entity; - const data = await this.parseExecutionData(bundle.data, options); + const ref = { workflowId: entity.workflowId, executionId: entity.id }; + const data = await this.parseExecutionData(ref, bundle.data, options); const serializedAnnotation = this.serializeAnnotation(annotation); if (entity.status === 'success' && bundle.data === '[]') { @@ -507,14 +608,19 @@ export class ExecutionPersistence { } private async parseExecutionData( + ref: ExecutionRef, data: string, options: { unflattenData?: boolean }, ): Promise { if (!options.unflattenData) return data; - const deserialized: unknown = await parseFlatted(data); - if (!deserialized) return undefined; - return migrateRunExecutionData(deserialized as IRunExecutionDataAll); + try { + const deserialized: unknown = await parseFlatted(data); + if (!deserialized) return undefined; + return migrateRunExecutionData(deserialized as IRunExecutionDataAll); + } catch (error) { + throw new CorruptedExecutionDataError(ref, error); + } } private serializeAnnotation(annotation: ExecutionEntity['annotation']) { diff --git a/packages/cli/src/metrics/__tests__/prometheus-metrics.service.test.ts b/packages/cli/src/metrics/__tests__/prometheus-metrics.service.test.ts index ec414d5a846..71b7c7bf954 100644 --- a/packages/cli/src/metrics/__tests__/prometheus-metrics.service.test.ts +++ b/packages/cli/src/metrics/__tests__/prometheus-metrics.service.test.ts @@ -6,7 +6,7 @@ import type { WorkflowRepository, LicenseMetricsRepository } from '@n8n/db'; import type express from 'express'; import promBundle from 'express-prom-bundle'; import { mock } from 'jest-mock-extended'; -import type { InstanceSettings } from 'n8n-core'; +import type { InstanceSettings, StorageConfig } from 'n8n-core'; import { EventMessageTypeNames } from 'n8n-workflow'; import promClient from 'prom-client'; @@ -35,6 +35,7 @@ describe('PrometheusMetricsService', () => { let instanceSettings: InstanceSettings; let workflowRepository: WorkflowRepository; let licenseMetricsRepository: LicenseMetricsRepository; + let storageConfig: StorageConfig; let prometheusMetricsService: PrometheusMetricsService; beforeEach(() => { @@ -56,6 +57,7 @@ describe('PrometheusMetricsService', () => { includeWorkflowExecutionDuration: false, includeWorkflowNameLabel: false, includeWorkflowStatistics: false, + includeExecutionDataMetrics: false, activeWorkflowCountInterval: 30, workflowStatisticsInterval: 30, }, @@ -78,6 +80,7 @@ describe('PrometheusMetricsService', () => { instanceSettings = mock({ instanceType: 'main' }); workflowRepository = mock(); licenseMetricsRepository = mock(); + storageConfig = mock({ modeTag: 'db' }); prometheusMetricsService = new PrometheusMetricsService( mock(), @@ -87,6 +90,7 @@ describe('PrometheusMetricsService', () => { instanceSettings, workflowRepository, licenseMetricsRepository, + storageConfig, ); promClient.Counter.prototype.inc = jest.fn(); @@ -103,6 +107,12 @@ describe('PrometheusMetricsService', () => { prometheusMetricsService.disableAllLabels(); }); + /** Capture the handler registered via `eventService.on(eventName, handler)`. */ + const getEventServiceHandler = (eventName: string) => { + const call = (eventService.on as jest.Mock).mock.calls.find((c) => c[0] === eventName); + return call ? call[1] : undefined; + }; + describe('constructor', () => { it('should enable metrics based on global config', async () => { const customGlobalConfig = { ...globalConfig }; @@ -115,6 +125,7 @@ describe('PrometheusMetricsService', () => { instanceSettings, mock(), mock(), + mock(), ); await customPrometheusMetricsService.init(app); @@ -826,12 +837,6 @@ describe('PrometheusMetricsService', () => { }); describe('token exchange metrics', () => { - // Helper to capture an eventService.on handler by event name - const getEventServiceHandler = (eventName: string) => { - const call = (eventService.on as jest.Mock).mock.calls.find((c) => c[0] === eventName); - return call ? call[1] : undefined; - }; - it('should register all 6 token exchange counters on init', async () => { await prometheusMetricsService.init(app); @@ -972,4 +977,101 @@ describe('PrometheusMetricsService', () => { expect(linkedCounter?.inc).toHaveBeenCalledWith(1); }); }); + + describe('execution data metrics', () => { + it('should register execution data metrics and set a storage-mode gauge when enabled', async () => { + prometheusMetricsService.enableMetric('executionData'); + await prometheusMetricsService.init(app); + + expect(promClient.Counter).toHaveBeenCalledWith({ + name: 'n8n_execution_data_reads_total', + help: 'Total number of execution data reads.', + labelNames: ['mode', 'result'], + }); + expect(promClient.Counter).toHaveBeenCalledWith({ + name: 'n8n_execution_data_writes_total', + help: 'Total number of execution data writes.', + labelNames: ['mode', 'result'], + }); + expect(promClient.Counter).toHaveBeenCalledWith( + expect.objectContaining({ name: 'n8n_execution_data_unreadable_bundles_total' }), + ); + expect(promClient.Histogram).toHaveBeenCalledWith( + expect.objectContaining({ name: 'n8n_execution_data_read_duration_seconds' }), + ); + expect(promClient.Histogram).toHaveBeenCalledWith( + expect.objectContaining({ name: 'n8n_execution_data_write_duration_seconds' }), + ); + expect(promClient.Gauge).toHaveBeenCalledWith( + expect.objectContaining({ name: 'n8n_execution_data_storage_mode', labelNames: ['mode'] }), + ); + + // @ts-expect-error private field + const gauge = prometheusMetricsService.gauges.executionDataStorageMode; + expect(gauge.set).toHaveBeenCalledWith({ mode: 'db' }, 0); + expect(gauge.set).toHaveBeenCalledWith({ mode: 'fs' }, 0); + expect(gauge.set).toHaveBeenCalledWith({ mode: 'db' }, 1); // configured mode (storageConfig.modeTag) + }); + + it('should not register execution data metrics when disabled', async () => { + await prometheusMetricsService.init(app); + + expect(promClient.Counter).not.toHaveBeenCalledWith( + expect.objectContaining({ name: 'n8n_execution_data_reads_total' }), + ); + }); + + it('should count a read and observe its duration (seconds) on a successful read event', async () => { + prometheusMetricsService.enableMetric('executionData'); + await prometheusMetricsService.init(app); + + getEventServiceHandler('execution-data-read')({ + mode: 'db', + durationMs: 200, + success: true, + unreadableBundles: 0, + }); + + // @ts-expect-error private field + const reads = prometheusMetricsService.counters.executionDataReadsTotal; + expect(reads?.inc).toHaveBeenCalledWith({ mode: 'db', result: 'success' }, 1); + // @ts-expect-error private field + const histogram = prometheusMetricsService.histograms.executionDataReadDuration; + expect(histogram.observe).toHaveBeenCalledWith({ mode: 'db' }, 0.2); + }); + + it('should count unreadable bundles reported by a read event', async () => { + prometheusMetricsService.enableMetric('executionData'); + await prometheusMetricsService.init(app); + + getEventServiceHandler('execution-data-read')({ + mode: 'fs', + durationMs: 5, + success: true, + unreadableBundles: 3, + }); + + // @ts-expect-error private field + const unreadable = prometheusMetricsService.counters.executionDataUnreadableBundlesTotal; + expect(unreadable?.inc).toHaveBeenCalledWith({ mode: 'fs' }, 3); + }); + + it('should count a failed write without observing its duration', async () => { + prometheusMetricsService.enableMetric('executionData'); + await prometheusMetricsService.init(app); + + getEventServiceHandler('execution-data-write')({ + mode: 'db', + durationMs: 10, + success: false, + }); + + // @ts-expect-error private field + const writes = prometheusMetricsService.counters.executionDataWritesTotal; + expect(writes?.inc).toHaveBeenCalledWith({ mode: 'db', result: 'failure' }, 1); + // @ts-expect-error private field + const histogram = prometheusMetricsService.histograms.executionDataWriteDuration; + expect(histogram.observe).not.toHaveBeenCalled(); + }); + }); }); diff --git a/packages/cli/src/metrics/__tests__/prometheus-metrics.service.unmocked.test.ts b/packages/cli/src/metrics/__tests__/prometheus-metrics.service.unmocked.test.ts index 95ee27cd509..abba0faf8f8 100644 --- a/packages/cli/src/metrics/__tests__/prometheus-metrics.service.unmocked.test.ts +++ b/packages/cli/src/metrics/__tests__/prometheus-metrics.service.unmocked.test.ts @@ -3,7 +3,7 @@ import { GlobalConfig } from '@n8n/config'; import type { WorkflowRepository, LicenseMetricsRepository } from '@n8n/db'; import type express from 'express'; import { mock } from 'jest-mock-extended'; -import type { InstanceSettings } from 'n8n-core'; +import type { InstanceSettings, StorageConfig } from 'n8n-core'; import promClient from 'prom-client'; import { EventMessageWorkflow } from '@/eventbus/event-message-classes/event-message-workflow'; @@ -21,6 +21,7 @@ const cacheService = mock(); const eventService = mock(); const instanceSettings = mock({ instanceType: 'main' }); const workflowRepository = mock(); +const storageConfig = mock({ modeTag: 'db' }); const app = mock(); const eventBus = new MessageEventBus(mock(), mock(), mock(), mock(), mock(), mock()); @@ -46,6 +47,7 @@ describe('workflow_success_total', () => { instanceSettings, workflowRepository, mock(), + storageConfig, ); await prometheusMetricsService.init(app); @@ -90,6 +92,7 @@ workflow_success_total{workflow_id="1234"} 1" instanceSettings, workflowRepository, mock(), + storageConfig, ); await prometheusMetricsService.init(app); @@ -131,6 +134,7 @@ workflow_success_total{workflow_name="wf_1234"} 1" instanceSettings, workflowRepository, mock(), + storageConfig, ); // ACT @@ -176,6 +180,7 @@ describe('Active workflow count', () => { instanceSettings, workflowRepository, mock(), + storageConfig, ); afterEach(() => { diff --git a/packages/cli/src/metrics/prometheus-metrics.service.ts b/packages/cli/src/metrics/prometheus-metrics.service.ts index 541d94f1e65..d1a9a77d89a 100644 --- a/packages/cli/src/metrics/prometheus-metrics.service.ts +++ b/packages/cli/src/metrics/prometheus-metrics.service.ts @@ -8,7 +8,7 @@ import { Service } from '@n8n/di'; import type express from 'express'; import promBundle from 'express-prom-bundle'; import { DateTime } from 'luxon'; -import { InstanceSettings } from 'n8n-core'; +import { InstanceSettings, StorageConfig } from 'n8n-core'; import { EventMessageTypeNames, jsonParse } from 'n8n-workflow'; import promClient, { type Counter, type Gauge, type Histogram } from 'prom-client'; import semverParse from 'semver/functions/parse'; @@ -21,6 +21,11 @@ import { CacheService } from '@/services/cache/cache.service'; import type { Includes, MetricCategory, MetricLabel } from './types'; +/** Bucket boundaries (in seconds) shared by all duration histograms. */ +const DURATION_BUCKETS_SECONDS = [ + 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60, 120, 300, 600, +]; + @Service() export class PrometheusMetricsService { constructor( @@ -31,12 +36,15 @@ export class PrometheusMetricsService { private readonly instanceSettings: InstanceSettings, private readonly workflowRepository: WorkflowRepository, private readonly licenseMetricsRepository: LicenseMetricsRepository, + private readonly storageConfig: StorageConfig, ) {} private readonly counters: { [key: string]: Counter | null } = {}; private tokenExchangeListenersRegistered = false; + private executionDataListenersRegistered = false; + private readonly gauges: Record> = {}; private readonly histograms: Record> = {}; @@ -53,6 +61,7 @@ export class PrometheusMetricsService { workflowExecutionDuration: this.globalConfig.endpoints.metrics.includeWorkflowExecutionDuration, workflowStatistics: this.globalConfig.endpoints.metrics.includeWorkflowStatistics, + executionData: this.globalConfig.endpoints.metrics.includeExecutionDataMetrics, }, labels: { credentialsType: this.globalConfig.endpoints.metrics.includeCredentialTypeLabel, @@ -79,6 +88,7 @@ export class PrometheusMetricsService { this.initActiveWorkflowCountMetric(); this.initWorkflowStatisticsMetrics(); this.initTokenExchangeMetrics(); + this.initExecutionDataMetrics(); this.mountMetricsEndpoint(app); } @@ -367,7 +377,7 @@ export class PrometheusMetricsService { name: this.prefix + 'workflow_execution_duration_seconds', help: 'Workflow execution duration in seconds.', labelNames, - buckets: [0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60, 120, 300, 600], + buckets: DURATION_BUCKETS_SECONDS, }); this.eventService.on('workflow-post-execute', ({ runData, workflow }) => { @@ -712,4 +722,84 @@ export class PrometheusMetricsService { this.counters.tokenExchangeIdentityLinkedTotal?.inc(1); }); } + + private initExecutionDataMetrics() { + if (!this.includes.metrics.executionData) return; + + const readsTotal = (this.counters.executionDataReadsTotal = new promClient.Counter({ + name: this.prefix + 'execution_data_reads_total', + help: 'Total number of execution data reads.', + labelNames: ['mode', 'result'], + })); + + const writesTotal = (this.counters.executionDataWritesTotal = new promClient.Counter({ + name: this.prefix + 'execution_data_writes_total', + help: 'Total number of execution data writes.', + labelNames: ['mode', 'result'], + })); + + const unreadableBundlesTotal = (this.counters.executionDataUnreadableBundlesTotal = + new promClient.Counter({ + name: this.prefix + 'execution_data_unreadable_bundles_total', + help: 'Total number of execution data bundles that were missing or corrupt on read.', + labelNames: ['mode'], + })); + + for (const mode of ['db', 'fs'] as const) { + for (const result of ['success', 'failure'] as const) { + readsTotal.inc({ mode, result }, 0); + writesTotal.inc({ mode, result }, 0); + } + unreadableBundlesTotal.inc({ mode }, 0); + } + + this.histograms.executionDataReadDuration = new promClient.Histogram({ + name: this.prefix + 'execution_data_read_duration_seconds', + help: 'Execution data read duration in seconds (fetch + deserialize).', + labelNames: ['mode'], + buckets: DURATION_BUCKETS_SECONDS, + }); + + this.histograms.executionDataWriteDuration = new promClient.Histogram({ + name: this.prefix + 'execution_data_write_duration_seconds', + help: 'Execution data write duration in seconds.', + labelNames: ['mode'], + buckets: DURATION_BUCKETS_SECONDS, + }); + + this.gauges.executionDataStorageMode = new promClient.Gauge({ + name: this.prefix + 'execution_data_storage_mode', + help: 'Configured execution data storage mode (1 for the active mode, 0 otherwise).', + labelNames: ['mode'], + }); + this.gauges.executionDataStorageMode.set({ mode: 'db' }, 0); + this.gauges.executionDataStorageMode.set({ mode: 'fs' }, 0); + this.gauges.executionDataStorageMode.set({ mode: this.storageConfig.modeTag }, 1); + + if (this.executionDataListenersRegistered) return; + this.executionDataListenersRegistered = true; + + this.eventService.on( + 'execution-data-read', + ({ mode, durationMs, success, unreadableBundles }) => { + this.counters.executionDataReadsTotal?.inc( + { mode, result: success ? 'success' : 'failure' }, + 1, + ); + if (success) + this.histograms.executionDataReadDuration?.observe({ mode }, durationMs / 1000); + if (unreadableBundles > 0) { + this.counters.executionDataUnreadableBundlesTotal?.inc({ mode }, unreadableBundles); + } + }, + ); + + this.eventService.on('execution-data-write', ({ mode, durationMs, success }) => { + this.counters.executionDataWritesTotal?.inc( + { mode, result: success ? 'success' : 'failure' }, + 1, + ); + if (success) this.histograms.executionDataWriteDuration?.observe({ mode }, durationMs / 1000); + }); + } } diff --git a/packages/cli/src/metrics/types.ts b/packages/cli/src/metrics/types.ts index 98b6794db79..4e481c892c5 100644 --- a/packages/cli/src/metrics/types.ts +++ b/packages/cli/src/metrics/types.ts @@ -5,7 +5,8 @@ export type MetricCategory = | 'logs' | 'queue' | 'workflowExecutionDuration' - | 'workflowStatistics'; + | 'workflowStatistics' + | 'executionData'; export type MetricLabel = | 'credentialsType' diff --git a/packages/cli/test/integration/prometheus-metrics.test.ts b/packages/cli/test/integration/prometheus-metrics.test.ts index 91394f0f3fb..9a339536539 100644 --- a/packages/cli/test/integration/prometheus-metrics.test.ts +++ b/packages/cli/test/integration/prometheus-metrics.test.ts @@ -42,6 +42,7 @@ globalConfig.endpoints.metrics = { activeWorkflowCountInterval: 60, includeWorkflowStatistics: false, workflowStatisticsInterval: 300, + includeExecutionDataMetrics: false, }; const server = setupTestServer({ endpointGroups: ['metrics'] }); @@ -489,4 +490,23 @@ describe('PrometheusMetricsService', () => { expect(lines).not.toContainEqual(expect.stringContaining('n8n_test_workflows')); expect(lines).not.toContainEqual(expect.stringContaining('n8n_test_credentials')); }); + + it('should return execution data metrics if enabled', async () => { + prometheusService.enableMetric('executionData'); + await prometheusService.init(server.app); + + const response = await agent.get('/metrics'); + + expect(response.status).toEqual(200); + + const lines = toLines(response); + + expect(lines).toContain('n8n_test_execution_data_reads_total{mode="db",result="success"} 0'); + expect(lines).toContain('n8n_test_execution_data_writes_total{mode="fs",result="failure"} 0'); + expect(lines).toContain('n8n_test_execution_data_unreadable_bundles_total{mode="db"} 0'); + + expect( + lines.some((l) => /^n8n_test_execution_data_storage_mode\{mode="(db|fs)"\} 1$/.test(l)), + ).toBe(true); + }); });