mirror of
https://github.com/n8n-io/n8n.git
synced 2026-10-12 02:19:43 +00:00
feat(core): Add database connection pool metrics (#32670)
This commit is contained in:
@@ -107,6 +107,10 @@ export class PrometheusMetricsConfig {
|
||||
/** How often (in seconds) to refresh the workflow info metric cache. */
|
||||
@Env('N8N_METRICS_WORKFLOW_INFO_METRIC_INTERVAL')
|
||||
workflowInfoMetricInterval: number = 60;
|
||||
|
||||
/** Whether to include metrics for the database connection pool (size, usage, wait queue, acquire latency). */
|
||||
@Env('N8N_METRICS_INCLUDE_DB_POOL_METRICS')
|
||||
includeDbPoolMetrics: boolean = false;
|
||||
}
|
||||
|
||||
@Config
|
||||
|
||||
@@ -253,6 +253,7 @@ describe('GlobalConfig', () => {
|
||||
includeFormMetrics: false,
|
||||
includeWorkflowInfoMetrics: false,
|
||||
workflowInfoMetricInterval: 60,
|
||||
includeDbPoolMetrics: false,
|
||||
},
|
||||
additionalNonUIRoutes: '',
|
||||
disableProductionWebhooksOnMainProcess: false,
|
||||
|
||||
@@ -2,6 +2,9 @@
|
||||
* Convert time from any time unit to any other unit
|
||||
*/
|
||||
export const Time = {
|
||||
nanoseconds: {
|
||||
toSeconds: 1 / 1_000_000_000,
|
||||
},
|
||||
milliseconds: {
|
||||
toHours: 1 / (60 * 60 * 1000),
|
||||
toMinutes: 1 / (60 * 1000),
|
||||
|
||||
+2
@@ -6,6 +6,7 @@ import type { ErrorReporter } from 'n8n-core';
|
||||
import { getContainerRuntimeClient } from 'testcontainers';
|
||||
import { mock } from 'vitest-mock-extended';
|
||||
|
||||
import type { DbConnectionMetrics } from '../db-connection-metrics';
|
||||
import { DbConnectionMonitor } from '../db-connection-monitor';
|
||||
|
||||
/**
|
||||
@@ -165,6 +166,7 @@ describe('DbConnectionMonitor recovery against real Postgres', () => {
|
||||
buildDatabaseConfig(),
|
||||
mock<Logger>(),
|
||||
mock<ErrorReporter>(),
|
||||
mock<DbConnectionMetrics>(),
|
||||
);
|
||||
// start() installs the obtainMasterConnection wrapper on the live driver.
|
||||
// Under vitest (NODE_ENV=test) it does not schedule background pings, so the
|
||||
|
||||
@@ -8,6 +8,7 @@ import { setTimeout as setTimeoutP } from 'timers/promises';
|
||||
import type { Mock, MockedFunction } from 'vitest';
|
||||
import { mock, mockDeep } from 'vitest-mock-extended';
|
||||
|
||||
import { DbConnectionMetrics } from '../db-connection-metrics';
|
||||
import { DbConnectionMonitor } from '../db-connection-monitor';
|
||||
|
||||
// The monitor uses `setTimeout` from `timers/promises` for recovery backoff.
|
||||
@@ -33,6 +34,7 @@ describe('DbConnectionMonitor', () => {
|
||||
connectionAcquisitionTimeoutMs: 30_000,
|
||||
});
|
||||
const logger = mock<Logger>();
|
||||
const dbConnectionMetrics = mock<DbConnectionMetrics>();
|
||||
const dataSource = mockDeep<DataSource>({ options: { type: 'postgres' } });
|
||||
|
||||
beforeEach(() => {
|
||||
@@ -47,6 +49,7 @@ describe('DbConnectionMonitor', () => {
|
||||
databaseConfig,
|
||||
logger,
|
||||
errorReporter,
|
||||
dbConnectionMetrics,
|
||||
);
|
||||
});
|
||||
|
||||
@@ -249,6 +252,7 @@ describe('DbConnectionMonitor', () => {
|
||||
mock<DatabaseConfig>({ pingIntervalSeconds: 1, pingTimeoutMs: 5_000 }),
|
||||
logger,
|
||||
errorReporter,
|
||||
dbConnectionMetrics,
|
||||
);
|
||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||||
const pingSpy = vi.spyOn(scheduledMonitor as any, 'ping');
|
||||
@@ -272,6 +276,7 @@ describe('DbConnectionMonitor', () => {
|
||||
mock<DatabaseConfig>({ pingIntervalSeconds: 1, pingTimeoutMs: 5_000 }),
|
||||
logger,
|
||||
errorReporter,
|
||||
dbConnectionMetrics,
|
||||
);
|
||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||||
const pingSpy = vi.spyOn(scheduledMonitor as any, 'ping');
|
||||
@@ -435,6 +440,7 @@ describe('DbConnectionMonitor', () => {
|
||||
misconfigured,
|
||||
logger,
|
||||
errorReporter,
|
||||
dbConnectionMetrics,
|
||||
);
|
||||
|
||||
// @ts-expect-error readonly property
|
||||
@@ -635,6 +641,7 @@ describe('DbConnectionMonitor', () => {
|
||||
misconfigured,
|
||||
logger,
|
||||
errorReporter,
|
||||
dbConnectionMetrics,
|
||||
);
|
||||
|
||||
misconfiguredMonitor.start();
|
||||
@@ -736,6 +743,7 @@ describe('DbConnectionMonitor', () => {
|
||||
databaseConfig,
|
||||
logger,
|
||||
errorReporter,
|
||||
dbConnectionMetrics,
|
||||
);
|
||||
|
||||
sqliteMonitor.start();
|
||||
@@ -922,6 +930,7 @@ describe('DbConnectionMonitor', () => {
|
||||
}),
|
||||
logger,
|
||||
errorReporter,
|
||||
dbConnectionMetrics,
|
||||
);
|
||||
const noTimeoutInternals = noTimeoutMonitor as unknown as {
|
||||
acquireConnection: (original: () => Promise<unknown>) => Promise<unknown>;
|
||||
@@ -1018,6 +1027,7 @@ describe('DbConnectionMonitor', () => {
|
||||
databaseConfig,
|
||||
logger,
|
||||
errorReporter,
|
||||
dbConnectionMetrics,
|
||||
);
|
||||
// @ts-expect-error readonly property
|
||||
dataSource.isInitialized = true;
|
||||
@@ -1048,4 +1058,47 @@ describe('DbConnectionMonitor', () => {
|
||||
expect(onConnectedChange).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
});
|
||||
|
||||
describe('acquire latency', () => {
|
||||
const buildPostgresMonitor = (metrics: DbConnectionMetrics, obtain: () => Promise<unknown>) => {
|
||||
const driver = { obtainMasterConnection: vi.fn(obtain), master: { on: vi.fn() } };
|
||||
const ds = {
|
||||
options: { type: 'postgres' },
|
||||
isInitialized: true,
|
||||
driver,
|
||||
} as unknown as DataSource;
|
||||
const freshMonitor = new DbConnectionMonitor(
|
||||
ds,
|
||||
vi.fn(),
|
||||
databaseConfig,
|
||||
logger,
|
||||
errorReporter,
|
||||
metrics,
|
||||
);
|
||||
// start() installs the obtainMasterConnection wrapper on the live driver.
|
||||
freshMonitor.start();
|
||||
return { driver };
|
||||
};
|
||||
|
||||
it('records the acquisition duration through the observer', async () => {
|
||||
const metrics = new DbConnectionMetrics();
|
||||
const observed: number[] = [];
|
||||
metrics.acquireDurationObserver = (seconds) => observed.push(seconds);
|
||||
|
||||
const { driver } = buildPostgresMonitor(metrics, async () => 'connection');
|
||||
|
||||
const result = await driver.obtainMasterConnection();
|
||||
|
||||
expect(result).toBe('connection');
|
||||
expect(observed).toHaveLength(1);
|
||||
expect(observed[0]).toBeGreaterThanOrEqual(0);
|
||||
});
|
||||
|
||||
it('acquires without timing when no observer is registered', async () => {
|
||||
const metrics = new DbConnectionMetrics(); // observer left undefined
|
||||
const { driver } = buildPostgresMonitor(metrics, async () => 'connection');
|
||||
|
||||
await expect(driver.obtainMasterConnection()).resolves.toBe('connection');
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
@@ -10,6 +10,7 @@ import { mock, mockDeep } from 'vitest-mock-extended';
|
||||
import * as migrationHelper from '../../migrations/migration-helpers';
|
||||
import type { Migration } from '../../migrations/migration-types';
|
||||
import { DbConnection } from '../db-connection';
|
||||
import type { DbConnectionMetrics } from '../db-connection-metrics';
|
||||
import { DbConnectionMonitor } from '../db-connection-monitor';
|
||||
import type { DbConnectionOptions } from '../db-connection-options';
|
||||
|
||||
@@ -28,6 +29,7 @@ describe('DbConnection', () => {
|
||||
const errorReporter = mock<ErrorReporter>();
|
||||
const databaseConfig = mock<DatabaseConfig>();
|
||||
const logger = mock<Logger>();
|
||||
const dbConnectionMetrics = mock<DbConnectionMetrics>();
|
||||
const dataSource = mockDeep<DataSource>({ options: { migrations } });
|
||||
const connectionOptions = mockDeep<DbConnectionOptions>();
|
||||
const postgresOptions: DataSourceOptions = {
|
||||
@@ -53,7 +55,13 @@ describe('DbConnection', () => {
|
||||
return dataSource;
|
||||
});
|
||||
|
||||
dbConnection = new DbConnection(errorReporter, connectionOptions, databaseConfig, logger);
|
||||
dbConnection = new DbConnection(
|
||||
errorReporter,
|
||||
connectionOptions,
|
||||
databaseConfig,
|
||||
logger,
|
||||
dbConnectionMetrics,
|
||||
);
|
||||
});
|
||||
|
||||
describe('init', () => {
|
||||
|
||||
@@ -0,0 +1,42 @@
|
||||
import type { DataSource } from '@n8n/typeorm';
|
||||
|
||||
import { readPoolStats } from '../db-pool-stats';
|
||||
|
||||
describe('readPoolStats', () => {
|
||||
it('reads Postgres pool counters with active = total - idle', () => {
|
||||
const dataSource = {
|
||||
options: { type: 'postgres' },
|
||||
driver: { master: { totalCount: 5, idleCount: 2, waitingCount: 3 } },
|
||||
} as unknown as DataSource;
|
||||
|
||||
expect(readPoolStats(dataSource)).toEqual({ active: 3, idle: 2, waiting: 3 });
|
||||
});
|
||||
|
||||
it('returns undefined when the Postgres pool is absent (mid-recovery)', () => {
|
||||
const dataSource = {
|
||||
options: { type: 'postgres' },
|
||||
driver: { master: undefined },
|
||||
} as unknown as DataSource;
|
||||
|
||||
expect(readPoolStats(dataSource)).toBeUndefined();
|
||||
});
|
||||
|
||||
it('reads sqlite-pooled stats via getPoolStats', () => {
|
||||
const stats = { active: 1, idle: 4, waiting: 0 };
|
||||
const dataSource = {
|
||||
options: { type: 'sqlite-pooled' },
|
||||
driver: { getPoolStats: () => stats },
|
||||
} as unknown as DataSource;
|
||||
|
||||
expect(readPoolStats(dataSource)).toEqual(stats);
|
||||
});
|
||||
|
||||
it('returns undefined for an unsupported driver type', () => {
|
||||
const dataSource = {
|
||||
options: { type: 'sqlite' },
|
||||
driver: {},
|
||||
} as unknown as DataSource;
|
||||
|
||||
expect(readPoolStats(dataSource)).toBeUndefined();
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,6 @@
|
||||
import { Service } from '@n8n/di';
|
||||
|
||||
@Service()
|
||||
export class DbConnectionMetrics {
|
||||
acquireDurationObserver?: (seconds: number) => void;
|
||||
}
|
||||
@@ -7,6 +7,8 @@ import type { ErrorReporter } from 'n8n-core';
|
||||
import { ensureError, OperationalError } from 'n8n-workflow';
|
||||
import { setTimeout as setTimeoutP } from 'timers/promises';
|
||||
|
||||
import type { DbConnectionMetrics } from './db-connection-metrics';
|
||||
|
||||
/** The chokepoint every TypeORM query funnels through to acquire a master connection. */
|
||||
type ObtainMasterConnection = PostgresDriver['obtainMasterConnection'];
|
||||
|
||||
@@ -76,6 +78,7 @@ export class DbConnectionMonitor {
|
||||
private readonly databaseConfig: DatabaseConfig,
|
||||
private readonly logger: Logger,
|
||||
private readonly errorReporter: ErrorReporter,
|
||||
private readonly dbConnectionMetrics: DbConnectionMetrics,
|
||||
initialConnected = true,
|
||||
) {
|
||||
this.connected = initialConnected;
|
||||
@@ -328,7 +331,7 @@ export class DbConnectionMonitor {
|
||||
await this.awaitRecovery();
|
||||
|
||||
try {
|
||||
return await original();
|
||||
return await this.timeAcquisition(original);
|
||||
} catch (error) {
|
||||
if (!this.isRecoverableConnectionError(error)) {
|
||||
throw error;
|
||||
@@ -346,10 +349,26 @@ export class DbConnectionMonitor {
|
||||
await this.awaitRecovery();
|
||||
// `original` may be bound to the previous (destroyed) driver.
|
||||
// Prefer the live one refreshed by the latest wrapConnectionAcquisition().
|
||||
return await (this.liveObtainMasterConnection ?? original)();
|
||||
return await this.timeAcquisition(this.liveObtainMasterConnection ?? original);
|
||||
}
|
||||
}
|
||||
|
||||
private async timeAcquisition(acquire: ObtainMasterConnection) {
|
||||
const observer = this.dbConnectionMetrics.acquireDurationObserver;
|
||||
if (!observer) return await acquire();
|
||||
|
||||
const start = process.hrtime.bigint();
|
||||
const connection = await acquire();
|
||||
const elapsedSeconds = Number(process.hrtime.bigint() - start) * Time.nanoseconds.toSeconds;
|
||||
try {
|
||||
observer(elapsedSeconds);
|
||||
} catch (error) {
|
||||
// Metrics must never break or leak a pooled connection, but report so it isn't silent.
|
||||
this.errorReporter.error(ensureError(error));
|
||||
}
|
||||
return connection;
|
||||
}
|
||||
|
||||
/**
|
||||
* Waits out an in-progress recovery before a connection is acquired, bounded by `connectionAcquisitionTimeoutMs`.
|
||||
* Resolves immediately when no recovery is pending.
|
||||
|
||||
@@ -6,8 +6,10 @@ import { DataSource } from '@n8n/typeorm';
|
||||
import { ErrorReporter } from 'n8n-core';
|
||||
import { DbConnectionTimeoutError, ensureError } from 'n8n-workflow';
|
||||
|
||||
import { DbConnectionMetrics } from './db-connection-metrics';
|
||||
import { DbConnectionMonitor } from './db-connection-monitor';
|
||||
import { DbConnectionOptions } from './db-connection-options';
|
||||
import { readPoolStats, type DbPoolStats } from './db-pool-stats';
|
||||
import { wrapMigration } from '../migrations/migration-helpers';
|
||||
import type { Migration } from '../migrations/migration-types';
|
||||
|
||||
@@ -32,6 +34,7 @@ export class DbConnection {
|
||||
private readonly connectionOptions: DbConnectionOptions,
|
||||
private readonly databaseConfig: DatabaseConfig,
|
||||
private readonly logger: Logger,
|
||||
private readonly dbConnectionMetrics: DbConnectionMetrics,
|
||||
) {
|
||||
this.dataSource = new DataSource(this.options);
|
||||
Container.set(DataSource, this.dataSource);
|
||||
@@ -42,6 +45,10 @@ export class DbConnection {
|
||||
return this.connectionOptions.getOptions();
|
||||
}
|
||||
|
||||
getPoolStats(): DbPoolStats | undefined {
|
||||
return readPoolStats(this.dataSource);
|
||||
}
|
||||
|
||||
async init(): Promise<void> {
|
||||
const { connectionState, options } = this;
|
||||
if (connectionState.connected) return;
|
||||
@@ -76,6 +83,7 @@ export class DbConnection {
|
||||
this.databaseConfig,
|
||||
this.logger,
|
||||
this.errorReporter,
|
||||
this.dbConnectionMetrics,
|
||||
connectionState.connected,
|
||||
);
|
||||
this.monitor.start();
|
||||
|
||||
@@ -0,0 +1,40 @@
|
||||
import type { DataSource } from '@n8n/typeorm';
|
||||
import type { PostgresDriver } from '@n8n/typeorm/driver/postgres/PostgresDriver';
|
||||
import type { SqliteReadWriteDriver } from '@n8n/typeorm/driver/sqlite-pooled/SqliteReadWriteDriver';
|
||||
|
||||
/** Point-in-time view of a connection pool, normalized across drivers. */
|
||||
export interface DbPoolStats {
|
||||
/** Connections currently checked out and running a query. */
|
||||
active: number;
|
||||
/** Connections sitting idle in the pool, ready for use. */
|
||||
idle: number;
|
||||
/** Callers queued waiting for a connection because the pool is saturated. */
|
||||
waiting: number;
|
||||
}
|
||||
|
||||
/**
|
||||
* Reads point-in-time pool stats from the active driver, or `undefined` when
|
||||
* unavailable. Kept in `@n8n/db` so driver/TypeORM internals don't leak into
|
||||
* consumers (e.g. the metrics layer).
|
||||
*
|
||||
* For SQLite, `getPoolStats` combines the read-only pool and the single write
|
||||
* connection (which serializes writes and is the usual bottleneck), so the
|
||||
* numbers reflect total pool pressure.
|
||||
*/
|
||||
export function readPoolStats(dataSource: DataSource): DbPoolStats | undefined {
|
||||
switch (dataSource.options.type) {
|
||||
case 'postgres': {
|
||||
const pool = (dataSource.driver as PostgresDriver).master;
|
||||
if (!pool) return undefined; // briefly absent mid-recovery
|
||||
return {
|
||||
active: pool.totalCount - pool.idleCount,
|
||||
idle: pool.idleCount,
|
||||
waiting: pool.waitingCount,
|
||||
};
|
||||
}
|
||||
case 'sqlite-pooled':
|
||||
return (dataSource.driver as SqliteReadWriteDriver).getPoolStats();
|
||||
default:
|
||||
return undefined;
|
||||
}
|
||||
}
|
||||
@@ -37,7 +37,9 @@ export { postgresMigrations } from './migrations/postgresdb';
|
||||
export { wrapMigration } from './migrations/migration-helpers';
|
||||
export * from './migrations/migration-types';
|
||||
export { DbConnection } from './connection/db-connection';
|
||||
export { DbConnectionMetrics } from './connection/db-connection-metrics';
|
||||
export { DbConnectionOptions } from './connection/db-connection-options';
|
||||
export type { DbPoolStats } from './connection/db-pool-stats';
|
||||
|
||||
export { AuthRolesService } from './services/auth.roles.service';
|
||||
export { DbLock, DbLockService } from './services/db-lock.service';
|
||||
|
||||
@@ -6,6 +6,7 @@ import promClient from 'prom-client';
|
||||
|
||||
import type { PrometheusActiveWorkflowMetricsService } from '../prometheus/active-workflow-metrics.service';
|
||||
import type { PrometheusCacheMetricsService } from '../prometheus/cache-metrics.service';
|
||||
import type { PrometheusDbPoolMetricsService } from '../prometheus/db-pool-metrics.service';
|
||||
import type { PrometheusDefaultMetricsService } from '../prometheus/default-metrics.service';
|
||||
import type { PrometheusDnsCacheMetricsService } from '../prometheus/dns-cache-metrics.service';
|
||||
import type { PrometheusEventBusMetricsService } from '../prometheus/event-bus-metrics.service';
|
||||
@@ -48,6 +49,7 @@ describe('PrometheusMetricsService', () => {
|
||||
let webhook: jest.Mocked<PrometheusWebhookAndFormMetricsService>;
|
||||
let workflowInfo: jest.Mocked<PrometheusWorkflowInfoMetricsService>;
|
||||
let instanceAi: jest.Mocked<PrometheusInstanceAiMetricsService>;
|
||||
let dbPool: jest.Mocked<PrometheusDbPoolMetricsService>;
|
||||
|
||||
let service: PrometheusMetricsService;
|
||||
|
||||
@@ -72,6 +74,7 @@ describe('PrometheusMetricsService', () => {
|
||||
webhook,
|
||||
workflowInfo,
|
||||
instanceAi,
|
||||
dbPool,
|
||||
);
|
||||
|
||||
beforeEach(() => {
|
||||
@@ -103,6 +106,7 @@ describe('PrometheusMetricsService', () => {
|
||||
webhook = mock<PrometheusWebhookAndFormMetricsService>({ enabled: true });
|
||||
workflowInfo = mock<PrometheusWorkflowInfoMetricsService>({ enabled: true });
|
||||
instanceAi = mock<PrometheusInstanceAiMetricsService>({ enabled: true });
|
||||
dbPool = mock<PrometheusDbPoolMetricsService>({ enabled: true });
|
||||
|
||||
service = buildService();
|
||||
});
|
||||
@@ -133,6 +137,7 @@ describe('PrometheusMetricsService', () => {
|
||||
expect(webhook.init).toHaveBeenCalledWith(app);
|
||||
expect(workflowInfo.init).toHaveBeenCalledWith(app);
|
||||
expect(instanceAi.init).toHaveBeenCalledWith(app);
|
||||
expect(dbPool.init).toHaveBeenCalledWith(app);
|
||||
});
|
||||
|
||||
it('should NOT call init on disabled collectors', () => {
|
||||
|
||||
@@ -0,0 +1,184 @@
|
||||
import type { DatabaseConfig, PrometheusMetricsConfig } from '@n8n/config';
|
||||
import { DbConnectionMetrics, type DbConnection, type DbPoolStats } from '@n8n/db';
|
||||
import promClient from 'prom-client';
|
||||
|
||||
import { PrometheusDbPoolMetricsService } from '../db-pool-metrics.service';
|
||||
|
||||
jest.mock('prom-client');
|
||||
|
||||
type GaugeConfig = {
|
||||
name: string;
|
||||
help: string;
|
||||
collect?: (this: { set: jest.Mock }) => void;
|
||||
};
|
||||
|
||||
const buildConfig = (overrides: Partial<PrometheusMetricsConfig> = {}) =>
|
||||
({ prefix: 'n8n_', includeDbPoolMetrics: true, ...overrides }) as PrometheusMetricsConfig;
|
||||
|
||||
const buildDatabaseConfig = (overrides: Partial<DatabaseConfig> = {}) =>
|
||||
({
|
||||
type: 'postgresdb',
|
||||
postgresdb: { poolSize: 2 },
|
||||
sqlite: { poolSize: 3 },
|
||||
...overrides,
|
||||
}) as unknown as DatabaseConfig;
|
||||
|
||||
const buildDbConnection = (stats: DbPoolStats | undefined) =>
|
||||
({ getPoolStats: () => stats }) as unknown as DbConnection;
|
||||
|
||||
describe('PrometheusDbPoolMetricsService', () => {
|
||||
let mockGaugeSet: jest.Mock;
|
||||
let mockHistogramObserve: jest.Mock;
|
||||
|
||||
beforeEach(() => {
|
||||
mockGaugeSet = jest.fn();
|
||||
mockHistogramObserve = jest.fn();
|
||||
promClient.Gauge.prototype.set = mockGaugeSet;
|
||||
promClient.Histogram.prototype.observe = mockHistogramObserve;
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
jest.clearAllMocks();
|
||||
});
|
||||
|
||||
const gaugeConfigs = () =>
|
||||
(promClient.Gauge as unknown as jest.Mock).mock.calls.map((c) => c[0] as GaugeConfig);
|
||||
|
||||
const findGauge = (name: string) => gaugeConfigs().find((g) => g.name === name);
|
||||
|
||||
const runCollect = (name: string) => {
|
||||
const setSpy = jest.fn();
|
||||
findGauge(name)?.collect?.call({ set: setSpy });
|
||||
return setSpy;
|
||||
};
|
||||
|
||||
describe('enabled', () => {
|
||||
it('is true when includeDbPoolMetrics is set', () => {
|
||||
const service = new PrometheusDbPoolMetricsService(
|
||||
buildConfig({ includeDbPoolMetrics: true }),
|
||||
buildDatabaseConfig(),
|
||||
buildDbConnection(undefined),
|
||||
new DbConnectionMetrics(),
|
||||
);
|
||||
expect(service.enabled).toBe(true);
|
||||
});
|
||||
|
||||
it('is false when includeDbPoolMetrics is unset', () => {
|
||||
const service = new PrometheusDbPoolMetricsService(
|
||||
buildConfig({ includeDbPoolMetrics: false }),
|
||||
buildDatabaseConfig(),
|
||||
buildDbConnection(undefined),
|
||||
new DbConnectionMetrics(),
|
||||
);
|
||||
expect(service.enabled).toBe(false);
|
||||
});
|
||||
});
|
||||
|
||||
describe('init', () => {
|
||||
it('creates the pool gauges and the acquire histogram with prefixed names', () => {
|
||||
new PrometheusDbPoolMetricsService(
|
||||
buildConfig(),
|
||||
buildDatabaseConfig(),
|
||||
buildDbConnection(undefined),
|
||||
new DbConnectionMetrics(),
|
||||
).init();
|
||||
|
||||
const names = gaugeConfigs().map((g) => g.name);
|
||||
expect(names).toEqual(
|
||||
expect.arrayContaining([
|
||||
'n8n_db_pool_connections_active',
|
||||
'n8n_db_pool_connections_idle',
|
||||
'n8n_db_pool_requests_pending',
|
||||
'n8n_db_pool_connections_max',
|
||||
]),
|
||||
);
|
||||
expect(promClient.Histogram).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ name: 'n8n_db_pool_acquire_seconds' }),
|
||||
);
|
||||
});
|
||||
|
||||
it('applies a custom prefix', () => {
|
||||
new PrometheusDbPoolMetricsService(
|
||||
buildConfig({ prefix: 'custom_' }),
|
||||
buildDatabaseConfig(),
|
||||
buildDbConnection(undefined),
|
||||
new DbConnectionMetrics(),
|
||||
).init();
|
||||
|
||||
expect(gaugeConfigs().map((g) => g.name)).toContain('custom_db_pool_connections_active');
|
||||
});
|
||||
|
||||
it('sets the max gauge from the configured Postgres pool size', () => {
|
||||
new PrometheusDbPoolMetricsService(
|
||||
buildConfig(),
|
||||
buildDatabaseConfig({ type: 'postgresdb' }),
|
||||
buildDbConnection(undefined),
|
||||
new DbConnectionMetrics(),
|
||||
).init();
|
||||
|
||||
expect(mockGaugeSet).toHaveBeenCalledWith(2);
|
||||
});
|
||||
|
||||
it('sets the max gauge from the configured SQLite pool size plus the write connection', () => {
|
||||
new PrometheusDbPoolMetricsService(
|
||||
buildConfig(),
|
||||
buildDatabaseConfig({ type: 'sqlite' }),
|
||||
buildDbConnection(undefined),
|
||||
new DbConnectionMetrics(),
|
||||
).init();
|
||||
|
||||
expect(mockGaugeSet).toHaveBeenCalledWith(4);
|
||||
});
|
||||
});
|
||||
|
||||
describe('pool gauges', () => {
|
||||
beforeEach(() => {
|
||||
new PrometheusDbPoolMetricsService(
|
||||
buildConfig(),
|
||||
buildDatabaseConfig(),
|
||||
buildDbConnection({ active: 3, idle: 2, waiting: 1 }),
|
||||
new DbConnectionMetrics(),
|
||||
).init();
|
||||
});
|
||||
|
||||
it('reports active connections', () => {
|
||||
expect(runCollect('n8n_db_pool_connections_active')).toHaveBeenCalledWith(3);
|
||||
});
|
||||
|
||||
it('reports idle connections', () => {
|
||||
expect(runCollect('n8n_db_pool_connections_idle')).toHaveBeenCalledWith(2);
|
||||
});
|
||||
|
||||
it('reports pending requests', () => {
|
||||
expect(runCollect('n8n_db_pool_requests_pending')).toHaveBeenCalledWith(1);
|
||||
});
|
||||
});
|
||||
|
||||
it('skips reporting when pool stats are unavailable', () => {
|
||||
new PrometheusDbPoolMetricsService(
|
||||
buildConfig(),
|
||||
buildDatabaseConfig(),
|
||||
buildDbConnection(undefined),
|
||||
new DbConnectionMetrics(),
|
||||
).init();
|
||||
|
||||
expect(runCollect('n8n_db_pool_connections_active')).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
describe('acquire-latency histogram', () => {
|
||||
it('registers an observer that feeds the histogram', () => {
|
||||
const dbConnectionMetrics = new DbConnectionMetrics();
|
||||
|
||||
new PrometheusDbPoolMetricsService(
|
||||
buildConfig(),
|
||||
buildDatabaseConfig(),
|
||||
buildDbConnection(undefined),
|
||||
dbConnectionMetrics,
|
||||
).init();
|
||||
|
||||
expect(dbConnectionMetrics.acquireDurationObserver).toBeDefined();
|
||||
dbConnectionMetrics.acquireDurationObserver?.(0.42);
|
||||
expect(mockHistogramObserve).toHaveBeenCalledWith(0.42);
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,74 @@
|
||||
import { DatabaseConfig, PrometheusMetricsConfig } from '@n8n/config';
|
||||
import { DbConnection, DbConnectionMetrics, type DbPoolStats } from '@n8n/db';
|
||||
import { Service } from '@n8n/di';
|
||||
import promClient from 'prom-client';
|
||||
|
||||
import type { PrometheusMetricsCollector } from './base';
|
||||
|
||||
@Service()
|
||||
export class PrometheusDbPoolMetricsService implements PrometheusMetricsCollector {
|
||||
constructor(
|
||||
private readonly promConfig: PrometheusMetricsConfig,
|
||||
private readonly dbConfig: DatabaseConfig,
|
||||
private readonly dbConnection: DbConnection,
|
||||
private readonly dbConnectionMetrics: DbConnectionMetrics,
|
||||
) {}
|
||||
|
||||
get enabled(): boolean {
|
||||
return this.promConfig.includeDbPoolMetrics;
|
||||
}
|
||||
|
||||
init() {
|
||||
const { prefix } = this.promConfig;
|
||||
const readStats = () => this.dbConnection.getPoolStats();
|
||||
|
||||
const makeGauge = (name: string, help: string, value: (stats: DbPoolStats) => number) =>
|
||||
new promClient.Gauge({
|
||||
name: `${prefix}db_pool_${name}`,
|
||||
help,
|
||||
collect() {
|
||||
const stats = readStats();
|
||||
if (stats) this.set(value(stats));
|
||||
},
|
||||
});
|
||||
|
||||
makeGauge(
|
||||
'connections_active',
|
||||
'Number of in-use connections in the database connection pool.',
|
||||
(s) => s.active,
|
||||
);
|
||||
makeGauge(
|
||||
'connections_idle',
|
||||
'Number of not-in-use connections in the database connection pool.',
|
||||
(s) => s.idle,
|
||||
);
|
||||
makeGauge(
|
||||
'requests_pending',
|
||||
'Number of requests waiting for an available database connection.',
|
||||
(s) => s.waiting,
|
||||
);
|
||||
|
||||
const maxGauge = new promClient.Gauge({
|
||||
name: `${prefix}db_pool_connections_max`,
|
||||
help: 'Max size of the database connection pool.',
|
||||
});
|
||||
maxGauge.set(this.maxPoolSize());
|
||||
|
||||
// Postgres only: the timing is fed by DbConnectionMonitor wrapping
|
||||
// `obtainMasterConnection` (a Postgres-only chokepoint). SQLite acquires
|
||||
// happen inside the `tarn` pool, which has no equivalent interception point.
|
||||
const acquireHistogram = new promClient.Histogram({
|
||||
name: `${prefix}db_pool_acquire_seconds`,
|
||||
help: 'Time spent acquiring a connection from the database pool (Postgres only).',
|
||||
buckets: [0.001, 0.005, 0.01, 0.05, 0.1, 0.5, 1, 5],
|
||||
});
|
||||
this.dbConnectionMetrics.acquireDurationObserver = (seconds) =>
|
||||
acquireHistogram.observe(seconds);
|
||||
}
|
||||
|
||||
private maxPoolSize(): number {
|
||||
if (this.dbConfig.type === 'postgresdb') return this.dbConfig.postgresdb.poolSize;
|
||||
// sqlite-pooled adds one write connection on top of the read pool.
|
||||
return this.dbConfig.sqlite.poolSize + 1;
|
||||
}
|
||||
}
|
||||
@@ -6,6 +6,7 @@ import promClient from 'prom-client';
|
||||
import { PrometheusActiveWorkflowMetricsService } from './active-workflow-metrics.service';
|
||||
import type { PrometheusMetricsCollector } from './base';
|
||||
import { PrometheusCacheMetricsService } from './cache-metrics.service';
|
||||
import { PrometheusDbPoolMetricsService } from './db-pool-metrics.service';
|
||||
import { PrometheusDefaultMetricsService } from './default-metrics.service';
|
||||
import { PrometheusDnsCacheMetricsService } from './dns-cache-metrics.service';
|
||||
import { PrometheusEventBusMetricsService } from './event-bus-metrics.service';
|
||||
@@ -49,6 +50,7 @@ export class PrometheusMetricsService {
|
||||
webhook: PrometheusWebhookAndFormMetricsService,
|
||||
workflowInfo: PrometheusWorkflowInfoMetricsService,
|
||||
instanceAi: PrometheusInstanceAiMetricsService,
|
||||
dbPool: PrometheusDbPoolMetricsService,
|
||||
) {
|
||||
this.logger = logger.scoped('metrics');
|
||||
this.collectors = [
|
||||
@@ -70,6 +72,7 @@ export class PrometheusMetricsService {
|
||||
webhook,
|
||||
workflowInfo,
|
||||
instanceAi,
|
||||
dbPool,
|
||||
];
|
||||
}
|
||||
|
||||
|
||||
Generated
+9
-13
@@ -136,8 +136,8 @@ catalogs:
|
||||
specifier: ^0.6.0
|
||||
version: 0.6.0
|
||||
'@n8n/typeorm':
|
||||
specifier: 0.3.20-17
|
||||
version: 0.3.20-17
|
||||
specifier: 0.3.20-18
|
||||
version: 0.3.20-18
|
||||
'@n8n_io/ai-assistant-sdk':
|
||||
specifier: 1.22.0
|
||||
version: 1.22.0
|
||||
@@ -1457,7 +1457,7 @@ importers:
|
||||
version: link:../permissions
|
||||
'@n8n/typeorm':
|
||||
specifier: 'catalog:'
|
||||
version: 0.3.20-17(@sentry/[email protected](@opentelemetry/[email protected](@opentelemetry/[email protected])))([email protected])([email protected])
|
||||
version: 0.3.20-18([email protected])([email protected])
|
||||
jest-mock-extended:
|
||||
specifier: ^3.0.4
|
||||
version: 3.0.4([email protected](@types/[email protected])([email protected](@swc/[email protected](@swc/[email protected]))(@types/[email protected])([email protected])))([email protected])
|
||||
@@ -1847,7 +1847,7 @@ importers:
|
||||
version: link:../permissions
|
||||
'@n8n/typeorm':
|
||||
specifier: 'catalog:'
|
||||
version: 0.3.20-17(@sentry/[email protected](@opentelemetry/[email protected](@opentelemetry/[email protected])))([email protected])([email protected])
|
||||
version: 0.3.20-18([email protected])([email protected])
|
||||
'@n8n/utils':
|
||||
specifier: workspace:*
|
||||
version: link:../utils
|
||||
@@ -3549,7 +3549,7 @@ importers:
|
||||
version: link:../@n8n/task-runner
|
||||
'@n8n/typeorm':
|
||||
specifier: 'catalog:'
|
||||
version: 0.3.20-17(@sentry/[email protected](@opentelemetry/[email protected](@opentelemetry/[email protected])))([email protected])([email protected])
|
||||
version: 0.3.20-18([email protected])([email protected])
|
||||
'@n8n/utils':
|
||||
specifier: workspace:*
|
||||
version: link:../@n8n/utils
|
||||
@@ -4163,7 +4163,7 @@ importers:
|
||||
version: link:../testing/janitor
|
||||
'@n8n/typeorm':
|
||||
specifier: 'catalog:'
|
||||
version: 0.3.20-17(@sentry/[email protected](@opentelemetry/[email protected](@opentelemetry/[email protected])))([email protected])([email protected])
|
||||
version: 0.3.20-18([email protected])([email protected])
|
||||
'@n8n/typescript-config':
|
||||
specifier: workspace:*
|
||||
version: link:../@n8n/typescript-config
|
||||
@@ -9259,18 +9259,15 @@ packages:
|
||||
sqlite3:
|
||||
optional: true
|
||||
|
||||
'@n8n/[email protected]7':
|
||||
resolution: {integrity: sha512-PgMNu3IqYMhgEjzqJ4SPAiReYgwXMpMPZx3BhE4M/895S2yaWmRJtHcI5Ol6Sys0rFqEQt+IbAo/th9cT7XdAg==}
|
||||
'@n8n/[email protected]8':
|
||||
resolution: {integrity: sha512-Fq9ur3X6k/4jD9VZvXYclnjCnOLst+vz3vucrjcQkeBb1VhY/nJwzhnC+uHtmGjDGxFsFRZUo+xM7JQW7ZEBSg==}
|
||||
engines: {node: '>=16.13.0'}
|
||||
peerDependencies:
|
||||
'@sentry/node': ^10.55.0
|
||||
pg: ^8.17.0
|
||||
pg-native: ^3.5.2
|
||||
pg-query-stream: ^4.10.3
|
||||
sqlite3: ^5.1.7
|
||||
peerDependenciesMeta:
|
||||
'@sentry/node':
|
||||
optional: true
|
||||
pg:
|
||||
optional: true
|
||||
pg-native:
|
||||
@@ -26629,7 +26626,7 @@ snapshots:
|
||||
transitivePeerDependencies:
|
||||
- supports-color
|
||||
|
||||
'@n8n/[email protected]7(@sentry/[email protected](@opentelemetry/[email protected](@opentelemetry/[email protected])))([email protected])([email protected])':
|
||||
'@n8n/[email protected]8([email protected])([email protected])':
|
||||
dependencies:
|
||||
app-root-path: 3.1.0
|
||||
async-mutex: 0.5.0
|
||||
@@ -26645,7 +26642,6 @@ snapshots:
|
||||
tslib: 2.8.1
|
||||
uuid: 13.0.1
|
||||
optionalDependencies:
|
||||
'@sentry/node': 10.55.0(@opentelemetry/[email protected](@opentelemetry/[email protected]))
|
||||
pg: 8.17.0
|
||||
sqlite3: 5.1.7
|
||||
transitivePeerDependencies:
|
||||
|
||||
+1
-1
@@ -54,7 +54,7 @@ catalog:
|
||||
'@modelcontextprotocol/ext-apps': 1.3.2
|
||||
'@modelcontextprotocol/sdk': 1.26.0
|
||||
'@mozilla/readability': ^0.6.0
|
||||
'@n8n/typeorm': 0.3.20-17
|
||||
'@n8n/typeorm': 0.3.20-18
|
||||
'@n8n_io/ai-assistant-sdk': 1.22.0
|
||||
'@openrouter/ai-sdk-provider': ^2.8.0
|
||||
'@rudderstack/rudder-sdk-node': 3.0.5
|
||||
|
||||
Reference in New Issue
Block a user