github-featured/triggerdotdev--trigger.dev
2
0
Fork 0
Code Issues Pull Requests Deployments Models Agent Notes Packages Projects Releases Wiki Activity

feat(webapp): per-client database pool metrics that survive the driver adapter (#4541)

Browse Source 
## What

Follow-up to #4539. The driver-adapter work is inert until a client
flips to the pg driver adapter, but the moment one does, our database
observability degrades: the OTel metrics pipeline reads pool stats from
Prisma's `$metrics`, which is owned by the Rust engine's `quaint` pool.
Under the adapter, `pg.Pool` owns the pool, so those gauges read zero.
The pipeline also only ever scraped a single client (the control-plane
writer singleton).

This PR makes database metrics driver-agnostic and per-client:

- Every configured client registers a metrics source: control-plane
writer/replica, run-ops writer/replica, legacy writer/replica.
Previously only the control-plane writer singleton was scraped.
- Each OTel instrument is observed per client with `db_client` and
`db_driver` (`quaint` | `pg-adapter`) attributes. `db_client` uses our
canonical datasource-role labels (`control-plane-writer`,
`control-plane-replica`, `run-ops-writer`, `run-ops-replica`,
`legacy-run-ops-writer`, `legacy-run-ops-replica`) — the same strings
used for the `db.datasource` span attribute, so a metric and a trace
point at the same pool.
- Pool figures come from the authoritative source per driver:
- **pg-adapter**: `pg.Pool` (`totalCount`/`idleCount`/`waitingCount`,
plus cumulative opened/closed from `connect`/`remove` events).
- **quaint**: the Rust engine's `$metrics` pool gauges/counters, exactly
as before.
- Query counters and duration histograms still come from `$metrics` for
both drivers (the Rust engine executes queries in both cases).
- New `db.pool.connections.waiting` gauge (pg.Pool exposes this; quaint
reports 0).
- Stops exporting Prisma metrics from the Prometheus `/metrics` route.
Pool observability now lives entirely in the OTel pipeline, per driver,
per client.

## Why

So we can flip any client (including the control-plane writer, the
primary desync-fix target) to the driver adapter without losing pool
visibility. Existing dashboards keyed on the same metric names keep
working; they gain a per-client dimension.

## Testing

Unit (`apps/webapp/app/utils/databaseMetrics.server.test.ts`): the pure
normalizer — quaint reads pool from `$metrics`; adapter reads pool from
`pg.Pool` and keeps engine query metrics; `busy` never goes negative;
graceful zeroing when `$metrics` is unavailable (adapter still reports
live pool figures).

Live smoke test against a prod-shaped local stack: three
physically-distinct Postgres DBs (control-plane, run-ops, legacy) behind
dual PgBouncers, split mode on, with a mix of adapter and quaint
clients. Reading the actual emitted OTel metrics, every pool shows up as
its own series:

```
db.pool.connections.total{db_client="control-plane-writer",  db_driver="pg-adapter"} = 1
db.pool.connections.total{db_client="control-plane-replica", db_driver="quaint"}     = 1
db.pool.connections.total{db_client="run-ops-writer",        db_driver="pg-adapter"} = 1
db.pool.connections.total{db_client="run-ops-replica",       db_driver="quaint"}     = 1
db.pool.connections.total{db_client="legacy-run-ops-writer", db_driver="quaint"}     = 1
db.pool.connections.total{db_client="legacy-run-ops-replica",db_driver="quaint"}     = 1
db.client.queries.total{db_client="control-plane-writer",db_driver="pg-adapter"} = incrementing
db.client.queries.duration.count{db_client="control-plane-writer",db_driver="pg-adapter"} = incrementing
```

Confirms: metrics are attributed per pool with the correct driver;
adapter pools' figures come from `pg.Pool`; and query counters/duration
histograms keep incrementing under the pg adapter. Also verified
`/metrics` (Prometheus) now returns zero `prisma_*` series while still
serving the app's own metrics.

`pnpm run typecheck --filter webapp` passes.

## Notes

- `/metrics` (Prometheus) no longer includes `prisma_*` series. Anything
scraping that endpoint for Prisma metrics should read the equivalent
`db.*` metrics from the OTel exporter instead.
- **PgBouncer + `?schema=` gotcha (separate from this PR, worth flagging
for rollout):** since #4539 parses `?schema=` from the DSN and passes `{
schema }` to the adapter, node-postgres sends `search_path` as a startup
parameter. A transaction-mode PgBouncer rejects that with `FATAL:
unsupported startup parameter: search_path`. Our prod control-plane DSNs
use the default `public` schema with no `?schema=` param, so this is
latent, but any client we flip to the adapter must not carry `?schema=`
in its DSN (or the pooler needs `ignore_startup_parameters =
search_path`).

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Eric Allam
2026-08-10 13:54:06 +01:00
committed by GitHub
parent bd8ce4a50f
commit 951d8e8d7b
6 changed files with 490 additions and 144 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.server-changes/db-pool-metrics-per-client.md
+6
View File
@@ -0,0 +1,6 @@
---
area: webapp
type: improvement
---
Database connection metrics are now reported for every configured database connection instead of only the primary one, and stay accurate regardless of connection type.
apps/webapp/app/db.server.ts
+120 -42
View File
@@ -24,6 +24,7 @@ import {
logTransactionInfrastructureError,
} from "./utils/prismaErrors";
import { singleton } from "./utils/singleton";
import { registerDatabaseMetricsSource } from "./utils/databaseMetrics.server";
import {
isSplitEnabled,
assertSplitRealtimeInterlock,
@@ -247,16 +248,16 @@ export function selectRunOpsTopology(
if (config.legacySharesControlPlane) {
legacyRunOps = controlPlane;
} else {
const legacyWriter = builders.buildLegacyWriter(config.legacyUrl, "run-ops-legacy-writer");
const legacyWriter = builders.buildLegacyWriter(config.legacyUrl, "legacy-run-ops-writer");
const legacyReplica: PrismaReplicaClient = config.legacyReplicaUrl
? builders.buildLegacyReplica(config.legacyReplicaUrl, "run-ops-legacy-reader")
? builders.buildLegacyReplica(config.legacyReplicaUrl, "legacy-run-ops-replica")
: legacyWriter;
legacyRunOps = { writer: legacyWriter, replica: legacyReplica };
}
const newWriter = builders.buildNewWriter(config.newUrl, "run-ops-new-writer");
const newWriter = builders.buildNewWriter(config.newUrl, "run-ops-writer");
const newReplica: RunOpsPrismaClient = config.newReplicaUrl
? builders.buildNewReplica(config.newReplicaUrl, "run-ops-new-reader")
? builders.buildNewReplica(config.newReplicaUrl, "run-ops-replica")
: newWriter;
return {
@@ -430,19 +431,25 @@ function getClient() {
return buildWriterClient({
url,
clientType: "writer",
clientType: "control-plane-writer",
poolTimeout: env.DATABASE_WRITER_POOL_TIMEOUT,
connectTimeout: env.DATABASE_WRITER_CONNECTION_TIMEOUT,
useDriverAdapter: env.CONTROL_PLANE_DATABASE_WRITER_DRIVER_ADAPTER === "1",
});
}
type DriverAdapterPool = {
adapter: PrismaPg;
pool: Pool;
poolCounters: { opened: () => number; closed: () => number };
};
function buildDriverAdapterPool(
connectionString: string,
clientType: string,
poolTimeoutSeconds: number,
connectionLimit: number
): PrismaPg {
): DriverAdapterPool {
const pool = new Pool({
connectionString,
max: connectionLimit,
@@ -457,6 +464,15 @@ function buildDriverAdapterPool(
});
});
let opened = 0;
let closed = 0;
pool.on("connect", () => {
opened += 1;
});
pool.on("remove", () => {
closed += 1;
});
let schema: string | undefined;
try {
schema = new URL(connectionString).searchParams.get("schema") ?? undefined;
@@ -464,7 +480,11 @@ function buildDriverAdapterPool(
schema = undefined;
}
return new PrismaPg(pool, { schema, disposeExternalPool: true });
return {
adapter: new PrismaPg(pool, { schema, disposeExternalPool: true }),
pool,
poolCounters: { opened: () => opened, closed: () => closed },
};
}
// Generalized writer builder shared by the control-plane client and the run-ops
@@ -548,21 +568,34 @@ export function buildWriterClient({
: []) satisfies Prisma.LogDefinition[]),
] satisfies Prisma.LogDefinition[];
const client = useDriverAdapter
? new PrismaClient({
adapter: buildDriverAdapterPool(
url,
clientType,
poolTimeout ?? env.DATABASE_POOL_TIMEOUT,
env.DATABASE_CONNECTION_LIMIT
),
log: logConfig,
})
const driverPool = useDriverAdapter
? buildDriverAdapterPool(
url,
clientType,
poolTimeout ?? env.DATABASE_POOL_TIMEOUT,
env.DATABASE_CONNECTION_LIMIT
)
: undefined;
const client = driverPool
? new PrismaClient({ adapter: driverPool.adapter, log: logConfig })
: new PrismaClient({
datasources: { db: { url: databaseUrl.href } },
log: logConfig,
});
registerDatabaseMetricsSource(
driverPool
? {
clientType,
usesDriverAdapter: true,
client,
pool: driverPool.pool,
poolCounters: driverPool.poolCounters,
}
: { clientType, usesDriverAdapter: false, client }
);
// Only use structured logging if we're not already logging to stdout
if (process.env.PRISMA_LOG_TO_STDOUT !== "1") {
client.$on("info", (log) => {
@@ -631,7 +664,7 @@ function getReplicaClient() {
return buildReplicaClient({
url,
clientType: "reader",
clientType: "control-plane-replica",
poolTimeout: env.DATABASE_READ_REPLICA_POOL_TIMEOUT,
connectTimeout: env.DATABASE_READ_REPLICA_CONNECTION_TIMEOUT,
useDriverAdapter: env.CONTROL_PLANE_DATABASE_REPLICA_DRIVER_ADAPTER === "1",
@@ -719,21 +752,34 @@ export function buildReplicaClient({
: []) satisfies Prisma.LogDefinition[]),
] satisfies Prisma.LogDefinition[];
const replicaClient = useDriverAdapter
? new PrismaClient({
adapter: buildDriverAdapterPool(
url,
clientType,
poolTimeout ?? env.DATABASE_POOL_TIMEOUT,
env.DATABASE_CONNECTION_LIMIT
),
log: logConfig,
})
const driverPool = useDriverAdapter
? buildDriverAdapterPool(
url,
clientType,
poolTimeout ?? env.DATABASE_POOL_TIMEOUT,
env.DATABASE_CONNECTION_LIMIT
)
: undefined;
const replicaClient = driverPool
? new PrismaClient({ adapter: driverPool.adapter, log: logConfig })
: new PrismaClient({
datasources: { db: { url: replicaUrl.href } },
log: logConfig,
});
registerDatabaseMetricsSource(
driverPool
? {
clientType,
usesDriverAdapter: true,
client: replicaClient,
pool: driverPool.pool,
poolCounters: driverPool.poolCounters,
}
: { clientType, usesDriverAdapter: false, client: replicaClient }
);
// Only use structured logging if we're not already logging to stdout
if (process.env.PRISMA_LOG_TO_STDOUT !== "1") {
replicaClient.$on("info", (log) => {
@@ -813,14 +859,18 @@ function buildRunOpsWriterClient({
}`
);
const client = useDriverAdapter
const driverPool = useDriverAdapter
? buildDriverAdapterPool(
url,
clientType,
env.RUN_OPS_DATABASE_WRITER_POOL_TIMEOUT ?? env.DATABASE_POOL_TIMEOUT,
env.DATABASE_CONNECTION_LIMIT
)
: undefined;
const client = driverPool
? new RunOpsPrismaClient({
adapter: buildDriverAdapterPool(
url,
clientType,
env.RUN_OPS_DATABASE_WRITER_POOL_TIMEOUT ?? env.DATABASE_POOL_TIMEOUT,
env.DATABASE_CONNECTION_LIMIT
),
adapter: driverPool.adapter,
log: [
{ emit: "event", level: "error" },
{ emit: "event", level: "info" },
@@ -844,6 +894,18 @@ function buildRunOpsWriterClient({
],
});
registerDatabaseMetricsSource(
driverPool
? {
clientType,
usesDriverAdapter: true,
client,
pool: driverPool.pool,
poolCounters: driverPool.poolCounters,
}
: { clientType, usesDriverAdapter: false, client }
);
if (process.env.PRISMA_LOG_TO_STDOUT !== "1") {
client.$on("info", (log) => logger.info("RunOpsPrismaClient info", { clientType, event: log }));
client.$on("warn", (log) => logger.warn("RunOpsPrismaClient warn", { clientType, event: log }));
@@ -894,14 +956,18 @@ function buildRunOpsReplicaClient({
}`
);
const client = useDriverAdapter
const driverPool = useDriverAdapter
? buildDriverAdapterPool(
url,
clientType,
env.RUN_OPS_DATABASE_READ_REPLICA_POOL_TIMEOUT ?? env.DATABASE_POOL_TIMEOUT,
env.RUN_OPS_DATABASE_READ_REPLICA_CONNECTION_LIMIT ?? env.DATABASE_CONNECTION_LIMIT
)
: undefined;
const client = driverPool
? new RunOpsPrismaClient({
adapter: buildDriverAdapterPool(
url,
clientType,
env.RUN_OPS_DATABASE_READ_REPLICA_POOL_TIMEOUT ?? env.DATABASE_POOL_TIMEOUT,
env.RUN_OPS_DATABASE_READ_REPLICA_CONNECTION_LIMIT ?? env.DATABASE_CONNECTION_LIMIT
),
adapter: driverPool.adapter,
log: [
{ emit: "event", level: "error" },
{ emit: "event", level: "info" },
@@ -925,6 +991,18 @@ function buildRunOpsReplicaClient({
],
});
registerDatabaseMetricsSource(
driverPool
? {
clientType,
usesDriverAdapter: true,
client,
pool: driverPool.pool,
poolCounters: driverPool.poolCounters,
}
: { clientType, usesDriverAdapter: false, client }
);
if (process.env.PRISMA_LOG_TO_STDOUT !== "1") {
client.$on("info", (log) => logger.info("RunOpsPrismaClient info", { clientType, event: log }));
client.$on("warn", (log) => logger.warn("RunOpsPrismaClient warn", { clientType, event: log }));
apps/webapp/app/routes/metrics.ts
+1 -12
View File
@@ -1,5 +1,4 @@
import type { LoaderFunctionArgs } from "@remix-run/server-runtime";
import { prisma } from "~/db.server";
import { metricsRegister } from "~/metrics.server";
export async function loader({ request }: LoaderFunctionArgs) {
@@ -13,17 +12,7 @@ export async function loader({ request }: LoaderFunctionArgs) {
}
}
// We need to remove empty lines from the prisma metrics, grafana doesn't like them
let prismaMetrics = "";
try {
prismaMetrics = (await prisma.$metrics.prometheus()).replace(/^\s*[\r\n]/gm, "");
} catch {
prismaMetrics = "";
}
const coreMetrics = await metricsRegister.metrics();
// Order matters, core metrics end with `# EOF`, prisma metrics don't
const metrics = prismaMetrics + coreMetrics;
const metrics = await metricsRegister.metrics();
return new Response(metrics, {
headers: {
apps/webapp/app/utils/databaseMetrics.server.test.ts
+141
View File
@@ -0,0 +1,141 @@
import { describe, expect, it } from "vitest";
import {
normalizeDatabaseMetrics,
type DatabaseMetricsSource,
type MetricHistogramValue,
} from "./databaseMetrics.server";
const durationHistogram: MetricHistogramValue = {
buckets: [
[1, 10],
[10, 5],
],
sum: 1234,
count: 15,
};
function quaintJson() {
return {
counters: [
{ key: "prisma_client_queries_total", value: 100 },
{ key: "prisma_datasource_queries_total", value: 250 },
{ key: "prisma_pool_connections_opened_total", value: 12 },
{ key: "prisma_pool_connections_closed_total", value: 3 },
],
gauges: [
{ key: "prisma_client_queries_active", value: 4 },
{ key: "prisma_client_queries_wait", value: 2 },
{ key: "prisma_pool_connections_open", value: 9 },
{ key: "prisma_pool_connections_busy", value: 4 },
{ key: "prisma_pool_connections_idle", value: 5 },
],
histograms: [{ key: "prisma_client_queries_duration_histogram_ms", value: durationHistogram }],
};
}
const stubClient = { $metrics: { json: async () => quaintJson() } };
describe("normalizeDatabaseMetrics", () => {
it("reads pool figures from $metrics for a quaint (Rust) client", () => {
const source: DatabaseMetricsSource = {
clientType: "writer",
usesDriverAdapter: false,
client: stubClient,
};
const result = normalizeDatabaseMetrics(source, quaintJson());
expect(result.driver).toBe("quaint");
expect(result.clientType).toBe("writer");
expect(result.engineMetricsAvailable).toBe(true);
expect(result.pool).toEqual({
open: 9,
busy: 4,
idle: 5,
waiting: 0,
openedTotal: 12,
closedTotal: 3,
});
expect(result.counters).toEqual({ queriesTotal: 100, datasourceQueriesTotal: 250 });
expect(result.gauges).toEqual({ queriesActive: 4, queriesWait: 2 });
expect(result.histograms.queriesDuration).toEqual(durationHistogram);
});
it("reads pool figures from pg.Pool for a driver-adapter client and keeps engine query metrics", () => {
const source: DatabaseMetricsSource = {
clientType: "control-plane-writer",
usesDriverAdapter: true,
client: stubClient,
pool: { totalCount: 8, idleCount: 3, waitingCount: 6 },
poolCounters: { opened: () => 20, closed: () => 12 },
};
const result = normalizeDatabaseMetrics(source, quaintJson());
expect(result.driver).toBe("pg-adapter");
expect(result.engineMetricsAvailable).toBe(true);
expect(result.pool).toEqual({
open: 8,
busy: 5,
idle: 3,
waiting: 6,
openedTotal: 20,
closedTotal: 12,
});
expect(result.counters).toEqual({ queriesTotal: 100, datasourceQueriesTotal: 250 });
});
it("does not report negative busy when idle exceeds total for an adapter pool", () => {
const source: DatabaseMetricsSource = {
clientType: "reader",
usesDriverAdapter: true,
client: stubClient,
pool: { totalCount: 2, idleCount: 5, waitingCount: 0 },
poolCounters: { opened: () => 0, closed: () => 0 },
};
const result = normalizeDatabaseMetrics(source, quaintJson());
expect(result.pool?.busy).toBe(0);
});
it("omits engine-derived metrics and pool when $metrics is unavailable for a quaint client", () => {
const source: DatabaseMetricsSource = {
clientType: "writer",
usesDriverAdapter: false,
client: stubClient,
};
const result = normalizeDatabaseMetrics(source, undefined);
expect(result.engineMetricsAvailable).toBe(false);
expect(result.pool).toBeUndefined();
expect(result.counters).toBeUndefined();
expect(result.gauges).toBeUndefined();
expect(result.histograms.queriesDuration).toBeUndefined();
});
it("keeps pg.Pool figures but omits engine metrics when $metrics is unavailable for an adapter client", () => {
const source: DatabaseMetricsSource = {
clientType: "control-plane-writer",
usesDriverAdapter: true,
client: stubClient,
pool: { totalCount: 7, idleCount: 2, waitingCount: 1 },
poolCounters: { opened: () => 9, closed: () => 2 },
};
const result = normalizeDatabaseMetrics(source, undefined);
expect(result.engineMetricsAvailable).toBe(false);
expect(result.pool).toEqual({
open: 7,
busy: 5,
idle: 2,
waiting: 1,
openedTotal: 9,
closedTotal: 2,
});
expect(result.counters).toBeUndefined();
expect(result.gauges).toBeUndefined();
});
});
apps/webapp/app/utils/databaseMetrics.server.ts
+152
View File
@@ -0,0 +1,152 @@
import { singleton } from "./singleton";
export type MetricHistogramValue = {
buckets: [number, number][];
sum: number;
count: number;
};
type PrismaMetricsJson = {
counters: Array<{ key: string; value: number }>;
gauges: Array<{ key: string; value: number }>;
histograms: Array<{ key: string; value: MetricHistogramValue }>;
};
type MetricsCapableClient = {
$metrics: { json: () => Promise<PrismaMetricsJson> };
};
type PoolLike = {
totalCount: number;
idleCount: number;
waitingCount: number;
};
export type DatabaseMetricsSource = {
clientType: string;
usesDriverAdapter: boolean;
client: MetricsCapableClient;
pool?: PoolLike;
poolCounters?: { opened: () => number; closed: () => number };
};
export type NormalizedPoolMetrics = {
open: number;
busy: number;
idle: number;
waiting: number;
openedTotal: number;
closedTotal: number;
};
export type NormalizedDatabaseMetrics = {
clientType: string;
driver: "pg-adapter" | "quaint";
engineMetricsAvailable: boolean;
pool?: NormalizedPoolMetrics;
counters?: { queriesTotal: number; datasourceQueriesTotal: number };
gauges?: { queriesActive: number; queriesWait: number };
histograms: {
queriesWait?: MetricHistogramValue;
queriesDuration?: MetricHistogramValue;
datasourceQueriesDuration?: MetricHistogramValue;
};
};
const sources = singleton("databaseMetricsSources", () => new Map<string, DatabaseMetricsSource>());
export function registerDatabaseMetricsSource(source: DatabaseMetricsSource): void {
sources.set(source.clientType, source);
}
export function listDatabaseMetricsSources(): ReadonlyArray<DatabaseMetricsSource> {
return Array.from(sources.values());
}
export function resetDatabaseMetricsSources(): void {
sources.clear();
}
function indexByKey(entries: Array<{ key: string; value: number }>): Record<string, number> {
const out: Record<string, number> = {};
for (const entry of entries) {
out[entry.key] = entry.value;
}
return out;
}
export function normalizeDatabaseMetrics(
source: DatabaseMetricsSource,
json: PrismaMetricsJson | undefined
): NormalizedDatabaseMetrics {
const driver = source.usesDriverAdapter ? ("pg-adapter" as const) : ("quaint" as const);
const counters = json ? indexByKey(json.counters) : undefined;
const gauges = json ? indexByKey(json.gauges) : undefined;
let pool: NormalizedPoolMetrics | undefined;
if (source.usesDriverAdapter && source.pool) {
const total = source.pool.totalCount;
const idle = source.pool.idleCount;
pool = {
open: total,
idle,
busy: Math.max(0, total - idle),
waiting: source.pool.waitingCount,
openedTotal: source.poolCounters?.opened() ?? 0,
closedTotal: source.poolCounters?.closed() ?? 0,
};
} else if (counters && gauges) {
pool = {
open: gauges["prisma_pool_connections_open"] ?? 0,
busy: gauges["prisma_pool_connections_busy"] ?? 0,
idle: gauges["prisma_pool_connections_idle"] ?? 0,
waiting: 0,
openedTotal: counters["prisma_pool_connections_opened_total"] ?? 0,
closedTotal: counters["prisma_pool_connections_closed_total"] ?? 0,
};
}
const result: NormalizedDatabaseMetrics = {
clientType: source.clientType,
driver,
engineMetricsAvailable: json !== undefined,
pool,
histograms: {},
};
if (json && counters && gauges) {
const histograms: Record<string, MetricHistogramValue> = {};
for (const histogram of json.histograms) {
histograms[histogram.key] = histogram.value;
}
result.counters = {
queriesTotal: counters["prisma_client_queries_total"] ?? 0,
datasourceQueriesTotal: counters["prisma_datasource_queries_total"] ?? 0,
};
result.gauges = {
queriesActive: gauges["prisma_client_queries_active"] ?? 0,
queriesWait: gauges["prisma_client_queries_wait"] ?? 0,
};
result.histograms = {
queriesWait: histograms["prisma_client_queries_wait_histogram_ms"],
queriesDuration: histograms["prisma_client_queries_duration_histogram_ms"],
datasourceQueriesDuration: histograms["prisma_datasource_queries_duration_histogram_ms"],
};
}
return result;
}
export async function collectDatabaseClientMetrics(): Promise<NormalizedDatabaseMetrics[]> {
return Promise.all(
Array.from(sources.values()).map(async (source) => {
let json: PrismaMetricsJson | undefined;
try {
json = await source.client.$metrics.json();
} catch {
json = undefined;
}
return normalizeDatabaseMetrics(source, json);
})
);
}
apps/webapp/app/v3/tracer.server.ts
+70 -90
View File
@@ -66,9 +66,8 @@ import { LoggerSpanExporter } from "./telemetry/loggerExporter.server";
import { CompactMetricExporter } from "./telemetry/compactMetricExporter.server";
import { logger } from "~/services/logger.server";
import { flattenAttributes } from "@trigger.dev/core/v3";
import { prisma } from "~/db.server";
import { metricsRegister } from "~/metrics.server";
import type { Prisma } from "@trigger.dev/database";
import { collectDatabaseClientMetrics } from "~/utils/databaseMetrics.server";
import { performance } from "node:perf_hooks";
export const SEMINTATTRS_FORCE_RECORDING = "forceRecording";
@@ -455,6 +454,10 @@ function configurePrismaMetrics({ meter }: { meter: Meter }) {
description: "Idle (free) connections in the pool",
unit: "connections",
});
const waitingGauge = meter.createObservableGauge("db.pool.connections.waiting", {
description: "Requests waiting to acquire a pool connection",
unit: "requests",
});
// Histogram statistics as gauges
const queriesWaitTimeCount = meter.createObservableGauge("db.client.queries.wait_time.count", {
@@ -505,105 +508,81 @@ function configurePrismaMetrics({ meter }: { meter: Meter }) {
}
);
// Single helper so we hit Prisma only once per scrape ---------------------
async function readPrismaMetrics() {
const metrics = await prisma.$metrics.json();
// Extract counter values
const counters: Record<string, number> = {};
for (const counter of metrics.counters) {
counters[counter.key] = counter.value;
}
// Extract gauge values
const gauges: Record<string, number> = {};
for (const gauge of metrics.gauges) {
gauges[gauge.key] = gauge.value;
}
// Extract histogram values
const histograms: Record<string, Prisma.MetricHistogram> = {};
for (const histogram of metrics.histograms) {
histograms[histogram.key] = histogram.value;
}
return {
counters: {
queriesTotal: counters["prisma_client_queries_total"] ?? 0,
datasourceQueriesTotal: counters["prisma_datasource_queries_total"] ?? 0,
connectionsOpenedTotal: counters["prisma_pool_connections_opened_total"] ?? 0,
connectionsClosedTotal: counters["prisma_pool_connections_closed_total"] ?? 0,
},
gauges: {
queriesActive: gauges["prisma_client_queries_active"] ?? 0,
queriesWait: gauges["prisma_client_queries_wait"] ?? 0,
connectionsOpen: gauges["prisma_pool_connections_open"] ?? 0,
connectionsBusy: gauges["prisma_pool_connections_busy"] ?? 0,
connectionsIdle: gauges["prisma_pool_connections_idle"] ?? 0,
},
histograms: {
queriesWait: histograms["prisma_client_queries_wait_histogram_ms"],
queriesDuration: histograms["prisma_client_queries_duration_histogram_ms"],
datasourceQueriesDuration: histograms["prisma_datasource_queries_duration_histogram_ms"],
},
};
}
meter.addBatchObservableCallback(
async (res) => {
let prismaMetrics: Awaited<ReturnType<typeof readPrismaMetrics>>;
let clients: Awaited<ReturnType<typeof collectDatabaseClientMetrics>>;
try {
prismaMetrics = await readPrismaMetrics();
clients = await collectDatabaseClientMetrics();
} catch {
return;
}
const { counters, gauges, histograms } = prismaMetrics;
// Observe counters
res.observe(queriesTotal, counters.queriesTotal);
res.observe(datasourceQueriesTotal, counters.datasourceQueriesTotal);
res.observe(connectionsOpenedTotal, counters.connectionsOpenedTotal);
res.observe(connectionsClosedTotal, counters.connectionsClosedTotal);
for (const client of clients) {
const attributes = { db_client: client.clientType, db_driver: client.driver };
const { pool, counters, gauges, histograms } = client;
// Observe gauges
res.observe(queriesActive, gauges.queriesActive);
res.observe(queriesWait, gauges.queriesWait);
res.observe(totalGauge, gauges.connectionsOpen);
res.observe(busyGauge, gauges.connectionsBusy);
res.observe(freeGauge, gauges.connectionsIdle);
if (pool) {
res.observe(connectionsOpenedTotal, pool.openedTotal, attributes);
res.observe(connectionsClosedTotal, pool.closedTotal, attributes);
res.observe(totalGauge, pool.open, attributes);
res.observe(busyGauge, pool.busy, attributes);
res.observe(freeGauge, pool.idle, attributes);
res.observe(waitingGauge, pool.waiting, attributes);
}
// Observe histogram statistics as gauges
if (histograms.queriesWait) {
res.observe(queriesWaitTimeCount, histograms.queriesWait.count);
res.observe(queriesWaitTimeSum, histograms.queriesWait.sum);
res.observe(
queriesWaitTimeMean,
histograms.queriesWait.count > 0
? histograms.queriesWait.sum / histograms.queriesWait.count
: 0
);
}
if (!counters || !gauges) {
continue;
}
if (histograms.queriesDuration) {
res.observe(queriesDurationCount, histograms.queriesDuration.count);
res.observe(queriesDurationSum, histograms.queriesDuration.sum);
res.observe(
queriesDurationMean,
histograms.queriesDuration.count > 0
? histograms.queriesDuration.sum / histograms.queriesDuration.count
: 0
);
}
res.observe(queriesTotal, counters.queriesTotal, attributes);
res.observe(datasourceQueriesTotal, counters.datasourceQueriesTotal, attributes);
res.observe(queriesActive, gauges.queriesActive, attributes);
res.observe(queriesWait, gauges.queriesWait, attributes);
if (histograms.datasourceQueriesDuration) {
res.observe(datasourceQueriesDurationCount, histograms.datasourceQueriesDuration.count);
res.observe(datasourceQueriesDurationSum, histograms.datasourceQueriesDuration.sum);
res.observe(
datasourceQueriesDurationMean,
histograms.datasourceQueriesDuration.count > 0
? histograms.datasourceQueriesDuration.sum / histograms.datasourceQueriesDuration.count
: 0
);
if (histograms.queriesWait) {
res.observe(queriesWaitTimeCount, histograms.queriesWait.count, attributes);
res.observe(queriesWaitTimeSum, histograms.queriesWait.sum, attributes);
res.observe(
queriesWaitTimeMean,
histograms.queriesWait.count > 0
? histograms.queriesWait.sum / histograms.queriesWait.count
: 0,
attributes
);
}
if (histograms.queriesDuration) {
res.observe(queriesDurationCount, histograms.queriesDuration.count, attributes);
res.observe(queriesDurationSum, histograms.queriesDuration.sum, attributes);
res.observe(
queriesDurationMean,
histograms.queriesDuration.count > 0
? histograms.queriesDuration.sum / histograms.queriesDuration.count
: 0,
attributes
);
}
if (histograms.datasourceQueriesDuration) {
res.observe(
datasourceQueriesDurationCount,
histograms.datasourceQueriesDuration.count,
attributes
);
res.observe(
datasourceQueriesDurationSum,
histograms.datasourceQueriesDuration.sum,
attributes
);
res.observe(
datasourceQueriesDurationMean,
histograms.datasourceQueriesDuration.count > 0
? histograms.datasourceQueriesDuration.sum /
histograms.datasourceQueriesDuration.count
: 0,
attributes
);
}
}
},
[
@@ -616,6 +595,7 @@ function configurePrismaMetrics({ meter }: { meter: Meter }) {
totalGauge,
busyGauge,
freeGauge,
waitingGauge,
queriesWaitTimeCount,
queriesWaitTimeSum,
queriesWaitTimeMean,