diff --git a/CLAUDE.md b/CLAUDE.md index 71434e5..6ce1d70 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -12,6 +12,19 @@ symphony is a **napi-rs cdylib** loaded by Node.js. The tokio multi-thread runti For consumers that want symphony as its own OS process rather than embedded in their Node app, the package ships a `symphony-server` bin. It reads a JSON config file (`{ version, proxies: [{ listeners, routes }] }` — one entry per port-set, since the route table is per-proxy), constructs a `SymphonyProxy` per entry, and **watches the config file** to hot-reload (route change → `updateConfig`; listener change → recreate that proxy). Cert material may be given inline (`certChain`/`privateKey`) or by path (`certChainFile`/`privateKeyFile`) — the path form is resolved in `server.ts` only, so the napi `CertConfig` stays inline-only. It writes a `status.json` (`{ pid, version, ports, ... }`) for supervisors, and handles `SIGHUP` (reload) / `SIGTERM`/`SIGINT` (graceful stop). host-manager uses this to supervise symphony out-of-process. +### Admin/metrics endpoint (`ts/admin.ts`) + +An optional `admin` block in the config file (`{ socketPath?, socketMode?, port?, host? }`) makes `symphony-server` expose `GET /metrics` (Prometheus text), `/metrics.json`, and `/health` over a Unix socket, a loopback TCP port, or both. It exists because an out-of-process symphony has no reachable napi `metrics()` — the endpoint is the only export path for that deployment. + +Three properties are load-bearing and easy to regress: + +- **It must never affect proxying.** A bind failure is logged and retried on a 5s timer instead of throwing out of `doReconcile()`. This is not defensiveness for its own sake: during a version upgrade host-manager runs both processes concurrently (the Rust listeners overlap via `SO_REUSEPORT`, which a Node HTTP server has no equivalent for), so the successor *will* lose the admin bind for a few seconds and must keep serving traffic anyway. +- **A stale Unix socket is reclaimed, a live one is not.** Three things protect this, and each was a real hole once: the probe counts a path reclaimable only on `ECONNREFUSED` (an `EACCES` from a restrictively-permissioned live socket is not evidence nobody is listening); the inode must actually be a socket (a `socketPath` misconfigured onto a regular file would otherwise be deleted); and the bind happens on a pid-unique temp path that is `rename`d into place, so there is no probe→unlink→bind window for a second process to delete a socket the first has already bound. On shutdown the published path is unlinked only while its inode is still the one we put there. Same family as the `status.json` ownership guard. +- **Counters are read per request**, not cached at reconcile, so a scrape never serves numbers frozen at the last config reload. +- **Totals are derived, never maintained alongside their parts.** `blocked`/`errors` are summed from the per-reason values in the same snapshot, and the proxy-wide blocked total from the listener values. A separate `total_blocked` incremented next to its reason counter is two non-atomic writes: a scrape landing between them sees a total that disagrees with its own breakdown, so the invariant would hold only while the proxy is idle — precisely when nobody is reading it. + +Prometheus shape: blocked/error counts are emitted **only** under their `reason` label (they sum to the unlabeled total, so a separate total would be a second representation of the same number), and the proxy-wide active gauge is `sum without(listener)`. `renderPrometheus` is exported from the package for embedded consumers. + The server also **watches the cert/key files referenced by the config** (grouped by parent dir, deduped, re-derived on every reconcile so watchers don't leak) → a debounced `reconcile()` on change, so an on-disk cert renewal is picked up live without a `config.json` write or restart. Two details make a listener-level cert rotation actually apply: the per-proxy `listenerSig` is computed over the *resolved* listeners (cert contents included), so a rotated `defaultCert`/mTLS file changes the signature and forces a recreate rather than a route-only hot-swap against the frozen `default_listener_tls`. Basename-filtered dir watching handles in-place / rename rotation (what host-manager does); k8s projected-volume `..data` symlink swaps are not yet covered. Cert-failure resilience lives in `router.rs::build_route_table`: a route whose cert can't be built (e.g. rustls `KeyMismatch` from a rotated key vs a stale inlined chain) is isolated — one bad tenant cert never aborts the whole table; on a hot-swap the last-good route is carried forward for that SNI (mid-rotation the old cert is still valid), and on initial build the SNI is simply dropped. ### Data flow @@ -46,7 +59,7 @@ TCP accept (SO_REUSEPORT per worker thread) | `src/proxy_conn.rs` | Per-connection handler: the full 7-step flow | | `src/protection.rs` | IP rate limiting, concurrency, CIDR lists, JA3 blocking | | `src/suspended.rs` | Pending-connection registry (DashMap + oneshot channels) | -| `src/metrics.rs` | AtomicU64 counters: active, accepted, errors, blocked | +| `src/metrics.rs` | AtomicU64 counters (active, accepted, bytes, per-reason blocks/errors) + `CountingStream` | | `src/error.rs` | SymphonyError enum → napi::Error conversion | --- @@ -151,6 +164,18 @@ napi `Buffer` contains raw pointers (`*mut napi_env__`, `*mut napi_ref__`) that 4. Add a new `kind` case in `parse_upstream_spec()` in `proxy.rs` 5. Add a test +### Adding a new metric + +1. Add the counter to `ListenerMetrics`/`GlobalMetrics` in `metrics.rs`, or a variant to the + `labeled_enum!` block for `BlockKind`/`ErrorKind` — the variant list drives the counter array, + the label, and the export, so there is no second list to update. +2. Increment it at the call site. A new `protection::BlockReason` variant will fail to compile + until `From<&BlockReason> for BlockKind` maps it — that is deliberate, so a new protection + check can't land in an unlabeled bucket. +3. Surface it in `JsProxyMetrics`/`JsListenerMetrics` (`proxy.rs`), `ts/types.ts`, and the + mapping in `ts/proxy.ts`. +4. Add the sample to `renderPrometheus` in `ts/admin.ts`, and a case in `__test__/metrics.spec.ts`. + ### Adding a new napi method 1. Implement in `proxy.rs` with `#[napi]` @@ -169,6 +194,7 @@ Tests live in `__test__/` and use Node's built-in `node:test` runner. - **`protection.spec.ts`** — rate limit token bucket exhaustion, CIDR blocklist in `blockedIps()` - **`suspended.spec.ts`** — hold → resolve → proxy, hold → null → close, hold → timeout → drop - **`mtls.spec.ts`** — mTLS termination + PROXY v2 TLV forwarding of the client cert chain (0xE2, SSL TLV 0x20); skips without openssl +- **`metrics.spec.ts`** — per-listener breakdown and byte counting, `renderPrometheus` output shape, the admin endpoint over UDS + TCP, and stale-socket reclaim after a `SIGKILL` Build and run: ```bash diff --git a/README.md b/README.md index 7b563ee..5ed115a 100644 --- a/README.md +++ b/README.md @@ -599,23 +599,115 @@ attestation roadmap, and the shared-responsibility split for DDoS — see ## Metrics & monitoring +### In-process (`proxy.metrics()`) + ```typescript const m = proxy.metrics(); -// m.activeConnections — connections being proxied right now -// m.blockedConnections — total blocked since start -// m.pendingSuspended — connections currently held waiting for resolveConnection() +// Proxy-wide +// m.activeConnections — connections being proxied right now +// m.blockedConnections — total rejected since start (protection + maxConnections) +// m.pendingSuspended — connections held waiting for resolveConnection() +// m.suspendedResolved — suspended connections that were resolved with a route +// m.suspendedUnresolved — suspended connections that timed out or were rejected +// m.routes — routes in the live table, including the default route +// m.failingRoutes — routes whose cert failed to build (see "Per-route certificates") + +// Per listener, in configuration order +for (const l of m.listeners) { + // l.address, l.mode ('tls' | 'http') + // l.activeConnections, l.accepted + // l.bytesReceived — bytes read from clients (client → upstream) + // l.bytesSent — bytes written to clients (upstream → client) + // Counted where the proxy sees the bytes. On a terminated-TLS route that is the plaintext + // stream, and the handshake — which precedes the counter — is excluded. On a passthrough + // route the proxy has no plaintext view and forwards wire bytes, so the handshake records + // are part of the stream and are counted. + // l.blockedByReason — [{ reason: 'rate_limited', count: 12 }, ...] + // l.errorsByReason — [{ reason: 'upstream_connect', count: 3 }, ...] +} const blocked = proxy.blockedIps(); // blocked.rateLimited — IPs with a depleted per-second or sustained token bucket // blocked.concurrencyLimited — IPs at their maxConcurrentPerIp limit // blocked.cidrBlocklist — the configured static CIDR blocklist // blocked.penaltyBoxed — IPs currently in the penalty box +``` + +Every reason is reported on every call, including reasons still at zero, so a dashboard series +exists before the first incident rather than appearing mid-outage. `l.blocked` / `l.errors` are +summed from the very reason values reported alongside them, and `m.blockedConnections` from the +listener values in the same snapshot — so a reading taken mid-traffic is internally consistent +rather than only adding up while the proxy is idle. + +**Block reasons:** `max_connections`, `cidr_blocked`, `ja3_blocked`, `ja4_blocked`, +`incomplete_handshake`, `no_sni`, `rate_limited`, `too_many_connections`, `penalty_boxed`. + +**Error reasons:** `no_route`, `route_rate_limited`, `suspend_unresolved`, `tls_handshake`, +`tls_missing_cert`, `upstream_connect`, `idle_timeout`, `stream`, `http_header`. + +> `idle_timeout` counts terminations by `idleTimeoutMs`, which today is a *total duration* cap +> rather than an idleness one — see [#34](https://github.com/HarperFast/symphony/issues/34). Busy +> long-lived connections land in this bucket, not just quiet ones. + +### Out-of-process (`symphony-server` admin endpoint) -setInterval(() => { - console.log('active:', proxy.metrics().activeConnections); -}, 10_000); +When symphony runs as its own process there is no JS API to call, so the server bin can expose +the same numbers over HTTP. Add an `admin` block to the config file: + +```json +{ + "version": 1, + "admin": { + "socketPath": "/run/symphony/admin.sock", + "socketMode": 432, + "port": 9095, + "host": "127.0.0.1" + }, + "proxies": [ ... ] +} ``` +Both bindings are optional; give either or both. Omit the `admin` block entirely and nothing is +exposed. `socketPath` may be relative to the config file's directory, and is chmodded to +`socketMode` (default `0o660`) after bind. `host` defaults to `127.0.0.1` — metrics carry no +tenant identifiers, but there is still no reason to publish them off-box. + +| Route | Response | +|---|---| +| `GET /metrics` | Prometheus text exposition (v0.0.4) | +| `GET /metrics.json` | the same snapshot as JSON | +| `GET /health` | `{ ok, pid, version, ports }` | + +``` +$ curl --unix-socket /run/symphony/admin.sock http://localhost/metrics +# HELP symphony_build_info Always 1; the version is carried in the label. +# TYPE symphony_build_info gauge +symphony_build_info{version="0.5.0"} 1 +... +symphony_listener_accepted_total{proxy="80,443",listener="0.0.0.0:443",mode="tls"} 148213 +symphony_listener_blocked_total{proxy="80,443",listener="0.0.0.0:443",mode="tls",reason="rate_limited"} 27 +symphony_listener_errors_total{proxy="80,443",listener="0.0.0.0:443",mode="tls",reason="upstream_connect"} 4 +``` + +The `proxy` label is the port-set of the proxy entry the listener belongs to (each config entry +gets its own route table). Blocked and error counts are only ever emitted with their `reason` +label — the labelled series sum to the total, so use `sum without(reason)` rather than looking +for a separate unlabelled metric. Likewise the proxy-wide active-connection gauge is +`sum without(listener) (symphony_listener_active_connections)`. + +The endpoint is strictly read-only and best-effort: it never blocks proxying, and a bind failure +is logged and retried every 5s rather than aborting startup. That matters during a version +upgrade, where the incumbent still holds the socket while the replacement is already serving +traffic through `SO_REUSEPORT` — the successor picks up the admin endpoint once the old process +exits. A socket file left behind by a `SIGKILL`ed process is reclaimed automatically, but only +after a connect probe proves nobody is listening on it. + +The Prometheus renderer is internal to the standalone server and is not exported from the +package root: a snapshot carries that process's pid, timestamps, and port-set grouping, which an +embedded consumer would have to synthesise. An embedded proxy has `proxy.metrics()` directly. If +a caller genuinely needs Prometheus text from an embedded proxy, open an issue — the right shape +is a `ProxyMetrics`-based renderer, not this one. + --- ## Hot config updates diff --git a/__test__/metrics.spec.ts b/__test__/metrics.spec.ts new file mode 100644 index 0000000..0f4dc8d --- /dev/null +++ b/__test__/metrics.spec.ts @@ -0,0 +1,639 @@ +/** + * Metrics coverage: the in-process `metrics()` breakdown, the Prometheus renderer, and the + * standalone server's admin endpoint over both a Unix socket and a loopback port. + * + * Requires the native addon to be built (npm run build:debug). + */ + +import assert from 'node:assert/strict'; +import { after, before, describe, it } from 'node:test'; +import { spawn, ChildProcess } from 'node:child_process'; +import * as fs from 'node:fs'; +import * as http from 'node:http'; +import * as os from 'node:os'; +import * as path from 'node:path'; +import * as net from 'node:net'; +import * as tls from 'node:tls'; +import { SymphonyProxy } from '../ts/index.js'; +// Not exported from the package root — the admin endpoint owns this shape (see ts/index.ts). +import { renderPrometheus, type MetricsSnapshot } from '../ts/admin.js'; +import { generateSelfSignedCert, getFreePort, startEchoServer, tlsRoundTrip, sleep } from './util.js'; + +const SERVER_JS = path.join(__dirname, '..', 'ts', 'server.js'); + +async function waitFor(predicate: () => boolean | Promise, timeoutMs = 5000, stepMs = 50): Promise { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (await predicate()) return; + await sleep(stepMs); + } + throw new Error('waitFor: timed out'); +} + +/** GET over a Unix socket or a TCP port. */ +function get( + target: { socketPath: string } | { port: number }, + urlPath: string, + method = 'GET' +): Promise<{ status: number; body: string; contentType: string }> { + return new Promise((resolve, reject) => { + const req = http.request({ ...target, path: urlPath, method }, (res) => { + let body = ''; + res.setEncoding('utf8'); + res.on('data', (c) => (body += c)); + res.on('end', () => + resolve({ status: res.statusCode ?? 0, body, contentType: String(res.headers['content-type'] ?? '') }) + ); + }); + req.on('error', reject); + req.end(); + }); +} + +function reasonCount(reasons: Array<{ reason: string; count: number }>, reason: string): number { + const match = reasons.find((r) => r.reason === reason); + assert.ok(match, `expected a '${reason}' entry; got ${reasons.map((r) => r.reason).join(', ')}`); + return match.count; +} + +describe('proxy.metrics()', () => { + const cert = generateSelfSignedCert('localhost'); + let echo: Awaited>; + let proxy: SymphonyProxy; + let tlsPort: number; + let httpPort: number; + + before(async () => { + echo = await startEchoServer(); + tlsPort = await getFreePort(); + httpPort = await getFreePort(); + proxy = new SymphonyProxy({ + listeners: [ + { host: '127.0.0.1', port: tlsPort }, + { host: '127.0.0.1', port: httpPort, mode: 'http' }, + ], + routes: [ + { + sni: 'localhost', + upstreams: [{ kind: 'tcp', host: '127.0.0.1', port: echo.port }], + terminateTls: true, + cert: { certChain: cert.cert, privateKey: cert.key }, + }, + ], + }); + await proxy.start(); + }); + + after(async () => { + await proxy.stop().catch(() => {}); + await echo.close().catch(() => {}); + }); + + it('reports one entry per configured listener, in order, with its mode', () => { + const m = proxy.metrics(); + assert.equal(m.listeners.length, 2); + assert.deepEqual( + m.listeners.map((l) => [l.address, l.mode]), + [ + [`127.0.0.1:${tlsPort}`, 'tls'], + [`127.0.0.1:${httpPort}`, 'http'], + ] + ); + }); + + it('reports the live route count', () => { + assert.equal(proxy.metrics().routes, 1); + assert.equal(proxy.metrics().failingRoutes, 0); + }); + + it('emits every block and error reason, including the ones still at zero', () => { + const listener = proxy.metrics().listeners[0]; + // A reason that only ever fires under protection config must still have a series. + assert.equal(reasonCount(listener.blockedByReason, 'rate_limited'), 0); + assert.equal(reasonCount(listener.errorsByReason, 'upstream_connect'), 0); + assert.ok(listener.blockedByReason.some((r) => r.reason === 'max_connections')); + }); + + it('counts bytes in both directions across a proxied session', async () => { + const before = proxy.metrics().listeners[0]; + const payload = Buffer.from('x'.repeat(4096)); + const echoed = await tlsRoundTrip({ port: tlsPort, servername: 'localhost', caCert: cert.cert, data: payload }); + assert.equal(echoed.length, payload.length); + + const after = proxy.metrics().listeners[0]; + assert.equal(after.accepted, before.accepted + 1); + // The counter wraps the client *after* the handshake, so on a terminated-TLS route it + // sees exactly the plaintext payload each way — no handshake and no record framing. + assert.equal(after.bytesReceived, before.bytesReceived + payload.length); + assert.equal(after.bytesSent, before.bytesSent + payload.length); + }); + + it('classifies an unroutable SNI as no_route rather than a generic error', async () => { + const before = proxy.metrics().listeners[0]; + // No route for this SNI and no default route → symphony drops the connection. + await tlsRoundTrip({ port: tlsPort, servername: 'nope.example.com', caCert: cert.cert, data: 'ping' }).catch( + () => undefined + ); + + await waitFor(() => proxy.metrics().listeners[0].errors > before.errors); + const after = proxy.metrics().listeners[0]; + assert.equal(reasonCount(after.errorsByReason, 'no_route'), reasonCount(before.errorsByReason, 'no_route') + 1); + // The per-reason series always sum to the unlabeled total. + assert.equal( + after.errorsByReason.reduce((sum, r) => sum + r.count, 0), + after.errors + ); + }); +}); + +// The three session outcomes are told apart by where the failure is raised, not by inspecting an +// io::ErrorKind — these lock that in, since a misclassification is invisible until an incident. +describe('proxy.metrics() error classification', () => { + const cert = generateSelfSignedCert('localhost'); + let echo: Awaited>; + let proxy: SymphonyProxy; + let tlsPort: number; + let deadPort: number; + + before(async () => { + echo = await startEchoServer(); + tlsPort = await getFreePort(); + // Reserved and then released, so a connect here is refused rather than hanging. + deadPort = await getFreePort(); + proxy = new SymphonyProxy({ + listeners: [{ host: '127.0.0.1', port: tlsPort, idleTimeoutMs: 300 }], + routes: [ + { + sni: 'idle.test', + upstreams: [{ kind: 'tcp', host: '127.0.0.1', port: echo.port }], + terminateTls: true, + cert: { certChain: cert.cert, privateKey: cert.key }, + }, + { + sni: 'dead.test', + upstreams: [{ kind: 'tcp', host: '127.0.0.1', port: deadPort }], + terminateTls: true, + cert: { certChain: cert.cert, privateKey: cert.key }, + }, + ], + }); + await proxy.start(); + }); + + after(async () => { + await proxy.stop().catch(() => {}); + await echo.close().catch(() => {}); + }); + + it('records an unreachable upstream as upstream_connect', async () => { + const before = reasonCount(proxy.metrics().listeners[0].errorsByReason, 'upstream_connect'); + await tlsRoundTrip({ port: tlsPort, servername: 'dead.test', caCert: cert.cert, data: 'ping' }).catch( + () => undefined + ); + await waitFor(() => reasonCount(proxy.metrics().listeners[0].errorsByReason, 'upstream_connect') > before); + }); + + it('records a session that goes quiet as idle_timeout, not a stream error', async () => { + const beforeIdle = reasonCount(proxy.metrics().listeners[0].errorsByReason, 'idle_timeout'); + const beforeStream = reasonCount(proxy.metrics().listeners[0].errorsByReason, 'stream'); + + // Connect, complete the handshake, then send nothing until the 300ms idle timeout fires. + const socket = tls.connect({ + port: tlsPort, + host: '127.0.0.1', + servername: 'idle.test', + ca: cert.cert, + rejectUnauthorized: false, + }); + try { + await new Promise((resolve, reject) => { + socket.once('secureConnect', resolve); + socket.once('error', reject); + }); + socket.on('error', () => {}); // the proxy dropping us is the expected outcome + await waitFor(() => reasonCount(proxy.metrics().listeners[0].errorsByReason, 'idle_timeout') > beforeIdle, 5000); + } finally { + socket.destroy(); + } + + assert.equal( + reasonCount(proxy.metrics().listeners[0].errorsByReason, 'stream'), + beforeStream, + 'an idle timeout must not also be counted as a stream error' + ); + }); +}); + +describe('renderPrometheus', () => { + const snapshot: MetricsSnapshot = { + pid: 42, + version: '9.9.9', + startedAt: '2026-01-01T00:00:00.000Z', + reloadedAt: '2026-01-01T00:01:00.000Z', + proxies: [ + { + ports: '80,443', + metrics: { + activeConnections: 3, + blockedConnections: 2, + pendingSuspended: 1, + suspendedResolved: 5, + suspendedUnresolved: 4, + routes: 7, + failingRoutes: 1, + listeners: [ + { + address: '0.0.0.0:443', + mode: 'tls', + activeConnections: 3, + accepted: 10, + blocked: 2, + errors: 1, + bytesReceived: 1024, + bytesSent: 2048, + blockedByReason: [ + { reason: 'rate_limited', count: 2 }, + { reason: 'no_sni', count: 0 }, + ], + errorsByReason: [{ reason: 'upstream_connect', count: 1 }], + }, + ], + }, + }, + ], + }; + + const output = renderPrometheus(snapshot); + const lines = output.split('\n'); + + it('declares HELP and TYPE exactly once per metric name', () => { + const typeLines = lines.filter((l) => l.startsWith('# TYPE ')); + const names = typeLines.map((l) => l.split(' ')[2]); + assert.deepEqual([...new Set(names)].length, names.length, `duplicate TYPE declarations in:\n${output}`); + // Both `outcome` samples share one declaration. + assert.equal(typeLines.filter((l) => l.includes('symphony_suspended_total')).length, 1); + }); + + // Samples of one metric name must be contiguous under a single HELP/TYPE pair. Emitting in + // proxy → listener call order would interleave them once a second proxy exists, and strict + // parsers reject or drop the split group. + it('keeps every metric name contiguous across multiple proxies', () => { + const second = structuredClone(snapshot.proxies[0]); + second.ports = '8443'; + second.metrics.listeners[0].address = '0.0.0.0:8443'; + const multi = renderPrometheus({ ...snapshot, proxies: [snapshot.proxies[0], second] }).split('\n'); + + const seen = new Set(); + let previous = ''; + for (const line of multi) { + if (!line || line.startsWith('#')) continue; + const name = line.slice(0, Math.min(...[line.indexOf('{'), line.indexOf(' ')].filter((i) => i >= 0))); + if (name !== previous) { + assert.ok(!seen.has(name), `samples for ${name} are split across the output`); + seen.add(name); + previous = name; + } + } + // Both proxies really are present, so the check above wasn't vacuous. + assert.ok(multi.includes('symphony_routes{proxy="80,443"} 7')); + assert.ok(multi.includes('symphony_routes{proxy="8443"} 7')); + }); + + it('labels every proxy-scoped and listener-scoped sample', () => { + assert.ok(lines.includes('symphony_routes{proxy="80,443"} 7')); + assert.ok(lines.includes('symphony_routes_failing{proxy="80,443"} 1')); + assert.ok(lines.includes('symphony_suspended_total{proxy="80,443",outcome="resolved"} 5')); + assert.ok(lines.includes('symphony_suspended_total{proxy="80,443",outcome="unresolved"} 4')); + assert.ok(lines.includes('symphony_listener_accepted_total{proxy="80,443",listener="0.0.0.0:443",mode="tls"} 10')); + assert.ok( + lines.includes('symphony_listener_bytes_received_total{proxy="80,443",listener="0.0.0.0:443",mode="tls"} 1024') + ); + }); + + it('emits blocked/error counts only under their reason label', () => { + assert.ok( + lines.includes( + 'symphony_listener_blocked_total{proxy="80,443",listener="0.0.0.0:443",mode="tls",reason="rate_limited"} 2' + ) + ); + // A zero-valued reason still gets a series, so it exists before the first incident. + assert.ok( + lines.includes( + 'symphony_listener_blocked_total{proxy="80,443",listener="0.0.0.0:443",mode="tls",reason="no_sni"} 0' + ) + ); + // No unlabeled duplicate of the same number. + assert.ok(!lines.some((l) => /^symphony_listener_blocked_total\{[^}]*\}\s/.test(l) && !l.includes('reason='))); + }); + + // Consumers bind this endpoint on loopback, which any local process can reach. That is only + // acceptable while the labels stay free of tenant identifiers — a per-SNI or per-route label + // would turn aggregate proxy health into a list of which customers a host fronts. This pins + // the allowed label set so adding one is a deliberate, visible decision. + it('never labels a sample with a tenant identifier', () => { + const allowed = new Set(['version', 'proxy', 'listener', 'mode', 'reason', 'outcome']); + for (const line of lines) { + const labelSet = line.match(/\{(.*)\}/)?.[1]; + if (!labelSet) continue; + // Keys sit at the start or after a comma; a label *value* may itself contain a comma + // (proxy="80,443"), so the boundary has to be matched rather than split on. + for (const [, , key] of labelSet.matchAll(/(^|,)([a-z_]+)="/g)) { + assert.ok(allowed.has(key), `unexpected metric label '${key}' — is it tenant-identifying?`); + } + } + }); + + it('carries the version in build_info and timestamps in seconds', () => { + assert.ok(lines.includes('symphony_build_info{version="9.9.9"} 1')); + assert.ok(lines.includes(`symphony_start_time_seconds ${Date.parse(snapshot.startedAt) / 1000}`)); + assert.ok(lines.includes(`symphony_config_reload_time_seconds ${Date.parse(snapshot.reloadedAt) / 1000}`)); + }); + + it('escapes label values', () => { + const escaped = renderPrometheus({ + ...snapshot, + version: 'a"b\\c', + proxies: [], + }); + assert.ok(escaped.includes('symphony_build_info{version="a\\"b\\\\c"} 1')); + }); +}); + +describe('symphony-server admin endpoint', () => { + const cert = generateSelfSignedCert('localhost'); + let dir: string; + let configPath: string; + let statusPath: string; + let socketPath: string; + let adminPort: number; + let proxyPort: number; + let echo: Awaited>; + let child: ChildProcess; + let stderr = ''; + let shuttingDown = false; + + before(async () => { + dir = fs.mkdtempSync(path.join(os.tmpdir(), 'symphony-metrics-test-')); + configPath = path.join(dir, 'config.json'); + statusPath = path.join(dir, 'status.json'); + socketPath = path.join(dir, 'admin.sock'); + echo = await startEchoServer(); + proxyPort = await getFreePort(); + adminPort = await getFreePort(); + + fs.writeFileSync( + configPath, + JSON.stringify({ + version: 1, + admin: { socketPath, port: adminPort, host: '127.0.0.1' }, + proxies: [ + { + listeners: [{ host: '127.0.0.1', port: proxyPort }], + routes: [ + { + sni: 'localhost', + upstreams: [{ kind: 'tcp', host: '127.0.0.1', port: echo.port }], + terminateTls: true, + cert: { certChain: cert.cert, privateKey: cert.key }, + }, + ], + }, + ], + }) + ); + + child = spawn(process.execPath, [SERVER_JS, '--config', configPath], { stdio: ['ignore', 'pipe', 'pipe'] }); + child.stderr?.on('data', (d) => (stderr += d.toString())); + child.on('exit', (code, sig) => { + if (!shuttingDown) stderr += `\n[child exited early code=${code} sig=${sig}]`; + }); + await waitFor(() => fs.existsSync(statusPath) && fs.existsSync(socketPath), 8000); + }); + + after(async () => { + shuttingDown = true; + if (child && child.exitCode === null && child.signalCode === null) { + child.kill('SIGTERM'); + await waitFor(() => child.exitCode !== null || child.signalCode !== null, 3000).catch(() => + child.kill('SIGKILL') + ); + } + await echo.close().catch(() => {}); + fs.rmSync(dir, { recursive: true, force: true }); + }); + + it('serves Prometheus text over the unix socket', async () => { + const res = await get({ socketPath }, '/metrics'); + assert.equal(res.status, 200, stderr); + assert.match(res.contentType, /text\/plain; version=0\.0\.4/); + assert.match(res.body, /^# HELP symphony_build_info /m); + assert.match( + res.body, + new RegExp( + `symphony_listener_accepted_total\\{proxy="${proxyPort}",listener="127\\.0\\.0\\.1:${proxyPort}",mode="tls"\\} \\d+` + ) + ); + }); + + it('serves the same snapshot as JSON over the loopback port', async () => { + const res = await get({ port: adminPort }, '/metrics.json'); + assert.equal(res.status, 200, stderr); + assert.match(res.contentType, /application\/json/); + const snapshot = JSON.parse(res.body) as MetricsSnapshot; + assert.equal(snapshot.proxies.length, 1); + assert.equal(snapshot.proxies[0].ports, String(proxyPort)); + assert.equal(snapshot.proxies[0].metrics.listeners[0].address, `127.0.0.1:${proxyPort}`); + }); + + it('reflects live traffic on the next scrape', async () => { + const before = JSON.parse((await get({ socketPath }, '/metrics.json')).body) as MetricsSnapshot; + await tlsRoundTrip({ port: proxyPort, servername: 'localhost', caCert: cert.cert, data: 'hello' }); + const after = JSON.parse((await get({ socketPath }, '/metrics.json')).body) as MetricsSnapshot; + + assert.equal(after.proxies[0].metrics.listeners[0].accepted, before.proxies[0].metrics.listeners[0].accepted + 1); + assert.ok(after.proxies[0].metrics.listeners[0].bytesSent > before.proxies[0].metrics.listeners[0].bytesSent); + }); + + it('restricts the unix socket to owner and group', () => { + assert.equal(fs.statSync(socketPath).mode & 0o777, 0o660); + }); + + it('answers /health with the pid and served ports', async () => { + const res = await get({ port: adminPort }, '/health'); + assert.equal(res.status, 200); + const health = JSON.parse(res.body) as { ok: boolean; pid: number; ports: number[] }; + assert.equal(health.ok, true); + assert.equal(health.pid, child.pid); + assert.deepEqual(health.ports, [proxyPort]); + }); + + it('404s an unknown path and 405s a non-GET', async () => { + assert.equal((await get({ port: adminPort }, '/nope')).status, 404); + assert.equal((await get({ port: adminPort }, '/metrics', 'POST')).status, 405); + }); + + it('ignores a query string on /metrics', async () => { + const res = await get({ port: adminPort }, '/metrics?foo=bar'); + assert.equal(res.status, 200); + }); + + // Shutdown deliberately does NOT unlink the published path — any check-then-unlink can delete + // a successor's live socket in the window between the two syscalls. What has to hold is that + // the endpoint stops answering; reclaiming the pathname is the next binder's job, atomically. + it('stops serving on shutdown and leaves the path safely reclaimable', async () => { + shuttingDown = true; + child.kill('SIGTERM'); + await waitFor(() => child.exitCode !== null || child.signalCode !== null, 5000); + + await assert.rejects( + () => get({ socketPath }, '/health'), + (err: NodeJS.ErrnoException) => err.code === 'ECONNREFUSED', + 'a dead endpoint must refuse connections, not hang or answer' + ); + }); +}); + +// A SIGKILLed process never runs its shutdown path, so the socket file survives it. The next +// process must reclaim it — but only after proving nobody is listening, since unlinking a live +// socket would silently steal the endpoint from a running symphony. +describe('symphony-server admin endpoint (stale socket recovery)', () => { + const cert = generateSelfSignedCert('localhost'); + let dir: string; + let socketPath: string; + let survivor: ChildProcess | null = null; + + function writeConfig(configPath: string, proxyPort: number, echoPort: number): void { + fs.writeFileSync( + configPath, + JSON.stringify({ + version: 1, + admin: { socketPath }, + proxies: [ + { + listeners: [{ host: '127.0.0.1', port: proxyPort }], + routes: [ + { + sni: 'localhost', + upstreams: [{ kind: 'tcp', host: '127.0.0.1', port: echoPort }], + terminateTls: true, + cert: { certChain: cert.cert, privateKey: cert.key }, + }, + ], + }, + ], + }) + ); + } + + before(() => { + dir = fs.mkdtempSync(path.join(os.tmpdir(), 'symphony-stale-sock-')); + socketPath = path.join(dir, 'admin.sock'); + }); + + after(async () => { + if (survivor && survivor.exitCode === null && survivor.signalCode === null) { + survivor.kill('SIGKILL'); + await waitFor(() => survivor!.exitCode !== null || survivor!.signalCode !== null, 3000).catch(() => {}); + } + fs.rmSync(dir, { recursive: true, force: true }); + }); + + it('reclaims a socket left behind by a SIGKILLed process', async () => { + const echo = await startEchoServer(); + try { + const firstConfig = path.join(dir, 'first.json'); + const firstStatus = path.join(dir, 'first-status.json'); + writeConfig(firstConfig, await getFreePort(), echo.port); + const first = spawn(process.execPath, [SERVER_JS, '--config', firstConfig, '--status', firstStatus], { + stdio: ['ignore', 'pipe', 'pipe'], + }); + await waitFor(() => fs.existsSync(socketPath), 8000); + + first.kill('SIGKILL'); + await waitFor(() => first.exitCode !== null || first.signalCode !== null, 5000); + assert.ok(fs.existsSync(socketPath), 'SIGKILL should leave the socket file behind'); + + // Second process: same socket path, now stale. + const secondConfig = path.join(dir, 'second.json'); + const secondStatus = path.join(dir, 'second-status.json'); + writeConfig(secondConfig, await getFreePort(), echo.port); + let stderr = ''; + survivor = spawn(process.execPath, [SERVER_JS, '--config', secondConfig, '--status', secondStatus], { + stdio: ['ignore', 'pipe', 'pipe'], + }); + survivor.stderr?.on('data', (d) => (stderr += d.toString())); + await waitFor(() => fs.existsSync(secondStatus), 8000); + + // The bind is retried on a timer if it loses the race, so allow a couple of cycles. + await waitFor(async () => { + const res = await get({ socketPath }, '/health').catch(() => null); + return res?.status === 200; + }, 15000); + const health = JSON.parse((await get({ socketPath }, '/health')).body) as { pid: number }; + assert.equal(health.pid, survivor.pid, `the new process must own the socket; stderr:\n${stderr}`); + } finally { + await echo.close().catch(() => {}); + } + }); + + // The inverse, and the one that actually costs something to get wrong: a socket that is still + // being served must survive a would-be reclaimer, whatever the probe happens to return. A + // permission-denied probe is not evidence that nobody is listening. + it('leaves a live socket alone even when the probe cannot connect to it', async () => { + const echo = await startEchoServer(); + const livePath = path.join(dir, 'live.sock'); + // Stand in for a running symphony: a real listening socket the reclaimer must not delete. + const incumbent = net.createServer(); + await new Promise((resolve) => incumbent.listen(livePath, resolve)); + const liveIno = fs.statSync(livePath).ino; + + try { + // 0o000 makes connect() fail with EACCES rather than ECONNREFUSED. Treating any probe + // error as "stale" would unlink this live socket. + fs.chmodSync(livePath, 0o000); + + const configPath = path.join(dir, 'contend.json'); + const statusPath = path.join(dir, 'contend-status.json'); + fs.writeFileSync( + configPath, + JSON.stringify({ + version: 1, + admin: { socketPath: livePath }, + proxies: [ + { + listeners: [{ host: '127.0.0.1', port: await getFreePort() }], + routes: [ + { + sni: 'localhost', + upstreams: [{ kind: 'tcp', host: '127.0.0.1', port: echo.port }], + terminateTls: true, + cert: { certChain: cert.cert, privateKey: cert.key }, + }, + ], + }, + ], + }) + ); + + const contender = spawn(process.execPath, [SERVER_JS, '--config', configPath, '--status', statusPath], { + stdio: ['ignore', 'pipe', 'pipe'], + }); + try { + // It boots and proxies regardless — losing the admin bind must never block startup. + await waitFor(() => fs.existsSync(statusPath), 8000); + // Give the retry timer a cycle to (wrongly) reclaim, if it were going to. + await sleep(1000); + + assert.ok(fs.existsSync(livePath), 'the live socket must not have been unlinked'); + assert.equal(fs.statSync(livePath).ino, liveIno, 'the live socket must not have been replaced'); + } finally { + contender.kill('SIGKILL'); + await waitFor(() => contender.exitCode !== null || contender.signalCode !== null, 3000).catch(() => {}); + } + } finally { + fs.chmodSync(livePath, 0o660); + await new Promise((resolve) => incumbent.close(() => resolve())); + await echo.close().catch(() => {}); + } + }); +}); diff --git a/package.json b/package.json index 3143905..08e5b35 100644 --- a/package.json +++ b/package.json @@ -31,7 +31,7 @@ "artifacts": "napi artifacts", "prepublishOnly": "napi prepublish -t npm", "version": "napi version", - "test": "tsc -p tsconfig.test.json && node --test --test-force-exit dist-test/__test__/protection.spec.js dist-test/__test__/proxy.spec.js dist-test/__test__/proxy-protocol-v2.spec.js dist-test/__test__/mtls.spec.js dist-test/__test__/suspended.spec.js dist-test/__test__/http-listener.spec.js dist-test/__test__/server.spec.js dist-test/__test__/h2-dispatch.spec.js", + "test": "tsc -p tsconfig.test.json && node --test --test-force-exit dist-test/__test__/protection.spec.js dist-test/__test__/proxy.spec.js dist-test/__test__/proxy-protocol-v2.spec.js dist-test/__test__/mtls.spec.js dist-test/__test__/suspended.spec.js dist-test/__test__/http-listener.spec.js dist-test/__test__/server.spec.js dist-test/__test__/h2-dispatch.spec.js dist-test/__test__/metrics.spec.js", "test:integration": "tsc -p tsconfig.test.json && node --test --test-force-exit dist-test/__test__/harper-integration.spec.js", "benchmark": "tsc -p tsconfig.test.json && node dist-test/__test__/benchmark.js", "benchmark:throughput": "tsc -p tsconfig.test.json && node dist-test/__test__/benchmark-throughput.js", diff --git a/src/http_listener.rs b/src/http_listener.rs index 4fe0770..0442141 100644 --- a/src/http_listener.rs +++ b/src/http_listener.rs @@ -17,6 +17,7 @@ use crate::http_proxy::{ host_header, read_http_headers, request_target, strip_body_framing, with_connection_close, }; use crate::listener::{make_reuseport_socket, set_rlimit_nofile}; +use crate::metrics::{BlockKind, CountingStream, ErrorKind}; use crate::proxy_conn::ConnContext; use crate::upstream::{self, UpstreamStream}; use std::net::SocketAddr; @@ -91,7 +92,7 @@ async fn accept_loop( let active = ctx.global_metrics.active_connections.load(std::sync::atomic::Ordering::Relaxed); if active >= max_connections as u64 { drop(stream); - ctx.listener_metrics.inc_blocked(); + ctx.listener_metrics.inc_blocked(BlockKind::MaxConnections); continue; } } @@ -125,12 +126,12 @@ async fn handle_http(mut stream: TcpStream, peer_addr: SocketAddr, ctx: Arc pair, Ok(Err(e)) => { tracing::debug!("http :80 header read error from {}: {e}", peer_addr.ip()); - ctx.listener_metrics.inc_error(); + ctx.listener_metrics.inc_error(ErrorKind::HttpHeader); return; } Err(_) => { tracing::debug!("http :80 header read timeout from {}", peer_addr.ip()); - ctx.listener_metrics.inc_error(); + ctx.listener_metrics.inc_error(ErrorKind::HttpHeader); return; } }; @@ -140,24 +141,30 @@ async fn handle_http(mut stream: TcpStream, peer_addr: SocketAddr, ctx: Arc. let Some(host) = host else { - let _ = write_simple_response(&mut stream, b"HTTP/1.1 400 Bad Request\r\nContent-Length: 0\r\nConnection: close\r\n\r\n").await; + let _ = write_simple_response(&mut stream, b"HTTP/1.1 400 Bad Request\r\nContent-Length: 0\r\nConnection: close\r\n\r\n", &ctx.listener_metrics).await; return; }; let target_str = std::str::from_utf8(target).unwrap_or("/"); @@ -166,7 +173,11 @@ async fn handle_http(mut stream: TcpStream, peer_addr: SocketAddr, ctx: Arc std::io::Result<()> { +) -> std::result::Result<(), ErrorKind> { let table = ctx.route_table.0.load(); let Some(route) = table.resolve(Some(host)) else { // No matching route — answer 404 so the ACME client gets a definitive answer // rather than a hung connection. - let _ = write_simple_response(client, b"HTTP/1.1 404 Not Found\r\nContent-Length: 0\r\nConnection: close\r\n\r\n").await; - return Ok(()); + let _ = write_simple_response(client, b"HTTP/1.1 404 Not Found\r\nContent-Length: 0\r\nConnection: close\r\n\r\n", &ctx.listener_metrics).await; + return Err(ErrorKind::NoRoute); }; // Honour the route's global rate limit the same way the TLS path does, so // a flood of /.well-known/acme-challenge/ requests can't bypass the cap. if let Some(rl) = &route.rate_limiter { if !rl.try_acquire() { - ctx.listener_metrics.inc_error(); - return Ok(()); + return Err(ErrorKind::RouteRateLimited); } } let upstream = upstream::connect(&route.destination, Some(peer_addr.ip()), UPSTREAM_CONNECT_TIMEOUT) .await - .map_err(|e| std::io::Error::other(e.to_string()))?; + .map_err(|e| { + tracing::debug!("acme upstream connect failed for {host}: {e}"); + ErrorKind::UpstreamConnect + })?; // ACME HTTP-01 challenges are GET requests with no body. Strip any // Content-Length / Transfer-Encoding headers so a client that lies about a @@ -210,19 +223,23 @@ async fn proxy_acme( // redirect path. let forwarded = with_connection_close(&strip_body_framing(headers)); - let result = match upstream { - UpstreamStream::Tcp(mut up) => proxy_one_shot(client, &mut up, &forwarded).await, - UpstreamStream::Uds { mut stream, _guard } => { - let r = proxy_one_shot(client, &mut stream, &forwarded).await; - drop(_guard); - r + // Scoped so the byte counter releases its borrow of `client` before the shutdown below. + let result = { + let mut counted = CountingStream::new(&mut *client, &ctx.listener_metrics); + match upstream { + UpstreamStream::Tcp(mut up) => proxy_one_shot(&mut counted, &mut up, &forwarded).await, + UpstreamStream::Uds { mut stream, _guard } => { + let r = proxy_one_shot(&mut counted, &mut stream, &forwarded).await; + drop(_guard); + r + } } }; // Always close the client socket after one request/response, regardless of // the upstream outcome. The HTTP-mode listener never reuses connections. let _ = client.shutdown().await; - result + result.map_err(|_| ErrorKind::Stream) } /// Send the (sanitized, body-less) request `headers` to `upstream`, then copy @@ -230,12 +247,13 @@ async fn proxy_acme( /// the request bytes flush — any bytes the client already pipelined past the /// header boundary are silently dropped, so no pipelined non-ACME payload can /// reach the backend. -async fn proxy_one_shot( - client: &mut TcpStream, +async fn proxy_one_shot( + client: &mut C, upstream: &mut U, headers: &[u8], ) -> std::io::Result<()> where + C: tokio::io::AsyncWrite + Unpin, U: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin, { upstream.write_all(headers).await?; @@ -243,8 +261,13 @@ where io::copy(upstream, client).await.map(|_| ()) } -async fn write_simple_response(stream: &mut TcpStream, response: &[u8]) -> std::io::Result<()> { +async fn write_simple_response( + stream: &mut TcpStream, + response: &[u8], + metrics: &crate::metrics::ListenerMetrics, +) -> std::io::Result<()> { stream.write_all(response).await?; + metrics.add_bytes_out(response.len() as u64); stream.shutdown().await } diff --git a/src/listener.rs b/src/listener.rs index a90fe65..13142c2 100644 --- a/src/listener.rs +++ b/src/listener.rs @@ -1,3 +1,4 @@ +use crate::metrics::BlockKind; use crate::proxy_conn::{ConnContext, handle}; use socket2::{Domain, Protocol, Socket, Type}; use std::net::SocketAddr; @@ -68,7 +69,7 @@ async fn accept_loop( if active >= max_connections as u64 { // Drop the stream — OS will send RST drop(stream); - ctx.listener_metrics.inc_blocked(); + ctx.listener_metrics.inc_blocked(BlockKind::MaxConnections); continue; } } diff --git a/src/metrics.rs b/src/metrics.rs index b4b29a0..c416154 100644 --- a/src/metrics.rs +++ b/src/metrics.rs @@ -1,11 +1,114 @@ +use crate::protection::BlockReason; +use std::pin::Pin; use std::sync::atomic::{AtomicU64, Ordering}; +use std::task::{Context, Poll}; +use tokio::io::{AsyncRead, AsyncWrite, ReadBuf}; -#[derive(Default)] +/// Declares a small closed enum plus the label text each variant carries into the exported +/// metrics. `ALL` drives both the per-variant counter array and the export, so a new variant +/// is automatically counted and exported — there is no separate list to keep in sync. +macro_rules! labeled_enum { + ($name:ident { $($(#[$doc:meta])* $variant:ident => $label:literal),+ $(,)? }) => { + #[derive(Clone, Copy, Debug, PartialEq, Eq)] + pub enum $name { + $($(#[$doc])* $variant),+ + } + + impl $name { + pub const ALL: &'static [$name] = &[$($name::$variant),+]; + pub const COUNT: usize = Self::ALL.len(); + + pub fn as_str(self) -> &'static str { + match self { + $(Self::$variant => $label),+ + } + } + } + }; +} + +labeled_enum!(BlockKind { + /// Listener-level `maxConnections` cap — refused in the accept loop, before protection runs. + MaxConnections => "max_connections", + CidrBlocked => "cidr_blocked", + Ja3Blocked => "ja3_blocked", + Ja4Blocked => "ja4_blocked", + IncompleteHandshake => "incomplete_handshake", + NoSni => "no_sni", + RateLimited => "rate_limited", + TooManyConnections => "too_many_connections", + PenaltyBoxed => "penalty_boxed", +}); + +// Exhaustive by construction: adding a BlockReason variant fails to compile until it is given +// a BlockKind, so a new protection check can never silently land in an unlabeled bucket. +impl From<&BlockReason> for BlockKind { + fn from(reason: &BlockReason) -> Self { + match reason { + BlockReason::CidrBlocked => Self::CidrBlocked, + BlockReason::Ja3Blocked => Self::Ja3Blocked, + BlockReason::Ja4Blocked => Self::Ja4Blocked, + BlockReason::IncompleteHandshake => Self::IncompleteHandshake, + BlockReason::NoSni => Self::NoSni, + BlockReason::RateLimited => Self::RateLimited, + BlockReason::TooManyConnections => Self::TooManyConnections, + BlockReason::PenaltyBoxed => Self::PenaltyBoxed, + } + } +} + +labeled_enum!(ErrorKind { + /// SNI matched no route and the listener has no default route. + NoRoute => "no_route", + /// The route's own rate limiter rejected the connection. + RouteRateLimited => "route_rate_limited", + /// A suspended connection was never resolved — timed out or rejected by JS. + SuspendUnresolved => "suspend_unresolved", + /// TLS handshake failed or timed out. + TlsHandshake => "tls_handshake", + /// Route asks for TLS termination but has no usable cert (e.g. cert build failed). + TlsMissingCert => "tls_missing_cert", + /// Could not establish the upstream connection. + UpstreamConnect => "upstream_connect", + /// The proxied session hit `idleTimeoutMs`. Note that today this fires on *total* duration, + /// not idleness — `forward()` wraps the copy in a hard `tokio::time::timeout` that does not + /// reset on activity (issue #34, pre-existing). Until that is fixed this counts busy + /// connections cut at the deadline, not quiet ones. + IdleTimeout => "idle_timeout", + /// I/O error while proxying an established session. + Stream => "stream", + /// HTTP-mode listener could not read the request head. + HttpHeader => "http_header", +}); + +/// Per-listener counters. Every field is `Relaxed` — these are monotonic counters and gauges +/// read out of band by `metrics()`, never used to make a decision that needs ordering. pub struct ListenerMetrics { pub active_connections: AtomicU64, pub total_accepted: AtomicU64, - pub total_blocked: AtomicU64, - pub total_errors: AtomicU64, + /// Bytes read from clients on this listener (client → upstream), counted where the proxy + /// sees them. On a terminated-TLS route that is the plaintext stream — the handshake happens + /// before the counter is installed and is excluded. On a passthrough route the proxy has no + /// plaintext view and simply forwards wire bytes, so the handshake records are part of the + /// stream and are counted. + pub bytes_in: AtomicU64, + /// Bytes written to clients on this listener (upstream → client). Same framing as `bytes_in`. + pub bytes_out: AtomicU64, + blocked_by_kind: [AtomicU64; BlockKind::COUNT], + errors_by_kind: [AtomicU64; ErrorKind::COUNT], +} + +impl Default for ListenerMetrics { + fn default() -> Self { + Self { + active_connections: AtomicU64::new(0), + total_accepted: AtomicU64::new(0), + bytes_in: AtomicU64::new(0), + bytes_out: AtomicU64::new(0), + blocked_by_kind: std::array::from_fn(|_| AtomicU64::new(0)), + errors_by_kind: std::array::from_fn(|_| AtomicU64::new(0)), + } + } } impl ListenerMetrics { @@ -18,20 +121,66 @@ impl ListenerMetrics { self.active_connections.fetch_sub(1, Ordering::Relaxed); } - pub fn inc_blocked(&self) { - self.total_blocked.fetch_add(1, Ordering::Relaxed); + pub fn inc_blocked(&self, kind: BlockKind) { + self.blocked_by_kind[kind as usize].fetch_add(1, Ordering::Relaxed); + } + + /// Direct byte accounting for paths that handle a whole message at once (the HTTP-mode + /// listener), where `CountingStream`'s per-connection batching would buy nothing — these + /// fire once or twice per connection, not per chunk. + pub fn add_bytes_in(&self, bytes: u64) { + self.bytes_in.fetch_add(bytes, Ordering::Relaxed); + } + + pub fn add_bytes_out(&self, bytes: u64) { + self.bytes_out.fetch_add(bytes, Ordering::Relaxed); + } + + pub fn inc_error(&self, kind: ErrorKind) { + self.errors_by_kind[kind as usize].fetch_add(1, Ordering::Relaxed); } - pub fn inc_error(&self) { - self.total_errors.fetch_add(1, Ordering::Relaxed); + /// Per-reason block counts, in `BlockKind::ALL` order. Zero-valued reasons are included + /// so an exported series exists from the first scrape rather than appearing mid-incident. + pub fn blocked_by_reason(&self) -> Vec<(&'static str, u64)> { + BlockKind::ALL + .iter() + .map(|k| (k.as_str(), self.blocked_by_kind[*k as usize].load(Ordering::Relaxed))) + .collect() + } + + /// Per-reason error counts, in `ErrorKind::ALL` order. See `blocked_by_reason`. + pub fn errors_by_reason(&self) -> Vec<(&'static str, u64)> { + ErrorKind::ALL + .iter() + .map(|k| (k.as_str(), self.errors_by_kind[*k as usize].load(Ordering::Relaxed))) + .collect() } } +/// Sums a per-reason breakdown into its total. +/// +/// The exported totals are derived from the same values the breakdown reports rather than kept +/// as separate counters. A standalone `total_blocked` incremented next to its reason counter is +/// two non-atomic writes, so a scrape landing between them observes a total that does not equal +/// the sum of its parts — the invariant would hold only while the proxy is idle, which is +/// exactly when nobody is looking. Deriving makes it structural, and removes an atomic from the +/// block/error paths. +pub fn total_of(counts: &[(&'static str, u64)]) -> u64 { + counts.iter().map(|(_, count)| count).sum() +} + +// No proxy-wide blocked counter: it is derived from the listeners in `metrics()` for the same +// reason the per-listener totals are derived from their reasons — a separately incremented copy +// can disagree with its parts under traffic. #[derive(Default)] pub struct GlobalMetrics { pub active_connections: AtomicU64, - pub total_blocked: AtomicU64, pub pending_suspended: AtomicU64, + /// Suspended connections that JS resolved with a route. + pub suspended_resolved: AtomicU64, + /// Suspended connections that timed out or were rejected with a null route. + pub suspended_unresolved: AtomicU64, } impl GlobalMetrics { @@ -43,15 +192,215 @@ impl GlobalMetrics { self.active_connections.fetch_sub(1, Ordering::Relaxed); } - pub fn inc_blocked(&self) { - self.total_blocked.fetch_add(1, Ordering::Relaxed); - } - pub fn inc_suspended(&self) { self.pending_suspended.fetch_add(1, Ordering::Relaxed); } - pub fn dec_suspended(&self) { + /// Leave the pending gauge, recording how the suspension ended. + pub fn dec_suspended(&self, resolved: bool) { self.pending_suspended.fetch_sub(1, Ordering::Relaxed); + if resolved { + self.suspended_resolved.fetch_add(1, Ordering::Relaxed); + } else { + self.suspended_unresolved.fetch_add(1, Ordering::Relaxed); + } + } +} + +/// Bytes a connection may accumulate locally before publishing to the shared listener counters. +/// The whole point of the local buffer is to keep the shared cache line off the per-chunk path +/// (see `CountingStream`), so this wants to be well above `copy_bidirectional`'s 8 KiB buffer — +/// at 256 KiB a saturated connection publishes ~32× less often than it would per chunk, while a +/// scrape still sees a busy connection's traffic within a fraction of a second. +const COUNTER_FLUSH_BYTES: u64 = 256 * 1024; + +/// Wraps the *client* side of a proxied session so byte counts accrue as the copy runs rather +/// than at completion. Counting `copy_bidirectional`'s return value instead would lose every +/// byte of any session that ends by idle timeout or reset — i.e. most long-lived ones. +/// +/// Counts are accumulated per connection and published to the shared `ListenerMetrics` only +/// every `COUNTER_FLUSH_BYTES` and on drop. A `fetch_add` per chunk would put a single shared +/// cache line in the path of every 8 KiB of proxied traffic, ping-ponging it across every core — +/// exactly the cross-core contention `SO_REUSEPORT` per worker exists to avoid. +/// +/// Because it wraps the client, the direction naming is from the proxy's point of view: bytes +/// read here came *from* the client, bytes written here go *to* the client. +pub struct CountingStream<'a, S> { + inner: S, + metrics: &'a ListenerMetrics, + pending_in: u64, + pending_out: u64, +} + +impl<'a, S> CountingStream<'a, S> { + pub fn new(inner: S, metrics: &'a ListenerMetrics) -> Self { + Self { inner, metrics, pending_in: 0, pending_out: 0 } + } + + fn record_in(&mut self, bytes: u64) { + self.pending_in += bytes; + if self.pending_in >= COUNTER_FLUSH_BYTES { + self.metrics.bytes_in.fetch_add(self.pending_in, Ordering::Relaxed); + self.pending_in = 0; + } + } + + fn record_out(&mut self, bytes: u64) { + self.pending_out += bytes; + if self.pending_out >= COUNTER_FLUSH_BYTES { + self.metrics.bytes_out.fetch_add(self.pending_out, Ordering::Relaxed); + self.pending_out = 0; + } + } +} + +// Publishes whatever is left when the session ends — including when the connection task is +// aborted at shutdown, since that drops the future and with it this stream. +impl Drop for CountingStream<'_, S> { + fn drop(&mut self) { + if self.pending_in > 0 { + self.metrics.bytes_in.fetch_add(self.pending_in, Ordering::Relaxed); + } + if self.pending_out > 0 { + self.metrics.bytes_out.fetch_add(self.pending_out, Ordering::Relaxed); + } + } +} + +impl AsyncRead for CountingStream<'_, S> { + fn poll_read(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll> { + let this = self.get_mut(); + let before = buf.filled().len(); + let result = Pin::new(&mut this.inner).poll_read(cx, buf); + if matches!(result, Poll::Ready(Ok(()))) { + let read = buf.filled().len().saturating_sub(before); + if read > 0 { + this.record_in(read as u64); + } + } + result + } +} + +impl AsyncWrite for CountingStream<'_, S> { + fn poll_write(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll> { + let this = self.get_mut(); + let result = Pin::new(&mut this.inner).poll_write(cx, buf); + if let Poll::Ready(Ok(written)) = &result { + this.record_out(*written as u64); + } + result + } + + fn poll_write_vectored( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + bufs: &[std::io::IoSlice<'_>], + ) -> Poll> { + let this = self.get_mut(); + let result = Pin::new(&mut this.inner).poll_write_vectored(cx, bufs); + if let Poll::Ready(Ok(written)) = &result { + this.record_out(*written as u64); + } + result + } + + fn is_write_vectored(&self) -> bool { + self.inner.is_write_vectored() + } + + fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.get_mut().inner).poll_flush(cx) + } + + fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.get_mut().inner).poll_shutdown(cx) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn block_and_error_kinds_have_distinct_labels() { + let mut labels: Vec<&str> = BlockKind::ALL.iter().map(|k| k.as_str()).collect(); + labels.sort_unstable(); + let count = labels.len(); + labels.dedup(); + assert_eq!(labels.len(), count, "BlockKind labels must be unique"); + + let mut labels: Vec<&str> = ErrorKind::ALL.iter().map(|k| k.as_str()).collect(); + labels.sort_unstable(); + let count = labels.len(); + labels.dedup(); + assert_eq!(labels.len(), count, "ErrorKind labels must be unique"); + } + + #[test] + fn per_reason_counts_sum_to_the_total() { + let m = ListenerMetrics::default(); + m.inc_blocked(BlockKind::RateLimited); + m.inc_blocked(BlockKind::RateLimited); + m.inc_blocked(BlockKind::NoSni); + m.inc_error(ErrorKind::UpstreamConnect); + + assert_eq!(total_of(&m.blocked_by_reason()), 3); + assert_eq!(total_of(&m.errors_by_reason()), 1); + + // Every reason is exported, including the ones still at zero. + assert_eq!(m.blocked_by_reason().len(), BlockKind::COUNT); + assert_eq!(m.errors_by_reason().len(), ErrorKind::COUNT); + } + + #[tokio::test] + async fn counting_stream_publishes_both_directions_on_drop() { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + let metrics = ListenerMetrics::default(); + // duplex gives a paired in-memory stream; write into `peer` to be read through the counter. + let (client, mut peer) = tokio::io::duplex(64); + let mut counted = CountingStream::new(client, &metrics); + + peer.write_all(b"hello").await.unwrap(); + let mut buf = [0u8; 5]; + counted.read_exact(&mut buf).await.unwrap(); + counted.write_all(b"world!").await.unwrap(); + + // Below the flush threshold, so the shared counters stay untouched until the session ends. + assert_eq!(metrics.bytes_in.load(Ordering::Relaxed), 0); + assert_eq!(metrics.bytes_out.load(Ordering::Relaxed), 0); + + drop(counted); + assert_eq!(metrics.bytes_in.load(Ordering::Relaxed), 5); + assert_eq!(metrics.bytes_out.load(Ordering::Relaxed), 6); + } + + // A long-lived connection must not withhold its traffic from scrapes until it closes. + #[tokio::test] + async fn counting_stream_publishes_once_past_the_flush_threshold() { + use tokio::io::AsyncWriteExt; + + let metrics = ListenerMetrics::default(); + let (client, mut peer) = tokio::io::duplex(COUNTER_FLUSH_BYTES as usize * 4); + let mut counted = CountingStream::new(client, &metrics); + + let chunk = vec![0u8; 8 * 1024]; + let mut written = 0u64; + while written < COUNTER_FLUSH_BYTES { + counted.write_all(&chunk).await.unwrap(); + written += chunk.len() as u64; + } + + assert_eq!( + metrics.bytes_out.load(Ordering::Relaxed), + written, + "crossing the threshold must publish everything accumulated so far" + ); + + // Drain so the duplex peer doesn't hold the buffer, then confirm drop double-counts nothing. + drop(counted); + peer.shutdown().await.ok(); + assert_eq!(metrics.bytes_out.load(Ordering::Relaxed), written); } } diff --git a/src/proxy.rs b/src/proxy.rs index 85d8ca3..144efea 100644 --- a/src/proxy.rs +++ b/src/proxy.rs @@ -1,6 +1,6 @@ use crate::http_listener::spawn_http_listeners; use crate::listener::spawn_listeners; -use crate::metrics::{GlobalMetrics, ListenerMetrics}; +use crate::metrics::{total_of, GlobalMetrics, ListenerMetrics}; use crate::protection::ProtectionState; use crate::proxy_conn::{ConnContext, JsEvent}; use crate::router::{ @@ -147,11 +147,47 @@ pub struct JsHotConfig { pub protection: Option>, } +/// A single labelled counter — one entry per block/error reason. +#[napi(object)] +pub struct JsLabeledCount { + pub reason: String, + pub count: f64, +} + +#[napi(object)] +pub struct JsListenerMetrics { + /// "host:port" — matches the `listener` field on emitted events. + pub address: String, + /// "tls" or "http". + pub mode: String, + pub active_connections: f64, + pub accepted: f64, + pub blocked: f64, + pub errors: f64, + /// Bytes read from clients (client → upstream). + pub bytes_received: f64, + /// Bytes written to clients (upstream → client). + pub bytes_sent: f64, + pub blocked_by_reason: Vec, + pub errors_by_reason: Vec, +} + +// Counters are reported as f64 because napi maps u64 to BigInt, which JSON.stringify cannot +// serialise. f64 is exact to 2^53, so a byte counter stays exact past 9 PB per listener. #[napi(object)] pub struct JsProxyMetrics { pub active_connections: f64, pub blocked_connections: f64, pub pending_suspended: f64, + /// Suspended connections that JS resolved with a route. + pub suspended_resolved: f64, + /// Suspended connections that timed out or were rejected. + pub suspended_unresolved: f64, + /// Routes currently in the live table, including the default route. + pub routes: f64, + /// Routes whose cert failed to build — dropped, or serving a carried-forward last-good cert. + pub failing_routes: f64, + pub listeners: Vec, } #[napi(object)] @@ -190,6 +226,13 @@ struct InternalListener { mode: ListenerMode, } +fn labeled_counts(counts: Vec<(&'static str, u64)>) -> Vec { + counts + .into_iter() + .map(|(reason, count)| JsLabeledCount { reason: reason.to_string(), count: count as f64 }) + .collect() +} + /// Per-listener runtime state. struct ListenerState { addr: String, @@ -553,10 +596,52 @@ impl SymphonyProxyWrap { #[napi] pub fn metrics(&self) -> JsProxyMetrics { + let table = self.route_table.0.load(); + + // `listeners` and `listener_states` are built in lockstep in the constructor and never + // mutated, so index i refers to the same listener in both. + // + // Each listener's totals are summed from the very reason values it reports, so a scrape + // taken mid-traffic is internally consistent — a separately maintained total would be a + // second write racing the first and could disagree with its own breakdown. + let listeners: Vec = self + .listeners + .iter() + .zip(self.listener_states.iter()) + .map(|(listener, state)| { + let blocked_by_reason = state.metrics.blocked_by_reason(); + let errors_by_reason = state.metrics.errors_by_reason(); + JsListenerMetrics { + address: state.addr.clone(), + mode: match listener.mode { + ListenerMode::Tls => "tls".to_string(), + ListenerMode::Http => "http".to_string(), + }, + active_connections: state.metrics.active_connections.load(Ordering::Relaxed) as f64, + accepted: state.metrics.total_accepted.load(Ordering::Relaxed) as f64, + blocked: total_of(&blocked_by_reason) as f64, + errors: total_of(&errors_by_reason) as f64, + bytes_received: state.metrics.bytes_in.load(Ordering::Relaxed) as f64, + bytes_sent: state.metrics.bytes_out.load(Ordering::Relaxed) as f64, + blocked_by_reason: labeled_counts(blocked_by_reason), + errors_by_reason: labeled_counts(errors_by_reason), + } + }) + .collect(); + + // Likewise derived, so the proxy-wide total always equals the sum of the listener values + // in this same snapshot. + let blocked_connections = listeners.iter().map(|l| l.blocked).sum(); + JsProxyMetrics { active_connections: self.global_metrics.active_connections.load(Ordering::Relaxed) as f64, - blocked_connections: self.global_metrics.total_blocked.load(Ordering::Relaxed) as f64, + blocked_connections, pending_suspended: self.global_metrics.pending_suspended.load(Ordering::Relaxed) as f64, + suspended_resolved: self.global_metrics.suspended_resolved.load(Ordering::Relaxed) as f64, + suspended_unresolved: self.global_metrics.suspended_unresolved.load(Ordering::Relaxed) as f64, + routes: table.route_count() as f64, + failing_routes: table.failing_route_count() as f64, + listeners, } } diff --git a/src/proxy_conn.rs b/src/proxy_conn.rs index 21757c1..8d10803 100644 --- a/src/proxy_conn.rs +++ b/src/proxy_conn.rs @@ -1,4 +1,4 @@ -use crate::metrics::{GlobalMetrics, ListenerMetrics}; +use crate::metrics::{BlockKind, CountingStream, ErrorKind, GlobalMetrics, ListenerMetrics}; use crate::protection::{IpState, ProtectionState}; use crate::router::{Destination, ForwardFingerprint, LiveRouteTable, SourceAddressMode}; use crate::sni; @@ -84,8 +84,7 @@ pub async fn handle(stream: TcpStream, peer_addr: SocketAddr, ctx: Arc { - ctx.listener_metrics.inc_blocked(); - ctx.global_metrics.inc_blocked(); + ctx.listener_metrics.inc_blocked(BlockKind::from(&reason)); emit(&ctx.js_emit, JsEvent::Blocked { ip: peer_ip.to_string(), reason: reason.as_str().to_string(), @@ -110,7 +109,7 @@ pub async fn handle(stream: TcpStream, peer_addr: SocketAddr, ctx: Arc r.clone(), None => { - ctx.listener_metrics.inc_error(); + ctx.listener_metrics.inc_error(ErrorKind::NoRoute); return; // No route and no default — drop } }; @@ -118,7 +117,7 @@ pub async fn handle(stream: TcpStream, peer_addr: SocketAddr, ctx: Arc r, _ => { ctx.suspended_registry.remove(id); - ctx.global_metrics.dec_suspended(); + ctx.global_metrics.dec_suspended(false); + ctx.listener_metrics.inc_error(ErrorKind::SuspendUnresolved); return; // Timed out or rejected } }; - ctx.global_metrics.dec_suspended(); + ctx.global_metrics.dec_suspended(true); EffectiveRoute { destination: resolved.destination, @@ -203,17 +203,17 @@ pub async fn handle(stream: TcpStream, peer_addr: SocketAddr, ctx: Arc { tracing::debug!("TLS handshake error from {peer_ip}: {e}"); - ctx.listener_metrics.inc_error(); + ctx.listener_metrics.inc_error(ErrorKind::TlsHandshake); return; } Err(_) => { tracing::debug!("TLS handshake timeout from {peer_ip}"); - ctx.listener_metrics.inc_error(); + ctx.listener_metrics.inc_error(ErrorKind::TlsHandshake); return; } } } else { - ctx.listener_metrics.inc_error(); + ctx.listener_metrics.inc_error(ErrorKind::TlsMissingCert); return; } } else { @@ -221,33 +221,39 @@ pub async fn handle(stream: TcpStream, peer_addr: SocketAddr, ctx: Arc, + client: tokio_rustls::server::TlsStream, dest: &Destination, sf: SourceForwarding<'_>, ctx: &ConnContext, -) -> std::io::Result<()> { +) -> std::result::Result<(), ErrorKind> { // TLS facts (incl. the verified mTLS client cert chain) forwarded via PROXY v2 // TLVs; only collected on routes that can carry them. let tls_forward = matches!(sf.mode, SourceAddressMode::ProxyProtocolV2) .then(|| collect_tls_forward(client.get_ref().1)); let sf = SourceForwarding { tls: tls_forward.as_ref(), ..sf }; - let mut upstream = upstream::connect(dest, Some(sf.peer_addr.ip()), ctx.upstream_connect_timeout) - .await - .map_err(|e| std::io::Error::other(e.to_string()))?; - // HTTP-header injection is only valid for a plaintext HTTP/1 upstream. An h2-negotiated // upstream receives binary frames, so text header insertion would corrupt them. + // Read before wrapping — the counter has no view of the TLS session. let l7_http1 = client.get_ref().1.alpn_protocol() != Some(b"h2".as_ref()); + let mut upstream = upstream::connect(dest, Some(sf.peer_addr.ip()), ctx.upstream_connect_timeout) + .await + .map_err(|e| { + tracing::debug!("upstream connect failed for {}: {e}", sf.peer_addr.ip()); + ErrorKind::UpstreamConnect + })?; + + let mut client = CountingStream::new(client, &ctx.listener_metrics); + match &mut upstream { UpstreamStream::Tcp(ref mut up) => forward(&mut client, up, &sf, l7_http1, ctx.idle_timeout).await, UpstreamStream::Uds { ref mut stream, .. } => { @@ -257,14 +263,19 @@ async fn proxy_via_tls( } async fn proxy_raw( - mut client: TcpStream, + client: TcpStream, dest: &Destination, sf: SourceForwarding<'_>, ctx: &ConnContext, -) -> std::io::Result<()> { +) -> std::result::Result<(), ErrorKind> { let mut upstream = upstream::connect(dest, Some(sf.peer_addr.ip()), ctx.upstream_connect_timeout) .await - .map_err(|e| std::io::Error::other(e.to_string()))?; + .map_err(|e| { + tracing::debug!("upstream connect failed for {}: {e}", sf.peer_addr.ip()); + ErrorKind::UpstreamConnect + })?; + + let mut client = CountingStream::new(client, &ctx.listener_metrics); // Passthrough forwards raw TLS bytes — never a plaintext HTTP/1 stream, so header injection // is disabled (only PROXY protocol carriers apply here). @@ -287,7 +298,7 @@ async fn forward( sf: &SourceForwarding<'_>, l7_http1: bool, idle: Duration, -) -> std::io::Result<()> +) -> std::result::Result<(), ErrorKind> where C: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin, U: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin, @@ -301,12 +312,15 @@ where crate::http_proxy::proxy_http1_rewriting(client, upstream, &rewrites).await } }; + // The idle timeout is reported as its own kind rather than inferred from an + // `io::ErrorKind::TimedOut`, which a peer's kernel-level ETIMEDOUT would also produce. if idle.is_zero() { - body.await + body.await.map_err(|_| ErrorKind::Stream) } else { - timeout(idle, body) - .await - .map_err(|_| std::io::Error::new(std::io::ErrorKind::TimedOut, "idle timeout"))? + match timeout(idle, body).await { + Ok(result) => result.map_err(|_| ErrorKind::Stream), + Err(_) => Err(ErrorKind::IdleTimeout), + } } } diff --git a/src/router.rs b/src/router.rs index f4a01e4..f9eccbf 100644 --- a/src/router.rs +++ b/src/router.rs @@ -164,6 +164,17 @@ pub struct RouteTable { } impl RouteTable { + /// Number of routes serving traffic, including the default route if one is configured. + pub fn route_count(&self) -> usize { + self.exact.len() + self.wildcard.len() + usize::from(self.default.is_some()) + } + + /// Number of SNIs whose cert failed to build in this table — either dropped, or serving a + /// carried-forward last-good cert. Non-zero means a rotation needs attention. + pub fn failing_route_count(&self) -> usize { + self.failing_snis.len() + } + pub fn resolve(&self, sni: Option<&str>) -> Option<&Route> { let Some(sni) = sni else { return self.default.as_ref(); diff --git a/ts/addon.d.ts b/ts/addon.d.ts index 9c0a5eb..702adf3 100644 --- a/ts/addon.d.ts +++ b/ts/addon.d.ts @@ -118,10 +118,40 @@ export interface JsHotConfig { */ protection?: Array } +/** A single labelled counter — one entry per block/error reason. */ +export interface JsLabeledCount { + reason: string + count: number +} +export interface JsListenerMetrics { + /** "host:port" — matches the `listener` field on emitted events. */ + address: string + /** "tls" or "http". */ + mode: string + activeConnections: number + accepted: number + blocked: number + errors: number + /** Bytes read from clients (client → upstream). */ + bytesReceived: number + /** Bytes written to clients (upstream → client). */ + bytesSent: number + blockedByReason: Array + errorsByReason: Array +} export interface JsProxyMetrics { activeConnections: number blockedConnections: number pendingSuspended: number + /** Suspended connections that JS resolved with a route. */ + suspendedResolved: number + /** Suspended connections that timed out or were rejected. */ + suspendedUnresolved: number + /** Routes currently in the live table, including the default route. */ + routes: number + /** Routes whose cert failed to build — dropped, or serving a carried-forward last-good cert. */ + failingRoutes: number + listeners: Array } export interface JsBlockedIpsInfo { rateLimited: Array diff --git a/ts/admin.ts b/ts/admin.ts new file mode 100644 index 0000000..8efe315 --- /dev/null +++ b/ts/admin.ts @@ -0,0 +1,507 @@ +//! Read-only admin/metrics endpoint for the standalone `symphony-server`. +//! +//! Consumers that run symphony out-of-process have no access to the napi `metrics()` call, so +//! the server exposes the same numbers over HTTP: +//! +//! GET /metrics Prometheus text exposition (v0.0.4) +//! GET /metrics.json the same snapshot as JSON +//! GET /health liveness (`{ ok, pid, version, ports }`) +//! +//! It binds a Unix socket, a loopback TCP port, or both. Everything here is best-effort: a +//! bind failure or a handler throw must never affect proxying, so failures are logged and +//! retried rather than propagated. + +import { createServer, type Server, type IncomingMessage, type ServerResponse } from 'node:http'; +import { connect } from 'node:net'; +import { chmodSync, existsSync, lstatSync, renameSync, unlinkSync } from 'node:fs'; +import type { ProxyMetrics } from './types.js'; + +/** One thing to listen on. `precheck` may reject the bind before a server is created. */ +interface BindTarget { + describe: string; + listen: (server: Server) => void; + precheck?: () => Promise; + /** Runs once bound; throwing fails the bind (the caller closes the server and retries). */ + onBound?: () => void; + /** Best-effort removal of anything `listen`/`onBound` left behind on a failed attempt. */ + cleanup?: () => void; +} + +export interface AdminConfig { + /** Unix socket path to listen on. Relative paths resolve against the config file's directory. */ + socketPath?: string; + /** Permissions applied to `socketPath` after bind. Default 0o660. */ + socketMode?: number; + /** TCP port to listen on. */ + port?: number; + /** Interface for `port`. Default 127.0.0.1 — do not expose metrics off-box without a reason. */ + host?: string; +} + +/** One running proxy's identity and current counters. */ +export interface ProxySnapshot { + /** Sorted listener ports for this proxy, as configured ("80,443"). */ + ports: string; + metrics: ProxyMetrics; +} + +export interface MetricsSnapshot { + pid: number; + version: string; + startedAt: string; + reloadedAt: string; + proxies: ProxySnapshot[]; +} + +const RETRY_MS = 5_000; +const KEEP_ALIVE_MS = 5_000; +const PROBE_TIMEOUT_MS = 1_000; +/** Hard ceiling on concurrent admin connections — a scrape endpoint needs a handful at most. */ +const MAX_ADMIN_CONNECTIONS = 16; + +// ── Prometheus rendering ────────────────────────────────────────────────────── + +// Only `\`, `"` and newline are special in a label value (exposition format v0.0.4). +function escapeLabel(value: string): string { + return value.replace(/\\/g, '\\\\').replace(/"/g, '\\"').replace(/\n/g, '\\n'); +} + +function labels(pairs: Record): string { + const rendered = Object.entries(pairs) + .map(([k, v]) => `${k}="${escapeLabel(v)}"`) + .join(','); + return rendered ? `{${rendered}}` : ''; +} + +/** + * Accumulates samples grouped by metric name. + * + * The exposition format requires every sample of a metric name to be contiguous, under a single + * HELP/TYPE pair. Emitting in call order would interleave them — with more than one proxy + * configured, `symphony_routes` for the second proxy would land after the first proxy's listener + * samples, and strict parsers reject or drop the split group. Grouping here means the callers + * below can stay in the natural proxy → listener iteration order. + */ +class Exposition { + private readonly groups = new Map(); + + sample( + name: string, + type: 'counter' | 'gauge', + help: string, + value: number, + tags: Record = {} + ): void { + let group = this.groups.get(name); + if (!group) this.groups.set(name, (group = { type, help, samples: [] })); + group.samples.push(`${name}${labels(tags)} ${value}`); + } + + toString(): string { + const lines: string[] = []; + for (const [name, { type, help, samples }] of this.groups) { + lines.push(`# HELP ${name} ${help}`, `# TYPE ${name} ${type}`, ...samples); + } + return `${lines.join('\n')}\n`; + } +} + +/** + * Render a snapshot as Prometheus text. + * + * Blocked and error counts are only ever emitted with their `reason` label — the per-reason + * series sum to the unlabeled total by construction, so a separate total would be a second + * representation of the same number. Use `sum without(reason)` for the total. Likewise the + * proxy-level active-connection gauge is `sum without(listener)` of the per-listener one. + */ +export function renderPrometheus(snapshot: MetricsSnapshot): string { + const out = new Exposition(); + + out.sample('symphony_build_info', 'gauge', 'Always 1; the version is carried in the label.', 1, { + version: snapshot.version, + }); + out.sample( + 'symphony_start_time_seconds', + 'gauge', + 'Unix time the server process started.', + Date.parse(snapshot.startedAt) / 1000 + ); + out.sample( + 'symphony_config_reload_time_seconds', + 'gauge', + 'Unix time of the last successful config reconcile.', + Date.parse(snapshot.reloadedAt) / 1000 + ); + + for (const { ports, metrics } of snapshot.proxies) { + const proxy = { proxy: ports }; + + out.sample( + 'symphony_routes', + 'gauge', + 'Routes in the live table, including the default route.', + metrics.routes, + proxy + ); + out.sample( + 'symphony_routes_failing', + 'gauge', + 'Routes whose cert failed to build — dropped, or serving a carried-forward last-good cert.', + metrics.failingRoutes, + proxy + ); + out.sample( + 'symphony_suspended_pending', + 'gauge', + 'Connections held awaiting resolveConnection().', + metrics.pendingSuspended, + proxy + ); + out.sample( + 'symphony_suspended_total', + 'counter', + 'Suspended connections by how they ended.', + metrics.suspendedResolved, + { + ...proxy, + outcome: 'resolved', + } + ); + out.sample( + 'symphony_suspended_total', + 'counter', + 'Suspended connections by how they ended.', + metrics.suspendedUnresolved, + { + ...proxy, + outcome: 'unresolved', + } + ); + + for (const l of metrics.listeners) { + const tags = { ...proxy, listener: l.address, mode: l.mode }; + + out.sample( + 'symphony_listener_active_connections', + 'gauge', + 'Connections currently being proxied.', + l.activeConnections, + tags + ); + out.sample('symphony_listener_accepted_total', 'counter', 'Connections accepted for proxying.', l.accepted, tags); + out.sample( + 'symphony_listener_bytes_received_total', + 'counter', + 'Bytes read from clients (client → upstream).', + l.bytesReceived, + tags + ); + out.sample( + 'symphony_listener_bytes_sent_total', + 'counter', + 'Bytes written to clients (upstream → client).', + l.bytesSent, + tags + ); + for (const { reason, count } of l.blockedByReason) { + out.sample( + 'symphony_listener_blocked_total', + 'counter', + 'Connections rejected before proxying, by reason.', + count, + { + ...tags, + reason, + } + ); + } + for (const { reason, count } of l.errorsByReason) { + out.sample('symphony_listener_errors_total', 'counter', 'Connections that failed, by reason.', count, { + ...tags, + reason, + }); + } + } + } + + return out.toString(); +} + +// ── Server ──────────────────────────────────────────────────────────────────── + +/** + * Whether `socketPath` is safe to remove and rebind. + * + * Reclaimable means two things, and both must hold — the cost of getting this wrong is deleting + * a *live* endpoint out from under a running process, which is the failure mode symphony's + * status.json ownership guard exists to prevent: + * + * - `ECONNREFUSED` specifically, not merely "the probe failed". A live socket with restrictive + * permissions (or one owned by another uid) refuses the probe with `EACCES`; treating every + * error as stale would unlink it. Anything that isn't a definitive "nobody is listening" + * leaves the path alone and the bind is retried instead. + * - The inode is actually a socket. Left to `EADDRINUSE` alone, a `socketPath` misconfigured + * onto a regular file (say, status.json) would see the connect fail and delete that file. + */ +function socketIsReclaimable(socketPath: string): Promise { + return new Promise((resolve) => { + let stats; + try { + stats = lstatSync(socketPath); + } catch { + return resolve(false); // vanished under us — let the bind retry decide + } + if (!stats.isSocket()) return resolve(false); + + const probe = connect(socketPath); + let settled = false; + const done = (reclaimable: boolean) => { + if (settled) return; + settled = true; + clearTimeout(timer); + probe.destroy(); + resolve(reclaimable); + }; + // A frozen owner or a full accept backlog can leave the connect hanging, which would + // stall the reconcile that called us. Treat a hang as "not reclaimable" — something is + // there, and refusing to touch it is the safe reading. + const timer = setTimeout(() => done(false), PROBE_TIMEOUT_MS); + timer.unref(); + probe.once('connect', () => done(false)); + probe.once('error', (err: NodeJS.ErrnoException) => done(err.code === 'ECONNREFUSED')); + }); +} + +/** + * Owns the admin listeners for the lifetime of the process. `update()` is idempotent: it is + * called on every reconcile and only rebuilds when the admin config actually changed. + */ +export class AdminServer { + private readonly snapshot: () => MetricsSnapshot; + private readonly log: (msg: string, ...rest: unknown[]) => void; + private readonly logErr: (msg: string, ...rest: unknown[]) => void; + private servers: Server[] = []; + private signature = ''; + private config: AdminConfig | null = null; + private retryTimer: NodeJS.Timeout | null = null; + private stopped = false; + // update(), stop(), and the retry timer all mutate the same listeners, and each of them + // awaits. Run them through one chain so a retry that is mid-bind can't publish its server + // after a later update() or stop() has already finished tearing things down. + private queue: Promise = Promise.resolve(); + + constructor( + snapshot: () => MetricsSnapshot, + log: (msg: string, ...rest: unknown[]) => void, + logErr: (msg: string, ...rest: unknown[]) => void + ) { + this.snapshot = snapshot; + this.log = log; + this.logErr = logErr; + } + + /** Serialize a mutation of the listener set; failures never break the chain. */ + private enqueue(operation: () => Promise): Promise { + this.queue = this.queue.then(operation, operation); + return this.queue; + } + + update(config: AdminConfig | undefined): Promise { + return this.enqueue(async () => { + if (this.stopped) return; + const signature = JSON.stringify(config ?? null); + if (signature === this.signature) return; + this.signature = signature; + this.config = config ?? null; + // Drop a pending retry from the previous config — otherwise it fires seconds later and + // tears down the listeners this call is about to bind. + this.clearRetry(); + await this.closeServers(); + if (config && (config.socketPath || config.port !== undefined)) await this.bind(); + }); + } + + stop(): Promise { + // Set before queueing, so an operation already waiting its turn bails instead of binding. + this.stopped = true; + this.clearRetry(); + return this.enqueue(() => this.closeServers()); + } + + private clearRetry(): void { + if (this.retryTimer) { + clearTimeout(this.retryTimer); + this.retryTimer = null; + } + } + + // A failed bind is retried rather than thrown: during a version upgrade the incumbent still + // holds the admin socket/port (the proxy listeners overlap via SO_REUSEPORT, but a Node HTTP + // server has no such luxury), so the successor binds a few seconds later once it exits. + private scheduleRetry(): void { + if (this.stopped || this.retryTimer) return; + this.retryTimer = setTimeout(() => { + this.retryTimer = null; + // Queued like everything else, so a retry can't interleave with an update or a stop. + void this.enqueue(() => this.bind()); + }, RETRY_MS); + this.retryTimer.unref(); + } + + private async bind(): Promise { + const config = this.config; + if (this.stopped || !config) return; + await this.closeServers(); + + const targets: BindTarget[] = []; + if (config.socketPath) targets.push(this.socketTarget(config.socketPath, config.socketMode ?? 0o660)); + if (config.port !== undefined) { + const host = config.host ?? '127.0.0.1'; + targets.push({ describe: `${host}:${config.port}`, listen: (server) => server.listen(config.port, host) }); + } + + for (const target of targets) { + try { + const server = await this.listenOne(target); + this.servers.push(server); + this.log(`admin endpoint listening on ${target.describe}`); + } catch (err) { + this.logErr(`could not bind admin endpoint on ${target.describe} (retrying):`, (err as Error).message); + // Drop any sibling that did bind so a retry starts from a clean slate and can't + // double-bind the one that succeeded. + await this.closeServers(); + this.scheduleRetry(); + return; + } + } + } + + /** + * Bind the Unix socket by listening on a pid-unique temporary path and `rename`-ing it onto + * the real one. + * + * The obvious shape — probe, `unlink`, `listen` — has a window between the probe and the + * unlink. Two processes can both find the path stale, and the second one's unlink then + * deletes the socket the first has already bound and is serving. `rename` is atomic and + * replaces the target in one step, so the path always names a socket somebody is listening + * on. The precheck stays, because rename would otherwise happily clobber a *live* incumbent + * during an upgrade overlap — there we want to lose and retry, not steal the endpoint. + * + * Nothing unlinks the published path, including on a clean shutdown. Any check-then-unlink + * can delete a *successor's* socket in the window between the two syscalls, and that loss is + * not self-repairing: `update()` returns early on an unchanged signature, so the successor + * would keep serving an unreachable socket until its config changed or it restarted. Leaving + * a stale pathname behind costs one inode in a 0o700 directory, and the next binder replaces + * it atomically after proving it stale. Cheap litter beats a silently dead endpoint. + */ + private socketTarget(path: string, mode: number): BindTarget { + const tempPath = `${path}.${process.pid}`; + return { + describe: path, + precheck: async () => { + if (existsSync(path) && !(await socketIsReclaimable(path))) { + throw new Error(`${path} is in use by another process`); + } + }, + listen: (server) => server.listen(tempPath), + onBound: () => { + chmodSync(tempPath, mode); + renameSync(tempPath, path); + }, + cleanup: () => { + try { + unlinkSync(tempPath); + } catch { + // never created, or already renamed into place + } + }, + }; + } + + private async listenOne(target: BindTarget): Promise { + await target.precheck?.(); + + const server = createServer((req, res) => this.handle(req, res)); + // Scrapes are short and infrequent, so reap idle connections quickly. Note this is a + // timeout, not a switch: setting it to 0 would disable the reaping, letting a client park + // arbitrarily many idle connections against the same process-wide fd budget the proxy + // listeners draw from. maxConnections caps that regardless of client behaviour. + server.keepAliveTimeout = KEEP_ALIVE_MS; + server.maxConnections = MAX_ADMIN_CONNECTIONS; + + try { + return await new Promise((resolve, reject) => { + const onError = (err: NodeJS.ErrnoException) => { + server.removeListener('listening', onListening); + reject(err); + }; + const onListening = () => { + server.removeListener('error', onError); + try { + target.onBound?.(); + } catch (err) { + // The socket is bound but not reachable at its published path — useless, and + // it would leak an fd and a temp inode. Fail so the retry starts clean. + server.close(); + reject(err); + return; + } + // Post-bind errors must not crash the process. + server.on('error', (err) => this.logErr(`admin endpoint error (${target.describe}):`, err)); + resolve(server); + }; + server.once('error', onError); + server.once('listening', onListening); + target.listen(server); + }); + } catch (err) { + target.cleanup?.(); + throw err; + } + } + + private handle(req: IncomingMessage, res: ServerResponse): void { + // Strip any query string; the endpoints take no parameters. + const path = (req.url ?? '/').split('?')[0]; + if (req.method !== 'GET' && req.method !== 'HEAD') { + res.writeHead(405, { allow: 'GET, HEAD' }).end(); + return; + } + + try { + const snapshot = this.snapshot(); + if (path === '/metrics') { + const body = renderPrometheus(snapshot); + res.writeHead(200, { 'content-type': 'text/plain; version=0.0.4; charset=utf-8' }).end(body); + } else if (path === '/metrics.json') { + res.writeHead(200, { 'content-type': 'application/json' }).end(JSON.stringify(snapshot)); + } else if (path === '/health') { + const ports = snapshot.proxies.flatMap((p) => p.ports.split(',').map(Number)); + res + .writeHead(200, { 'content-type': 'application/json' }) + .end(JSON.stringify({ ok: true, pid: snapshot.pid, version: snapshot.version, ports })); + } else { + res.writeHead(404, { 'content-type': 'text/plain' }).end('not found\n'); + } + } catch (err) { + this.logErr('admin request failed:', (err as Error).message); + res.writeHead(500, { 'content-type': 'text/plain' }).end('internal error\n'); + } + } + + private async closeServers(): Promise { + const servers = this.servers; + this.servers = []; + await Promise.all( + servers.map( + (server) => + new Promise((resolve) => { + server.close(() => resolve()); + // close() waits for in-flight requests; a scrape holding the socket must not + // stall shutdown. + server.closeAllConnections(); + }) + ) + ); + } +} diff --git a/ts/index.ts b/ts/index.ts index 5c31b80..a128ef7 100644 --- a/ts/index.ts +++ b/ts/index.ts @@ -12,6 +12,8 @@ export type { RateLimitConfig, HotConfig, ProxyMetrics, + ListenerMetrics, + LabeledCount, BlockedIpsInfo, SuspendedConnection, ResolveRoute, @@ -20,3 +22,9 @@ export type { SuspendedEvent, ErrorEvent, } from './types.js'; +// `renderPrometheus` and the standalone server's snapshot shape are deliberately NOT exported +// from the package root. They are an implementation detail of the symphony-server admin +// endpoint — a snapshot carries that process's pid, timestamps, and port-set grouping, which an +// embedded consumer would have to synthesise. Exporting them would make that shape a +// compatibility commitment with no caller asking for it; add a `ProxyMetrics`-based renderer +// interface if one ever does. diff --git a/ts/proxy.ts b/ts/proxy.ts index c44adbe..f8da7ca 100644 --- a/ts/proxy.ts +++ b/ts/proxy.ts @@ -239,6 +239,22 @@ export class SymphonyProxy extends EventEmitter { activeConnections: m.activeConnections, blockedConnections: m.blockedConnections, pendingSuspended: m.pendingSuspended, + suspendedResolved: m.suspendedResolved, + suspendedUnresolved: m.suspendedUnresolved, + routes: m.routes, + failingRoutes: m.failingRoutes, + listeners: m.listeners.map((l) => ({ + address: l.address, + mode: l.mode as 'tls' | 'http', + activeConnections: l.activeConnections, + accepted: l.accepted, + blocked: l.blocked, + errors: l.errors, + bytesReceived: l.bytesReceived, + bytesSent: l.bytesSent, + blockedByReason: l.blockedByReason.map((c) => ({ reason: c.reason, count: c.count })), + errorsByReason: l.errorsByReason.map((c) => ({ reason: c.reason, count: c.count })), + })), }; } diff --git a/ts/server.ts b/ts/server.ts index 4649cd3..26d5f93 100644 --- a/ts/server.ts +++ b/ts/server.ts @@ -3,6 +3,7 @@ import { readFileSync, writeFileSync, renameSync, unlinkSync, watch } from 'node import { dirname, isAbsolute, join, basename } from 'node:path'; import { SymphonyProxy } from './index.js'; import type { ProxyConfig, ListenerConfig, RouteConfig, CertConfig, MtlsConfig } from './index.js'; +import { AdminServer, type AdminConfig, type MetricsSnapshot } from './admin.js'; // package.json sits at the package root, which is 1 level above dist/server.js (production // layout) or 2 levels above dist-test/ts/server.js (test layout) — try both, like loadAddon. @@ -56,6 +57,8 @@ interface FileProxyConfig { interface ConfigFile { version?: number; + /** Optional read-only metrics/health endpoint. Omit to expose nothing. */ + admin?: AdminConfig; proxies: FileProxyConfig[]; } @@ -145,7 +148,9 @@ class ServerState { private readonly statusPath: string; private readonly baseDir: string; private readonly active = new Map(); + private readonly admin: AdminServer; private startedAt = ''; + private reloadedAt = ''; private reloading: Promise = Promise.resolve(); private watcher: ReturnType | null = null; // Cert/key files referenced by the current config, watched for rotation. Keyed by @@ -159,6 +164,19 @@ class ServerState { this.configPath = configPath; this.statusPath = statusPath; this.baseDir = dirname(configPath); + this.admin = new AdminServer(() => this.metricsSnapshot(), log, logErr); + } + + // Read live on each request rather than cached, so a scrape never serves counters frozen at + // the last reconcile. + private metricsSnapshot(): MetricsSnapshot { + return { + pid: process.pid, + version: pkg.version, + startedAt: this.startedAt, + reloadedAt: this.reloadedAt, + proxies: [...this.active].map(([ports, entry]) => ({ ports, metrics: entry.proxy.metrics() })), + }; } private readConfig(): ConfigFile | null { @@ -337,6 +355,15 @@ class ServerState { // renewal on disk (no config.json write) triggers a reconcile and live reload. this.updateCertWatchers(config); + // Rebinds only if the admin block changed. Never throws — the endpoint is + // observability, and losing it must not abort a reconcile that already applied routes. + const admin = config.admin && { + ...config.admin, + socketPath: config.admin.socketPath ? resolvePath(config.admin.socketPath, this.baseDir) : undefined, + }; + await this.admin.update(admin).catch((err) => logErr('admin endpoint update failed:', (err as Error).message)); + + this.reloadedAt = new Date().toISOString(); this.writeStatus(); } @@ -346,7 +373,7 @@ class ServerState { pid: process.pid, version: pkg.version, startedAt: this.startedAt, - reloadedAt: new Date().toISOString(), + reloadedAt: this.reloadedAt, configPath: this.configPath, ports, }; @@ -386,6 +413,7 @@ class ServerState { } for (const w of this.certWatchers.values()) w.close(); this.certWatchers.clear(); + await this.admin.stop().catch((err) => logErr('stopping admin endpoint:', (err as Error).message)); await this.reloading.catch(() => {}); for (const [key, entry] of this.active) { await entry.proxy.stop().catch((err) => logErr(`stopping proxy [${key}]:`, err)); diff --git a/ts/types.ts b/ts/types.ts index b741791..438afa3 100644 --- a/ts/types.ts +++ b/ts/types.ts @@ -268,6 +268,28 @@ export interface HotConfig { // ── Metrics ─────────────────────────────────────────────────────────────────── +/** A counter broken out by reason. Every reason is reported, including those still at zero. */ +export interface LabeledCount { + reason: string; + count: number; +} + +export interface ListenerMetrics { + /** "host:port" — matches the `listener` field on emitted events. */ + address: string; + mode: 'tls' | 'http'; + activeConnections: number; + accepted: number; + blocked: number; + errors: number; + /** Bytes read from clients (client → upstream). */ + bytesReceived: number; + /** Bytes written to clients (upstream → client). */ + bytesSent: number; + blockedByReason: LabeledCount[]; + errorsByReason: LabeledCount[]; +} + export interface ProxyMetrics { /** Number of connections currently being proxied. */ activeConnections: number; @@ -275,6 +297,16 @@ export interface ProxyMetrics { blockedConnections: number; /** Connections currently held waiting for resolveConnection(). */ pendingSuspended: number; + /** Suspended connections that were resolved with a route. */ + suspendedResolved: number; + /** Suspended connections that timed out or were rejected. */ + suspendedUnresolved: number; + /** Routes in the live table, including the default route. */ + routes: number; + /** Routes whose cert failed to build — dropped, or serving a carried-forward last-good cert. */ + failingRoutes: number; + /** Per-listener breakdown, in configuration order. */ + listeners: ListenerMetrics[]; } export interface BlockedIpsInfo {