chore: integrate Prometheus

apps/node-fastify/.env.sample

@@ -26,4 +26,7 @@ VERIFY_WEBHOOK_SIGNATURE=true
 SENTRY_DSN=
 
 # Optional, when specified, Sentry-supported tag 'Environment' is set. Defaults to 'development'
-SENTRY_ENVIRONMENT=
+SENTRY_ENVIRONMENT=
+
+# Optional, allow disabling Prometheus metrics endpoint and collection of default metrics. Defaults to true.
+PROMETHEUS_METRICS_ENABLED=true

apps/node-fastify/package.json

@@ -18,16 +18,17 @@
   "author": "Supabase",
   "license": "Apache-2.0",
   "dependencies": {
-    "@fastify/autoload": "^6.3.0",
-    "@fastify/type-provider-typebox": "^6.0.0",
-    "@sentry/node": "^10.0.0",
-    "dotenv": "^17.0.0",
-    "fastify": "^5.7.3",
-    "orb-sync-lib": "*"
+    "@fastify/autoload": "^6.3.1",
+    "@fastify/type-provider-typebox": "^6.1.0",
+    "@sentry/node": "^10.39.0",
+    "dotenv": "^17.3.1",
+    "fastify": "^5.7.4",
+    "orb-sync-lib": "*",
+    "fastify-metrics": "^12.1.0"
   },
   "devDependencies": {
-    "pino-pretty": "^13.0.0",
-    "tsx": "^4.19.3",
+    "pino-pretty": "^13.1.3",
+    "tsx": "^4.21.0",
     "vitest": "^3.2.4"
   }
 }

apps/node-fastify/src/app.ts

@@ -1,13 +1,15 @@
 // Needs to be imported here at the very beginning so that auto-instrumentation works for the imported modules
 import './instrument';
 
-import fastify, { FastifyInstance, FastifyServerOptions } from 'fastify';
+import fastify, { FastifyInstance, FastifyServerOptions, type FastifyError } from 'fastify';
+import fastifyMetrics from 'fastify-metrics';
 import autoload from '@fastify/autoload';
 import path from 'node:path';
 import { OrbSync } from 'orb-sync-lib';
 import { getConfig } from './utils/config';
 import * as Sentry from '@sentry/node';
 import pino from 'pino';
+import prometheus from './prometheus';
 
 export async function createApp(
   opts: FastifyServerOptions = {},
@@ -43,6 +45,25 @@ export async function createApp(
     }
   });
 
