feat(core): Add metrics option to cache (#6846)

* add metrics to cache

* use events for metrics

* pr comments / broken test

* lint fix

* update the test

* improve tests

* Update packages/cli/src/config/schema.ts

* disable flaky test

* lint fix

---------

Co-authored-by: कारतोफ्फेलस्क्रिप्ट™ <aditya@netroy.in>
Co-authored-by: कारतोफ्फेलस्क्रिप्ट™ <netroy@users.noreply.github.com>

packages/cli/src/services/cache.service.ts

@@ -5,15 +5,23 @@ import type { MemoryCache } from 'cache-manager';
 import type { RedisCache } from 'cache-manager-ioredis-yet';
 import { jsonStringify } from 'n8n-workflow';
 import { getDefaultRedisClient, getRedisPrefix } from './redis/RedisServiceHelper';
+import EventEmitter from 'events';
 
 @Service()
-export class CacheService {
+export class CacheService extends EventEmitter {
 	/**
 	 * Keys and values:
 	 * - `'cache:workflow-owner:${workflowId}'`: `User`
 	 */
 	private cache: RedisCache | MemoryCache | undefined;
 
+	metricsCounterEvents = {
+		cacheHit: 'metrics.cache.hit',
+
+		cacheMiss: 'metrics.cache.miss',
+		cacheUpdate: 'metrics.cache.update',
+	};
+
 	isRedisCache(): boolean {
 		return (this.cache as RedisCache)?.store?.isCacheable !== undefined;
 	}
@@ -85,9 +93,12 @@ export class CacheService {
 		}
 		const value = await this.cache?.store.get(key);
 		if (value !== undefined) {
+			this.emit(this.metricsCounterEvents.cacheHit);
 			return value;
 		}
+		this.emit(this.metricsCounterEvents.cacheMiss);
 		if (options.refreshFunction) {
+			this.emit(this.metricsCounterEvents.cacheUpdate);
 			const refreshValue = await options.refreshFunction(key);
 			await this.set(key, refreshValue, options.refreshTtl);
 			return refreshValue;
@@ -124,8 +135,10 @@ export class CacheService {
 			values = keys.map(() => undefined);
 		}
 		if (!values.includes(undefined)) {
+			this.emit(this.metricsCounterEvents.cacheHit);
 			return values;
 		}
+		this.emit(this.metricsCounterEvents.cacheMiss);
 		if (options.refreshFunctionEach) {
 			for (let i = 0; i < keys.length; i++) {
 				if (values[i] === undefined) {
@@ -145,6 +158,7 @@ export class CacheService {
 			return values;
 		}
 		if (options.refreshFunctionMany) {
+			this.emit(this.metricsCounterEvents.cacheUpdate);
 			const refreshValues: unknown[] = await options.refreshFunctionMany(keys);
 			if (keys.length !== refreshValues.length) {
 				throw new Error('refreshFunctionMany must return the same number of values as keys');
@@ -195,7 +209,6 @@ export class CacheService {
 		if (values.length === 0) {
 			return;
 		}
-		// eslint-disable-next-line @typescript-eslint/naming-convention
 		const nonNullValues = values.filter(
 			([key, value]) => value !== undefined && value !== null && key && key.length > 0,
 		);

packages/cli/src/services/metrics.service.ts

@@ -0,0 +1,160 @@
+import config from '@/config';
+import { N8N_VERSION } from '@/constants';
+import type express from 'express';
+import promBundle from 'express-prom-bundle';
+import promClient, { type Counter } from 'prom-client';
+import semverParse from 'semver/functions/parse';
+import { Service } from 'typedi';
+import EventEmitter from 'events';
+import { LoggerProxy } from 'n8n-workflow';
+
+import { CacheService } from '@/services/cache.service';
+import type { EventMessageTypes } from '@/eventbus/EventMessageClasses';
+import {
+	METRICS_EVENT_NAME,
+	getLabelsForEvent,
+} from '@/eventbus/MessageEventBusDestination/Helpers.ee';
+import { eventBus } from '@/eventbus';
+
+@Service()
+export class MetricsService extends EventEmitter {
+	constructor(private readonly cacheService: CacheService) {
+		super();
+	}
+
+	counters: Record<string, Counter<string> | null> = {};
+
+	async configureMetrics(app: express.Application) {
+		promClient.register.clear(); // clear all metrics in case we call this a second time
+		this.setupDefaultMetrics();
+		this.setupN8nVersionMetric();
+		this.setupCacheMetrics();
+		this.setupMessageEventBusMetrics();
+		this.setupApiMetrics(app);
+		this.mountMetricsEndpoint(app);
+	}
+
+	private 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,
+			);
+		}
+	}
+
+	private setupDefaultMetrics() {
+		if (config.getEnv('endpoints.metrics.includeDefaultMetrics')) {
+			promClient.collectDefaultMetrics();
+		}
+	}
+
+	private 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);
+		}
+	}
+
+	mountMetricsEndpoint(app: express.Application) {
+		app.get('/metrics', async (req: express.Request, res: express.Response) => {
+			const metrics = await promClient.register.metrics();
+			res.setHeader('Content-Type', promClient.register.contentType);
+			res.send(metrics).end();
+		});
+	}
+
+	private setupCacheMetrics() {
+		if (!config.getEnv('endpoints.metrics.includeCacheMetrics')) {
+			return;
+		}
+		this.counters.cacheHitsTotal = new promClient.Counter({
+			name: config.getEnv('endpoints.metrics.prefix') + 'cache_hits_total',
+			help: 'Total number of cache hits.',
+			labelNames: ['cache'],
+		});
+		this.counters.cacheHitsTotal.inc(0);
+		this.cacheService.on(this.cacheService.metricsCounterEvents.cacheHit, (amount: number = 1) => {
+			this.counters.cacheHitsTotal?.inc(amount);
+		});
+
+		this.counters.cacheMissesTotal = new promClient.Counter({
+			name: config.getEnv('endpoints.metrics.prefix') + 'cache_misses_total',
+			help: 'Total number of cache misses.',
+			labelNames: ['cache'],
+		});
+		this.counters.cacheMissesTotal.inc(0);
+		this.cacheService.on(this.cacheService.metricsCounterEvents.cacheMiss, (amount: number = 1) => {
+			this.counters.cacheMissesTotal?.inc(amount);
+		});
+
+		this.counters.cacheUpdatesTotal = new promClient.Counter({
+			name: config.getEnv('endpoints.metrics.prefix') + 'cache_updates_total',
+			help: 'Total number of cache updates.',
+			labelNames: ['cache'],
+		});
+		this.counters.cacheUpdatesTotal.inc(0);
+		this.cacheService.on(
+			this.cacheService.metricsCounterEvents.cacheUpdate,
+			(amount: number = 1) => {
+				this.counters.cacheUpdatesTotal?.inc(amount);
+			},
+		);
+	}
+
+	private getCounterForEvent(event: EventMessageTypes): Counter<string> | null {
+		if (!promClient) return null;
+		if (!this.counters[event.eventName]) {
+			const prefix = config.getEnv('endpoints.metrics.prefix');
+			const metricName =
+				prefix + event.eventName.replace('n8n.', '').replace(/\./g, '_') + '_total';
+
+			if (!promClient.validateMetricName(metricName)) {
+				LoggerProxy.debug(`Invalid metric name: ${metricName}. Ignoring it!`);
+				this.counters[event.eventName] = null;
+				return null;
+			}
+
+			const counter = new promClient.Counter({
+				name: metricName,
+				help: `Total number of ${event.eventName} events.`,
+				labelNames: Object.keys(getLabelsForEvent(event)),
+			});
+			counter.inc(0);
+			this.counters[event.eventName] = counter;
+		}
+
+		return this.counters[event.eventName];
+	}
+
+	private setupMessageEventBusMetrics() {
+		if (!config.getEnv('endpoints.metrics.includeMessageEventBusMetrics')) {
+			return;
+		}
+		eventBus.on(METRICS_EVENT_NAME, (event: EventMessageTypes) => {
+			const counter = this.getCounterForEvent(event);
+			if (!counter) return;
+			counter.inc(1);
+		});
+	}
+}