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


<!-- This is an auto-generated description by cubic. -->
<a
href="https://cubic.dev/pr/twentyhq/twenty/pull/23251?utm_source=github"
target="_blank" rel="noopener noreferrer"
data-no-image-dialog="true"><picture><source
media="(prefers-color-scheme: dark)"
srcset="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"><source
media="(prefers-color-scheme: light)"
srcset="https://www.cubic.dev/buttons/review-in-cubic-light.svg"><img
alt="Review in cubic"
src="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"></picture></a>
<!-- End of auto-generated description by cubic. -->

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: {

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<string, GaugeCallback>;
+  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);
+  });
+});

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<DatabasePoolName, Pool>();
+  private readonly instrumentedDrivers = new WeakSet<PostgresDriver>();
+  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);
+  }
+}

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 {}