+  // Ensure errors get logged
+  app.setErrorHandler((error, _request, reply) => {
+    const err = error as FastifyError;
+
+    const statusCode = err.statusCode || 500;
+
+    // Only log server errors
+    if (statusCode >= 500) {
+      app.log.error(error);
+      if (config.PROMETHEUS_METRICS_ENABLED) {
+        prometheus.metrics.internalServerErrorsCounter.inc();
+      }
+    }
+
+    reply.status(statusCode).send({
+      error: statusCode >= 500 ? 'Internal Server Error' : err.message,
+    });
+  });
+
   /**
    * Expose all routes in './routes'
    * Use compiled routes in test environment
@@ -54,6 +75,22 @@ export async function createApp(
     dir: routesDir,
   });
 
+  if (config.PROMETHEUS_METRICS_ENABLED) {
+    logger.info('Prometheus metrics enabled, registering /metrics endpoint');
+    await app.register(fastifyMetrics, {
+      endpoint: '/metrics',
+      defaultMetrics: {
+        enabled: true,
+      },
+      routeMetrics: {
+        registeredRoutesOnly: false,
+        // if we don't set this group, every single unknown route will get multiple metrics, possibly overflowing our metrics store
+        invalidRouteGroup: 'invalidRoute',
+      },
+      promClient: prometheus.client,
+    });
+  }
+
   const orbSync =
     orbSyncInstance ||
     new OrbSync({
@@ -65,6 +102,20 @@ export async function createApp(
       logger,
     });
 
+  // Capture pg pool stats every few seconds
+  if (config.PROMETHEUS_METRICS_ENABLED) {
+    setInterval(() => {
+      const pool = orbSync.postgresClient.pool;
+      const total = pool.totalCount;
+      const idle = pool.idleCount;
+      const waiting = pool.waitingCount;
+
+      prometheus.metrics.pgPoolTotalGauge.set(total);
+      prometheus.metrics.pgPoolIdleGauge.set(idle);
+      prometheus.metrics.pgPoolWaitingGauge.set(waiting);
+    }, 5_000);
+  }
+
   app.decorate('orbSync', orbSync);
 
   await app.ready();

apps/node-fastify/src/prometheus.ts

@@ -0,0 +1,63 @@
+import client from 'prom-client';
+
+// Avoid duplicate metric registration across test runs or multiple app instances
+function getOrCreateCounter(config: { name: string; help: string; labelNames?: string[] }) {
+  const existing = client.register.getSingleMetric(config.name);
+  return (existing as client.Counter) ?? new client.Counter(config);
+}
+
+function getOrCreateHistogram(config: { name: string; help: string; labelNames?: string[]; buckets?: number[] }) {
+  const existing = client.register.getSingleMetric(config.name);
+  return (existing as client.Histogram) ?? new client.Histogram(config);
+}
+
+function getOrCreateGauge(config: { name: string; help: string; labelNames?: string[] }) {
+  const existing = client.register.getSingleMetric(config.name);
+  return (existing as client.Gauge) ?? new client.Gauge(config);
+}
+
+const webhooksProcessedCounter = getOrCreateCounter({
+  name: 'orb_sync_webhooks_processed_total',
+  help: 'Total number of webhooks processed',
+  labelNames: ['event'],
+});
+
+const webhookDelayMsHistogram = getOrCreateHistogram({
+  name: 'orb_sync_webhook_delay_ms',
+  help: 'Delay between when an event is emitted and when it is processed by the API in milliseconds',
+  labelNames: ['event'],
+});
+
+const internalServerErrorsCounter = getOrCreateCounter({
+  name: 'orb_sync_internal_server_errors_total',
+  help: 'Total number of errors encountered',
+});
+
+const pgPoolTotalGauge = getOrCreateGauge({
+  name: 'orb_sync_pg_pool_total',
+  help: 'Total number of clients in the Postgres pool',
+});
+
+const pgPoolIdleGauge = getOrCreateGauge({
+  name: 'orb_sync_pg_pool_idle',
+  help: 'Number of idle clients in the Postgres pool',
+});
+
+const pgPoolWaitingGauge = getOrCreateGauge({
+  name: 'orb_sync_pg_pool_waiting',
+  help: 'Number of waiting requests for a client from the Postgres pool',
+});
+
+
+
+export default {
+  client,
+  metrics: {
+    webhooksProcessedCounter,
+    webhookDelayMsHistogram,
+    internalServerErrorsCounter,
+    pgPoolTotalGauge,
+    pgPoolIdleGauge,
+    pgPoolWaitingGauge,
+  },
+};

apps/node-fastify/src/routes/webhooks.ts

@@ -1,13 +1,17 @@
 import type { FastifyInstance } from 'fastify';
+import prometheus from '../prometheus';
 
 export default async function routes(fastify: FastifyInstance) {
   fastify.post('/webhooks', {
-    bodyLimit: 5e6, // 5 MB
+    bodyLimit: 10e6, // 10 MB
     handler: async (request, reply) => {
       const headers = request.headers;
       const body: { raw: Buffer } = request.body as { raw: Buffer };
 
-      await fastify.orbSync.processWebhook(body.raw.toString(), headers);
+      const { eventType, timeSinceEventCreatedMs } = await fastify.orbSync.processWebhook(body.raw.toString(), headers);
+
+      prometheus.metrics.webhooksProcessedCounter.inc({ event: eventType });
+      prometheus.metrics.webhookDelayMsHistogram.observe({ event: eventType }, timeSinceEventCreatedMs);
 
       return reply.send({ received: true });
     },

apps/node-fastify/src/test/metrics.test.ts

@@ -0,0 +1,37 @@
+'use strict';
+import { FastifyInstance } from 'fastify';
+import { beforeAll, describe, test, expect, afterAll } from 'vitest';
+import { createApp } from '../app';
+import pino from 'pino';
+
+describe('/metrics', () => {
+  let app: FastifyInstance;
+
+  beforeAll(async () => {
+    const logger = pino({ level: 'silent' });
+
+    app = await createApp(
+      {
+        loggerInstance: logger,
+        disableRequestLogging: true,
+        requestIdHeader: 'Request-Id',
+      },
+      logger
+    );
+  });
+
+  afterAll(async () => {
+    await app.close();
+  });
+
+  test('is alive', async () => {
+    const response = await app.inject({
+      url: `/metrics`,
+      method: 'GET',
+    });
+    const text = response.body;
+    expect(response.statusCode).toBe(200);
+    expect(text).toContain('orb_sync_webhooks_processed_total');
+    expect(text).toContain('process_cpu_seconds_total');
+  });
+});