feat(admin): add Prometheus-backed /admin/metrics SSE endpoint

.changeset/admin-metrics-sse-endpoint.md

@@ -0,0 +1,5 @@
+---
+"nostream": minor
+---
+
+feat: add Prometheus-backed admin metrics SSE endpoint

.env.example

@@ -37,6 +37,11 @@ WORKER_COUNT=2 # Defaults to CPU count. Use 1 or 2 for local testing.
 # OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4318
 # OTEL_METRIC_EXPORT_INTERVAL_MS=10000
 
+# --- OBSERVABILITY (Prometheus for Admin Metrics) ---
+# PROMETHEUS_URL=http://127.0.0.1:9090
+# PROMETHEUS_QUERY_TIMEOUT_MS=5000
+# ADMIN_METRICS_SSE_INTERVAL_MS=5000
+
 # --- RELAY PRIVATE KEY (Optional) ---
 # RELAY_PRIVATE_KEY=your_hex_private_key
 

docker-compose.yml

@@ -60,6 +60,8 @@ services:
       OTEL_SERVICE_NAME: ${OTEL_SERVICE_NAME:-nostream}
       OTEL_EXPORTER_OTLP_ENDPOINT: ${OTEL_EXPORTER_OTLP_ENDPOINT:-http://otel-collector:4318}
       OTEL_METRIC_EXPORT_INTERVAL_MS: ${OTEL_METRIC_EXPORT_INTERVAL_MS:-10000}
+      PROMETHEUS_URL: ${PROMETHEUS_URL:-http://prometheus:9090}
+      ADMIN_METRICS_SSE_INTERVAL_MS: ${ADMIN_METRICS_SSE_INTERVAL_MS:-5000}
       # Enable DEBUG for troubleshooting. Examples:
       # DEBUG: "primary:*"
       # DEBUG: "worker:*"

src/controllers/admin/get-metrics-controller.ts

@@ -0,0 +1,66 @@
+import { Request, Response } from 'express'
+
+import { IController } from '../../@types/controllers'
+import { createLogger } from '../../factories/logger-factory'
+import { collectAdminMetricsSnapshot, getAdminMetricsSseIntervalMs } from '../../utils/admin-metrics'
+
+const logger = createLogger('get-admin-metrics-controller')
+
+export class GetAdminMetricsController implements IController {
+  public async handleRequest(request: Request, response: Response): Promise<void> {
+    response.status(200)
+    response.setHeader('content-type', 'text/event-stream; charset=utf-8')
+    response.setHeader('cache-control', 'no-cache, no-transform')
+    response.setHeader('connection', 'keep-alive')
+    response.flushHeaders()
+
+    let closed = false
+    let inFlight = false
+    let timer: ReturnType<typeof setInterval> | undefined
+
+    const cleanup = () => {
+      closed = true
+      if (timer) {
+        clearInterval(timer)
+        timer = undefined
+      }
+    }
+
+    request.on('close', cleanup)
+
+    const sendSnapshot = async () => {
+      if (closed || response.writableEnded || inFlight) {
+        return
+      }
+
+      inFlight = true
+
+      try {
+        const snapshot = await collectAdminMetricsSnapshot()
+        if (closed || response.writableEnded) {
+          return
+        }
+
+        response.write(`data: ${JSON.stringify(snapshot)}\n\n`)
+      } catch (error) {
+        logger.warn('failed to collect admin metrics snapshot: %o', error)
+        if (!closed && !response.writableEnded) {
+          response.write(`event: error\ndata: ${JSON.stringify({ error: 'failed to collect metrics' })}\n\n`)
+        }
+      } finally {
+        inFlight = false
+      }
+    }
+
+    await sendSnapshot()
+
+    if (closed || response.writableEnded) {
+      return
+    }
+
+    const intervalMs = getAdminMetricsSseIntervalMs()
+    timer = setInterval(() => {
+      void sendSnapshot()
+    }, intervalMs)
+  }
+}

src/factories/controllers/get-admin-metrics-controller-factory.ts

@@ -0,0 +1,6 @@
+import { GetAdminMetricsController } from '../../controllers/admin/get-metrics-controller'
+import { IController } from '../../@types/controllers'
+
+export const createGetAdminMetricsController = (): IController => {
+  return new GetAdminMetricsController()
+}

src/routes/admin/index.ts

@@ -1,6 +1,7 @@
 import { json, Router } from 'express'
 
 import { createGetAdminHealthController } from '../../factories/controllers/get-admin-health-controller-factory'
+import { createGetAdminMetricsController } from '../../factories/controllers/get-admin-metrics-controller-factory'
 import { createGetAdminSessionController } from '../../factories/controllers/get-admin-session-controller-factory'
 import { createPostAdminLoginController } from '../../factories/controllers/post-admin-login-controller-factory'
 import { adminAuthMiddleware } from '../../handlers/request-handlers/admin-auth-middleware'
@@ -21,5 +22,6 @@
 router.post('/login', adminLoginRateLimitMiddleware, json(), withAdminController(createPostAdminLoginController))
 router.get('/session', adminRateLimitMiddleware, adminAuthMiddleware, withAdminController(createGetAdminSessionController))
 router.get('/health', adminRateLimitMiddleware, adminAuthMiddleware, withAdminController(createGetAdminHealthController))
+router.get('/metrics', adminRateLimitMiddleware, adminAuthMiddleware, withAdminController(createGetAdminMetricsController))
 
 export default router

src/utils/admin-metrics.ts

@@ -0,0 +1,147 @@
+import { AdminDependencyHealth, collectAdminHealthSnapshot } from './admin-health'
+import { queryPrometheusInstant } from './prometheus-client'
+
+export interface AdminMetricsValues {
+  eventsPerSecond: number
+  eventsRejectedPerSecond: number
+  eventsAcceptedTotal: number
+  eventsRejectedTotal: number
+  activeConnections: number
+  cpuLoadPercent: number
+  memoryUsedMb: number
+}
+
+export interface AdminMetricsSnapshot {
+  timestamp: number
+  status: 'ok' | 'degraded' | 'unavailable'
+  metrics: AdminMetricsValues
+  health: {
+    database: AdminDependencyHealth
+    redis: AdminDependencyHealth
+  }
+  prometheus: {
+    available: boolean
+    error?: string
+  }
+}
+
+const PROMQL = {
+  eventsAcceptedRate: 'sum(rate(nostream_events_accepted_total[1m]))',
+  eventsRejectedRate: 'sum(rate(nostream_events_rejected_total[1m]))',
+  eventsAcceptedTotal: 'sum(nostream_events_accepted_total)',
+  eventsRejectedTotal: 'sum(nostream_events_rejected_total)',
+  activeConnections: 'sum(nostream_websocket_connections)',
+  cpuLoadPercent: 'avg(nostream_process_cpu_load_percent)',
+  memoryUsedMb: 'sum(nostream_process_memory_used_mb)',
+} as const
+
+const toMetricValue = (value: number | undefined): number => {
+  return value ?? 0
+}
+
+let cachedSnapshot: AdminMetricsSnapshot | undefined
+let cachedAt = 0
+let snapshotInFlight: Promise<AdminMetricsSnapshot> | undefined
+
+export const resetAdminMetricsSnapshotCache = (): void => {
+  cachedSnapshot = undefined
+  cachedAt = 0
+  snapshotInFlight = undefined
+}
+
+const collectAdminMetricsSnapshotUncached = async (): Promise<AdminMetricsSnapshot> => {
+  const [
+    health,
+    eventsAcceptedRate,
+    eventsRejectedRate,
+    eventsAcceptedTotal,
+    eventsRejectedTotal,
+    activeConnections,
+    cpuLoadPercent,
+    memoryUsedMb,
+  ] = await Promise.all([
+    collectAdminHealthSnapshot(),
+    queryPrometheusInstant(PROMQL.eventsAcceptedRate),
+    queryPrometheusInstant(PROMQL.eventsRejectedRate),
+    queryPrometheusInstant(PROMQL.eventsAcceptedTotal),
+    queryPrometheusInstant(PROMQL.eventsRejectedTotal),
+    queryPrometheusInstant(PROMQL.activeConnections),
+    queryPrometheusInstant(PROMQL.cpuLoadPercent),
+    queryPrometheusInstant(PROMQL.memoryUsedMb),
+  ])
+
+  const prometheusAvailable =
+    eventsAcceptedRate !== undefined ||
+    eventsRejectedRate !== undefined ||
+    eventsAcceptedTotal !== undefined ||
+    eventsRejectedTotal !== undefined ||
+    activeConnections !== undefined ||
+    cpuLoadPercent !== undefined ||
+    memoryUsedMb !== undefined
+
+  const metricsHealthy = health.database.ok && health.redis.ok
+  let status: AdminMetricsSnapshot['status'] = 'ok'
+
+  if (!prometheusAvailable) {
+    status = 'unavailable'
+  } else if (!metricsHealthy) {
+    status = 'degraded'
+  }
+
+  return {
+    timestamp: Date.now(),
+    status,
+    metrics: {
+      eventsPerSecond: toMetricValue(eventsAcceptedRate),
+      eventsRejectedPerSecond: toMetricValue(eventsRejectedRate),
+      eventsAcceptedTotal: toMetricValue(eventsAcceptedTotal),
+      eventsRejectedTotal: toMetricValue(eventsRejectedTotal),
+      activeConnections: toMetricValue(activeConnections),
+      cpuLoadPercent: toMetricValue(cpuLoadPercent),
+      memoryUsedMb: toMetricValue(memoryUsedMb),
+    },
+    health: {
+      database: health.database,
+      redis: health.redis,
+    },
+    prometheus: {
+      available: prometheusAvailable,
+      ...(prometheusAvailable ? {} : { error: 'Prometheus query returned no data' }),
+    },
+  }
+}
+
+export const collectAdminMetricsSnapshot = async (): Promise<AdminMetricsSnapshot> => {
+  const cacheTtlMs = getAdminMetricsSseIntervalMs()
+  const now = Date.now()
+
+  if (cachedSnapshot && now - cachedAt < cacheTtlMs) {
+    return cachedSnapshot
+  }
+
+  if (snapshotInFlight) {
+    return snapshotInFlight
+  }
+
+  snapshotInFlight = collectAdminMetricsSnapshotUncached()
+    .then((snapshot) => {
+      cachedSnapshot = snapshot
+      cachedAt = Date.now()
+      return snapshot
+    })
+    .finally(() => {
+      snapshotInFlight = undefined
+    })
+
+  return snapshotInFlight
+}
+
+export const getAdminMetricsSseIntervalMs = (): number => {
+  const intervalCandidate = Number(process.env.ADMIN_METRICS_SSE_INTERVAL_MS || 5000)
+
+  if (!Number.isFinite(intervalCandidate) || intervalCandidate < 1000) {
+    return 5000
+  }
+
+  return intervalCandidate
+}

src/utils/prometheus-client.ts

@@ -0,0 +1,83 @@
+import axios from 'axios'
+
+import { createLogger } from '../factories/logger-factory'
+
+const logger = createLogger('prometheus-client')
+
+const DEFAULT_PROMETHEUS_URL = 'http://127.0.0.1:9090'
+
+interface PrometheusInstantQueryResponse {
+  status: 'success' | 'error'
+  data?: {
+    resultType: 'vector' | 'scalar' | string
+    result:
+      | Array<{
+          value?: [number, string]
+        }>
+      | [number, string]
+  }
+  error?: string
+  errorType?: string
+}
+
+export const getPrometheusBaseUrl = (): string => {
+  const configured = process.env.PROMETHEUS_URL?.trim()
+  if (!configured) {
+    return DEFAULT_PROMETHEUS_URL
+  }
+
+  return configured.replace(/\/+$/, '')
+}
+
+export const parsePrometheusInstantQueryScalar = (response: PrometheusInstantQueryResponse): number | undefined => {
+  if (response.status !== 'success') {
+    return undefined
+  }
+
+  const data = response.data
+  let rawValue: string | undefined
+
+  if (data?.resultType === 'scalar' && Array.isArray(data.result) && data.result.length >= 2) {
+    rawValue = String(data.result[1])
+  } else if (Array.isArray(data?.result)) {
+    const vectorResult = data.result as Array<{ value?: [number, string] }>
+    rawValue = vectorResult[0]?.value?.[1]
+  }
+
+  if (rawValue === undefined) {
+    return undefined
+  }
+
+  const parsed = Number(rawValue)
+  return Number.isFinite(parsed) ? parsed : undefined
+}
+
+export const queryPrometheusInstant = async (query: string): Promise<number | undefined> => {
+  const baseUrl = getPrometheusBaseUrl()
+
+  try {
+    const timeoutCandidate = Number(process.env.PROMETHEUS_QUERY_TIMEOUT_MS)
+    const timeoutMs = Number.isFinite(timeoutCandidate) && timeoutCandidate > 0 ? timeoutCandidate : 5000
+
+    const response = await axios.get<PrometheusInstantQueryResponse>(`${baseUrl}/api/v1/query`, {
+      params: { query },
+      timeout: timeoutMs,
+      validateStatus: () => true,
+    })
+
+    if (response.status !== 200) {
+      logger.warn('prometheus query failed with HTTP %d: %s', response.status, query)
+      return undefined
+    }
+
+    if (response.data.status !== 'success') {
+      logger.warn('prometheus query error for %s: %s', query, response.data.error ?? response.data.errorType ?? 'unknown')
+      return undefined
+    }
+
+    return parsePrometheusInstantQueryScalar(response.data)
+  } catch (error) {
+    logger.warn('prometheus query request failed for %s: %o', query, error)
+    return undefined
+  }
+}