feat(core): Add per-workflow execution data write bytes metric (#34425)

packages/cli/src/events/maps/execution-data.event-map.ts

@@ -19,6 +19,7 @@ export type ExecutionDataEventMap = {
 	};
 	'execution-data-write': {
 		mode: ExecutionDataStorageLocation;
+		workflowId: string;
 		durationMs: number;
 		success: boolean;
 		jsonSizeBytes: number;

packages/cli/src/executions/__tests__/execution-persistence.test.ts

@@ -148,7 +148,7 @@ describe('ExecutionPersistence', () => {
 				);
 				expect(eventService.emit).toHaveBeenCalledWith(
 					'execution-data-write',
-					expect.objectContaining({ jsonSizeBytes: 4321 }),
+					expect.objectContaining({ jsonSizeBytes: 4321, workflowId: 'workflow-123' }),
 				);
 			});
 
@@ -1143,7 +1143,7 @@ describe('ExecutionPersistence', () => {
 				);
 				expect(eventService.emit).toHaveBeenCalledWith(
 					'execution-data-write',
-					expect.objectContaining({ jsonSizeBytes: 2048 }),
+					expect.objectContaining({ jsonSizeBytes: 2048, workflowId: 'wf-1' }),
 				);
 			});
 
@@ -1168,7 +1168,7 @@ describe('ExecutionPersistence', () => {
 				);
 				expect(eventService.emit).toHaveBeenCalledWith(
 					'execution-data-write',
-					expect.objectContaining({ jsonSizeBytes: 1536 }),
+					expect.objectContaining({ jsonSizeBytes: 1536, workflowId: 'wf-1' }),
 				);
 			});
 
@@ -2031,7 +2031,12 @@ describe('ExecutionPersistence', () => {
 
 			expect(eventService.emit).toHaveBeenCalledWith(
 				'execution-data-write',
-				expect.objectContaining({ mode: 'db', success: true, durationMs: expect.any(Number) }),
+				expect.objectContaining({
+					mode: 'db',
+					workflowId: 'workflow-123',
+					success: true,
+					durationMs: expect.any(Number),
+				}),
 			);
 		});
 
@@ -2044,7 +2049,7 @@ describe('ExecutionPersistence', () => {
 
 			expect(eventService.emit).toHaveBeenCalledWith(
 				'execution-data-write',
-				expect.objectContaining({ mode: 'db', success: false }),
+				expect.objectContaining({ mode: 'db', workflowId: 'workflow-123', success: false }),
 			);
 		});
 

packages/cli/src/executions/execution-persistence.ts

