From e5c9fcf0582f91d4c792986062688eebf28acf70 Mon Sep 17 00:00:00 2001 From: Weiko Date: Fri, 24 Jul 2026 13:03:12 +0200 Subject: [PATCH] Add PostgreSQL connection pool pressure metrics (#23251) ## Summary - Add pool gauges for total, idle, waiting, and maximum connections - Record PostgreSQL connection acquisition duration and failures - Instrument core, workspace primary, and optional replica data sources - Add unit tests covering gauges, acquisition timing, failures, and deduplication Review in cubic --- .../typeorm/database-gauge.service.ts | 10 + .../database-pool-metrics.service.spec.ts | 232 ++++++++++++++++++ .../typeorm/database-pool-metrics.service.ts | 114 +++++++++ .../src/database/typeorm/typeorm.module.ts | 5 +- .../global-workspace-datasource.module.ts | 2 + .../global-workspace-datasource.service.ts | 13 + 6 files changed, 374 insertions(+), 2 deletions(-) create mode 100644 packages/twenty-server/src/database/typeorm/database-pool-metrics.service.spec.ts create mode 100644 packages/twenty-server/src/database/typeorm/database-pool-metrics.service.ts diff --git a/packages/twenty-server/src/database/typeorm/database-gauge.service.ts b/packages/twenty-server/src/database/typeorm/database-gauge.service.ts index 18e75bbd4b..100b1adfc3 100644 --- a/packages/twenty-server/src/database/typeorm/database-gauge.service.ts +++ b/packages/twenty-server/src/database/typeorm/database-gauge.service.ts @@ -3,6 +3,10 @@ import { InjectDataSource } from '@nestjs/typeorm'; import { DataSource } from 'typeorm'; +import { + DatabasePoolMetricsService, + DatabasePoolName, +} from 'src/database/typeorm/database-pool-metrics.service'; import { MetricsService } from 'src/engine/core-modules/metrics/metrics.service'; @Injectable() @@ -13,9 +17,15 @@ export class DatabaseGaugeService implements OnModuleInit { @InjectDataSource() private readonly dataSource: DataSource, private readonly metricsService: MetricsService, + private readonly databasePoolMetricsService: DatabasePoolMetricsService, ) {} onModuleInit() { + this.databasePoolMetricsService.registerDataSource({ + poolName: DatabasePoolName.Core, + dataSource: this.dataSource, + }); + this.metricsService.createObservableGauge({ metricName: 'twenty_database_up', options: { diff --git a/packages/twenty-server/src/database/typeorm/database-pool-metrics.service.spec.ts b/packages/twenty-server/src/database/typeorm/database-pool-metrics.service.spec.ts new file mode 100644 index 0000000000..fa3e09d436 --- /dev/null +++ b/packages/twenty-server/src/database/typeorm/database-pool-metrics.service.spec.ts @@ -0,0 +1,232 @@ +import { type Pool } from 'pg'; +import { type DataSource } from 'typeorm'; +import { type PostgresDriver } from 'typeorm/driver/postgres/PostgresDriver'; + +import { + DatabasePoolMetricsService, + DatabasePoolName, +} from 'src/database/typeorm/database-pool-metrics.service'; +import { type MetricsService } from 'src/engine/core-modules/metrics/metrics.service'; + +type GaugeCallback = () => Promise< + Array<{ + value: number; + attributes: { + pool: DatabasePoolName; + }; + }> +>; + +const createDataSource = ({ + totalCount = 0, + idleCount = 0, + waitingCount = 0, + max = 10, + obtainMasterConnection = jest + .fn() + .mockResolvedValue([{}, jest.fn()] as [unknown, jest.Mock]), +}: { + totalCount?: number; + idleCount?: number; + waitingCount?: number; + max?: number; + obtainMasterConnection?: jest.Mock; +} = {}) => { + const pool = { + totalCount, + idleCount, + waitingCount, + options: { + max, + }, + } as Pool; + const driver = { + master: pool, + obtainMasterConnection, + } as unknown as PostgresDriver; + const dataSource = { + driver, + } as unknown as DataSource; + + return { + dataSource, + driver, + obtainMasterConnection, + }; +}; + +describe('DatabasePoolMetricsService', () => { + let service: DatabasePoolMetricsService; + let gaugeCallbacks: Map; + let histogramRecord: jest.Mock; + + beforeEach(() => { + gaugeCallbacks = new Map(); + histogramRecord = jest.fn(); + + const metricsService = { + getMeter: jest.fn().mockReturnValue({ + createHistogram: jest.fn().mockReturnValue({ + record: histogramRecord, + }), + }), + createMultiObservableGauge: jest + .fn() + .mockImplementation(({ metricName, callback }) => { + gaugeCallbacks.set(metricName, callback); + }), + } as unknown as MetricsService; + + service = new DatabasePoolMetricsService(metricsService); + }); + + it('reports pool connection state for every registered data source', async () => { + const core = createDataSource({ + totalCount: 10, + idleCount: 3, + waitingCount: 2, + max: 10, + }); + const workspace = createDataSource({ + totalCount: 8, + idleCount: 5, + waitingCount: 0, + max: 12, + }); + + service.registerDataSource({ + poolName: DatabasePoolName.Core, + dataSource: core.dataSource, + }); + service.registerDataSource({ + poolName: DatabasePoolName.WorkspacePrimary, + dataSource: workspace.dataSource, + }); + + await expect( + gaugeCallbacks.get('twenty_database_pool_total_connections')?.(), + ).resolves.toEqual([ + { + value: 10, + attributes: { + pool: DatabasePoolName.Core, + }, + }, + { + value: 8, + attributes: { + pool: DatabasePoolName.WorkspacePrimary, + }, + }, + ]); + await expect( + gaugeCallbacks.get('twenty_database_pool_idle_connections')?.(), + ).resolves.toEqual([ + { + value: 3, + attributes: { + pool: DatabasePoolName.Core, + }, + }, + { + value: 5, + attributes: { + pool: DatabasePoolName.WorkspacePrimary, + }, + }, + ]); + await expect( + gaugeCallbacks.get('twenty_database_pool_waiting_requests')?.(), + ).resolves.toEqual([ + { + value: 2, + attributes: { + pool: DatabasePoolName.Core, + }, + }, + { + value: 0, + attributes: { + pool: DatabasePoolName.WorkspacePrimary, + }, + }, + ]); + await expect( + gaugeCallbacks.get('twenty_database_pool_max_connections')?.(), + ).resolves.toEqual([ + { + value: 10, + attributes: { + pool: DatabasePoolName.Core, + }, + }, + { + value: 12, + attributes: { + pool: DatabasePoolName.WorkspacePrimary, + }, + }, + ]); + }); + + it('records connection acquisition duration', async () => { + const dataSource = createDataSource({ + obtainMasterConnection: jest.fn().mockImplementation(async () => { + await new Promise((resolve) => setTimeout(resolve, 250)); + + return [{}, jest.fn()]; + }), + }); + + service.registerDataSource({ + poolName: DatabasePoolName.WorkspacePrimary, + dataSource: dataSource.dataSource, + }); + + const connectionPromise = dataSource.driver.obtainMasterConnection(); + + await jest.advanceTimersByTimeAsync(250); + await connectionPromise; + + expect(histogramRecord).toHaveBeenCalledWith(0.25, { + pool: DatabasePoolName.WorkspacePrimary, + }); + }); + + it('records failed connection acquisitions', async () => { + const error = new Error('connection failed'); + const dataSource = createDataSource({ + obtainMasterConnection: jest.fn().mockRejectedValue(error), + }); + + service.registerDataSource({ + poolName: DatabasePoolName.Core, + dataSource: dataSource.dataSource, + }); + + await expect(dataSource.driver.obtainMasterConnection()).rejects.toThrow( + error, + ); + expect(histogramRecord).toHaveBeenCalledWith(0, { + pool: DatabasePoolName.Core, + }); + }); + + it('does not instrument a data source more than once', async () => { + const dataSource = createDataSource(); + + service.registerDataSource({ + poolName: DatabasePoolName.Core, + dataSource: dataSource.dataSource, + }); + service.registerDataSource({ + poolName: DatabasePoolName.Core, + dataSource: dataSource.dataSource, + }); + + await dataSource.driver.obtainMasterConnection(); + + expect(dataSource.obtainMasterConnection).toHaveBeenCalledTimes(1); + expect(histogramRecord).toHaveBeenCalledTimes(1); + }); +}); diff --git a/packages/twenty-server/src/database/typeorm/database-pool-metrics.service.ts b/packages/twenty-server/src/database/typeorm/database-pool-metrics.service.ts new file mode 100644 index 0000000000..09614ca3a6 --- /dev/null +++ b/packages/twenty-server/src/database/typeorm/database-pool-metrics.service.ts @@ -0,0 +1,114 @@ +import { Injectable } from '@nestjs/common'; + +import { type Histogram } from '@opentelemetry/api'; +import { type Pool } from 'pg'; +import { type DataSource } from 'typeorm'; +import { type PostgresDriver } from 'typeorm/driver/postgres/PostgresDriver'; + +import { MetricsService } from 'src/engine/core-modules/metrics/metrics.service'; + +export enum DatabasePoolName { + Core = 'core', + WorkspacePrimary = 'workspace_primary', + WorkspaceReplica = 'workspace_replica', +} + +const ACQUISITION_DURATION_BUCKETS_SECONDS = [ + 0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, +]; + +const POOL_GAUGES = [ + { + metricName: 'twenty_database_pool_total_connections', + description: 'Current total number of PostgreSQL pool connections', + getValue: (pool: Pool) => pool.totalCount, + }, + { + metricName: 'twenty_database_pool_idle_connections', + description: 'Current number of idle PostgreSQL pool connections', + getValue: (pool: Pool) => pool.idleCount, + }, + { + metricName: 'twenty_database_pool_waiting_requests', + description: + 'Current number of requests waiting for a PostgreSQL pool connection', + getValue: (pool: Pool) => pool.waitingCount, + }, + { + metricName: 'twenty_database_pool_max_connections', + description: 'Maximum number of PostgreSQL pool connections', + getValue: (pool: Pool) => pool.options.max, + }, +] as const; + +@Injectable() +export class DatabasePoolMetricsService { + private readonly pools = new Map(); + private readonly instrumentedDrivers = new WeakSet(); + private readonly acquisitionDurationHistogram: Histogram; + + constructor(private readonly metricsService: MetricsService) { + this.acquisitionDurationHistogram = this.metricsService + .getMeter() + .createHistogram('twenty_database_pool_acquisition_duration_seconds', { + description: + 'Time spent acquiring a connection from the PostgreSQL pool', + unit: 's', + advice: { + explicitBucketBoundaries: ACQUISITION_DURATION_BUCKETS_SECONDS, + }, + }); + + for (const gauge of POOL_GAUGES) { + this.metricsService.createMultiObservableGauge({ + metricName: gauge.metricName, + options: { + description: gauge.description, + }, + callback: async () => + Array.from(this.pools, ([poolName, pool]) => ({ + value: gauge.getValue(pool), + attributes: { + pool: poolName, + }, + })), + }); + } + } + + registerDataSource({ + poolName, + dataSource, + }: { + poolName: DatabasePoolName; + dataSource: DataSource; + }): void { + const driver = dataSource.driver as PostgresDriver; + const pool = driver.master as Pool; + + this.pools.set(poolName, pool); + + if (this.instrumentedDrivers.has(driver)) { + return; + } + + const obtainMasterConnection = driver.obtainMasterConnection.bind(driver); + + driver.obtainMasterConnection = async () => { + const start = performance.now(); + + try { + return await obtainMasterConnection(); + } finally { + this.acquisitionDurationHistogram.record( + (performance.now() - start) / 1000, + { + pool: poolName, + }, + ); + } + }; + + this.instrumentedDrivers.add(driver); + } +} diff --git a/packages/twenty-server/src/database/typeorm/typeorm.module.ts b/packages/twenty-server/src/database/typeorm/typeorm.module.ts index cd47754df2..3838a07541 100644 --- a/packages/twenty-server/src/database/typeorm/typeorm.module.ts +++ b/packages/twenty-server/src/database/typeorm/typeorm.module.ts @@ -5,6 +5,7 @@ import { DataSource, type DataSourceOptions } from 'typeorm'; import { typeORMCoreModuleOptions } from 'src/database/typeorm/core/core.datasource'; import { DatabaseGaugeService } from 'src/database/typeorm/database-gauge.service'; +import { DatabasePoolMetricsService } from 'src/database/typeorm/database-pool-metrics.service'; import { MetricsModule } from 'src/engine/core-modules/metrics/metrics.module'; import { installUpgradeAwareRepositoryProxy } from 'src/engine/twenty-orm/upgrade-aware/install-upgrade-aware-repository-proxy'; @@ -23,7 +24,7 @@ import { installUpgradeAwareRepositoryProxy } from 'src/engine/twenty-orm/upgrad }), MetricsModule, ], - providers: [DatabaseGaugeService], - exports: [], + providers: [DatabasePoolMetricsService, DatabaseGaugeService], + exports: [DatabasePoolMetricsService], }) export class TypeORMModule {} diff --git a/packages/twenty-server/src/engine/twenty-orm/global-workspace-datasource/global-workspace-datasource.module.ts b/packages/twenty-server/src/engine/twenty-orm/global-workspace-datasource/global-workspace-datasource.module.ts index 6d941fcc76..c9dba20597 100644 --- a/packages/twenty-server/src/engine/twenty-orm/global-workspace-datasource/global-workspace-datasource.module.ts +++ b/packages/twenty-server/src/engine/twenty-orm/global-workspace-datasource/global-workspace-datasource.module.ts @@ -1,6 +1,7 @@ import { Global, Module } from '@nestjs/common'; import { TypeOrmModule } from '@nestjs/typeorm'; +import { TypeORMModule } from 'src/database/typeorm/typeorm.module'; import { ApplicationEntity } from 'src/engine/core-modules/application/application.entity'; import { TwentyConfigModule } from 'src/engine/core-modules/twenty-config/twenty-config.module'; import { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity'; @@ -22,6 +23,7 @@ import { WorkspaceEventEmitterModule } from 'src/engine/workspace-event-emitter/ @Global() @Module({ imports: [ + TypeORMModule, TypeOrmModule.forFeature([ WorkspaceEntity, ObjectMetadataEntity, diff --git a/packages/twenty-server/src/engine/twenty-orm/global-workspace-datasource/global-workspace-datasource.service.ts b/packages/twenty-server/src/engine/twenty-orm/global-workspace-datasource/global-workspace-datasource.service.ts index 863c6e6051..a77ac70d8c 100644 --- a/packages/twenty-server/src/engine/twenty-orm/global-workspace-datasource/global-workspace-datasource.service.ts +++ b/packages/twenty-server/src/engine/twenty-orm/global-workspace-datasource/global-workspace-datasource.service.ts @@ -8,6 +8,10 @@ import { InjectDataSource } from '@nestjs/typeorm'; import { isDefined } from 'twenty-shared/utils'; import { DataSource } from 'typeorm'; +import { + DatabasePoolMetricsService, + DatabasePoolName, +} from 'src/database/typeorm/database-pool-metrics.service'; import { TwentyConfigService } from 'src/engine/core-modules/twenty-config/twenty-config.service'; import { GlobalWorkspaceDataSource } from 'src/engine/twenty-orm/global-workspace-datasource/global-workspace-datasource'; import { WorkspaceEventEmitter } from 'src/engine/workspace-event-emitter/workspace-event-emitter'; @@ -25,6 +29,7 @@ export class GlobalWorkspaceDataSourceService private readonly workspaceEventEmitter: WorkspaceEventEmitter, @InjectDataSource() private readonly coreDataSource: DataSource, + private readonly databasePoolMetricsService: DatabasePoolMetricsService, ) {} async onModuleInit(): Promise { @@ -57,6 +62,10 @@ export class GlobalWorkspaceDataSourceService ); await this.globalWorkspaceDataSource.initialize(); + this.databasePoolMetricsService.registerDataSource({ + poolName: DatabasePoolName.WorkspacePrimary, + dataSource: this.globalWorkspaceDataSource, + }); const shouldInitializeReplicaDataSource = isDefined( this.twentyConfigService.get('PG_DATABASE_REPLICA_URL'), @@ -91,6 +100,10 @@ export class GlobalWorkspaceDataSourceService this.coreDataSource, ); await this.globalWorkspaceDataSourceReplica.initialize(); + this.databasePoolMetricsService.registerDataSource({ + poolName: DatabasePoolName.WorkspaceReplica, + dataSource: this.globalWorkspaceDataSourceReplica, + }); } }