diff --git a/specs/aggregate-prom-client-metrics-across-cluster-wor.md b/specs/aggregate-prom-client-metrics-across-cluster-wor.md new file mode 100644 index 000000000..99fc21d84 --- /dev/null +++ b/specs/aggregate-prom-client-metrics-across-cluster-wor.md @@ -0,0 +1,284 @@ +# Spec: Aggregate prom-client metrics across cluster workers for `/metrics` + +Internal task id: task-58ee1acb (bean) +Branch: `night/aggregate-prom-client-metrics-across-cluster-wor` + +> **Status: specification only.** This document describes the planned change. No +> implementation is included in this PR yet. + +--- + +## Context & problem + +VTEX IO runtimes run as a Node.js `cluster`: the master (`src/service/master.ts`) +forks `min(cpus, MAX_WORKERS=4)` workers (`src/service/loaders.ts` `getWorkers`, +`src/constants.ts` `MAX_WORKERS`), each worker listening on the same shared port +(`HTTP_SERVER_PORT`, `src/service/index.ts`). Every worker keeps its **own** +prom-client default registry (`register` from `prom-client`): the request +counters are created per-worker in +`src/service/metrics/requestMetricsMiddleware.ts` via +`src/service/tracing/metrics/instruments.ts` (`new Counter(REQUESTS_TOTAL)` etc.), +and `collectDefaultMetrics()` is called per-worker in +`src/service/worker/runtime/builtIn/middlewares.ts`. + +The `/metrics` endpoint is served by `prometheusLoggerMiddleware` +(`src/service/worker/runtime/builtIn/middlewares.ts:24-42`). It answers with +`register.metrics()` — i.e. **only the local registry of whichever worker the OS +round-robin handed the scrape connection to**. Because Node's keep-alive socket +timeout (5s) is shorter than the Prometheus scrape interval, almost every scrape +lands on a *different* worker. + +Consequence: Prometheus sees, under a single `instance` label, an interleaved +braid of 4 independent monotonic counters. Each time a scrape hits a worker whose +local count is lower than the previously scraped worker's, Prometheus treats the +drop as a **counter reset**, so `rate()` / `increase()` over `runtime_*` counters +(notably `runtime_http_requests_total{handler,status_code}`) are inflated by +orders of magnitude (~×40 measured on a live pod). Full analysis: +`metrics-discrepancy-report.html`, sections 3 and 9.1; this task is +recommendation 1 in section 9. + +The pinned `prom-client@^14.2.0` ships `AggregatorRegistry` and a cluster +IPC protocol precisely for this: workers report their registry to the master over +`cluster` messaging, and the master merges them into a single monotonic view. + +### Relevant existing facts (verified in-repo) + +- `prom-client@14.2.0` exports `AggregatorRegistry` with: + - instance method `clusterMetrics(): Promise` (master-side; requests + every connected worker's registry-as-JSON, aggregates, returns exposition + text). + - static `AggregatorRegistry.aggregate(metricsArr)` and `setRegistries(regs)`. + - The constructor calls `addListeners()`, which is **idempotent** and installs: + - in the **master**: a `cluster.on('message')` listener collecting + `prom-client:getMetricsRes` responses. + - in a **worker**: a `process.on('message')` listener that answers + `prom-client:getMetricsReq` with `registries.map(r => r.getMetricsAsJSON())`. + - Default `registries = [globalRegistry]`, which is exactly the `register` our + instruments and `collectDefaultMetrics()` write to — so **no** `setRegistries` + call is needed. +- `clusterMetrics()` is only useful in the **master** (it iterates + `cluster.workers`). But `/metrics` is served **inside a worker** (Koa + middleware). So the answering worker must obtain the aggregate from the master + over IPC. +- Existing IPC message plumbing to mirror: + - Master → worker and worker → master string/tagged messages already exist: + `UP_SIGNAL` (`src/constants.ts`), `statusTrack` / `broadcastStatusTrack` + (`src/service/worker/runtime/statusTrack.ts`), routed by `onMessage` in + `src/service/master.ts` and `onMessage(serviceJSON)` in + `src/service/worker/index.ts`. + - Both `onMessage` handlers currently `logger.warn(...)` on any unrecognized + message. prom-client's own `getMetricsReq`/`getMetricsRes` messages and our + new aggregate messages will also reach these handlers, so both must be taught + to ignore messages they don't own instead of warning. +- Scrapes of `/metrics` are excluded from request counters today because + `prometheusLoggerMiddleware` runs **before** the counting middlewares in the + Koa chain (`src/service/worker/index.ts`) and returns without calling `next()` + for the `/metrics` path. This must be preserved. +- Single-worker mode: `getWorkers()` returns `1` when `LINKED` + (`src/service/loaders.ts`). `serviceJSON.workers` carries the resolved count. + +--- + +## Proposed approach + +**Serve `/metrics` from an aggregate that the master builds from all workers, +requested by the answering worker over the existing cluster IPC.** + +Flow (multi-worker mode, `serviceJSON.workers > 1`): + +1. Master, at `startMaster`, constructs a single `AggregatorRegistry` + (`new AggregatorRegistry()`), enabling both prom-client's master-side + collector listener and our own request handler. +2. Each worker, at `startWorker`, constructs an `AggregatorRegistry` too, so + prom-client's worker-side `getMetricsReq` responder is installed in every + worker. (Its own registry is still the default `register`.) +3. When a worker receives `GET /metrics`, instead of returning + `register.metrics()`, it sends a tagged IPC request to the master + (`process.send({ type: AGG_METRICS_REQ, id })`) and awaits a matching + `AGG_METRICS_RES` (correlation id + a bounded timeout), then serves the + returned exposition string with `Content-Type: register.contentType`. +4. The master, on `AGG_METRICS_REQ`, calls `aggregatorRegistry.clusterMetrics()`. + prom-client fans `getMetricsReq` out to **all** connected workers (including + the one that asked), collects each worker's registry JSON, merges with the + per-metric aggregators, and resolves to a single exposition string. The master + replies to the requesting worker only: + `worker.send({ type: AGG_METRICS_RES, id, body })` (or `{ id, error }`). + +Because the merge is produced from every worker's registry gathered in one pass, +the answering worker is **not** double-counted (it is one of the N sources, not a +source plus a separate local merge). + +Single-worker mode (`serviceJSON.workers === 1`, includes `LINKED`): the +middleware keeps its **current** behavior exactly — `await eventLoopLagMeasurer +.updateInstrumentsAndReset()` then `ctx.body = await register.metrics()`. No IPC, +no aggregation. This keeps LINKED/dev and the `workers:1` production case +unchanged and avoids IPC overhead when there is nothing to aggregate. + +Robustness: if the IPC round-trip errors or times out (prom-client's +`clusterMetrics` has its own 5s timeout; we add a slightly larger guard), the +worker falls back to serving its **local** `register.metrics()` so `/metrics` +never returns empty/500. This is logged, not fatal. + +### New/changed module boundaries + +A small dedicated module (proposed `src/service/metrics/clusterMetricsAggregator.ts`) +owns: +- message-type constants (`AGG_METRICS_REQ`, `AGG_METRICS_RES`) and their type + guards; +- master setup: lazily create the singleton `AggregatorRegistry`, expose + `handleWorkerMetricsRequest(worker, message)` that calls `clusterMetrics()` and + replies; +- worker setup: `ensureWorkerAggregatorRegistry()` (constructs the registry so the + prom-client responder is installed) and `requestAggregatedMetrics(): Promise` + (send + await correlated response, with timeout + fallback), plus + `handleMasterMetricsResponse(message)` to resolve pending promises. + +Keeping this in one module keeps `master.ts`, `worker/index.ts` and +`middlewares.ts` changes minimal and testable in isolation with IPC mocks. + +--- + +## Files / components to change + +| File | Change | +| --- | --- | +| `src/service/metrics/clusterMetricsAggregator.ts` *(new)* | Message constants + guards; master-side `AggregatorRegistry` singleton + request handler; worker-side request/response correlation, timeout, and local fallback. | +| `src/service/master.ts` | In `startMaster`, initialize the aggregator (construct `AggregatorRegistry`). In `onMessage`, route `AGG_METRICS_REQ` to `handleWorkerMetricsRequest`; ignore prom-client's `getMetricsRes`/`getMetricsReq` messages instead of `logger.warn`. | +| `src/service/worker/index.ts` | In `startWorker`, call `ensureWorkerAggregatorRegistry()`. In `onMessage(serviceJSON)`, route `AGG_METRICS_RES` to `handleMasterMetricsResponse`; ignore prom-client's `getMetricsReq` messages instead of `logger.warn`. | +| `src/service/worker/runtime/builtIn/middlewares.ts` | In `prometheusLoggerMiddleware`, when `serviceJSON.workers > 1` request the aggregate via IPC and serve it; when `== 1` keep current local-registry path. Needs access to the worker count (pass `serviceJSON`/`workers` into the middleware factory — it is constructed in `startWorker` which already has `serviceJSON`). Keep the early-return-before-counting behavior and the `COLOSSUS_ROUTE_ID` guard. | +| `src/constants.ts` *(maybe)* | Add message-type string constants if we prefer them centralized next to `UP_SIGNAL`; otherwise they live in the new module. | +| Test files under `__tests__/` *(new)* | See Test plan. | + +No changes to metric names, labels, help text, buckets, or the set of collected +metrics. `runtime_http_requests_total{handler,status_code}` and all other +`runtime_*` / `io_*` series keep identical identity. + +--- + +## How each `AC:` line will be satisfied + +1. **Two consecutive scrapes never show a decreasing `runtime_http_requests_total` + series (monotonic regardless of which worker answers).** + After the change every scrape is served from the *same* aggregate (sum over all + workers) instead of one arbitrary worker's local subset. A Jest test drives the + aggregation with IPC mocks / fake worker registries: it simulates the scrape + being answered by worker A then worker B, asserts both responses are the merged + view, and asserts no series value decreases between the two scrapes. + +2. **`/metrics` equals the sum across workers (a counter incremented once per + worker in N workers reports N).** + `AggregatorRegistry.aggregate` sums counter samples with matching labels across + registries. Test: build N registries, `inc()` the same + `runtime_http_requests_total{handler,status_code}` once in each, run them through + the aggregation, assert the merged output reports exactly `N` for that series. + +3. **`workers === 1` (LINKED) serves the default registry content unchanged.** + The middleware branches on `serviceJSON.workers`: for `1` it executes the + existing code path (`updateInstrumentsAndReset()` + `register.metrics()`) with no + IPC. Test: invoke `prometheusLoggerMiddleware` built with `workers: 1` against a + mock `/metrics` ctx and assert the body equals `register.metrics()` and no IPC + send occurred. + +4. **Default process metrics (`collectDefaultMetrics`) remain present.** + `collectDefaultMetrics()` still runs per worker into the default registry, which + is what each worker reports to the aggregator; the aggregate therefore includes + them (e.g. `process_cpu_seconds_total`, `nodejs_*`). Test asserts a known + default-metric name appears in the aggregated output, and (single-worker path) in + the local output. + +5. **`yarn test` passes.** New tests are added; existing tests remain green + (message-routing guards keep `onMessage` behavior for known messages). + +6. **`yarn lint` passes.** New code follows `tslint-config-vtex` + (alphabetized object literals, import order) as in surrounding files. + +7. **`yarn build` passes.** Types: use `AggregatorRegistry` and message + interfaces; `clusterMetrics()` returns `Promise`; middleware factory + signature updated to receive the worker count. + +--- + +## Risks & alternatives considered + +**Assumptions (no human available to confirm):** +- The default global registry (`register`) is the only registry we need to + aggregate. Verified: instruments and `collectDefaultMetrics` all target the + default registry, so `AggregatorRegistry`'s default `registries=[globalRegistry]` + is correct and `setRegistries` is unnecessary. +- `serviceJSON.workers` is the reliable multi/single switch (it already resolves + `LINKED → 1` and caps at `MAX_WORKERS`). Chosen over reading `LINKED` directly so + a real `workers:1` production config also takes the unchanged path. +- Serving `/metrics` from the worker (proxying to master over IPC) is preferred + over moving the endpoint to the master, because the master is not a Koa HTTP + server and does not listen on `HTTP_SERVER_PORT`; relocating the endpoint would + be a much larger, riskier change. + +**Risks & mitigations:** +- *IPC handler collisions / log noise.* prom-client adds its own + `process.on('message')` / `cluster.on('message')` listeners; these fire + alongside our `onMessage` handlers, which today `warn` on unknown messages. Both + handlers will be updated to silently ignore messages they don't own + (prom-client's `getMetricsReq`/`getMetricsRes` and unmatched aggregate ids). Risk + of behavioral drift on existing known messages is avoided by only adding + branches, not altering existing ones. +- *Latency / timeout.* `clusterMetrics()` has a built-in 5s timeout. A slow/dead + worker could delay a scrape. Mitigation: bounded wait on the worker side plus + fallback to local `register.metrics()` so `/metrics` still returns 200 promptly; + the fallback is logged. Prometheus scrape timeout is typically ≥10s. +- *Event-loop-lag gauges.* `eventLoopLagMeasurer.updateInstrumentsAndReset()` is + called only on the answering worker and lag is a **gauge**; + `AggregatorRegistry` sums gauges across workers by default, which is not a + meaningful aggregation for lag, and non-answering workers won't have refreshed + their lag gauge at scrape time. This is a pre-existing per-worker artifact and is + **out of scope** for this task (which targets `runtime_*` counters). Documented + here so it is a conscious decision; a follow-up could set a custom aggregator + (`max`/`average`) or `omit` for these gauges. The answering worker still calls + `updateInstrumentsAndReset()` in both paths to preserve current behavior. +- *Ordering.* Master must have its `AggregatorRegistry` before workers ask. + Guaranteed: it is created in `startMaster` before any worker can accept traffic. +- *`io_http_requests_current` gauge* is also summed across workers; that is + arguably the desired "current concurrent requests across the replica" and matches + or improves current behavior; noted, not changed. + +**Alternatives considered:** +- *Move `/metrics` to the master process* (separate tiny HTTP listener): + cleanest conceptually but adds a new listener/port and process responsibilities; + rejected as too invasive for this fix. +- *Shared memory / external store for counters*: over-engineered; prom-client + already provides the IPC aggregation protocol. +- *Sticky-session the scrape to one worker*: doesn't fix the underlying + per-worker split (still under-reports 3/4 of traffic). + +--- + +## Test plan + +Framework: existing Jest + ts-jest. New specs under `__tests__/` next to the +touched code (e.g. `src/service/metrics/__tests__/clusterMetricsAggregator.test.ts` +and a middleware test alongside `builtIn/`). + +1. **Monotonic across which-worker-answers (AC 1).** Construct several in-memory + registries with differing `runtime_http_requests_total{handler,status_code}` + values. Exercise the aggregation twice, simulating the scrape being answered by + different workers each time (IPC mocked). Assert every series in scrape #2 is + `>=` the same series in scrape #1 (no decrease). +2. **Sum across workers (AC 2).** N registries, each `inc()` the same series once; + assert merged output reports `N` for that series and that per-worker distinct + label sets are all present and summed correctly. +3. **Single-worker unchanged (AC 3).** Call `prometheusLoggerMiddleware` built with + `workers: 1`; mock a `/metrics` ctx; assert body === `register.metrics()`, + `Content-Type` set to `register.contentType`, and that no `process.send` IPC + occurred. Also assert non-`/metrics` and `COLOSSUS_ROUTE_ID` requests still call + `next()` and are not counted. +4. **Default metrics present (AC 4).** Assert a known `collectDefaultMetrics` + series name (e.g. `process_cpu_seconds_total` / a `nodejs_` metric) appears in + both the aggregated output and the single-worker output. +5. **IPC request/response correlation + timeout fallback.** Mock master/worker + `send`/`on('message')`: assert a worker `AGG_METRICS_REQ` gets a correlated + `AGG_METRICS_RES`, the pending promise resolves with the body, and that on + timeout/error the middleware falls back to local `register.metrics()`. +6. **Message routing guards.** Assert master `onMessage` and worker `onMessage` + route the new messages correctly and no longer `warn` on prom-client's + `getMetricsReq`/`getMetricsRes`, while still handling `UP_SIGNAL` / + `statusTrack` as before. +7. **Gates.** `yarn test`, `yarn lint`, `yarn build` all pass (AC 5–7). diff --git a/src/service/master.ts b/src/service/master.ts index 84f7f6603..2e21c891d 100644 --- a/src/service/master.ts +++ b/src/service/master.ts @@ -3,6 +3,12 @@ import { constants } from 'os' import { INSPECT_DEBUGGER_PORT, LINKED, UP_SIGNAL } from '../constants' import { isLog, logOnceToDevConsole } from './logger' +import { + handleWorkerMetricsRequest, + initMasterAggregatorRegistry, + isAggMetricsRequest, + isPromClientMessage, +} from './metrics/clusterMetricsAggregator' import { logger } from './worker/listeners' import { broadcastStatusTrack, isStatusTrackBroadcast, trackStatus } from './worker/runtime/statusTrack' import { ServiceJSON } from './worker/runtime/typings' @@ -15,6 +21,10 @@ const onMessage = (worker: Worker, message: any) => { } else if (isStatusTrackBroadcast(message)) { trackStatus() broadcastStatusTrack() + } else if (isAggMetricsRequest(message)) { + handleWorkerMetricsRequest(worker, message) + } else if (isPromClientMessage(message)) { + // Handled by prom-client's own cluster listener; ignore here. } else { logger.warn({ content: message, @@ -77,6 +87,12 @@ export const startMaster = (service: ServiceJSON) => { process.env.DETERMINISTIC_VARY = 'true' } + // Set up the master-side Prometheus aggregator so workers can request a + // merged, monotonic /metrics view across the whole cluster over IPC. + if (numWorkers > 1) { + initMasterAggregatorRegistry() + } + // Setup dubugger if (LINKED) { cluster.setupMaster({ inspectPort: INSPECT_DEBUGGER_PORT }) diff --git a/src/service/metrics/__tests__/clusterMetricsAggregator.test.ts b/src/service/metrics/__tests__/clusterMetricsAggregator.test.ts new file mode 100644 index 000000000..4cf1cf792 --- /dev/null +++ b/src/service/metrics/__tests__/clusterMetricsAggregator.test.ts @@ -0,0 +1,199 @@ +import { AggregatorRegistry, Counter, Registry } from 'prom-client' + +import { + __resetForTests, + AGG_METRICS_REQ, + AGG_METRICS_RES, + AggMetricsResMessage, + handleMasterMetricsResponse, + handleWorkerMetricsRequest, + isAggMetricsRequest, + isAggMetricsResponse, + isPromClientMessage, + requestAggregatedMetrics, +} from '../clusterMetricsAggregator' + +const REQUESTS_TOTAL = { + help: 'The total number of HTTP requests.', + labelNames: ['status_code', 'handler'], + name: 'runtime_http_requests_total', +} + +// Build an isolated registry that mimics one worker's default registry with a +// runtime_http_requests_total counter incremented `count` times. +const buildWorkerRegistry = (count: number, labels = { handler: 'render', status_code: '200' }) => { + const registry = new Registry() + const counter = new Counter({ ...REQUESTS_TOTAL, registers: [registry] }) + for (let i = 0; i < count; i++) { + counter.inc(labels) + } + return registry +} + +// Parse `runtime_http_requests_total{...} ` samples out of exposition text. +const parseSeries = (text: string, metric: string): Record => { + const out: Record = {} + text + .split('\n') + .filter((line) => line.startsWith(metric) && !line.startsWith('# ')) + .forEach((line) => { + const match = line.match(/^(\S+)\s+([\d.eE+-]+)$/) + if (match) { + out[match[1]] = Number(match[2]) + } + }) + return out +} + +const aggregateRegistries = async (registries: Array): Promise => { + const jsons = await Promise.all(registries.map((r) => r.getMetricsAsJSON())) + const merged = AggregatorRegistry.aggregate(jsons) + return merged.metrics() +} + +describe('clusterMetricsAggregator', () => { + afterEach(() => { + __resetForTests() + jest.useRealTimers() + delete (process as any).send + }) + + describe('message guards', () => { + it('recognizes aggregate request/response and prom-client messages', () => { + expect(isAggMetricsRequest({ id: 1, type: AGG_METRICS_REQ })).toBe(true) + expect(isAggMetricsRequest({ type: AGG_METRICS_REQ })).toBe(false) + expect(isAggMetricsResponse({ body: 'x', id: 1, type: AGG_METRICS_RES })).toBe(true) + expect(isAggMetricsResponse('UP')).toBe(false) + expect(isPromClientMessage({ type: 'prom-client:getMetricsReq' })).toBe(true) + expect(isPromClientMessage({ type: 'prom-client:getMetricsRes' })).toBe(true) + expect(isPromClientMessage('UP')).toBe(false) + expect(isPromClientMessage({ statusTrack: true })).toBe(false) + }) + }) + + describe('aggregation across workers', () => { + // AC2: a counter incremented once per worker in N workers reports N. + it('sums a series across N workers', async () => { + const N = 4 + const registries = Array.from({ length: N }, () => buildWorkerRegistry(1)) + const output = await aggregateRegistries(registries) + const series = parseSeries(output, 'runtime_http_requests_total') + const key = Object.keys(series).find((k) => k.includes('runtime_http_requests_total'))! + expect(series[key]).toBe(N) + }) + + it('sums differing per-worker counts', async () => { + const registries = [buildWorkerRegistry(5), buildWorkerRegistry(10), buildWorkerRegistry(2), buildWorkerRegistry(8)] + const output = await aggregateRegistries(registries) + const series = parseSeries(output, 'runtime_http_requests_total') + const key = Object.keys(series).find((k) => k.includes('runtime_http_requests_total'))! + expect(series[key]).toBe(25) + }) + + it('preserves and sums distinct label sets from different workers', async () => { + const registries = [ + buildWorkerRegistry(3, { handler: 'render', status_code: '200' }), + buildWorkerRegistry(4, { handler: 'render', status_code: '500' }), + buildWorkerRegistry(1, { handler: 'render', status_code: '200' }), + ] + const output = await aggregateRegistries(registries) + const series = parseSeries(output, 'runtime_http_requests_total') + const ok = Object.keys(series).find((k) => k.includes('status_code="200"'))! + const err = Object.keys(series).find((k) => k.includes('status_code="500"'))! + expect(series[ok]).toBe(4) + expect(series[err]).toBe(4) + }) + + // AC1: two consecutive scrapes never show a decreasing value, regardless of + // which worker answers (the merge is the same aggregate either way). + it('never decreases across two consecutive scrapes', async () => { + const workerCounts = [5, 10, 2, 8] + const registriesScrape1 = workerCounts.map((c) => buildWorkerRegistry(c)) + const scrape1 = parseSeries(await aggregateRegistries(registriesScrape1), 'runtime_http_requests_total') + + // Between scrapes each worker serves more traffic; counters only grow. + const registriesScrape2 = workerCounts.map((c, i) => buildWorkerRegistry(c + i + 1)) + const scrape2 = parseSeries(await aggregateRegistries(registriesScrape2), 'runtime_http_requests_total') + + Object.keys(scrape1).forEach((key) => { + expect(scrape2[key]).toBeGreaterThanOrEqual(scrape1[key]) + }) + }) + + // AC4: default process metrics remain present in the aggregated output. + it('includes default process metrics collected per worker', async () => { + const registryWithDefaults = new Registry() + // Emulate a default metric present in a worker registry. + const cpu = new Counter({ + help: 'Total user and system CPU time spent in seconds.', + name: 'process_cpu_seconds_total', + registers: [registryWithDefaults], + }) + cpu.inc(1) + const output = await aggregateRegistries([registryWithDefaults, buildWorkerRegistry(1)]) + expect(output).toContain('process_cpu_seconds_total') + }) + }) + + describe('worker IPC request/response', () => { + it('sends AGG_METRICS_REQ and resolves with the master body', async () => { + const sent: any[] = [] + ;(process as any).send = (msg: any) => sent.push(msg) + + const promise = requestAggregatedMetrics() + + expect(sent).toHaveLength(1) + expect(sent[0].type).toBe(AGG_METRICS_REQ) + const { id } = sent[0] + + handleMasterMetricsResponse({ body: 'AGGREGATED_BODY', id, type: AGG_METRICS_RES }) + await expect(promise).resolves.toBe('AGGREGATED_BODY') + }) + + it('falls back to the local registry on timeout', async () => { + jest.useFakeTimers() + ;(process as any).send = () => undefined + + const promise = requestAggregatedMetrics() + jest.advanceTimersByTime(6000) + const body = await promise + // Local register.metrics() output; just assert it is a (possibly empty) string. + expect(typeof body).toBe('string') + }) + + it('falls back to the local registry when the master replies with an error', async () => { + const sent: any[] = [] + ;(process as any).send = (msg: any) => sent.push(msg) + + const promise = requestAggregatedMetrics() + const { id } = sent[0] + const errorMsg: AggMetricsResMessage = { error: 'boom', id, type: AGG_METRICS_RES } + handleMasterMetricsResponse(errorMsg) + const body = await promise + expect(typeof body).toBe('string') + }) + + it('ignores responses with unknown correlation ids', () => { + expect(() => handleMasterMetricsResponse({ body: 'x', id: 9999, type: AGG_METRICS_RES })).not.toThrow() + }) + + it('serves local metrics when process.send is unavailable', async () => { + delete (process as any).send + const body = await requestAggregatedMetrics() + expect(typeof body).toBe('string') + }) + }) + + describe('master request handler', () => { + it('replies to the requesting worker with a body', async () => { + const worker: any = { send: jest.fn() } + await handleWorkerMetricsRequest(worker, { id: 7, type: AGG_METRICS_REQ }) + expect(worker.send).toHaveBeenCalledTimes(1) + const reply = worker.send.mock.calls[0][0] + expect(reply.type).toBe(AGG_METRICS_RES) + expect(reply.id).toBe(7) + // No cluster workers in the test process → prom-client aggregates to ''. + expect(typeof reply.body).toBe('string') + }) + }) +}) diff --git a/src/service/metrics/clusterMetricsAggregator.ts b/src/service/metrics/clusterMetricsAggregator.ts new file mode 100644 index 000000000..f0ea0528b --- /dev/null +++ b/src/service/metrics/clusterMetricsAggregator.ts @@ -0,0 +1,194 @@ +import { Worker } from 'cluster' +import { AggregatorRegistry, register } from 'prom-client' + +import { logger } from '../worker/listeners' + +/** + * Cluster-wide Prometheus metrics aggregation. + * + * In multi-worker mode each cluster worker keeps its own prom-client default + * registry. Serving `/metrics` from a single (round-robin selected) worker + * therefore exposes only that worker's local counters, which makes Prometheus + * treat the per-scrape braid of independent counters as counter resets and + * inflates `rate()`/`increase()` by orders of magnitude. + * + * This module uses prom-client's `AggregatorRegistry` cluster IPC protocol so + * that the worker answering a scrape asks the master for a merged, monotonic + * view built from every worker's registry in one pass (no double-counting of + * the answering worker). + */ + +/** Tagged IPC message: worker -> master, "please build the cluster aggregate". */ +export const AGG_METRICS_REQ = 'vtex-api:aggMetricsReq' +/** Tagged IPC message: master -> worker, carries the aggregate (or an error). */ +export const AGG_METRICS_RES = 'vtex-api:aggMetricsRes' + +/** + * Upper bound for a worker waiting on the master aggregate. prom-client's own + * `clusterMetrics()` has a 5s timeout collecting worker registries; we guard + * slightly above that and fall back to the local registry on expiry so that + * `/metrics` never hangs or returns empty. + */ +const AGG_METRICS_TIMEOUT_MS = 6000 + +export interface AggMetricsReqMessage { + type: typeof AGG_METRICS_REQ + id: number +} + +export interface AggMetricsResMessage { + type: typeof AGG_METRICS_RES + id: number + body?: string + error?: string +} + +export const isAggMetricsRequest = (message: any): message is AggMetricsReqMessage => + message != null && message.type === AGG_METRICS_REQ && typeof message.id === 'number' + +export const isAggMetricsResponse = (message: any): message is AggMetricsResMessage => + message != null && message.type === AGG_METRICS_RES && typeof message.id === 'number' + +/** + * prom-client's cluster protocol emits its own tagged IPC messages + * (`prom-client:getMetricsReq` / `prom-client:getMetricsRes`). They are handled + * by prom-client's own cluster listeners, so our `onMessage` handlers must + * ignore them instead of warning about "unknown" messages. + */ +export const isPromClientMessage = (message: any): boolean => + message != null && typeof message.type === 'string' && message.type.startsWith('prom-client:') + +// --------------------------------------------------------------------------- +// Master side +// --------------------------------------------------------------------------- + +let masterAggregatorRegistry: AggregatorRegistry | undefined + +/** + * Constructs the master-side `AggregatorRegistry` (idempotent). Its constructor + * installs prom-client's master `cluster.on('message')` collector, which is + * what makes `clusterMetrics()` work. Must be called in the master before any + * worker can answer a scrape. + */ +export const initMasterAggregatorRegistry = (): AggregatorRegistry => { + if (!masterAggregatorRegistry) { + masterAggregatorRegistry = new AggregatorRegistry() + } + return masterAggregatorRegistry +} + +/** + * Handles an `AGG_METRICS_REQ` from a worker: builds the cluster aggregate and + * replies to that worker only. prom-client fans the request out to *all* + * connected workers (including the requester), so the requester is one of the N + * merged sources and is never double-counted. + */ +export const handleWorkerMetricsRequest = async (worker: Worker, message: AggMetricsReqMessage): Promise => { + const aggregator = initMasterAggregatorRegistry() + try { + const body = await aggregator.clusterMetrics() + worker.send({ type: AGG_METRICS_RES, id: message.id, body }) + } catch (err) { + worker.send({ type: AGG_METRICS_RES, id: message.id, error: (err as Error)?.message ?? String(err) }) + } +} + +// --------------------------------------------------------------------------- +// Worker side +// --------------------------------------------------------------------------- + +let workerAggregatorRegistry: AggregatorRegistry | undefined + +interface PendingRequest { + resolve: (body: string) => void + timer: NodeJS.Timeout +} + +const pendingRequests = new Map() +let requestCounter = 0 + +/** + * Constructs the worker-side `AggregatorRegistry` (idempotent). Its constructor + * installs prom-client's worker `process.on('message')` responder that answers + * the master's `getMetricsReq` with this worker's registry JSON. + */ +export const ensureWorkerAggregatorRegistry = (): AggregatorRegistry => { + if (!workerAggregatorRegistry) { + workerAggregatorRegistry = new AggregatorRegistry() + } + return workerAggregatorRegistry +} + +/** + * Requests the cluster aggregate from the master over IPC and resolves with the + * merged exposition string. On timeout or any error it falls back to this + * worker's local `register.metrics()` so `/metrics` always answers promptly. + */ +export const requestAggregatedMetrics = async (): Promise => { + if (typeof process.send !== 'function') { + return register.metrics() + } + + const id = requestCounter++ + + return new Promise((resolve) => { + const timer = setTimeout(() => { + pendingRequests.delete(id) + logger.warn({ + message: 'Timed out waiting for aggregated cluster metrics; falling back to local registry', + pid: process.pid, + }) + register.metrics().then(resolve).catch(() => resolve('')) + }, AGG_METRICS_TIMEOUT_MS) + + pendingRequests.set(id, { resolve, timer }) + + try { + process.send!({ type: AGG_METRICS_REQ, id }) + } catch (err) { + clearTimeout(timer) + pendingRequests.delete(id) + logger.warn({ + content: (err as Error)?.message ?? String(err), + message: 'Failed to request aggregated cluster metrics; falling back to local registry', + pid: process.pid, + }) + register.metrics().then(resolve).catch(() => resolve('')) + } + }) +} + +/** + * Resolves the pending `requestAggregatedMetrics` promise correlated by id when + * the master replies. On error it falls back to the local registry. + */ +export const handleMasterMetricsResponse = (message: AggMetricsResMessage): void => { + const pending = pendingRequests.get(message.id) + if (!pending) { + return + } + + pendingRequests.delete(message.id) + clearTimeout(pending.timer) + + if (message.error != null || message.body == null) { + logger.warn({ + content: message.error, + message: 'Master failed to aggregate cluster metrics; falling back to local registry', + pid: process.pid, + }) + register.metrics().then(pending.resolve).catch(() => pending.resolve('')) + return + } + + pending.resolve(message.body) +} + +/** Test-only helper to reset module state between test cases. */ +export const __resetForTests = () => { + masterAggregatorRegistry = undefined + workerAggregatorRegistry = undefined + pendingRequests.forEach((p) => clearTimeout(p.timer)) + pendingRequests.clear() + requestCounter = 0 +} diff --git a/src/service/worker/index.ts b/src/service/worker/index.ts index 7fd861dbf..1de57eff9 100644 --- a/src/service/worker/index.ts +++ b/src/service/worker/index.ts @@ -13,6 +13,12 @@ import { MetricsAccumulator } from '../../metrics/MetricsAccumulator' import { getService } from '../loaders' import { logOnceToDevConsole } from '../logger/console' import { LogLevel } from '../logger/loggerTypes' +import { + ensureWorkerAggregatorRegistry, + handleMasterMetricsResponse, + isAggMetricsResponse, + isPromClientMessage, +} from '../metrics/clusterMetricsAggregator' import { addOtelRequestMetricsMiddleware } from '../metrics/otelRequestMetricsMiddleware' import { addRequestMetricsMiddleware } from '../metrics/requestMetricsMiddleware' import { TracerSingleton } from '../tracing/TracerSingleton' @@ -78,6 +84,10 @@ const onMessage = (service: ServiceJSON) => (message: any) => { logAvailableRoutes(service) } else if (isStatusTrack(message)) { trackStatus() + } else if (isAggMetricsResponse(message)) { + handleMasterMetricsResponse(message) + } else if (isPromClientMessage(message)) { + // Handled by prom-client's own worker listener; ignore here. } else { logger.warn({ content: message, @@ -218,12 +228,19 @@ export const startWorker = (serviceJSON: ServiceJSON) => { addProcessListeners() const tracer = TracerSingleton.getTracer() + + // In multi-worker mode install prom-client's worker-side cluster responder so + // the master can collect this worker's registry for the aggregated /metrics. + if (serviceJSON.workers > 1) { + ensureWorkerAggregatorRegistry() + } + const app = new Koa() app.proxy = true app .use(error) .use(Instrumentation.Middlewares.ContextMiddlewares.Koa.ContextPropagationMiddleware()) - .use(prometheusLoggerMiddleware()) + .use(prometheusLoggerMiddleware(serviceJSON.workers)) .use(addTracingMiddleware(tracer)) .use(addRequestMetricsMiddleware()) .use(addOtelRequestMetricsMiddleware()) diff --git a/src/service/worker/runtime/builtIn/__tests__/middlewares.test.ts b/src/service/worker/runtime/builtIn/__tests__/middlewares.test.ts new file mode 100644 index 000000000..f8208a37e --- /dev/null +++ b/src/service/worker/runtime/builtIn/__tests__/middlewares.test.ts @@ -0,0 +1,98 @@ +import { register } from 'prom-client' + +import * as clusterMetricsAggregator from '../../../../metrics/clusterMetricsAggregator' +import { prometheusLoggerMiddleware } from '../middlewares' + +const buildCtx = (path: string, headers: Record = {}) => { + const setHeaders: Record = {} + return { + body: undefined as any, + get: (key: string) => headers[key] ?? '', + request: { path }, + set: (key: string, value: string) => { + setHeaders[key] = value + }, + setHeaders, + status: undefined as any, + } +} + +describe('prometheusLoggerMiddleware', () => { + beforeEach(() => { + // collectDefaultMetrics() and the lag measurer register into the default + // registry on every middleware construction; clear it between cases. + register.clear() + }) + + afterEach(() => { + jest.restoreAllMocks() + }) + + // AC3: workers=1 (LINKED) serves the default registry content unchanged. + it('serves the local default registry in single-worker mode without IPC', async () => { + const sendSpy = jest.fn() + ;(process as any).send = sendSpy + const aggSpy = jest.spyOn(clusterMetricsAggregator, 'requestAggregatedMetrics') + const localMetricsSpy = jest.spyOn(register, 'metrics').mockResolvedValue('LOCAL_REGISTRY') + + const middleware = prometheusLoggerMiddleware(1) + const ctx: any = buildCtx('/metrics') + const next = jest.fn() + + await middleware(ctx, next) + + expect(localMetricsSpy).toHaveBeenCalledTimes(1) + expect(ctx.body).toBe('LOCAL_REGISTRY') + expect(ctx.status).toBe(200) + expect(ctx.setHeaders['Content-Type']).toBe(register.contentType) + expect(aggSpy).not.toHaveBeenCalled() + expect(sendSpy).not.toHaveBeenCalled() + expect(next).not.toHaveBeenCalled() + + delete (process as any).send + }) + + // AC4: default process metrics are present in the single-worker output. + it('includes default process metrics in single-worker output', async () => { + const middleware = prometheusLoggerMiddleware(1) + const ctx: any = buildCtx('/metrics') + await middleware(ctx, () => Promise.resolve()) + expect(ctx.body).toContain('process_cpu_seconds_total') + }) + + it('requests the cluster aggregate in multi-worker mode', async () => { + const aggSpy = jest + .spyOn(clusterMetricsAggregator, 'requestAggregatedMetrics') + .mockResolvedValue('AGGREGATED') + + const middleware = prometheusLoggerMiddleware(4) + const ctx: any = buildCtx('/metrics') + + await middleware(ctx, () => Promise.resolve()) + + expect(aggSpy).toHaveBeenCalledTimes(1) + expect(ctx.body).toBe('AGGREGATED') + expect(ctx.status).toBe(200) + expect(ctx.setHeaders['Content-Type']).toBe(register.contentType) + }) + + it('does not count /metrics scrapes and passes through other routes', async () => { + const aggSpy = jest.spyOn(clusterMetricsAggregator, 'requestAggregatedMetrics') + const middleware = prometheusLoggerMiddleware(4) + + const nonMetricsCtx: any = buildCtx('/some-route') + const next1 = jest.fn().mockResolvedValue(undefined) + await middleware(nonMetricsCtx, next1) + expect(next1).toHaveBeenCalledTimes(1) + expect(nonMetricsCtx.body).toBeUndefined() + + // Requests carrying a colossus route id must pass through, not be answered. + const routedCtx: any = buildCtx('/metrics', { 'x-colossus-route-id': 'my-route' }) + const next2 = jest.fn().mockResolvedValue(undefined) + await middleware(routedCtx, next2) + expect(next2).toHaveBeenCalledTimes(1) + expect(routedCtx.body).toBeUndefined() + + expect(aggSpy).not.toHaveBeenCalled() + }) +}) diff --git a/src/service/worker/runtime/builtIn/middlewares.ts b/src/service/worker/runtime/builtIn/middlewares.ts index 6c0561a54..be530cca4 100644 --- a/src/service/worker/runtime/builtIn/middlewares.ts +++ b/src/service/worker/runtime/builtIn/middlewares.ts @@ -1,6 +1,7 @@ import { collectDefaultMetrics, register } from 'prom-client' import { HeaderKeys } from '../../../../constants' import { MetricsLogger } from '../../../logger/metricsLogger' +import { requestAggregatedMetrics } from '../../../metrics/clusterMetricsAggregator' import { EventLoopLagMeasurer } from '../../../tracing/metrics/measurers/EventLoopLagMeasurer' import { ServiceContext } from '../typings' import { Recorder } from '../utils/recorder' @@ -21,11 +22,16 @@ export const addMetricsLoggerMiddleware = () => { } } -export const prometheusLoggerMiddleware = () => { +export const prometheusLoggerMiddleware = (workers = 1) => { collectDefaultMetrics() const eventLoopLagMeasurer = new EventLoopLagMeasurer() eventLoopLagMeasurer.start() + // In multi-worker mode each worker holds only its own registry, so /metrics + // must serve the cluster-wide aggregate requested from the master over IPC. + // In single-worker mode (LINKED / workers:1) the local registry is complete. + const isMultiWorker = workers > 1 + return async (ctx: ServiceContext, next: () => Promise) => { if (ctx.request.path !== '/metrics') { return next() @@ -38,7 +44,7 @@ export const prometheusLoggerMiddleware = () => { await eventLoopLagMeasurer.updateInstrumentsAndReset() ctx.set('Content-Type', register.contentType) - ctx.body = await register.metrics() + ctx.body = isMultiWorker ? await requestAggregatedMetrics() : await register.metrics() ctx.status = 200 } }