From 69c0c5c354b9b0e8a0bab885367c9182ae55ecae Mon Sep 17 00:00:00 2001 From: Deep-Axe Date: Mon, 26 Oct 2026 10:49:56 +0530 Subject: [PATCH] Add per-server metrics and a Prometheus endpoint Derive tool-call metrics from the audit log with one aggregation query shared by PostgreSQL and SQLite: outcome counts per upstream and a fixed latency histogram from 5 ms to 15 s. The gateway serves Prometheus metrics on a separate in-cluster port (gateway.metrics.port, default 9464) that the Ingress does not route: call counters by outcome, a latency histogram, and upstream up, status and last-failure gauges. The pod carries prometheus.io scrape annotations. Administrators see each server's calls, error rate and average and p95 latency for the last 24 hours on its Connections card and at GET /api/metrics?hours=<1-168>. Members do not receive team-wide metrics. --- README.md | 24 ++- apps/dashboard/src/main.tsx | 31 ++++ charts/storelink/templates/apps.yaml | 18 +- charts/storelink/values.yaml | 3 + docs/system-design.md | 1 + docs/verification.md | 19 +- services/gateway/src/dashboard-api.ts | 27 +++ services/gateway/src/main.ts | 28 +++ services/gateway/src/metrics.ts | 234 +++++++++++++++++++++++++ services/gateway/src/postgres-store.ts | 7 + services/gateway/src/registry.ts | 14 ++ services/gateway/src/store.ts | 7 + tests/metrics.test.ts | 184 +++++++++++++++++++ 13 files changed, 585 insertions(+), 12 deletions(-) create mode 100644 services/gateway/src/metrics.ts create mode 100644 tests/metrics.test.ts diff --git a/README.md b/README.md index 577fb89..29df124 100644 --- a/README.md +++ b/README.md @@ -153,6 +153,27 @@ Each answer lists the tools it called, and **Activity** shows the matching audit - **Access:** invite members, and issue or revoke API keys. - **Activity:** the audit log of every tool call, showing user, server, tool, time, duration and outcome. +### Metrics + +Administrators see each server's calls, error rate and average and p95 latency for the last 24 hours on its **Connections** card. The same summary is available as JSON from `GET /api/metrics?hours=<1-168>`. + +The gateway serves Prometheus metrics on port `9464` at `/metrics`. The port is reachable inside the cluster only; the Ingress does not route it, and the pod carries `prometheus.io/scrape` annotations. To inspect it locally: + +```sh +kubectl -n storelink port-forward svc/storelink-gateway 9464:9464 +curl -s localhost:9464/metrics +``` + +| Metric | Type | Labels | +| --------------------------------------------------- | --------- | ----------------------------------- | +| `storelink_tool_calls_total` | counter | `team`, `server`, `slug`, `outcome` | +| `storelink_tool_call_duration_seconds` | histogram | `team`, `server`, `slug` | +| `storelink_upstream_up` | gauge | `team`, `server`, `slug` | +| `storelink_upstream_status` | gauge | `team`, `server`, `slug`, `status` | +| `storelink_upstream_last_failure_timestamp_seconds` | gauge | `team`, `server`, `slug` | + +For example, the error rate per server over five minutes is `sum by (server) (rate(storelink_tool_calls_total{outcome=~"failed|timeout|uncertain"}[5m])) / sum by (server) (rate(storelink_tool_calls_total{outcome!="denied"}[5m]))`. + ## Configuration Common chart values: @@ -165,6 +186,7 @@ Common chart values: | `chat.limits.perUserConcurrent` / `perTeamConcurrent` / `globalConcurrent` | `1` / `4` / `16` | Concurrent chat turns | | `gateway.extraUpstreamOrigins` | `[]` | Cluster-internal origins allowed as upstreams | | `gateway.healthIntervalMs` | `1000` | Interval between upstream health probes | +| `gateway.metrics.enabled`, `gateway.metrics.port` | `true`, `9464` | Prometheus metrics listener inside the cluster | | `store.host`, `gateway.host` | `store.storelink.localhost`, `dashboard.storelink.localhost` | Public hostnames | | `store.timezone` | `Asia/Kolkata` | Store timezone, which defines "today" | | `storage.className` | `local-path` | Storage class for the persistent volumes | @@ -297,7 +319,7 @@ The smoke tests drive Google Chrome through Playwright. `npm run dev:dashboard` | #1 VPS deployment | Documented and templated, not deployed | | #6 Security hardening | Partial | | #7 Scaling | Partial: rate and concurrency limits | -| #8 Observability | Partial: health and audit history | +| #8 Observability | Implemented | | #9 Upgrades and rollback | Documented and rehearsed | | #2–#5, multi-store bonus | Not attempted | diff --git a/apps/dashboard/src/main.tsx b/apps/dashboard/src/main.tsx index deba989..c996ec9 100644 --- a/apps/dashboard/src/main.tsx +++ b/apps/dashboard/src/main.tsx @@ -26,6 +26,15 @@ type Server = { lastSuccessAt?: string; lastFailureAt?: string; lastError?: string; + metrics?: { + calls: number; + errors: number; + timeouts: number; + denied: number; + errorRate: number; + avgLatencyMs: number; + p95LatencyMs: number; + }; }; type Member = { id: string; @@ -441,6 +450,28 @@ function Connections({ me, run }: { me: Me; run: Run }) { )} + {server.metrics && ( + <> +
+
Calls (24h)
+
{server.metrics.calls}
+
+
+
Error rate
+
+ {server.metrics.errorRate}% · {server.metrics.errors}{' '} + errors, {server.metrics.timeouts} timeouts +
+
+
+
Latency
+
+ {server.metrics.avgLatencyMs} ms avg ·{' '} + {server.metrics.p95LatencyMs} ms p95 +
+
+ + )} {me.user.role === 'admin' && (
diff --git a/charts/storelink/templates/apps.yaml b/charts/storelink/templates/apps.yaml index 61ab41e..6eed635 100644 --- a/charts/storelink/templates/apps.yaml +++ b/charts/storelink/templates/apps.yaml @@ -9,6 +9,9 @@ spec: template: metadata: labels: { storelink/component: gateway, app.kubernetes.io/instance: {{ .Release.Name }} } + {{- if .Values.gateway.metrics.enabled }} + annotations: { prometheus.io/scrape: "true", prometheus.io/port: {{ .Values.gateway.metrics.port | quote }}, prometheus.io/path: /metrics } + {{- end }} spec: serviceAccountName: {{ include "storelink.name" . }}-gateway automountServiceAccountToken: true @@ -49,13 +52,20 @@ spec: - { name: CHAT_PER_TEAM_CONCURRENT, value: {{ .Values.chat.limits.perTeamConcurrent | quote }} } - { name: CHAT_GLOBAL_CONCURRENT, value: {{ .Values.chat.limits.globalConcurrent | quote }} } - { name: HEALTH_INTERVAL_MS, value: {{ .Values.gateway.healthIntervalMs | quote }} } + {{- if .Values.gateway.metrics.enabled }} + - { name: METRICS_PORT, value: {{ .Values.gateway.metrics.port | quote }} } + {{- end }} {{- if .Values.chat.enabled }} - { name: OPENAI_API_KEY_FILE, value: /secrets/openai-api-key } {{- end }} - { name: WOO_MCP_URL, value: "http://{{ include "storelink.name" . }}-woo-mcp:3001/mcp" } - { name: GATEWAY_ORIGIN, value: {{ .Values.gateway.publicUrl | quote }} } - { name: TEAM_ID, value: {{ .Values.teamId | quote }} } - ports: [{ name: http, containerPort: 3000 }] + ports: + - { name: http, containerPort: 3000 } + {{- if .Values.gateway.metrics.enabled }} + - { name: metrics, containerPort: {{ .Values.gateway.metrics.port }} } + {{- end }} startupProbe: { httpGet: { path: /health/live, port: http }, periodSeconds: 2, failureThreshold: 60 } readinessProbe: { httpGet: { path: /health/ready, port: http }, periodSeconds: 5, timeoutSeconds: 5 } livenessProbe: { httpGet: { path: /health/live, port: http }, periodSeconds: 10 } @@ -88,7 +98,11 @@ metadata: name: {{ include "storelink.name" . }}-gateway spec: selector: { storelink/component: gateway, app.kubernetes.io/instance: {{ .Release.Name }} } - ports: [{ port: 3000, targetPort: http }] + ports: + - { name: http, port: 3000, targetPort: http } + {{- if .Values.gateway.metrics.enabled }} + - { name: metrics, port: {{ .Values.gateway.metrics.port }}, targetPort: metrics } + {{- end }} --- apiVersion: apps/v1 kind: Deployment diff --git a/charts/storelink/values.yaml b/charts/storelink/values.yaml index 7f4e17a..7723f1a 100644 --- a/charts/storelink/values.yaml +++ b/charts/storelink/values.yaml @@ -27,6 +27,9 @@ store: gateway: extraUpstreamOrigins: [] healthIntervalMs: 1000 + metrics: + enabled: true + port: 9464 host: dashboard.storelink.localhost publicUrl: http://dashboard.storelink.localhost resources: diff --git a/docs/system-design.md b/docs/system-design.md index 5a24951..db4dd91 100644 --- a/docs/system-design.md +++ b/docs/system-design.md @@ -41,6 +41,7 @@ The gateway (`services/gateway/src/gateway.ts`) uses `@modelcontextprotocol/sdk` - **Namespacing and collision handling:** Public tool names follow `__` (capped at 64 characters). Each upstream has a unique slug per team (up to 16 characters), and tool names with special characters or length over 46 characters receive a deterministic `_` alias. The gateway stores the `:__` route map in memory and dispatches calls by looking up the target upstream and original tool name. Colliding upstream tool names such as `get_status` on both `time` and `ops` are exposed as `time__get_status` and `ops__get_status`. - **Timeouts and bounds:** Upstream TCP connections have a 2-second timeout (`undici.Agent`), tool discovery has a 3-second deadline (up to 5 pages / 200 tools), and tool calls have a 15-second deadline (256 KiB response cap). Arguments are validated against the upstream's JSON Schema with Ajv before dispatch. - **Health probing and failure isolation:** The gateway pings each connected upstream every second (`gateway.healthIntervalMs`, 1-second deadline) and refreshes full tool catalogs every 15 seconds. If a health probe, discovery or tool call fails with a transport or timeout error, only that upstream is marked `Unreachable`, its routes are removed from the active catalog, and the `Unreachable` state is held for a 2-second retry window before rediscovery. Upstream MCP client transports reset on probe or discovery failure so a restarted pod receives a fresh session. Status transitions and timestamps are persisted to PostgreSQL (`upstream_events` and `upstream_servers`). +- **Metrics:** Call metrics are derived from `audit_calls`, which already records every call's server, outcome and duration (`services/gateway/src/metrics.ts`). One aggregation query, shared by PostgreSQL and the SQLite test store, counts outcomes and fills a fixed latency histogram (5 ms to 15 s buckets). The gateway serves the result in Prometheus format on a separate in-cluster port (`gateway.metrics.port`, default `9464`) that the Ingress does not route, so tool names and call volumes are not public. Administrators also see a 24-hour summary per server (calls, error rate, average and p95 latency, with p95 interpolated from the histogram buckets) in the dashboard and at `GET /api/metrics`. Members do not receive team-wide metrics, matching the audit log's visibility. - **Downstream sessions and catalog notifications:** `/mcp` uses stateful Streamable HTTP sessions (`services/gateway/src/mcp.ts`) with JSON POST responses and a GET SSE stream. When a team's tool catalog changes (server added, edited, disabled, removed, failed or recovered), the gateway sends `notifications/tools/list_changed` only to active sessions belonging to that team. Every request rechecks user and API key status; sessions expire after 15 minutes idle (capped at 8 per user / 256 globally). - **Mutations and auditing:** Write tools (`create_product`, `update_order_status`) are annotated with `readOnlyHint: false` and are never automatically retried. If a connection drops or times out after a write request is sent, the adapter and gateway return and audit the outcome as `uncertain` so callers check state before retrying. Every tool call writes a `started` audit record to PostgreSQL before dispatch and updates it with the duration and outcome (`success`, `failed`, `timeout`, `uncertain`, or `denied`). If the audit write fails, the tool call is aborted. diff --git a/docs/verification.md b/docs/verification.md index ac0f494..2091332 100644 --- a/docs/verification.md +++ b/docs/verification.md @@ -2,7 +2,7 @@ ## Automated test suite -`npm test` runs 37 deterministic unit and protocol tests across ten test files without making paid OpenAI calls: +`npm test` runs 40 deterministic unit and protocol tests across eleven test files without making paid OpenAI calls: - **`tests/gateway.test.ts`:** API key HMAC digest lookup, revoked-key attribution after restart, `__` tool collision routing (`time__get_status` vs `ops__get_status`), immediate route removal on disable, hung upstream timeout isolation, uncertain mutation auditing without retries, argument schema validation, cross-team route isolation, audit-failure fail-closed behavior, and full MCP SDK Streamable HTTP exchanges. - **`tests/mcp-notifications.test.ts`:** Stateful Streamable HTTP + GET SSE sessions, team-scoped `notifications/tools/list_changed` delivery, suppression of unchanged catalog refreshes, cross-user/cross-key session rejection (`403`), revoked-key notification cutoff, per-user/global session limits (`429`), `DELETE` cleanup, and idle session expiry (`404`). @@ -10,6 +10,7 @@ - **`tests/upstream-policy.test.ts` & `tests/upstream-network.test.ts`:** Public HTTPS registration without a Helm allowlist change, rejection of loopback/private/link-local/metadata/mixed DNS answers, socket-level DNS rebinding prevention, and 2-second TCP connect timeouts that do not truncate long-running tool calls. - **`tests/woocommerce.test.ts`:** OAuth 1.0a HMAC-SHA256 canonical host signing, direct HTTP transport for both HTTP and HTTPS public store URLs, redirect/HTML/auth error rejection without mutation retry, and service token / CORS enforcement. - **`tests/chat.test.ts`, `tests/chat-admission.test.ts` & `tests/dashboard-api.test.ts`:** Scripted OpenAI Responses multi-round tool continuation, mid-turn catalog/authz rechecks, round/call/context bounds, DST-aware store day windows, categorized provider/timeout errors, atomic conversation persistence, and per-user/team/global admission limits that do not charge rejected requests against the hourly quota. +- **`tests/metrics.test.ts`:** Per-server outcome counts and latency aggregation from the audit log, team and time-window filters, histogram-based p95 interpolation, and Prometheus exposition (counters, cumulative histogram buckets, upstream gauges, label escaping). - **`tests/scripts.test.ts`:** `kubectl` binary resolution order and host-wide `flock` cluster lock contention (`exit 73`) and release. ## Cluster smoke checks @@ -38,14 +39,14 @@ Claude Code discovered the tools of all three servers (`mcp__storelink__woo__*`, Mapping against the assignment brief's numbered stand-out items: -| Item | Status | Implementation | -| -------------------------------------------- | ------------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| **#1 Production-like VPS deployment** | Documented & templated (not deployed to a live VPS) | `charts/storelink/values-prod.yaml`, [README production guide](../README.md#production-deployment-k3s), [system design](system-design.md#production-changes) | -| **#6 Security hardening** | Partial | SSRF & socket-level DNS rebinding checks (`services/gateway/src/upstream-policy.ts`, `tests/upstream-policy.test.ts`); single-Secret least-privilege RBAC (`charts/storelink/templates/registry-rbac.yaml`, `services/gateway/src/credentials.ts`); ingress `NetworkPolicy` isolation on MCP servers and databases (`charts/storelink/templates/network-policies.yaml`); non-root app containers (`runAsNonRoot: true`, UID `1000`). Default-deny egress, non-root WordPress/DB images, and cluster Secret encryption at rest are not enabled by default. | -| **#7 Scaling** | Partial (rate limits only; single gateway replica) | Per-user hourly chat quota (`15/hr`), per-user/team/global concurrent turn limits (`services/gateway/src/chat-admission.ts`), per-user API request throttling (`180/min`), and login/invite rate limits (`services/gateway/src/dashboard-api.ts`). | -| **#8 Observability** | Partial (health & audit history; no Prometheus metrics) | 1-second upstream ping probes, persisted `lastError` / `lastFailureAt` / `lastSuccessAt` status fields, `upstream_events` transition history in the Connections UI (`services/gateway/src/registry.ts`), and per-call duration/outcome/reason tracking in `audit_calls`. | -| **#9 Upgrades and rollback** | Implemented & rehearsed | Revision-scoped store bootstrap Job (`charts/storelink/templates/store.yaml`), backward-compatible additive SQL migrations (`services/gateway/migrations/`), and an upgrade/rollback procedure rehearsed on the local cluster, with backups verified by restoring them into scratch databases, in [README](../README.md#upgrades-and-rollback) and [system design](system-design.md#upgrades-and-rollback). | -| **Bonus (Multi-store provisioning) & #2–#5** | Not attempted | Deferred in favor of completing the core platform and the items above. | +| Item | Status | Implementation | +| -------------------------------------------- | --------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| **#1 Production-like VPS deployment** | Documented & templated (not deployed to a live VPS) | `charts/storelink/values-prod.yaml`, [README production guide](../README.md#production-deployment-k3s), [system design](system-design.md#production-changes) | +| **#6 Security hardening** | Partial | SSRF & socket-level DNS rebinding checks (`services/gateway/src/upstream-policy.ts`, `tests/upstream-policy.test.ts`); single-Secret least-privilege RBAC (`charts/storelink/templates/registry-rbac.yaml`, `services/gateway/src/credentials.ts`); ingress `NetworkPolicy` isolation on MCP servers and databases (`charts/storelink/templates/network-policies.yaml`); non-root app containers (`runAsNonRoot: true`, UID `1000`). Default-deny egress, non-root WordPress/DB images, and cluster Secret encryption at rest are not enabled by default. | +| **#7 Scaling** | Partial (rate limits only; single gateway replica) | Per-user hourly chat quota (`15/hr`), per-user/team/global concurrent turn limits (`services/gateway/src/chat-admission.ts`), per-user API request throttling (`180/min`), and login/invite rate limits (`services/gateway/src/dashboard-api.ts`). | +| **#8 Observability** | Implemented | Prometheus `/metrics` on in-cluster port `9464` with per-server call counters by outcome, a latency histogram, and upstream up/status/last-failure gauges (`services/gateway/src/metrics.ts`, `services/gateway/src/main.ts`); 24-hour per-server calls, error rate and avg/p95 latency on dashboard cards and `GET /api/metrics`; 1-second health probes with persisted `lastError` and `upstream_events` history; per-call outcome, reason and duration in `audit_calls`. Tests: `tests/metrics.test.ts`. | +| **#9 Upgrades and rollback** | Implemented & rehearsed | Revision-scoped store bootstrap Job (`charts/storelink/templates/store.yaml`), backward-compatible additive SQL migrations (`services/gateway/migrations/`), and an upgrade/rollback procedure rehearsed on the local cluster, with backups verified by restoring them into scratch databases, in [README](../README.md#upgrades-and-rollback) and [system design](system-design.md#upgrades-and-rollback). | +| **Bonus (Multi-store provisioning) & #2–#5** | Not attempted | Deferred in favor of completing the core platform and the items above. | Additional platform features: diff --git a/services/gateway/src/dashboard-api.ts b/services/gateway/src/dashboard-api.ts index e36b49e..2edb115 100644 --- a/services/gateway/src/dashboard-api.ts +++ b/services/gateway/src/dashboard-api.ts @@ -16,6 +16,7 @@ import { CredentialInputError } from './credentials.ts'; import type { Gateway } from './gateway.ts'; import { runChat, ChatInputError, validateChatInput } from './chat.ts'; import type { Model } from './chat.ts'; +import { summarizeCalls } from './metrics.ts'; class HttpError extends Error { status: number; @@ -237,6 +238,32 @@ export function dashboardApi(options: Options) { ).rows, ); }); + router.get('/metrics', async (req, res) => { + admin(res); + const hours = z.coerce + .number() + .int() + .min(1) + .max(168) + .default(24) + .parse(req.query.hours); + const { teamId } = actor(res); + const stats = await store.callStats({ + teamId, + since: new Date(Date.now() - hours * 60 * 60 * 1000), + }); + res.json({ + hours, + servers: [...gateway.upstreams.values()] + .filter((upstream) => upstream.teamId === teamId) + .map((upstream) => ({ + id: upstream.id, + slug: upstream.slug, + status: upstream.status, + ...summarizeCalls(stats.find((s) => s.serverId === upstream.id)), + })), + }); + }); router.get('/tools', async (_req, res) => res.json(await gateway.listTools(actor(res))), ); diff --git a/services/gateway/src/main.ts b/services/gateway/src/main.ts index f55b0fc..f8f0da3 100644 --- a/services/gateway/src/main.ts +++ b/services/gateway/src/main.ts @@ -1,4 +1,5 @@ import express from 'express'; +import { createServer } from 'node:http'; import { existsSync } from 'node:fs'; import { resolve } from 'node:path'; import { expressMcpApp } from '../../../packages/shared/src/mcp-http.ts'; @@ -12,6 +13,7 @@ import { FileCredentials, KubernetesCredentials } from './credentials.ts'; import { Registry } from './registry.ts'; import { dashboardApi } from './dashboard-api.ts'; import { openaiModel } from './chat.ts'; +import { renderPrometheus } from './metrics.ts'; const usePostgres = Boolean( process.env.DATABASE_URL || process.env.DATABASE_URL_FILE, @@ -129,6 +131,31 @@ server.on('error', () => { console.error('Unable to bind gateway listener'); process.exitCode = 1; }); +// Prometheus metrics on a separate port that the Ingress does not route. +const metricsPort = Number(process.env.METRICS_PORT ?? 0); +const metricsServer = metricsPort + ? createServer((req, res) => { + if (req.method !== 'GET' || req.url !== '/metrics') { + res.writeHead(404).end(); + return; + } + void (async () => { + try { + const body = renderPrometheus(await store.callStats(), [ + ...gateway.upstreams.values(), + ]); + res + .writeHead(200, { + 'content-type': 'text/plain; version=0.0.4; charset=utf-8', + }) + .end(body); + } catch { + console.error('Metrics collection failed'); + res.writeHead(503).end(); + } + })(); + }).listen(metricsPort, process.env.BIND_ADDRESS ?? '127.0.0.1') + : undefined; let refreshing = false; const refresh = setInterval(() => { if (refreshing) return; @@ -180,6 +207,7 @@ for (const signal of ['SIGINT', 'SIGTERM']) clearInterval(refresh); clearInterval(health); server.close(); + metricsServer?.close(); await mcpHandler.close(); await registry?.close(); await developmentClient?.close(); diff --git a/services/gateway/src/metrics.ts b/services/gateway/src/metrics.ts new file mode 100644 index 0000000..cdf207a --- /dev/null +++ b/services/gateway/src/metrics.ts @@ -0,0 +1,234 @@ +// Tool-call metrics derived from the audit log, plus Prometheus text exposition. + +// Upper bounds (ms) of the latency histogram buckets; the gateway call deadline is 15 s. +export const latencyBucketsMs = [ + 5, 10, 25, 50, 100, 250, 500, 1000, 2500, 5000, 10000, 15000, +] as const; + +const completedOutcomes = `('success','failed','timeout','uncertain')`; + +export type CallStatsFilter = { teamId?: string; since?: Date }; + +export type ServerCallStats = { + teamId: string; + serverId: string; + success: number; + failed: number; + timeout: number; + uncertain: number; + denied: number; + durationSumMs: number; + // Cumulative counts of completed calls at or below each latencyBucketsMs bound. + buckets: number[]; +}; + +export type ServerCallSummary = { + calls: number; + errors: number; + timeouts: number; + denied: number; + errorRate: number; + avgLatencyMs: number; + p95LatencyMs: number; +}; + +export type UpstreamState = { + id: string; + teamId: string; + slug: string; + status: string; + lastFailureAt?: string; +}; + +// One query for both PostgreSQL ($n placeholders) and SQLite (? placeholders). +export function callStatsQuery( + style: 'postgres' | 'sqlite', + filter: CallStatsFilter, +): { text: string; values: string[] } { + const values: string[] = []; + const param = (value: string) => { + values.push(value); + return style === 'postgres' ? `$${values.length}` : '?'; + }; + const where = ['server_id IS NOT NULL', `outcome <> 'started'`]; + if (filter.teamId) where.push(`team_id = ${param(filter.teamId)}`); + if (filter.since) + where.push(`timestamp >= ${param(filter.since.toISOString())}`); + const count = (condition: string) => + `SUM(CASE WHEN ${condition} THEN 1 ELSE 0 END)`; + const buckets = latencyBucketsMs.map( + (bound, index) => + `${count(`outcome IN ${completedOutcomes} AND duration_ms <= ${bound}`)} AS le_${index}`, + ); + return { + text: `SELECT team_id, server_id, + ${count(`outcome = 'success'`)} AS success, + ${count(`outcome = 'failed'`)} AS failed, + ${count(`outcome = 'timeout'`)} AS timeout, + ${count(`outcome = 'uncertain'`)} AS uncertain, + ${count(`outcome = 'denied'`)} AS denied, + SUM(CASE WHEN outcome IN ${completedOutcomes} THEN duration_ms ELSE 0 END) AS duration_sum, + ${buckets.join(',\n ')} + FROM audit_calls + WHERE ${where.join(' AND ')} + GROUP BY team_id, server_id + ORDER BY team_id, server_id`, + values, + }; +} + +export function parseCallStats( + rows: Record[], +): ServerCallStats[] { + const number = (value: unknown) => Number(value ?? 0); + return rows.map((row) => ({ + teamId: String(row.team_id ?? ''), + serverId: String(row.server_id), + success: number(row.success), + failed: number(row.failed), + timeout: number(row.timeout), + uncertain: number(row.uncertain), + denied: number(row.denied), + durationSumMs: number(row.duration_sum), + buckets: latencyBucketsMs.map((_, index) => number(row[`le_${index}`])), + })); +} + +const completed = (stats: ServerCallStats) => + stats.success + stats.failed + stats.timeout + stats.uncertain; + +// Linear interpolation inside the matching bucket, as Prometheus histogram_quantile does. +export function estimateQuantile( + quantile: number, + buckets: number[], + total: number, +): number { + if (total === 0) return 0; + const rank = quantile * total; + for (let index = 0; index < buckets.length; index++) { + if (buckets[index] >= rank) { + const lower = index === 0 ? 0 : latencyBucketsMs[index - 1]; + const below = index === 0 ? 0 : buckets[index - 1]; + const inBucket = buckets[index] - below; + const fraction = inBucket === 0 ? 1 : (rank - below) / inBucket; + return Math.round(lower + (latencyBucketsMs[index] - lower) * fraction); + } + } + return latencyBucketsMs[latencyBucketsMs.length - 1]; +} + +export function summarizeCalls(stats?: ServerCallStats): ServerCallSummary { + if (!stats) + return { + calls: 0, + errors: 0, + timeouts: 0, + denied: 0, + errorRate: 0, + avgLatencyMs: 0, + p95LatencyMs: 0, + }; + const calls = completed(stats); + const errors = stats.failed + stats.timeout + stats.uncertain; + return { + calls, + errors, + timeouts: stats.timeout, + denied: stats.denied, + errorRate: calls ? Math.round((errors / calls) * 1000) / 10 : 0, + avgLatencyMs: calls ? Math.round(stats.durationSumMs / calls) : 0, + p95LatencyMs: estimateQuantile(0.95, stats.buckets, calls), + }; +} + +const escapeLabel = (value: string) => + value.replace(/\\/g, '\\\\').replace(/\n/g, '\\n').replace(/"/g, '\\"'); +const labels = (pairs: Record) => + `{${Object.entries(pairs) + .map(([key, value]) => `${key}="${escapeLabel(value)}"`) + .join(',')}}`; + +export function renderPrometheus( + stats: ServerCallStats[], + upstreams: UpstreamState[], +): string { + const slugs = new Map(upstreams.map((u) => [`${u.teamId}:${u.id}`, u.slug])); + const series = (s: ServerCallStats) => ({ + team: s.teamId, + server: s.serverId, + slug: slugs.get(`${s.teamId}:${s.serverId}`) ?? s.serverId, + }); + const lines: string[] = []; + const metric = (name: string, type: string, help: string) => + lines.push(`# HELP ${name} ${help}`, `# TYPE ${name} ${type}`); + + metric( + 'storelink_tool_calls_total', + 'counter', + 'Tool calls routed to an upstream, by outcome.', + ); + for (const s of stats) + for (const outcome of [ + 'success', + 'failed', + 'timeout', + 'uncertain', + 'denied', + ] as const) + lines.push( + `storelink_tool_calls_total${labels({ ...series(s), outcome })} ${s[outcome]}`, + ); + + metric( + 'storelink_tool_call_duration_seconds', + 'histogram', + 'Duration of completed tool calls.', + ); + for (const s of stats) { + const base = series(s); + latencyBucketsMs.forEach((bound, index) => + lines.push( + `storelink_tool_call_duration_seconds_bucket${labels({ ...base, le: String(bound / 1000) })} ${s.buckets[index]}`, + ), + ); + lines.push( + `storelink_tool_call_duration_seconds_bucket${labels({ ...base, le: '+Inf' })} ${completed(s)}`, + `storelink_tool_call_duration_seconds_sum${labels(base)} ${s.durationSumMs / 1000}`, + `storelink_tool_call_duration_seconds_count${labels(base)} ${completed(s)}`, + ); + } + + metric( + 'storelink_upstream_up', + 'gauge', + 'Whether the upstream is connected (1) or not (0).', + ); + for (const u of upstreams) + lines.push( + `storelink_upstream_up${labels({ team: u.teamId, server: u.id, slug: u.slug })} ${u.status === 'Connected' ? 1 : 0}`, + ); + + metric( + 'storelink_upstream_status', + 'gauge', + 'Current upstream status; the matching status label is 1.', + ); + for (const u of upstreams) + for (const status of ['Connected', 'Unreachable', 'Disabled']) + lines.push( + `storelink_upstream_status${labels({ team: u.teamId, server: u.id, slug: u.slug, status })} ${u.status === status ? 1 : 0}`, + ); + + metric( + 'storelink_upstream_last_failure_timestamp_seconds', + 'gauge', + 'Unix time of the most recent upstream failure.', + ); + for (const u of upstreams) + if (u.lastFailureAt) + lines.push( + `storelink_upstream_last_failure_timestamp_seconds${labels({ team: u.teamId, server: u.id, slug: u.slug })} ${Math.floor(Date.parse(u.lastFailureAt) / 1000)}`, + ); + + return `${lines.join('\n')}\n`; +} diff --git a/services/gateway/src/postgres-store.ts b/services/gateway/src/postgres-store.ts index 6f8f886..768d6d1 100644 --- a/services/gateway/src/postgres-store.ts +++ b/services/gateway/src/postgres-store.ts @@ -3,6 +3,8 @@ import { readFileSync, readdirSync } from 'node:fs'; import { Pool } from 'pg'; import type { Identity } from '../../../packages/shared/src/api-keys.ts'; import type { AuditResult, GatewayStore, KeyRecord } from './store.ts'; +import { callStatsQuery, parseCallStats } from './metrics.ts'; +import type { CallStatsFilter, ServerCallStats } from './metrics.ts'; export class PostgresStore implements GatewayStore { pool: Pool; @@ -137,6 +139,11 @@ export class PostgresStore implements GatewayStore { ); } + async callStats(filter: CallStatsFilter = {}): Promise { + const { text, values } = callStatsQuery('postgres', filter); + return parseCallStats((await this.pool.query(text, values)).rows); + } + async auditRows(teamId: string): Promise[]> { const result = await this.pool.query( 'SELECT * FROM audit_calls WHERE team_id=$1 ORDER BY sequence DESC LIMIT 200', diff --git a/services/gateway/src/registry.ts b/services/gateway/src/registry.ts index 439ab92..c420d7e 100644 --- a/services/gateway/src/registry.ts +++ b/services/gateway/src/registry.ts @@ -5,6 +5,7 @@ import type { Gateway } from './gateway.ts'; import { upstreamClient } from './upstream.ts'; import type { Credential, CredentialStore } from './credentials.ts'; import { validateCredential } from './credentials.ts'; +import { summarizeCalls } from './metrics.ts'; import { UpstreamPolicy } from './upstream-policy.ts'; export { RegistryInputError } from './upstream-policy.ts'; @@ -147,6 +148,18 @@ export class Registry { 'SELECT * FROM upstream_servers WHERE team_id=$1 ORDER BY slug', [actor.teamId], ); + // Team-wide call volume is administrator information, like the audit log. + const stats = + actor.role === 'admin' + ? new Map( + ( + await this.store.callStats({ + teamId: actor.teamId, + since: new Date(Date.now() - 24 * 60 * 60 * 1000), + }) + ).map((s) => [s.serverId, s]), + ) + : undefined; return rows.map((row) => { const runtime = this.gateway.upstreams.get(row.id); return { @@ -159,6 +172,7 @@ export class Registry { lastSuccessAt: runtime?.lastSuccessAt ?? row.last_success_at, lastFailureAt: runtime?.lastFailureAt ?? row.last_failure_at, lastError: runtime?.lastError ?? row.last_error, + metrics: stats ? summarizeCalls(stats.get(row.id)) : undefined, tools: [...this.gateway.routes.values()] .filter((route) => route.upstream.id === row.id) .map((route) => route.tool.name), diff --git a/services/gateway/src/store.ts b/services/gateway/src/store.ts index cd2cdc8..197dc5e 100644 --- a/services/gateway/src/store.ts +++ b/services/gateway/src/store.ts @@ -3,6 +3,8 @@ import { chmodSync, mkdirSync } from 'node:fs'; import { dirname } from 'node:path'; import { DatabaseSync } from 'node:sqlite'; import type { Identity } from '../../../packages/shared/src/api-keys.ts'; +import { callStatsQuery, parseCallStats } from './metrics.ts'; +import type { CallStatsFilter, ServerCallStats } from './metrics.ts'; export type KeyRecord = Identity & { revokedAt: string | null }; export type AuditOutcome = @@ -158,6 +160,11 @@ export class DevelopmentStore { return this.db.prepare('SELECT * FROM audit_calls ORDER BY rowid').all(); } + callStats(filter: CallStatsFilter = {}): ServerCallStats[] { + const { text, values } = callStatsQuery('sqlite', filter); + return parseCallStats(this.db.prepare(text).all(...values)); + } + close(): void { this.db.close(); } diff --git a/tests/metrics.test.ts b/tests/metrics.test.ts new file mode 100644 index 0000000..8392b19 --- /dev/null +++ b/tests/metrics.test.ts @@ -0,0 +1,184 @@ +import assert from 'node:assert/strict'; +import { randomBytes } from 'node:crypto'; +import { test } from 'node:test'; +import { generateApiKey, hashApiKey } from '../packages/shared/src/api-keys.ts'; +import { DevelopmentStore } from '../services/gateway/src/store.ts'; +import { Gateway } from '../services/gateway/src/gateway.ts'; +import type { ToolClient } from '../services/gateway/src/gateway.ts'; +import { + estimateQuantile, + latencyBucketsMs, + renderPrometheus, + summarizeCalls, +} from '../services/gateway/src/metrics.ts'; + +const tool = (name: string) => ({ + name, + inputSchema: { + type: 'object' as const, + properties: {}, + additionalProperties: false, + }, + annotations: { readOnlyHint: true }, +}); + +function client(behaviour: 'ok' | 'fail' | 'hang'): ToolClient { + return { + async listTools() { + return { tools: [tool('get_status')] }; + }, + async callTool() { + if (behaviour === 'fail') throw new TypeError('fetch failed'); + if (behaviour === 'hang') await new Promise(() => {}); + return { content: [{ type: 'text', text: 'ok' }] }; + }, + }; +} + +async function setup() { + const store = new DevelopmentStore(':memory:'); + const pepper = randomBytes(32); + const gateway = new Gateway(store, pepper, { discoveryMs: 50, callMs: 40 }); + const key = generateApiKey(); + store.addUser('admin-a', 'team', 'admin'); + store.addKey('admin-a', hashApiKey(key, pepper)); + for (const [id, behaviour] of [ + ['time', 'ok'], + ['ops', 'fail'], + ['slow', 'hang'], + ] as const) + gateway.register({ + id, + teamId: 'team', + slug: id, + client: client(behaviour), + }); + await gateway.refreshAll(); + return { store, gateway, actor: await gateway.authenticate(key) }; +} + +test('call statistics aggregate outcomes and latency per upstream from the audit log', async (t) => { + const { store, gateway, actor } = await setup(); + t.after(() => store.close()); + for (let i = 0; i < 3; i++) await gateway.call(actor, 'time__get_status', {}); + await gateway.call(actor, 'ops__get_status', {}); + await gateway.call(actor, 'slow__get_status', {}); + + const stats = new Map(store.callStats().map((s) => [s.serverId, s])); + assert.equal(stats.get('time')?.success, 3); + assert.equal(stats.get('ops')?.failed, 1); + assert.equal(stats.get('slow')?.timeout, 1); + + const time = summarizeCalls(stats.get('time')); + assert.equal(time.calls, 3); + assert.equal(time.errorRate, 0); + const slow = summarizeCalls(stats.get('slow')); + assert.equal(slow.errors, 1); + assert.equal(slow.timeouts, 1); + assert.equal(slow.errorRate, 100); + assert.ok(slow.avgLatencyMs >= 40); + assert.ok(slow.p95LatencyMs > 0); + + // Rows outside the window and other teams are excluded. + assert.equal(store.callStats({ teamId: 'other' }).length, 0); + assert.equal( + store.callStats({ since: new Date(Date.now() + 60_000) }).length, + 0, + ); + assert.equal(store.callStats({ teamId: 'team' }).length, 3); +}); + +test('p95 is interpolated within the matching histogram bucket', () => { + // 100 calls: 90 at <=5 ms, 10 between 50 and 100 ms. + const buckets = latencyBucketsMs.map((bound) => + bound < 100 ? 90 : 100, + ) as number[]; + assert.equal(estimateQuantile(0.95, buckets, 100), 75); + assert.equal(estimateQuantile(0.5, buckets, 100), 3); + assert.equal( + estimateQuantile( + 0.95, + buckets.map(() => 0), + 0, + ), + 0, + ); + // Calls slower than the largest bound report that bound. + assert.equal( + estimateQuantile( + 0.95, + buckets.map(() => 0), + 10, + ), + latencyBucketsMs[latencyBucketsMs.length - 1], + ); + assert.deepEqual(summarizeCalls(undefined), { + calls: 0, + errors: 0, + timeouts: 0, + denied: 0, + errorRate: 0, + avgLatencyMs: 0, + p95LatencyMs: 0, + }); +}); + +test('Prometheus output exposes counters, a cumulative histogram and upstream state', async (t) => { + const { store, gateway, actor } = await setup(); + t.after(() => store.close()); + await gateway.call(actor, 'time__get_status', {}); + await gateway.call(actor, 'ops__get_status', {}); + gateway.upstreams.get('slow')!.slug = 'slow "lane"'; + const text = renderPrometheus(store.callStats(), [ + ...gateway.upstreams.values(), + ]); + + assert.match(text, /# TYPE storelink_tool_calls_total counter/); + assert.match( + text, + /storelink_tool_calls_total\{team="team",server="time",slug="time",outcome="success"\} 1/, + ); + assert.match( + text, + /storelink_tool_calls_total\{team="team",server="ops",slug="ops",outcome="failed"\} 1/, + ); + assert.match(text, /# TYPE storelink_tool_call_duration_seconds histogram/); + assert.match( + text, + /storelink_tool_call_duration_seconds_bucket\{team="team",server="time",slug="time",le="\+Inf"\} 1/, + ); + assert.match( + text, + /storelink_tool_call_duration_seconds_count\{team="team",server="time",slug="time"\} 1/, + ); + // Bucket counts never decrease as the bound grows. + const timeBuckets = [ + ...text.matchAll( + /storelink_tool_call_duration_seconds_bucket\{team="team",server="time",slug="time",le="[^"]+"\} (\d+)/g, + ), + ].map((match) => Number(match[1])); + assert.equal(timeBuckets.length, latencyBucketsMs.length + 1); + assert.deepEqual( + timeBuckets, + [...timeBuckets].sort((a, b) => a - b), + ); + assert.match( + text, + /storelink_upstream_up\{team="team",server="time",slug="time"\} 1/, + ); + assert.match( + text, + /storelink_upstream_up\{team="team",server="ops",slug="ops"\} 0/, + ); + assert.match( + text, + /storelink_upstream_status\{team="team",server="ops",slug="ops",status="Unreachable"\} 1/, + ); + assert.match( + text, + /storelink_upstream_last_failure_timestamp_seconds\{team="team",server="ops",slug="ops"\} \d+/, + ); + // Label values are escaped. + assert.match(text, /slug="slow \\"lane\\""/); + assert.ok(text.endsWith('\n')); +});