feat(core): Add Prometheus metrics for n8n events and api invocations (experimental) (#5177)

* create prometheus metrics from events

* feat(core): Add more Prometheus metrics (experimental) (#5187)

* refactor(core): Add Prometheus labels to relevant metrics

* feat(core): Add more Prometheus metrics (experimental)

* add 'v' prefix to value of version label

Co-authored-by: Cornelius Suermann <cornelius@n8n.io>

packages/cli/package.json

@@ -136,6 +136,7 @@
     "dotenv": "^8.0.0",
     "express": "^4.16.4",
     "express-openapi-validator": "^4.13.6",
+    "express-prom-bundle": "^6.6.0",
     "fast-glob": "^3.2.5",
     "flatted": "^3.2.4",
     "google-timezones-json": "^1.0.2",

packages/cli/src/InternalHooks.ts

@@ -182,6 +182,7 @@ export class InternalHooksClass implements IInternalHooksClass {
 		workflow: IWorkflowBase,
 		nodeName: string,
 	): Promise<void> {
+		const nodeInWorkflow = workflow.nodes.find((node) => node.name === nodeName);
 		void eventBus.sendNodeEvent({
 			eventName: 'n8n.node.started',
 			payload: {
@@ -189,6 +190,7 @@ export class InternalHooksClass implements IInternalHooksClass {
 				nodeName,
 				workflowId: workflow.id?.toString(),
 				workflowName: workflow.name,
+				nodeType: nodeInWorkflow?.type,
 			},
 		});
 	}
@@ -198,6 +200,7 @@ export class InternalHooksClass implements IInternalHooksClass {
 		workflow: IWorkflowBase,
 		nodeName: string,
 	): Promise<void> {
+		const nodeInWorkflow = workflow.nodes.find((node) => node.name === nodeName);
 		void eventBus.sendNodeEvent({
 			eventName: 'n8n.node.finished',
 			payload: {
@@ -205,6 +208,7 @@ export class InternalHooksClass implements IInternalHooksClass {
 				nodeName,
 				workflowId: workflow.id?.toString(),
 				workflowName: workflow.name,
+				nodeType: nodeInWorkflow?.type,
 			},
 		});
 	}

packages/cli/src/Server.ts

@@ -73,7 +73,6 @@ import jwt from 'jsonwebtoken';
 import jwks from 'jwks-rsa';
 // @ts-ignore
 import timezones from 'google-timezones-json';
-import promClient, { Registry } from 'prom-client';
 import history from 'connect-history-api-fallback';
 
 import config from '@/config';
@@ -154,6 +153,7 @@ import { licenseController } from './license/license.controller';
 import { corsMiddleware } from './middlewares/cors';
 import { initEvents } from './events';
 import { AbstractServer } from './AbstractServer';
+import { configureMetrics } from './metrics';
 
 const exec = promisify(callbackExec);
 
@@ -321,15 +321,7 @@ class Server extends AbstractServer {
 	}
 
 	async configure(): Promise<void> {
-		const enableMetrics = config.getEnv('endpoints.metrics.enable');
-		let register: Registry;
-
-		if (enableMetrics) {
-			const prefix = config.getEnv('endpoints.metrics.prefix');
-			register = new promClient.Registry();
-			register.setDefaultLabels({ prefix });
-			promClient.collectDefaultMetrics({ register });
-		}
+		configureMetrics(this.app);
 
 		this.frontendSettings.isNpmAvailable = await exec('npm --version')
 			.then(() => true)
@@ -590,17 +582,6 @@ class Server extends AbstractServer {
 			this.app.use(`/${this.restEndpoint}/nodes`, nodesController);
 		}
 
-		// ----------------------------------------
-		// Metrics
-		// ----------------------------------------
-		if (enableMetrics) {
-			this.app.get('/metrics', async (req: express.Request, res: express.Response) => {
-				const response = await register.metrics();
-				res.setHeader('Content-Type', register.contentType);
-				ResponseHelper.sendSuccessResponse(res, response, true, 200);
-			});
-		}
-
 		// ----------------------------------------
 		// Workflow
 		// ----------------------------------------

packages/cli/src/config/schema.ts

@@ -573,7 +573,7 @@ export const schema = {
 				format: 'Boolean',
 				default: false,
 				env: 'N8N_METRICS',
-				doc: 'Enable metrics endpoint',
+				doc: 'Enable /metrics endpoint. Default: false',
 			},
 			prefix: {
 				format: String,
@@ -581,6 +581,54 @@ export const schema = {
 				env: 'N8N_METRICS_PREFIX',
 				doc: 'An optional prefix for metric names. Default: n8n_',
 			},
+			includeDefaultMetrics: {
+				format: Boolean,
+				default: true,
+				env: 'N8N_METRICS_INCLUDE_DEFAULT_METRICS',
+				doc: 'Whether to expose default system and node.js metrics. Default: true',
+			},
+			includeWorkflowIdLabel: {
+				format: Boolean,
+				default: false,
+				env: 'N8N_METRICS_INCLUDE_WORKFLOW_ID_LABEL',
+				doc: 'Whether to include a label for the workflow ID on workflow metrics. Default: false',
+			},
+			includeNodeTypeLabel: {
+				format: Boolean,
+				default: false,
+				env: 'N8N_METRICS_INCLUDE_NODE_TYPE_LABEL',
+				doc: 'Whether to include a label for the node type on node metrics. Default: false',
+			},
+			includeCredentialTypeLabel: {
+				format: Boolean,
+				default: false,
+				env: 'N8N_METRICS_INCLUDE_CREDENTIAL_TYPE_LABEL',
+				doc: 'Whether to include a label for the credential type on credential metrics. Default: false',
+			},
+			includeApiEndpoints: {
+				format: Boolean,
+				default: false,
+				env: 'N8N_METRICS_INCLUDE_API_ENDPOINTS',
+				doc: 'Whether to expose metrics for API endpoints. Default: false',
+			},
+			includeApiPathLabel: {
+				format: Boolean,
+				default: false,
+				env: 'N8N_METRICS_INCLUDE_API_PATH_LABEL',
+				doc: 'Whether to include a label for the path of API invocations. Default: false',
+			},
+			includeApiMethodLabel: {
+				format: Boolean,
+				default: false,
+				env: 'N8N_METRICS_INCLUDE_API_METHOD_LABEL',
+				doc: 'Whether to include a label for the HTTP method (GET, POST, ...) of API invocations. Default: false',
+			},
+			includeApiStatusCodeLabel: {
+				format: Boolean,
+				default: false,
+				env: 'N8N_METRICS_INCLUDE_API_STATUS_CODE_LABEL',
+				doc: 'Whether to include a label for the HTTP status code (200, 404, ...) of API invocations. Default: false',
+			},
 		},
 		rest: {
 			format: String,

packages/cli/src/eventbus/EventMessageClasses/EventMessageAudit.ts

@@ -35,6 +35,11 @@ export interface EventPayloadAudit extends AbstractEventPayload {
 	userEmail?: string;
 	firstName?: string;
 	lastName?: string;
+	credentialName?: string;
+	credentialType?: string;
+	credentialId?: string;
+	workflowId?: string;
+	workflowName?: string;
 }
 
 export interface EventMessageAuditOptions extends AbstractEventMessageOptions {

packages/cli/src/eventbus/EventMessageClasses/EventMessageNode.ts

@@ -11,6 +11,11 @@ export type EventNamesNodeType = typeof eventNamesNode[number];
 // --------------------------------------
 export interface EventPayloadNode extends AbstractEventPayload {
 	msg?: string;
+	executionId: string;
+	nodeName: string;
+	workflowId?: string;
+	workflowName: string;
+	nodeType?: string;
 }
 
 export interface EventMessageNodeOptions extends AbstractEventMessageOptions {

packages/cli/src/eventbus/MessageEventBus/MessageEventBus.ts

@@ -6,7 +6,10 @@ import { MessageEventBusLogWriter } from '../MessageEventBusWriter/MessageEventB
 import EventEmitter from 'events';
 import config from '@/config';
 import * as Db from '@/Db';
-import { messageEventBusDestinationFromDb } from '../MessageEventBusDestination/Helpers.ee';
+import {
+	messageEventBusDestinationFromDb,
+	incrementPrometheusMetric,
+} from '../MessageEventBusDestination/Helpers.ee';
 import uniqby from 'lodash.uniqby';
 import { EventMessageConfirmSource } from '../EventMessageClasses/EventMessageConfirm';
 import {
@@ -205,6 +208,10 @@ class MessageEventBus extends EventEmitter {
 	}
 
 	private async emitMessage(msg: EventMessageTypes) {
+		if (config.getEnv('endpoints.metrics.enable')) {
+			await incrementPrometheusMetric(msg);
+		}
+
 		// generic emit for external modules to capture events
 		// this is for internal use ONLY and not for use with custom destinations!
 		this.emit('message', msg);

packages/cli/src/eventbus/MessageEventBusDestination/Helpers.ee.ts

@@ -1,6 +1,13 @@
 /* eslint-disable import/no-cycle */
-import { MessageEventBusDestinationTypeNames } from 'n8n-workflow';
 import type { EventDestinations } from '@/databases/entities/MessageEventBusDestinationEntity';
+import { promClient } from '@/metrics';
+import {
+	EventMessageTypeNames,
+	LoggerProxy,
+	MessageEventBusDestinationTypeNames,
+} from 'n8n-workflow';
+import config from '../../config';
+import type { EventMessageTypes } from '../EventMessageClasses';
 import type { MessageEventBusDestination } from './MessageEventBusDestination.ee';
 import { MessageEventBusDestinationSentry } from './MessageEventBusDestinationSentry.ee';
 import { MessageEventBusDestinationSyslog } from './MessageEventBusDestinationSyslog.ee';
@@ -24,3 +31,85 @@ export function messageEventBusDestinationFromDb(
 	}
 	return null;
 }
+
+const prometheusCounters: Record<string, promClient.Counter<string> | null> = {};
+
+function getMetricNameForEvent(event: EventMessageTypes): string {
+	const prefix = config.getEnv('endpoints.metrics.prefix');
+	return prefix + event.eventName.replace('n8n.', '').replace(/\./g, '_') + '_total';
+}
+
+function getLabelValueForNode(nodeType: string): string {
+	return nodeType.replace('n8n-nodes-', '').replace(/\./g, '_');
+}
+
+function getLabelValueForCredential(credentialType: string): string {
+	return credentialType.replace(/\./g, '_');
+}
+
+function getLabelsForEvent(event: EventMessageTypes): Record<string, string> {
+	switch (event.__type) {
+		case EventMessageTypeNames.audit:
+			if (event.eventName.startsWith('n8n.audit.user.credentials')) {
+				return config.getEnv('endpoints.metrics.includeCredentialTypeLabel')
+					? {
+							credential_type: getLabelValueForCredential(
+								event.payload.credentialType ?? 'unknown',
+							),
+					  }
+					: {};
+			}
+
+			if (event.eventName.startsWith('n8n.audit.workflow')) {
+				return config.getEnv('endpoints.metrics.includeWorkflowIdLabel')
+					? { workflow_id: event.payload.workflowId?.toString() ?? 'unknown' }
+					: {};
+			}
+			break;
+
+		case EventMessageTypeNames.node:
+			return config.getEnv('endpoints.metrics.includeNodeTypeLabel')
+				? { node_type: getLabelValueForNode(event.payload.nodeType ?? 'unknown') }
+				: {};
+
+		case EventMessageTypeNames.workflow:
+			return config.getEnv('endpoints.metrics.includeWorkflowIdLabel')
+				? { workflow_id: event.payload.workflowId?.toString() ?? 'unknown' }
+				: {};
+	}
+
+	return {};
+}
+
+function getCounterSingletonForEvent(event: EventMessageTypes) {
+	if (!prometheusCounters[event.eventName]) {
+		const metricName = getMetricNameForEvent(event);
+
+		if (!promClient.validateMetricName(metricName)) {
+			LoggerProxy.debug(`Invalid metric name: ${metricName}. Ignoring it!`);
+			prometheusCounters[event.eventName] = null;
+			return null;
+		}
+
+		const counter = new promClient.Counter({
+			name: metricName,
+			help: `Total number of ${event.eventName} events.`,
+			labelNames: Object.keys(getLabelsForEvent(event)),
+		});
+
+		promClient.register.registerMetric(counter);
+		prometheusCounters[event.eventName] = counter;
+	}
+
+	return prometheusCounters[event.eventName];
+}
+
+export async function incrementPrometheusMetric(event: EventMessageTypes): Promise<void> {
+	const counter = getCounterSingletonForEvent(event);
+
+	if (!counter) {
+		return;
+	}
+
+	counter.inc(getLabelsForEvent(event));
+}

packages/cli/src/eventbus/eventBusRoutes.ts

@@ -32,6 +32,7 @@ import {
 } from 'n8n-workflow';
 import { User } from '../databases/entities/User';
 import * as ResponseHelper from '@/ResponseHelper';
+import { EventMessageNode, EventMessageNodeOptions } from './EventMessageClasses/EventMessageNode';
 
 export const eventBusRouter = express.Router();
 
@@ -116,6 +117,9 @@ eventBusRouter.post(
 				case EventMessageTypeNames.audit:
 					msg = new EventMessageAudit(req.body as EventMessageAuditOptions);
 					break;
+				case EventMessageTypeNames.node:
+					msg = new EventMessageNode(req.body as EventMessageNodeOptions);
+					break;
 				case EventMessageTypeNames.generic:
 				default:
 					msg = new EventMessageGeneric(req.body);

packages/cli/src/metrics/index.ts

@@ -0,0 +1,71 @@
+/* eslint-disable @typescript-eslint/no-use-before-define */
+import config from '@/config';
+import { N8N_VERSION } from '@/constants';
+import * as ResponseHelper from '@/ResponseHelper';
+import express from 'express';
+import promBundle from 'express-prom-bundle';
+import promClient from 'prom-client';
+import semverParse from 'semver/functions/parse';
+
+export { promClient };
+
+export function configureMetrics(app: express.Application) {
+	if (!config.getEnv('endpoints.metrics.enable')) {
+		return;
+	}
+
+	setupDefaultMetrics();
+	setupN8nVersionMetric();
+	setupApiMetrics(app);
+	mountMetricsEndpoint(app);
+}
+
+function setupN8nVersionMetric() {
+	const n8nVersion = semverParse(N8N_VERSION || '0.0.0');
+
+	if (n8nVersion) {
+		const versionGauge = new promClient.Gauge({
+			name: config.getEnv('endpoints.metrics.prefix') + 'version_info',
+			help: 'n8n version info.',
+			labelNames: ['version', 'major', 'minor', 'patch'],
+		});
+
+		versionGauge.set(
+			{
+				version: 'v' + n8nVersion.version,
+				major: n8nVersion.major,
+				minor: n8nVersion.minor,
+				patch: n8nVersion.patch,
+			},
+			1,
+		);
+	}
+}
+
+function setupDefaultMetrics() {
+	if (config.getEnv('endpoints.metrics.includeDefaultMetrics')) {
+		promClient.collectDefaultMetrics();
+	}
+}
+
+function setupApiMetrics(app: express.Application) {
+	if (config.getEnv('endpoints.metrics.includeApiEndpoints')) {
+		const metricsMiddleware = promBundle({
+			autoregister: false,
+			includeUp: false,
+			includePath: config.getEnv('endpoints.metrics.includeApiPathLabel'),
+			includeMethod: config.getEnv('endpoints.metrics.includeApiMethodLabel'),
+			includeStatusCode: config.getEnv('endpoints.metrics.includeApiStatusCodeLabel'),
+		});
+
+		app.use(['/rest/', '/webhook/', 'webhook-test/', '/api/'], metricsMiddleware);
+	}
+}
+
+function mountMetricsEndpoint(app: express.Application) {
+	app.get('/metrics', async (req: express.Request, res: express.Response) => {
+		const response = await promClient.register.metrics();
+		res.setHeader('Content-Type', promClient.register.contentType);
+		ResponseHelper.sendSuccessResponse(res, response, true, 200);
+	});
+}