Add Blink Prometheus listener (#11509)

bskylink/src/bin.ts

@@ -10,6 +10,7 @@ async function main() {
     httpLogger.info(
       {
         port: cfg.service.port,
+        metricsPort: cfg.service.metricsPort,
         safelinkEnabled: cfg.service.safelinkEnabled,
         hasDbUrl: !!cfg.db.url,
         hasDbMigrationUrl: !!cfg.db.migrationUrl,
@@ -33,16 +34,18 @@ async function main() {
 
     if (link.ctx.cfg.service.safelinkEnabled) {
       httpLogger.info('Starting Safelink client')
-      link.ctx.safelinkClient.runFetchEvents()
+      void link.ctx.safelinkClient.runFetchEvents()
     }
 
     await link.start()
     httpLogger.info('Link service is running')
 
-    process.on('SIGTERM', async () => {
-      httpLogger.info('Link service is stopping')
-      await link.destroy()
-      httpLogger.info('Link service is stopped')
+    process.on('SIGTERM', () => {
+      void (async () => {
+        httpLogger.info('Link service is stopping')
+        await link.destroy()
+        httpLogger.info('Link service is stopped')
+      })()
     })
   } catch (error) {
     httpLogger.error(

bskylink/src/config.ts

@@ -7,6 +7,7 @@ export type Config = {
 
 export type ServiceConfig = {
   port: number
+  metricsPort: number
   version?: string
   hostnames: string[]
   hostnamesSet: Set<string>
@@ -33,6 +34,7 @@ export type DbPoolConfig = {
 
 export type Environment = {
   port?: number
+  metricsPort?: number
   version?: string
   hostnames: string[]
   appHostname?: string
@@ -52,6 +54,7 @@ export type Environment = {
 export const readEnv = (): Environment => {
   return {
     port: envInt('LINK_PORT'),
+    metricsPort: envInt('LINK_METRICS_PORT'),
     version: envStr('LINK_VERSION'),
     hostnames: envList('LINK_HOSTNAMES'),
     appHostname: envStr('LINK_APP_HOSTNAME'),
@@ -74,6 +77,7 @@ export const readEnv = (): Environment => {
 export const envToCfg = (env: Environment): Config => {
   const serviceCfg: ServiceConfig = {
     port: env.port ?? 3000,
+    metricsPort: env.metricsPort ?? 9090,
     version: env.version,
     hostnames: env.hostnames,
     hostnamesSet: new Set(env.hostnames),

bskylink/src/index.test.ts

@@ -0,0 +1,49 @@
+import assert from 'node:assert'
+import {type AddressInfo} from 'node:net'
+import {test} from 'node:test'
+
+import {envToCfg} from './config.js'
+import {LinkService} from './index.js'
+
+const testConfig = () =>
+  envToCfg({
+    dbPostgresUrl: 'postgres://localhost:1/blink',
+    hostnames: ['go.bsky.app'],
+    metricsPort: 0,
+    port: 0,
+    safelinkAgentIdentifier: 'test',
+    safelinkAgentPass: 'test',
+    safelinkPdsUrl: 'https://example.com',
+  })
+
+void test('serves and terminates the Prometheus listener', async () => {
+  const service = await LinkService.create(testConfig())
+
+  try {
+    await service.start()
+    const {port} = service.metricsServer?.address() as AddressInfo
+    const res = await fetch(`http://127.0.0.1:${port}/metrics`)
+
+    assert.strictEqual(res.status, 200)
+    assert.match(res.headers.get('content-type') ?? '', /text\/plain/)
+    const metrics = await res.text()
+    assert.match(metrics, /process_cpu_user_seconds_total/)
+    assert.match(metrics, /nodejs_eventloop_lag_max_seconds/)
+    assert.match(metrics, /bskylink_db_pool_connections\{state="idle"\} 0/)
+    assert.match(metrics, /bskylink_db_pool_connections\{state="in_use"\} 0/)
+    assert.match(metrics, /bskylink_db_pool_max_connections 10/)
+    assert.match(metrics, /bskylink_db_pool_waiting_requests 0/)
+    assert.doesNotMatch(metrics, /http_request_duration_seconds/)
+  } finally {
+    await service.destroy()
+  }
+
+  assert.strictEqual(service.metricsServer?.listening, false)
+})
+
+void test('isolates the Prometheus registry per service', async () => {
+  const first = await LinkService.create(testConfig())
+  const second = await LinkService.create(testConfig())
+
+  await Promise.all([first.destroy(), second.destroy()])
+})

bskylink/src/index.ts

@@ -4,10 +4,12 @@ import type http from 'node:http'
 import cors from 'cors'
 import express from 'express'
 import {createHttpTerminator, type HttpTerminator} from 'http-terminator'
+import {type Registry} from 'prom-client'
 
 import {type Config} from './config.js'
 import {AppContext} from './context.js'
 import i18n from './i18n.js'
+import {createPrometheusRegistry} from './prometheus.js'
 import {default as routes, errorHandler} from './routes/index.js'
 
 export * from './config.js'
@@ -16,12 +18,17 @@ export * from './logger.js'
 
 export class LinkService {
   public server?: http.Server
+  public metricsServer?: http.Server
   private terminator?: HttpTerminator
+  private metricsTerminator?: HttpTerminator
+  private metricsRegistry: Registry
 
   constructor(
     public app: express.Application,
     public ctx: AppContext,
-  ) {}
+  ) {
+    this.metricsRegistry = createPrometheusRegistry(ctx)
+  }
 
   static async create(cfg: Config): Promise<LinkService> {
     let app = express()
@@ -41,11 +48,26 @@ export class LinkService {
     this.server.keepAliveTimeout = 90000
     this.terminator = createHttpTerminator({server: this.server})
     await events.once(this.server, 'listening')
+
+    const metricsApp = express()
+    metricsApp.get('/metrics', (_req, res, next) => {
+      res.set('Content-Type', this.metricsRegistry.contentType)
+      this.metricsRegistry.metrics().then(metrics => res.end(metrics), next)
+    })
+    this.metricsServer = metricsApp.listen(this.ctx.cfg.service.metricsPort)
+    this.metricsTerminator = createHttpTerminator({
+      server: this.metricsServer,
+      gracefulTerminationTimeout: 2000,
+    })
+    await events.once(this.metricsServer, 'listening')
   }
 
   async destroy() {
     this.ctx.abortController.abort()
-    await this.terminator?.terminate()
+    await Promise.all([
+      this.terminator?.terminate(),
+      this.metricsTerminator?.terminate(),
+    ])
     await this.ctx.db.close()
     this.ctx.metrics.stop()
   }

bskylink/src/prometheus.ts

@@ -0,0 +1,54 @@
+import {collectDefaultMetrics, Gauge, Registry} from 'prom-client'
+
+import {type AppContext} from './context.js'
+
+let runtimeRegistry: Registry | undefined
+
+const getRuntimeRegistry = (): Registry => {
+  if (!runtimeRegistry) {
+    runtimeRegistry = new Registry()
+
+    // Beyla already exports HTTP RED metrics and traces for Blink. These
+    // process metrics cover the runtime-only failure modes it cannot see,
+    // particularly event-loop stalls, GC pauses, and V8 heap pressure.
+    collectDefaultMetrics({register: runtimeRegistry})
+  }
+
+  return runtimeRegistry
+}
+
+export const createPrometheusRegistry = (ctx: AppContext): Registry => {
+  const poolRegistry = new Registry()
+
+  new Gauge<'state'>({
+    name: 'bskylink_db_pool_connections',
+    help: 'PostgreSQL client connections by usage state.',
+    labelNames: ['state'],
+    registers: [poolRegistry],
+    collect() {
+      const {idleCount, totalCount} = ctx.db.cfg.pool
+      this.set({state: 'idle'}, idleCount)
+      this.set({state: 'in_use'}, totalCount - idleCount)
+    },
+  })
+
+  new Gauge({
+    name: 'bskylink_db_pool_max_connections',
+    help: 'Configured maximum PostgreSQL client connections.',
+    registers: [poolRegistry],
+    collect() {
+      this.set(ctx.cfg.db.pool.size)
+    },
+  })
+
+  new Gauge({
+    name: 'bskylink_db_pool_waiting_requests',
+    help: 'Requests waiting for a PostgreSQL client connection.',
+    registers: [poolRegistry],
+    collect() {
+      this.set(ctx.db.cfg.pool.waitingCount)
+    },
+  })
+
+  return Registry.merge([getRuntimeRegistry(), poolRegistry])
+}