@@ -113,7 +113,7 @@ export class ExecutionPersistence {
 				const executionId = String(identifiers[0].id);
 				const ref = { workflowId: id, executionId };
 
-				const jsonSizeBytes = await this.trackWrite(storedAt, async () => {
+				const jsonSizeBytes = await this.trackWrite(storedAt, ref.workflowId, async () => {
 					const bundle: ExecutionDataPayload = {
 						data: stringify(rawData),
 						workflowData: workflowSnapshot,
@@ -761,7 +761,7 @@ export class ExecutionPersistence {
 				(workflowVersionId !== null || mode === 'db')
 			) {
 				const binaryDataSizeBytes = sumBinaryDataBytes(data);
-				const jsonSizeBytes = await this.trackWrite(mode, async () => {
+				const jsonSizeBytes = await this.trackWrite(mode, ref.workflowId, async () => {
 					const bundle: ExecutionDataPayload = {
 						data: stringify(data),
 						workflowData: this.toWorkflowSnapshot(workflowData),
@@ -786,7 +786,7 @@ export class ExecutionPersistence {
 			const existing = await this.trackRead(mode, async () => await this.readData(mode, ref, tx));
 			if (!existing) throw new MissingExecutionDataError(ref);
 
-			const jsonSizeBytes = await this.trackWrite(mode, async () => {
+			const jsonSizeBytes = await this.trackWrite(mode, ref.workflowId, async () => {
 				const bundle: ExecutionDataPayload = {
 					data: data !== undefined ? stringify(data) : existing.data,
 					workflowData: workflowData
@@ -889,6 +889,7 @@ export class ExecutionPersistence {
 	 */
 	private async trackWrite(
 		mode: ExecutionDataStorageLocation,
+		workflowId: string,
 		op: () => Promise<number>,
 	): Promise<number> {
 		const start = Date.now();
@@ -901,6 +902,7 @@ export class ExecutionPersistence {
 		} finally {
 			this.eventService.emit('execution-data-write', {
 				mode,
+				workflowId,
 				durationMs: Date.now() - start,
 				success,
 				jsonSizeBytes,

packages/cli/src/metrics/prometheus/__tests__/execution-data-metrics.service.test.ts

@@ -16,6 +16,7 @@ describe('PrometheusExecutionDataMetricsService', () => {
 	const config = mockInstance(PrometheusMetricsConfig, {
 		prefix: 'n8n_',
 		includeExecutionDataMetrics: true,
+		includeWorkflowIdLabel: false,
 	});
 	const eventService = mock<EventService>();
 	const storageConfig = mock<StorageConfig>({ modeTag: 'db' });
@@ -29,7 +30,11 @@ describe('PrometheusExecutionDataMetricsService', () => {
 	}
 
 	beforeEach(() => {
-		Object.assign(config, { prefix: 'n8n_', includeExecutionDataMetrics: true });
+		Object.assign(config, {
+			prefix: 'n8n_',
+			includeExecutionDataMetrics: true,
+			includeWorkflowIdLabel: false,
+		});
 		Object.assign(storageConfig, { modeTag: 'db' });
 		service = new PrometheusExecutionDataMetricsService(config, eventService, storageConfig);
 		mockCounterInc = vi.fn();
@@ -120,6 +125,27 @@ describe('PrometheusExecutionDataMetricsService', () => {
 			});
 		});
 
+		it('should create execution_data_write_bytes_total counter with only the mode label by default', () => {
+			service.init();
+
+			expect(promClient.Counter).toHaveBeenCalledWith({
+				name: 'n8n_execution_data_write_bytes_total',
+				help: 'Total execution data bytes written, by storage mode.',
+				labelNames: ['mode'],
+			});
+		});
+
+		it('should create execution_data_write_bytes_total counter with workflow_id label when includeWorkflowIdLabel is true', () => {
+			config.includeWorkflowIdLabel = true;
+			service.init();
+
+			expect(promClient.Counter).toHaveBeenCalledWith({
+				name: 'n8n_execution_data_write_bytes_total',
+				help: 'Total execution data bytes written, by storage mode.',
+				labelNames: ['mode', 'workflow_id'],
+			});
+		});
+
 		it('should create execution_data_storage_mode gauge', () => {
 			service.init();
 
@@ -269,7 +295,13 @@ describe('PrometheusExecutionDataMetricsService', () => {
 
 			expect(handler).toBeDefined();
 
-			handler!({ mode: 'db', durationMs: 100, success: true, jsonSizeBytes: 2048 });
+			handler!({
+				mode: 'db',
+				workflowId: 'wf-1',
+				durationMs: 100,
+				success: true,
+				jsonSizeBytes: 2048,
+			});
 
 			expect(mockCounterInc).toHaveBeenCalledWith({ mode: 'db', result: 'success' }, 1);
 		});
@@ -280,7 +312,13 @@ describe('PrometheusExecutionDataMetricsService', () => {
 
 			expect(handler).toBeDefined();
 
-			handler!({ mode: 'db', durationMs: 100, success: true, jsonSizeBytes: 2048 });
+			handler!({
+				mode: 'db',
+				workflowId: 'wf-1',
+				durationMs: 100,
+				success: true,
+				jsonSizeBytes: 2048,
+			});
 
 			expect(mockHistogramObserve).toHaveBeenCalledWith({ mode: 'db' }, 0.1);
 		});
@@ -291,18 +329,69 @@ describe('PrometheusExecutionDataMetricsService', () => {
 
 			expect(handler).toBeDefined();
 
-			handler!({ mode: 'db', durationMs: 100, success: true, jsonSizeBytes: 2048 });
+			handler!({
+				mode: 'db',
+				workflowId: 'wf-1',
+				durationMs: 100,
+				success: true,
+				jsonSizeBytes: 2048,
+			});
 
 			expect(mockHistogramObserve).toHaveBeenCalledWith({ mode: 'db' }, 2048);
 		});
 
+		it('should increment write bytes counter by jsonSizeBytes with only the mode label by default', () => {
+			service.init();
+			const handler = getEventHandler('execution-data-write');
+
+			expect(handler).toBeDefined();
+
+			handler!({
+				mode: 'db',
+				workflowId: 'wf-1',
+				durationMs: 100,
+				success: true,
+				jsonSizeBytes: 2048,
+			});
+
+			expect(mockCounterInc).toHaveBeenCalledWith({ mode: 'db' }, 2048);
+			expect(mockCounterInc).not.toHaveBeenCalledWith(
+				expect.objectContaining({ workflow_id: expect.anything() }),
+				expect.anything(),
+			);
+		});
+
+		it('should increment write bytes counter with workflow_id label when includeWorkflowIdLabel is true', () => {
+			config.includeWorkflowIdLabel = true;
+			service.init();
+			const handler = getEventHandler('execution-data-write');
+
+			expect(handler).toBeDefined();
+
+			handler!({
+				mode: 'fs',
+				workflowId: 'wf-1',
+				durationMs: 100,
+				success: true,
+				jsonSizeBytes: 4096,
+			});
+
+			expect(mockCounterInc).toHaveBeenCalledWith({ mode: 'fs', workflow_id: 'wf-1' }, 4096);
+		});
+
 		it('should increment writes counter with result:failure on failure', () => {
 			service.init();
 			const handler = getEventHandler('execution-data-write');
 
 			expect(handler).toBeDefined();
 
-			handler!({ mode: 'fs', durationMs: 10, success: false, jsonSizeBytes: 0 });
+			handler!({
+				mode: 'fs',
+				workflowId: 'wf-1',
+				durationMs: 10,
+				success: false,
+				jsonSizeBytes: 0,
+			});
 
 			expect(mockCounterInc).toHaveBeenCalledWith({ mode: 'fs', result: 'failure' }, 1);
 		});
@@ -313,9 +402,37 @@ describe('PrometheusExecutionDataMetricsService', () => {
 
 			expect(handler).toBeDefined();
 
-			handler!({ mode: 'db', durationMs: 10, success: false, jsonSizeBytes: 0 });
+			handler!({
+				mode: 'db',
+				workflowId: 'wf-1',
+				durationMs: 10,
+				success: false,
+				jsonSizeBytes: 0,
+			});
 
 			expect(mockHistogramObserve).not.toHaveBeenCalled();
 		});
+
+		it('should NOT increment write bytes counter on failure', () => {
+			service.init();
+			// Capture the handler before clearing mocks
+			const handler = getEventHandler('execution-data-write');
+			// Reset after seeding to isolate event handler behavior from seeding calls
+			vi.clearAllMocks();
+
+			expect(handler).toBeDefined();
+
+			handler!({
+				mode: 'fs',
+				workflowId: 'wf-1',
+				durationMs: 10,
+				success: false,
+				jsonSizeBytes: 0,
+			});
+
+			// Only the writes counter (mode + result) increments; the bytes counter stays untouched.
+			expect(mockCounterInc).toHaveBeenCalledTimes(1);
+			expect(mockCounterInc).toHaveBeenCalledWith({ mode: 'fs', result: 'failure' }, 1);
+		});
 	});
 });

packages/cli/src/metrics/prometheus/execution-data-metrics.service.ts

@@ -9,7 +9,7 @@ import type { PrometheusMetricsCollector } from './base';
 import { DURATION_BUCKETS_SECONDS, SIZE_BUCKETS_BYTES } from './constant';
 
 /**
- * Tracks execution data read/write counts, durations, and unreadable bundle counts.
+ * Tracks execution data read/write counts, durations, written bytes, and unreadable bundle counts.
  * Also exposes the active storage mode (db or fs) as a static gauge.
  */
 @Service()
@@ -72,6 +72,17 @@ export class PrometheusExecutionDataMetricsService implements PrometheusMetricsC
 			buckets: SIZE_BUCKETS_BYTES,
 		});
 
+		const writeBytesLabelNames = ['mode'];
+		if (this.config.includeWorkflowIdLabel) {
+			writeBytesLabelNames.push('workflow_id');
+		}
+
+		const writeBytesTotal = new promClient.Counter({
+			name: `${this.config.prefix}execution_data_write_bytes_total`,
+			help: 'Total execution data bytes written, by storage mode.',
+			labelNames: writeBytesLabelNames,
+		});
+
 		const storageModeGauge = new promClient.Gauge({
 			name: `${this.config.prefix}execution_data_storage_mode`,
 			help: 'Configured execution data storage mode (1 for the active mode, 0 otherwise).',
@@ -96,12 +107,21 @@ export class PrometheusExecutionDataMetricsService implements PrometheusMetricsC
 			},
 		);
 
-		this.eventService.on('execution-data-write', ({ mode, durationMs, success, jsonSizeBytes }) => {
-			writesTotal.inc({ mode, result: success ? 'success' : 'failure' }, 1);
-			if (success) {
-				writeDurationHistogram.observe({ mode }, durationMs / 1000);
-				writeSizeHistogram.observe({ mode }, jsonSizeBytes);
-			}
-		});
+		this.eventService.on(
+			'execution-data-write',
+			({ mode, workflowId, durationMs, success, jsonSizeBytes }) => {
+				writesTotal.inc({ mode, result: success ? 'success' : 'failure' }, 1);
+				if (success) {
+					writeDurationHistogram.observe({ mode }, durationMs / 1000);
+					writeSizeHistogram.observe({ mode }, jsonSizeBytes);
+
+					const labels: Record<string, string> = { mode };
+					if (this.config.includeWorkflowIdLabel) {
+						labels.workflow_id = workflowId;
+					}
+					writeBytesTotal.inc(labels, jsonSizeBytes);
+				}
+			},
+		);
 	}
 }