diff --git a/hub/.env.example b/hub/.env.example index ed2821cd..98392530 100644 --- a/hub/.env.example +++ b/hub/.env.example @@ -4,6 +4,10 @@ PORT=3001 # API logging threshold: trace, debug, info, warn, error, fatal, or silent. LOG_LEVEL=info +# Explicit spans only. See docs/telemetry.md before enabling export. +OTEL_TRACES_EXPORTER=none +# OTEL_EXPORTER_OTLP_TRACES_ENDPOINT=http://127.0.0.1:4318/v1/traces + # Independent worker concurrency. Both default to one (supported range: 1-100). ON_DEMAND_JOB_CONCURRENCY=1 SCHEDULED_JOB_CONCURRENCY=1 diff --git a/hub/AGENTS.md b/hub/AGENTS.md index afced341..95893269 100644 --- a/hub/AGENTS.md +++ b/hub/AGENTS.md @@ -6,6 +6,7 @@ Hub is being rebuilt in TypeScript. The app starts locally, but its product feat - [Code architecture conventions](docs/backend-architecture.md) define repositories, integrations, utilities, domains, and allowed imports. - [Coding conventions](docs/coding-conventions.md) explain frontend layers, formatting, and interface rules. - [Database conventions](docs/database.md) define Prisma files, model section order, UUID keys, timestamps, and database naming. +- [Telemetry](docs/telemetry.md) defines explicit spans, safe attributes, operator configuration, and API and worker lifetime. - [Development](docs/development.md) and [testing](docs/testing-and-operations.md) own local and CI commands. - [Storybook](docs/storybook.md#ui-changes-and-pull-request-evidence) owns presentation coverage, cleanup, and PR screenshots. diff --git a/hub/docs/README.md b/hub/docs/README.md index 79034d43..5ef13e81 100644 --- a/hub/docs/README.md +++ b/hub/docs/README.md @@ -8,6 +8,7 @@ Hub is being rebuilt. The [Hub overview](../README.md) describes what the baseli - [Development](development.md): local API and frontend setup. - [Database](database.md): Prisma schema conventions, migrations, and the shared client. - [Logging](logging.md): application events, automatic request records, and payload safety. +- [Telemetry](telemetry.md): explicit OpenTelemetry spans, safe attributes, and process lifetime. - [Background jobs](background-jobs.md): PostgreSQL queues, workers, schedules, setup, and test isolation. - [WebSocket updates](websockets.md): local broadcasts, subscriptions, access hooks, and connection limits. - [Testing and operations](testing-and-operations.md): baseline checks and their limits. diff --git a/hub/docs/architecture.md b/hub/docs/architecture.md index d5bf7066..65dad84c 100644 --- a/hub/docs/architecture.md +++ b/hub/docs/architecture.md @@ -22,6 +22,8 @@ The [Pino integration](logging.md) writes structured application and HTTP logs t The [background-job foundation](background-jobs.md) uses BullMQ with PostgreSQL and separate on-demand and scheduled worker processes. Setup migrates BullMQ's own schema independently of Prisma. Product processors and schedules are not registered yet. +The [telemetry integration](telemetry.md) provides explicit server-side OpenTelemetry spans with optional operator-configured OTLP export. API and worker entrypoints own startup and flush telemetry after draining active work. Automatic instrumentation and product analytics are not implemented. + The [WebSocket foundation](websockets.md) attaches to the API's HTTP server and broadcasts to local channel subscribers without persistence. The application rejects upgrades until access checks are supplied. Product channels, client recovery, and communication between processes are not implemented. ## Frontend diff --git a/hub/docs/development.md b/hub/docs/development.md index 3d344861..2e6efa8c 100644 --- a/hub/docs/development.md +++ b/hub/docs/development.md @@ -42,6 +42,8 @@ Vite also forwards WebSocket upgrades under `/api`. The `/api/realtime` endpoint The API reads an optional `PORT` environment variable, defaulting to 3001. If you change it, update the local Vite proxy target to match. The editor forwards ports 3001, 5173, and 6006 from the devcontainer. API startup reads `.env` when present; Compose's environment takes precedence. +On `SIGINT` or `SIGTERM`, the API stops accepting requests and allows active requests up to 30 seconds to finish, then flushes [optional telemetry](telemetry.md) within five seconds. No automatic request tracing is installed. + ## Background workers The API and frontend do not start background consumers. After starting PostgreSQL, run `npm run setup` once, then start `npm run worker:on-demand` and `npm run worker:scheduled` in separate terminals. Both use `DATABASE_URL`; their concurrency settings are in `.env.example`. diff --git a/hub/docs/telemetry.md b/hub/docs/telemetry.md new file mode 100644 index 00000000..4a180646 --- /dev/null +++ b/hub/docs/telemetry.md @@ -0,0 +1,99 @@ +# Telemetry + +Hub operators can send explicitly recorded server-side traces to an OpenTelemetry Protocol (OTLP) recipient of their choice. Tracing is disabled by default. Configuration belongs to the operator's process environment, not to an API endpoint or browser setting. + +The foundation provides two functions under `server/integrations/opentelemetry/`: `startTelemetry()` and `withTelemetrySpan()`. It does not automatically trace requests, database queries, outbound HTTP, or jobs. The health endpoint does not produce spans. Browser tracking and durable product analytics are not implemented. + +## Configure export + +Set these variables in the API and worker environments, or in a private local `.env`: + +```sh +OTEL_TRACES_EXPORTER=otlp +OTEL_SERVICE_NAME=replica-hub +OTEL_EXPORTER_OTLP_TRACES_ENDPOINT=http://127.0.0.1:4318/v1/traces +OTEL_EXPORTER_OTLP_TRACES_PROTOCOL=http/protobuf +OTEL_TRACES_SAMPLER=parentbased_traceidratio +OTEL_TRACES_SAMPLER_ARG=1 +``` + +The recipient can be a provider that accepts OTLP over HTTP/protobuf or an operator-managed OpenTelemetry Collector that forwards to other platforms. Use HTTPS for remote recipients. This integration does not include provider-specific SDKs, configure a Collector, or prove that a provider displays accepted traces. + +| Variable | Contract | +| --- | --- | +| `OTEL_TRACES_EXPORTER` | `none` is the default; `otlp` enables export. Other values fail startup. | +| `OTEL_SDK_DISABLED` | `true`, ignoring case, disables telemetry regardless of other settings. With export enabled, other nonempty values must be `false`. | +| `OTEL_SERVICE_NAME` | Defaults to `replica-hub`. Use a static service label of at most 100 printable characters, not secrets or personal data. | +| `OTEL_EXPORTER_OTLP_TRACES_ENDPOINT` | Explicit absolute HTTP(S) URL with a request path. Credentials, query strings, and fragments are rejected. | +| `OTEL_EXPORTER_OTLP_TRACES_PROTOCOL` | Only `http/protobuf` is supported. | +| `OTEL_EXPORTER_OTLP_TRACES_HEADERS` | Optional comma-separated URL-encoded `key=value` authentication headers. For example, `Authorization=Bearer%20example`. Keep real values private. | +| `OTEL_TRACES_SAMPLER` | `parentbased_traceidratio` is the default. `always_on` and `always_off` are also supported. Other samplers fail startup. | +| `OTEL_TRACES_SAMPLER_ARG` | A finite fraction from `0` through `1`, defaulting to `1`, controls the root sampling ratio. It is validated but has no effect for `always_on` and `always_off`. | + +Common `OTEL_EXPORTER_OTLP_ENDPOINT`, `OTEL_EXPORTER_OTLP_PROTOCOL`, and `OTEL_EXPORTER_OTLP_HEADERS` settings are also accepted. Trace-specific settings take precedence, even when empty. The common endpoint is a base URL; Hub appends `/v1/traces` to its path. Enabled export requires an explicit endpoint. Disabled telemetry does not validate unused export settings or create an exporter. + +Hub sets its own resource, sampler, batching, and context policies rather than loading arbitrary SDK environment configuration. `OTEL_RESOURCE_ATTRIBUTES`, other sampler names, automatic instrumentation, and custom propagation are not part of this contract. Do not infer support for every standard OpenTelemetry variable. + +## Process lifetime + +`startTelemetry(process.env, logger)` validates enabled configuration and registers one tracer provider and asynchronous context manager per process. The logger is supplied by composition code; this integration does not import the Pino integration. A second enabled initialization before shutdown is rejected. + +The API and worker entrypoints already call it before accepting requests or starting consumers. Static imports are sufficient because this foundation does not patch dependencies. `createApp()` does not initialize global telemetry or install signal handlers. + +The returned `{ shutdown() }` belongs to the entrypoint, not to domains or request handlers. The entrypoint waits for API requests and WebSockets, or active worker jobs, to drain, then awaits shutdown in `finally`. API and worker drain limits are 30 seconds; telemetry shutdown adds at most five seconds. Operators must allow both windows before terminating a process. Startup failures and shutdown timeouts exit unsuccessfully. + +Shutdown flushes queued spans and closes the SDK. It is idempotent, including when telemetry is disabled. Normal batching holds at most 256 spans, exports at most 64 per batch, and attempts export at one-second intervals with a two-second export timeout. Queue saturation or failed delivery can lose spans. Export failures produce safe static logs, not errors in the feature being traced. Shutdown failure is reported separately and rejects the shutdown call. + +## Trace a ported feature + +Call `withTelemetrySpan()` around one meaningful operation in a route or domain. From a function directory under `server/domains/build/`, wrapping an existing `importBuild()` operation would look like: + +```ts +import { withTelemetrySpan } from "../../../integrations/opentelemetry/with-telemetry-span/index.ts" + +return withTelemetrySpan( + "build.import", + { "replica.build.id": buildId }, + () => importBuild(buildId), +) +``` + +The callback runs once. The helper awaits it, returns its original result, and finishes the span on success or failure. A failure marks the span as failed and rethrows the exact original value. It does not attach exception messages, stacks, or thrown objects. A failed operation is not converted into a successful result. + +Nested calls become parent and child spans across `await`; concurrently running operations keep their contexts separate. Disabled or unsampled telemetry still runs the callback normally. Completed spans receive success or failure status without additional diagnostic payloads. + +Use static operation names such as `build.import`, not record IDs, branch names, URLs, or user input. Names contain at most 100 lowercase ASCII letters, digits, and separating `.`, `_`, or `-` characters, beginning with a letter. Format validation cannot determine whether a name is genuinely static; reviewers must check the call site. + +Attributes accept only these optional UUID strings: + +- `replica.repository.id` +- `replica.prototype.id` +- `replica.build.id` +- `replica.thread.id` +- `replica.comment.id` + +The integration validates names and attributes before invoking the callback, even when tracing is disabled. Unknown keys and malformed IDs reject with a safe error instead of silently recording unexpected data. An invalid instrumentation call is a programming error, not an exporter outage. To add an attribute, update the integration's type, runtime schema, this guide, and privacy coverage together. + +Do not pass request or response bodies, headers, cookies, credentials, SQL, job arguments, review text, profile fields, or raw exceptions. IDs are pseudonymous, not anonymous. Exported data needs operator-controlled access and retention. Pino redaction does not sanitize traces. Use this wrapper rather than directly modifying SDK spans or importing the OpenTelemetry API in feature code. + +## Architecture and future ports + +Keep configuration and telemetry-specific helpers inside `integrations/opentelemetry/`. Use one function directory with colocated types and tests; do not add a barrel export, telemetry domain, or telemetry utilities outside the integration. + +Follow the [backend import boundaries](backend-architecture.md#import-boundaries). Domains and routes may call the integration. Repositories and other integrations must receive a tracing callback from composition code rather than import OpenTelemetry integration functions or types. Business decisions about what to trace stay with the feature. + +There is no request-header extraction, outbound context injection, baggage handling, or job-context transport yet. Each independently started operation is a root unless it already has an active in-process parent. Adding cross-process propagation requires a real feature boundary and an explicit policy for trusted incoming context; never assume client-supplied trace headers are trusted. Automatic instrumentation may need earlier initialization or ESM loader support and must be verified for the selected libraries before changing startup. + +Sampled, best-effort traces explain execution. They are not durable usage counts, security audits, or delivery guarantees. Port product analytics separately when its domains, access checks, transactional events, and delivery requirements exist. + +## Verify a port + +Add a colocated test that proves the operation's span, safe attributes, result, and failure behavior. Keep the actual SDK and asynchronous context handling real; replace only export transport when network delivery is not under test. Release the runtime and any test-owned global providers or context managers after each case. + +From `hub/`, run the foundation checks with: + +```sh +npm test -- server/integrations/opentelemetry server/run-api +``` + +For an actual recipient check, configure a local OTLP receiver and invoke an instrumented operation. Visiting the health route alone cannot verify export. A receiver accepting a batch does not establish that an analytics or APM report contains it. diff --git a/hub/docs/websockets.md b/hub/docs/websockets.md index 5b2530ae..3fcdafa8 100644 --- a/hub/docs/websockets.md +++ b/hub/docs/websockets.md @@ -4,6 +4,8 @@ Hub can send live updates to the client app and Copilot plugin through the `ws` The application attaches `/api/realtime` to the same HTTP server as Express and passes its broadcast function to `createApp()`. Vite forwards WebSocket upgrades under `/api` during development. The endpoint rejects connections until application access checks are supplied. There are no comment channels or product events yet. +`runApi()` owns the shared server and signal handlers. It waits for HTTP requests and WebSockets to close before the entrypoint flushes [telemetry](telemetry.md). A failure closing either transport is reported after both cleanup operations settle, unless the overall API shutdown deadline expires. + ## Server operations `attachRealtimeServer({ server, path, logger, limits?, authenticate?, authorizeSubscription? })` receives an existing Node HTTP server and an injected Pino logger. It does not start listening or open another port. It returns an instance with two operations: diff --git a/hub/package-lock.json b/hub/package-lock.json index 4b6b9cd1..2f62433a 100644 --- a/hub/package-lock.json +++ b/hub/package-lock.json @@ -9,6 +9,11 @@ "version": "0.1.0", "hasInstallScript": true, "dependencies": { + "@opentelemetry/api": "1.10.0", + "@opentelemetry/context-async-hooks": "3.0.0", + "@opentelemetry/exporter-trace-otlp-proto": "0.300.0", + "@opentelemetry/resources": "3.0.0", + "@opentelemetry/sdk-trace": "3.0.0", "@primer/octicons-react": "19.40.0", "@primer/react-brand": "0.78.0", "@prisma/adapter-pg": "7.10.1", @@ -1943,6 +1948,221 @@ "@emnapi/runtime": "^1.7.1 || ^2.0.0-alpha.4" } }, + "node_modules/@opentelemetry/api": { + "version": "1.10.0", + "resolved": "https://registry.npmjs.org/@opentelemetry/api/-/api-1.10.0.tgz", + "integrity": "sha512-lMkyes8iHrG+mk6WH7X70MdCLI2RS0FwNBY1PyQ8UemaijE2m5mv5A0Niy3HtbYTFrgkUEQjFMXmSAk2hQDuPQ==", + "license": "Apache-2.0", + "engines": { + "node": ">=8.0.0" + } + }, + "node_modules/@opentelemetry/context-async-hooks": { + "version": "3.0.0", + "resolved": "https://registry.npmjs.org/@opentelemetry/context-async-hooks/-/context-async-hooks-3.0.0.tgz", + "integrity": "sha512-qDZeKSgTT+qTaL99eRbpvWO5+oZ0MKvMH4pqvzi6xYKyhiMTDBxBeTLI6BpdeMt58T5TpxHJvFl9P52i6CDWGA==", + "license": "Apache-2.0", + "engines": { + "node": ">=22.15.0" + }, + "peerDependencies": { + "@opentelemetry/api": ">=1.0.0 <1.11.0" + } + }, + "node_modules/@opentelemetry/core": { + "version": "3.0.0", + "resolved": "https://registry.npmjs.org/@opentelemetry/core/-/core-3.0.0.tgz", + "integrity": "sha512-whOLlww0R/RLtCWGBXhK1fdvL9t2/TL0fqdjZvcaoUUuNilRPjXecqQdzyVwV68s6XEDA5oYbm7/bZT62GXF4A==", + "license": "Apache-2.0", + "dependencies": { + "@opentelemetry/semantic-conventions": "^1.29.0" + }, + "engines": { + "node": ">=22.15.0" + }, + "peerDependencies": { + "@opentelemetry/api": ">=1.10.0 <1.11.0" + } + }, + "node_modules/@opentelemetry/exporter-trace-otlp-proto": { + "version": "0.300.0", + "resolved": "https://registry.npmjs.org/@opentelemetry/exporter-trace-otlp-proto/-/exporter-trace-otlp-proto-0.300.0.tgz", + "integrity": "sha512-ZCWrexCUsfp5o5Eti91uvTlXMfszknzrPt3VM+jAk/Q5T53S/ix6mF07ndSHAhreQ5bKmBg3r1Ar4wkqQ1XlPQ==", + "license": "Apache-2.0", + "dependencies": { + "@opentelemetry/otlp-exporter-base": "0.300.0", + "@opentelemetry/otlp-transformer": "0.300.0", + "@opentelemetry/sdk-trace": "3.0.0" + }, + "engines": { + "node": ">=22.15.0" + }, + "peerDependencies": { + "@opentelemetry/api": "^1.3.0" + } + }, + "node_modules/@opentelemetry/exporter-trace-otlp-proto/node_modules/@opentelemetry/otlp-transformer": { + "version": "0.300.0", + "resolved": "https://registry.npmjs.org/@opentelemetry/otlp-transformer/-/otlp-transformer-0.300.0.tgz", + "integrity": "sha512-T55lRmjRwvXWyTFdy/BbJ3T5RRsq3bAwQgxxA7qHFxgRd4+pZzRk02ZwOFsYK5072OXGYjYFNJcp68eQV4i0Ig==", + "license": "Apache-2.0", + "dependencies": { + "@opentelemetry/core": "3.0.0", + "@opentelemetry/resources": "3.0.0", + "@opentelemetry/sdk-logs": "3.0.0", + "@opentelemetry/sdk-metrics": "3.0.0", + "@opentelemetry/sdk-trace": "3.0.0" + }, + "engines": { + "node": ">=22.15.0" + }, + "peerDependencies": { + "@opentelemetry/api": ">=1.10.0 <1.11.0" + } + }, + "node_modules/@opentelemetry/exporter-trace-otlp-proto/node_modules/@opentelemetry/otlp-transformer/node_modules/@opentelemetry/sdk-logs": { + "version": "3.0.0", + "resolved": "https://registry.npmjs.org/@opentelemetry/sdk-logs/-/sdk-logs-3.0.0.tgz", + "integrity": "sha512-IysNZWrBk2PqYw4qU9PZcC7pjrZLa+KLk7vcLmPPNBUr97MgeamWhX0DUHsmxwJv3dnjYM6mLRmP8mw69ld+dA==", + "license": "Apache-2.0", + "dependencies": { + "@opentelemetry/core": "3.0.0", + "@opentelemetry/resources": "3.0.0", + "@opentelemetry/semantic-conventions": "^1.29.0" + }, + "engines": { + "node": ">=22.15.0" + }, + "peerDependencies": { + "@opentelemetry/api": ">=1.10.0 <1.11.0" + } + }, + "node_modules/@opentelemetry/exporter-trace-otlp-proto/node_modules/@opentelemetry/otlp-transformer/node_modules/@opentelemetry/sdk-metrics": { + "version": "3.0.0", + "resolved": "https://registry.npmjs.org/@opentelemetry/sdk-metrics/-/sdk-metrics-3.0.0.tgz", + "integrity": "sha512-YozyS2rIlizvUNLZwcWxH0x2FhX02/mcZwzM8hsl8yvQk+xeKmAc7vcSieWiO4oSwIB20B8eS/zwdopAd/J5zQ==", + "license": "Apache-2.0", + "dependencies": { + "@opentelemetry/core": "3.0.0", + "@opentelemetry/resources": "3.0.0" + }, + "engines": { + "node": ">=22.15.0" + }, + "peerDependencies": { + "@opentelemetry/api": ">=1.10.0 <1.11.0" + } + }, + "node_modules/@opentelemetry/otlp-exporter-base": { + "version": "0.300.0", + "resolved": "https://registry.npmjs.org/@opentelemetry/otlp-exporter-base/-/otlp-exporter-base-0.300.0.tgz", + "integrity": "sha512-D+PhzYURtv6p7Kk1rUrilLtV+ZvS218tI0jiKTV5z1agOeELw78C+cFd3o9UUshUx1j+EIpFM62yGk8BuydQvg==", + "license": "Apache-2.0", + "dependencies": { + "@opentelemetry/core": "3.0.0", + "@opentelemetry/otlp-transformer": "0.300.0" + }, + "engines": { + "node": ">=22.15.0" + }, + "peerDependencies": { + "@opentelemetry/api": "^1.3.0" + } + }, + "node_modules/@opentelemetry/otlp-exporter-base/node_modules/@opentelemetry/otlp-transformer": { + "version": "0.300.0", + "resolved": "https://registry.npmjs.org/@opentelemetry/otlp-transformer/-/otlp-transformer-0.300.0.tgz", + "integrity": "sha512-T55lRmjRwvXWyTFdy/BbJ3T5RRsq3bAwQgxxA7qHFxgRd4+pZzRk02ZwOFsYK5072OXGYjYFNJcp68eQV4i0Ig==", + "license": "Apache-2.0", + "dependencies": { + "@opentelemetry/core": "3.0.0", + "@opentelemetry/resources": "3.0.0", + "@opentelemetry/sdk-logs": "3.0.0", + "@opentelemetry/sdk-metrics": "3.0.0", + "@opentelemetry/sdk-trace": "3.0.0" + }, + "engines": { + "node": ">=22.15.0" + }, + "peerDependencies": { + "@opentelemetry/api": ">=1.10.0 <1.11.0" + } + }, + "node_modules/@opentelemetry/otlp-exporter-base/node_modules/@opentelemetry/otlp-transformer/node_modules/@opentelemetry/sdk-logs": { + "version": "3.0.0", + "resolved": "https://registry.npmjs.org/@opentelemetry/sdk-logs/-/sdk-logs-3.0.0.tgz", + "integrity": "sha512-IysNZWrBk2PqYw4qU9PZcC7pjrZLa+KLk7vcLmPPNBUr97MgeamWhX0DUHsmxwJv3dnjYM6mLRmP8mw69ld+dA==", + "license": "Apache-2.0", + "dependencies": { + "@opentelemetry/core": "3.0.0", + "@opentelemetry/resources": "3.0.0", + "@opentelemetry/semantic-conventions": "^1.29.0" + }, + "engines": { + "node": ">=22.15.0" + }, + "peerDependencies": { + "@opentelemetry/api": ">=1.10.0 <1.11.0" + } + }, + "node_modules/@opentelemetry/otlp-exporter-base/node_modules/@opentelemetry/otlp-transformer/node_modules/@opentelemetry/sdk-metrics": { + "version": "3.0.0", + "resolved": "https://registry.npmjs.org/@opentelemetry/sdk-metrics/-/sdk-metrics-3.0.0.tgz", + "integrity": "sha512-YozyS2rIlizvUNLZwcWxH0x2FhX02/mcZwzM8hsl8yvQk+xeKmAc7vcSieWiO4oSwIB20B8eS/zwdopAd/J5zQ==", + "license": "Apache-2.0", + "dependencies": { + "@opentelemetry/core": "3.0.0", + "@opentelemetry/resources": "3.0.0" + }, + "engines": { + "node": ">=22.15.0" + }, + "peerDependencies": { + "@opentelemetry/api": ">=1.10.0 <1.11.0" + } + }, + "node_modules/@opentelemetry/resources": { + "version": "3.0.0", + "resolved": "https://registry.npmjs.org/@opentelemetry/resources/-/resources-3.0.0.tgz", + "integrity": "sha512-4762KVmlDroRIt+4Ra8AnwcQqdXlkAR55jDrhzna8bzhJx1mUzvHW/Lh/OemFuY9+ouM4xaDc6CbUmoJmM6BFg==", + "license": "Apache-2.0", + "dependencies": { + "@opentelemetry/core": "3.0.0", + "@opentelemetry/semantic-conventions": "^1.29.0" + }, + "engines": { + "node": ">=22.15.0" + }, + "peerDependencies": { + "@opentelemetry/api": ">=1.10.0 <1.11.0" + } + }, + "node_modules/@opentelemetry/sdk-trace": { + "version": "3.0.0", + "resolved": "https://registry.npmjs.org/@opentelemetry/sdk-trace/-/sdk-trace-3.0.0.tgz", + "integrity": "sha512-4T0DbDGdtsjNmYK5/lNalHro4fsMZ0UZ3JE+C8fcqYP25xaJC22FHg8kdJNJ21srjJY5XoT7WNFCtj88NNTl4Q==", + "license": "Apache-2.0", + "dependencies": { + "@opentelemetry/core": "3.0.0", + "@opentelemetry/resources": "3.0.0", + "@opentelemetry/semantic-conventions": "^1.29.0" + }, + "engines": { + "node": ">=22.15.0" + }, + "peerDependencies": { + "@opentelemetry/api": ">=1.10.0 <1.11.0" + } + }, + "node_modules/@opentelemetry/semantic-conventions": { + "version": "1.43.0", + "resolved": "https://registry.npmjs.org/@opentelemetry/semantic-conventions/-/semantic-conventions-1.43.0.tgz", + "integrity": "sha512-eSYWTm620tTk45EKSedaUL8MFYI8hW164hIXsgIHyxu3VobUB3fFCu5t0hQby6OoWRPsG1KkKUG2M5UadiLiVg==", + "license": "Apache-2.0", + "engines": { + "node": ">=14" + } + }, "node_modules/@oxc-parser/binding-android-arm-eabi": { "version": "0.127.0", "resolved": "https://registry.npmjs.org/@oxc-parser/binding-android-arm-eabi/-/binding-android-arm-eabi-0.127.0.tgz", diff --git a/hub/package.json b/hub/package.json index 1334554c..c294691a 100644 --- a/hub/package.json +++ b/hub/package.json @@ -46,6 +46,11 @@ "typecheck": "tsc --noEmit && tsc --project ../.github/extensions/replica-prototype/tsconfig.json" }, "dependencies": { + "@opentelemetry/api": "1.10.0", + "@opentelemetry/context-async-hooks": "3.0.0", + "@opentelemetry/exporter-trace-otlp-proto": "0.300.0", + "@opentelemetry/resources": "3.0.0", + "@opentelemetry/sdk-trace": "3.0.0", "@prisma/adapter-pg": "7.10.1", "@prisma/client": "7.10.1", "@primer/octicons-react": "19.40.0", diff --git a/hub/server/index.ts b/hub/server/index.ts index 5136dded..e1fb35ce 100644 --- a/hub/server/index.ts +++ b/hub/server/index.ts @@ -1,31 +1,22 @@ import "dotenv/config" -import { createServer } from "node:http" - -import { createApp } from "./create-app/index.ts" -import { log } from "./integrations/pino/log/index.ts" +import { startTelemetry } from "./integrations/opentelemetry/start-telemetry/index.ts" import { logger } from "./integrations/pino/index.ts" -import { attachRealtimeServer } from "./integrations/ws/attach-realtime-server/index.ts" -import { shutdownServer } from "./shutdown-server/index.ts" +import { runApi } from "./run-api/index.ts" const port = Number(process.env.PORT ?? 3001) if (!Number.isInteger(port) || port < 1 || port > 65535) { throw new Error("PORT must be an integer between 1 and 65535.") } -const server = createServer() -const realtime = attachRealtimeServer({ server, path: "/api/realtime", logger }) -server.on("request", createApp({ broadcast: realtime.broadcast })) -server.listen(port, "127.0.0.1", () => { - log("info", "Hub API listening", { host: "127.0.0.1", port }) -}) - -let shuttingDown = false -function shutdown() { - if (shuttingDown) return - shuttingDown = true - void shutdownServer(server, realtime) +try { + const telemetry = startTelemetry(process.env, logger) + try { + await runApi({ port, logger }) + } finally { + await telemetry.shutdown() + } +} catch { + logger.fatal("Hub API failed; check API configuration, telemetry, and shutdown time.") + process.exit(1) } - -process.once("SIGINT", shutdown) -process.once("SIGTERM", shutdown) diff --git a/hub/server/integrations/opentelemetry/start-telemetry/index.test.ts b/hub/server/integrations/opentelemetry/start-telemetry/index.test.ts new file mode 100644 index 00000000..d71bf838 --- /dev/null +++ b/hub/server/integrations/opentelemetry/start-telemetry/index.test.ts @@ -0,0 +1,278 @@ +import { context, diag, propagation, ROOT_CONTEXT, trace } from "@opentelemetry/api" +import { OTLPTraceExporter } from "@opentelemetry/exporter-trace-otlp-proto" +import type { ReadableSpan, SpanExporter } from "@opentelemetry/sdk-trace" +import { afterEach, beforeEach, describe, expect, it, onTestFinished, vi } from "vitest" + +vi.mock("@opentelemetry/exporter-trace-otlp-proto", () => ({ + OTLPTraceExporter: vi.fn(), +})) + +function resetGlobals() { + trace.disable() + context.disable() + diag.disable() + propagation.disable() +} + +function mockTransport() { + const spans: ReadableSpan[] = [] + const shutdown = vi.fn(async () => {}) + const exportSpans = vi.fn((batch, callback) => { + spans.push(...batch) + callback({ code: 0 }) + }) + vi.mocked(OTLPTraceExporter).mockImplementation( + class { + export = exportSpans + shutdown = shutdown + forceFlush = async () => {} + } as unknown as typeof OTLPTraceExporter, + ) + return { spans, shutdown, exportSpans } +} + +async function settleShutdown(telemetry: { shutdown(): Promise }) { + try { + await telemetry.shutdown() + } catch (error) { + expect(error).toBeInstanceOf(Error) + expect(String(error)).not.toMatch(/secret|https:|private\.example/) + } +} + +function expectSafeLogs(calls: unknown[][]) { + expect(calls.length).toBeGreaterThan(0) + for (const call of calls) expect(call).toEqual([expect.any(String)]) + expect(JSON.stringify(calls)).not.toMatch(/secret|https:|private\.example/) +} + +beforeEach(resetGlobals) +afterEach(resetGlobals) + +describe("startTelemetry()", () => { + it.each([{}, { OTEL_SDK_DISABLED: "TrUe", OTEL_TRACES_EXPORTER: "invalid" }])( + "returns an idle synchronous handle when tracing is disabled: %j", + async (env) => { + const logger = { error: vi.fn(), warn: vi.fn() } + const { startTelemetry } = await import(".") + const telemetry = startTelemetry(env, logger) + onTestFinished(() => telemetry.shutdown()) + + expect(telemetry).not.toBeInstanceOf(Promise) + expect(OTLPTraceExporter).not.toHaveBeenCalled() + await expect(telemetry.shutdown()).resolves.toBeUndefined() + await expect(telemetry.shutdown()).resolves.toBeUndefined() + expect(logger.error).not.toHaveBeenCalled() + expect(logger.warn).not.toHaveBeenCalled() + }, + ) + + it("uses explicit trace configuration before common configuration", async () => { + const transport = mockTransport() + const { startTelemetry } = await import(".") + const { withTelemetrySpan } = await import("../with-telemetry-span") + const telemetry = startTelemetry( + { + OTEL_TRACES_EXPORTER: "otlp", + OTEL_EXPORTER_OTLP_TRACES_PROTOCOL: "http/protobuf", + OTEL_EXPORTER_OTLP_PROTOCOL: "grpc", + OTEL_EXPORTER_OTLP_TRACES_ENDPOINT: "https://collector.example/custom/traces", + OTEL_EXPORTER_OTLP_ENDPOINT: "not-an-endpoint", + OTEL_EXPORTER_OTLP_TRACES_HEADERS: "authorization=Bearer%20test%3Dtoken,x%2Dlabel=a%2Cb", + OTEL_EXPORTER_OTLP_HEADERS: "invalid", + OTEL_SERVICE_NAME: "hub-test", + OTEL_RESOURCE_ATTRIBUTES: "private.secret=resource-secret", + }, + { error: vi.fn(), warn: vi.fn() }, + ) + onTestFinished(() => telemetry.shutdown()) + + await withTelemetrySpan("test.operation", {}, () => {}) + await telemetry.shutdown() + + expect(OTLPTraceExporter).toHaveBeenCalledWith( + expect.objectContaining({ + url: "https://collector.example/custom/traces", + headers: { authorization: "Bearer test=token", "x-label": "a,b" }, + }), + ) + expect(transport.spans).toHaveLength(1) + expect(transport.spans[0]?.resource.attributes["service.name"]).toBe("hub-test") + expect(transport.spans[0]?.resource.attributes).not.toHaveProperty("private.secret") + expect(JSON.stringify(transport.spans[0]?.resource.attributes)).not.toContain("resource-secret") + }) + + it("appends the trace path to a common base endpoint", async () => { + const transport = mockTransport() + const { startTelemetry } = await import(".") + const { withTelemetrySpan } = await import("../with-telemetry-span") + const telemetry = startTelemetry( + { + OTEL_TRACES_EXPORTER: "otlp", + OTEL_EXPORTER_OTLP_ENDPOINT: "http://collector.example/base/", + OTEL_EXPORTER_OTLP_HEADERS: "x-token=test%20value", + }, + { error: vi.fn(), warn: vi.fn() }, + ) + onTestFinished(() => telemetry.shutdown()) + + await withTelemetrySpan("test.operation", {}, () => {}) + await telemetry.shutdown() + + expect(OTLPTraceExporter).toHaveBeenCalledWith( + expect.objectContaining({ + url: "http://collector.example/base/v1/traces", + headers: { "x-token": "test value" }, + }), + ) + expect(transport.spans[0]?.resource.attributes["service.name"]).toBe("replica-hub") + }) + + it.each([ + ["unsupported exporter", { OTEL_TRACES_EXPORTER: "secret-exporter" }], + ["missing endpoint", {}], + ["credentials", { OTEL_EXPORTER_OTLP_ENDPOINT: "https://secret-user:secret-pass@example.com" }], + ["query", { OTEL_EXPORTER_OTLP_ENDPOINT: "https://example.com?token=secret-query" }], + ["unsupported protocol", { OTEL_EXPORTER_OTLP_PROTOCOL: "secret-protocol" }], + ["encoded header newline", { OTEL_EXPORTER_OTLP_HEADERS: "x-token=secret%0D%0Avalue" }], + ["ratio greater than one", { OTEL_TRACES_SAMPLER_ARG: "1.1" }], + ])("rejects %s synchronously without leaking supplied configuration", async (_, invalid) => { + const logger = { error: vi.fn(), warn: vi.fn() } + const { startTelemetry } = await import(".") + const env: NodeJS.ProcessEnv = { + OTEL_TRACES_EXPORTER: "otlp", + OTEL_EXPORTER_OTLP_ENDPOINT: "https://example.com/secret-base", + ...invalid, + } + if (Object.keys(invalid).length === 0) delete env.OTEL_EXPORTER_OTLP_ENDPOINT + let thrown: unknown + + try { + const telemetry = startTelemetry(env, logger) + onTestFinished(() => telemetry.shutdown()) + } catch (error) { + thrown = error + } + + expect(thrown).toBeInstanceOf(Error) + expect(String(thrown)).not.toMatch(/secret|https:|1\.1/) + expect(JSON.stringify([logger.error.mock.calls, logger.warn.mock.calls])).not.toContain( + "secret", + ) + expect(OTLPTraceExporter).not.toHaveBeenCalled() + }) + + it.each([ + ["parentbased_traceidratio", "0"], + ["always_off", undefined], + ])("does not export with sampler %s and ratio %s", async (sampler, ratio) => { + const transport = mockTransport() + const { startTelemetry } = await import(".") + const { withTelemetrySpan } = await import("../with-telemetry-span") + const telemetry = startTelemetry( + { + OTEL_TRACES_EXPORTER: "otlp", + OTEL_EXPORTER_OTLP_ENDPOINT: "https://collector.example", + OTEL_TRACES_SAMPLER: sampler, + OTEL_TRACES_SAMPLER_ARG: ratio, + }, + { error: vi.fn(), warn: vi.fn() }, + ) + onTestFinished(() => telemetry.shutdown()) + + await withTelemetrySpan("test.sample", {}, () => {}) + await telemetry.shutdown() + + expect(transport.spans).toHaveLength(0) + }) + + it("rejects duplicate initialization and shuts down idempotently", async () => { + const transport = mockTransport() + const { startTelemetry } = await import(".") + const { withTelemetrySpan } = await import("../with-telemetry-span") + const env = { + OTEL_TRACES_EXPORTER: "otlp", + OTEL_EXPORTER_OTLP_ENDPOINT: "https://collector.example", + } + const logger = { error: vi.fn(), warn: vi.fn() } + const telemetry = startTelemetry(env, logger) + onTestFinished(() => telemetry.shutdown()) + + expect(() => startTelemetry(env, logger)).toThrow() + await withTelemetrySpan("test.first", {}, () => {}) + await Promise.all([telemetry.shutdown(), telemetry.shutdown()]) + expect(transport.shutdown).toHaveBeenCalledTimes(1) + await withTelemetrySpan("test.disabled", {}, () => {}) + + expect(transport.spans.map((span) => span.name)).toEqual(["test.first"]) + expect(context.active()).toBe(ROOT_CONTEXT) + }) + + it("reports failed exports without exposing exporter errors or configuration", async () => { + const transport = mockTransport() + transport.exportSpans.mockImplementation((_, callback) => + callback({ code: 1, error: new Error("secret-export-token https://private.example") }), + ) + const logger = { error: vi.fn(), warn: vi.fn() } + const { startTelemetry } = await import(".") + const { withTelemetrySpan } = await import("../with-telemetry-span") + const telemetry = startTelemetry( + { + OTEL_TRACES_EXPORTER: "otlp", + OTEL_EXPORTER_OTLP_ENDPOINT: "https://private.example", + OTEL_EXPORTER_OTLP_HEADERS: "authorization=secret-auth", + }, + logger, + ) + onTestFinished(() => settleShutdown(telemetry)) + + await withTelemetrySpan("test.export", {}, () => {}) + await expect(telemetry.shutdown()).rejects.toThrow(/shutdown/i) + await expect(telemetry.shutdown()).rejects.not.toThrow(/secret|https:|private\.example/) + + expectSafeLogs([...logger.error.mock.calls, ...logger.warn.mock.calls]) + }) + + it("reports shutdown rejection without leaking diagnostics", async () => { + const transport = mockTransport() + transport.shutdown.mockRejectedValue(new Error("secret-shutdown-token")) + const logger = { error: vi.fn(), warn: vi.fn() } + const { startTelemetry } = await import(".") + const env = { + OTEL_TRACES_EXPORTER: "otlp", + OTEL_EXPORTER_OTLP_ENDPOINT: "https://collector.example", + } + const telemetry = startTelemetry(env, logger) + onTestFinished(() => settleShutdown(telemetry)) + + await expect(telemetry.shutdown()).rejects.toThrow(/shutdown/i) + await expect(telemetry.shutdown()).rejects.not.toThrow(/secret|https:|private\.example/) + expect(transport.shutdown).toHaveBeenCalledTimes(1) + expectSafeLogs([...logger.error.mock.calls, ...logger.warn.mock.calls]) + }) + + it("bounds shutdown at five seconds when an exporter never settles", async () => { + vi.useFakeTimers() + const transport = mockTransport() + transport.shutdown.mockImplementation(() => new Promise(() => {})) + const logger = { error: vi.fn(), warn: vi.fn() } + const { startTelemetry } = await import(".") + const telemetry = startTelemetry( + { + OTEL_TRACES_EXPORTER: "otlp", + OTEL_EXPORTER_OTLP_ENDPOINT: "https://collector.example", + }, + logger, + ) + onTestFinished(() => settleShutdown(telemetry)) + const shuttingDown = telemetry.shutdown() + void shuttingDown.catch(() => {}) + + await vi.advanceTimersByTimeAsync(5000) + await expect(shuttingDown).rejects.toThrow(/shutdown/i) + await expect(shuttingDown).rejects.not.toThrow(/secret|https:|private\.example/) + + expectSafeLogs([...logger.error.mock.calls, ...logger.warn.mock.calls]) + expect(vi.getTimerCount()).toBe(0) + }) +}) diff --git a/hub/server/integrations/opentelemetry/start-telemetry/index.ts b/hub/server/integrations/opentelemetry/start-telemetry/index.ts new file mode 100644 index 00000000..1a59be7f --- /dev/null +++ b/hub/server/integrations/opentelemetry/start-telemetry/index.ts @@ -0,0 +1,191 @@ +import { context, diag, DiagLogLevel, trace } from "@opentelemetry/api" +import { AsyncLocalStorageContextManager } from "@opentelemetry/context-async-hooks" +import { OTLPTraceExporter } from "@opentelemetry/exporter-trace-otlp-proto" +import { resourceFromAttributes } from "@opentelemetry/resources" +import { + AlwaysOffSampler, + AlwaysOnSampler, + BatchSpanProcessor, + ParentBasedSampler, + TraceIdRatioBasedSampler, + TracerProvider, +} from "@opentelemetry/sdk-trace" + +import type { TelemetryLogger, TelemetryRuntime } from "./types.ts" + +let started = false + +export function startTelemetry(env: NodeJS.ProcessEnv, logger: TelemetryLogger): TelemetryRuntime { + if (env.OTEL_SDK_DISABLED?.toLowerCase() === "true") { + return { async shutdown() {} } + } + const exporterName = env.OTEL_TRACES_EXPORTER ?? "none" + if (exporterName === "none") return { async shutdown() {} } + if (exporterName !== "otlp") { + throw new Error("OTEL_TRACES_EXPORTER must be none or otlp.") + } + if (env.OTEL_SDK_DISABLED && env.OTEL_SDK_DISABLED.toLowerCase() !== "false") { + throw new Error("OTEL_SDK_DISABLED must be true or false.") + } + if (started) throw new Error("Telemetry has already started in this process.") + + const protocol = + env.OTEL_EXPORTER_OTLP_TRACES_PROTOCOL ?? env.OTEL_EXPORTER_OTLP_PROTOCOL ?? "http/protobuf" + if (protocol !== "http/protobuf") { + throw new Error("Telemetry supports only the http/protobuf OTLP protocol.") + } + const endpoint = env.OTEL_EXPORTER_OTLP_TRACES_ENDPOINT ?? env.OTEL_EXPORTER_OTLP_ENDPOINT + let url: URL + try { + if (!endpoint) throw new Error() + url = new URL(endpoint) + if ( + !["http:", "https:"].includes(url.protocol) || + !url.hostname || + url.username || + url.password || + url.search || + url.hash || + (env.OTEL_EXPORTER_OTLP_TRACES_ENDPOINT !== undefined && url.pathname === "/") + ) { + throw new Error() + } + } catch { + throw new Error( + "Telemetry requires an absolute HTTP(S) endpoint without credentials, query, or fragment; a trace-specific endpoint must include a request path.", + ) + } + if (env.OTEL_EXPORTER_OTLP_TRACES_ENDPOINT === undefined) { + url.pathname = `${url.pathname.replace(/\/$/, "")}/v1/traces` + } + + const headers: Record = {} + const encodedHeaders = env.OTEL_EXPORTER_OTLP_TRACES_HEADERS ?? env.OTEL_EXPORTER_OTLP_HEADERS + if (encodedHeaders !== undefined) { + try { + for (const entry of encodedHeaders.split(",")) { + const separator = entry.indexOf("=") + if (separator < 1) throw new Error() + const key = decodeURIComponent(entry.slice(0, separator).trim()) + const value = decodeURIComponent(entry.slice(separator + 1).trim()) + if ( + !/^[!#$%&'*+.^_`|~0-9a-z-]+$/i.test(key) || + !value.trim() || + !/^[\x20-\x7e]+$/.test(value) || + Object.hasOwn(headers, key.toLowerCase()) + ) { + throw new Error() + } + Object.defineProperty(headers, key.toLowerCase(), { + value, + enumerable: true, + }) + } + } catch { + throw new Error("Telemetry headers must be comma-separated URL-encoded key=value entries.") + } + } + + const samplerName = env.OTEL_TRACES_SAMPLER ?? "parentbased_traceidratio" + if (!["parentbased_traceidratio", "always_on", "always_off"].includes(samplerName)) { + throw new Error( + "OTEL_TRACES_SAMPLER must be parentbased_traceidratio, always_on, or always_off.", + ) + } + const ratioText = env.OTEL_TRACES_SAMPLER_ARG ?? "1" + const ratio = Number(ratioText) + if (!ratioText.trim() || !Number.isFinite(ratio) || ratio < 0 || ratio > 1) { + throw new Error("OTEL_TRACES_SAMPLER_ARG must be a number between 0 and 1.") + } + const serviceName = env.OTEL_SERVICE_NAME ?? "replica-hub" + if (!serviceName.trim() || !/^[\x20-\x7e]{1,100}$/.test(serviceName)) { + throw new Error("OTEL_SERVICE_NAME must contain 1 through 100 printable characters.") + } + + const exporter = new OTLPTraceExporter({ + url: url.href, + headers, + timeoutMillis: 2_000, + }) + const provider = new TracerProvider({ + resource: resourceFromAttributes({ "service.name": serviceName }), + sampler: + samplerName === "always_on" + ? new AlwaysOnSampler() + : samplerName === "always_off" + ? new AlwaysOffSampler() + : new ParentBasedSampler({ root: new TraceIdRatioBasedSampler(ratio) }), + spanLimits: { attributeCountLimit: 16, attributeValueLengthLimit: 100 }, + spanProcessors: [ + new BatchSpanProcessor({ + exporter, + maxQueueSize: 256, + maxExportBatchSize: 64, + scheduledDelayMillis: 1_000, + exportTimeoutMillis: 2_000, + }), + ], + }) + const manager = new AsyncLocalStorageContextManager().enable() + if (!context.setGlobalContextManager(manager)) { + manager.disable() + void provider.shutdown().catch(() => { + logger.error("Telemetry initialization cleanup failed.") + }) + throw new Error("A telemetry context manager is already registered.") + } + if (!trace.setGlobalTracerProvider(provider)) { + context.disable() + manager.disable() + void provider.shutdown().catch(() => { + logger.error("Telemetry initialization cleanup failed.") + }) + throw new Error("A telemetry provider is already registered.") + } + // SDK diagnostics may include recipient credentials or private exception text. + diag.setLogger( + { + error() { + logger.error("Telemetry export failed; spans may have been dropped.") + }, + warn() { + logger.warn("Telemetry SDK warning; check telemetry configuration and delivery.") + }, + info() {}, + debug() {}, + verbose() {}, + }, + DiagLogLevel.WARN, + ) + started = true + let shutdown: Promise | undefined + + return { + shutdown() { + shutdown ??= (async () => { + let timer: ReturnType | undefined + try { + await Promise.race([ + provider.shutdown(), + new Promise((_, reject) => { + timer = setTimeout(() => reject(new Error()), 5_000) + }), + ]) + } catch { + logger.error( + "Telemetry shutdown failed or exceeded its deadline; pending spans may be lost.", + ) + throw new Error("Telemetry shutdown failed or exceeded its deadline.") + } finally { + clearTimeout(timer) + trace.disable() + context.disable() + manager.disable() + diag.disable() + started = false + } + })() + return shutdown + }, + } +} diff --git a/hub/server/integrations/opentelemetry/start-telemetry/types.ts b/hub/server/integrations/opentelemetry/start-telemetry/types.ts new file mode 100644 index 00000000..b910b02e --- /dev/null +++ b/hub/server/integrations/opentelemetry/start-telemetry/types.ts @@ -0,0 +1,7 @@ +import type { Logger } from "pino" + +export type TelemetryLogger = Pick + +export type TelemetryRuntime = { + shutdown(): Promise +} diff --git a/hub/server/integrations/opentelemetry/with-telemetry-span/index.test.ts b/hub/server/integrations/opentelemetry/with-telemetry-span/index.test.ts new file mode 100644 index 00000000..da9919b1 --- /dev/null +++ b/hub/server/integrations/opentelemetry/with-telemetry-span/index.test.ts @@ -0,0 +1,207 @@ +import { context, diag, propagation, ROOT_CONTEXT, SpanStatusCode, trace } from "@opentelemetry/api" +import { OTLPTraceExporter } from "@opentelemetry/exporter-trace-otlp-proto" +import type { ReadableSpan, SpanExporter } from "@opentelemetry/sdk-trace" +import { afterEach, beforeEach, describe, expect, it, onTestFinished, vi } from "vitest" + +vi.mock("@opentelemetry/exporter-trace-otlp-proto", () => ({ + OTLPTraceExporter: vi.fn(), +})) + +function resetGlobals() { + trace.disable() + context.disable() + diag.disable() + propagation.disable() +} + +function mockTransport() { + const spans: ReadableSpan[] = [] + vi.mocked(OTLPTraceExporter).mockImplementation( + class { + export: SpanExporter["export"] = (batch, callback) => { + spans.push(...batch) + callback({ code: 0 }) + } + shutdown = async () => {} + forceFlush = async () => {} + } as unknown as typeof OTLPTraceExporter, + ) + return spans +} + +beforeEach(resetGlobals) +afterEach(resetGlobals) + +describe("withTelemetrySpan()", () => { + it("runs operations while disabled and preserves the exact result", async () => { + const { withTelemetrySpan } = await import(".") + const result = { id: "result" } + const operation = vi.fn(() => result) + + await expect(withTelemetrySpan("test.disabled", {}, operation)).resolves.toBe(result) + + expect(operation).toHaveBeenCalledTimes(1) + expect(OTLPTraceExporter).not.toHaveBeenCalled() + expect(trace.getActiveSpan()).toBeUndefined() + }) + + it.each([ + ["name containing private data", "test.secret@example.com", {}], + ["unknown attribute", "test.operation", { "http.url": "https://secret.example" }], + ["invalid UUID", "test.operation", { "replica.repository.id": "secret-repository" }], + ])("rejects %s before running an operation, even when disabled", async (_, name, attributes) => { + const { withTelemetrySpan } = await import(".") + const operation = vi.fn() + let thrown: unknown + + try { + await withTelemetrySpan( + name, + attributes as Parameters[1], + operation, + ) + } catch (error) { + thrown = error + } + + expect(thrown).toBeInstanceOf(Error) + expect(String(thrown)).not.toMatch(/secret|https:/) + expect(operation).not.toHaveBeenCalled() + expect(OTLPTraceExporter).not.toHaveBeenCalled() + }) + + it("exports approved IDs, preserves the result, and ends successful spans", async () => { + const spans = mockTransport() + const { startTelemetry } = await import("../start-telemetry") + const { withTelemetrySpan } = await import(".") + const telemetry = startTelemetry( + { + OTEL_TRACES_EXPORTER: "otlp", + OTEL_EXPORTER_OTLP_ENDPOINT: "https://collector.example", + }, + { error: vi.fn(), warn: vi.fn() }, + ) + onTestFinished(() => telemetry.shutdown()) + const attributes = { + "replica.repository.id": "123e4567-e89b-42d3-a456-426614174000", + "replica.prototype.id": "123e4567-e89b-42d3-a456-426614174001", + "replica.build.id": "123e4567-e89b-42d3-a456-426614174002", + "replica.thread.id": "123e4567-e89b-42d3-a456-426614174003", + "replica.comment.id": "123e4567-e89b-42d3-a456-426614174004", + } + const result = { preserved: true } + + await expect( + withTelemetrySpan("build.import", attributes, () => Promise.resolve(result)), + ).resolves.toBe(result) + await telemetry.shutdown() + + expect(spans).toHaveLength(1) + expect(spans[0]).toMatchObject({ + attributes, + status: { code: SpanStatusCode.OK }, + ended: true, + events: [], + }) + expect(trace.getActiveSpan()).toBeUndefined() + }) + + it("keeps parents isolated across concurrent async operations and restores the caller context", async () => { + const spans = mockTransport() + const { startTelemetry } = await import("../start-telemetry") + const { withTelemetrySpan } = await import(".") + const telemetry = startTelemetry( + { + OTEL_TRACES_EXPORTER: "otlp", + OTEL_EXPORTER_OTLP_ENDPOINT: "https://collector.example", + }, + { error: vi.fn(), warn: vi.fn() }, + ) + onTestFinished(() => telemetry.shutdown()) + const firstStarted = { resolve() {} } + const releaseFirst = { resolve() {} } + const firstStartedPromise = new Promise((resolve) => { + firstStarted.resolve = resolve + }) + const firstCanFinish = new Promise((resolve) => { + releaseFirst.resolve = resolve + }) + + const first = withTelemetrySpan("test.first", {}, async () => { + const parent = trace.getActiveSpan() + firstStarted.resolve() + await firstCanFinish + await withTelemetrySpan("test.first.child", {}, async () => { + await Promise.resolve() + expect(trace.getActiveSpan()).not.toBe(parent) + }) + expect(trace.getActiveSpan()).toBe(parent) + }) + await firstStartedPromise + expect(trace.getActiveSpan()).toBeUndefined() + try { + await withTelemetrySpan("test.second", {}, async () => { + await Promise.resolve() + await withTelemetrySpan("test.second.child", {}, () => {}) + }) + } finally { + releaseFirst.resolve() + await first + } + await telemetry.shutdown() + + const firstParent = spans.find((span) => span.name === "test.first") + const secondParent = spans.find((span) => span.name === "test.second") + const firstChild = spans.find((span) => span.name === "test.first.child") + const secondChild = spans.find((span) => span.name === "test.second.child") + expect(spans).toHaveLength(4) + expect(firstParent?.parentSpanContext).toBeUndefined() + expect(secondParent?.parentSpanContext).toBeUndefined() + expect(firstChild?.parentSpanContext?.spanId).toBe(firstParent?.spanContext().spanId) + expect(secondChild?.parentSpanContext?.spanId).toBe(secondParent?.spanContext().spanId) + expect(firstChild?.spanContext().traceId).toBe(firstParent?.spanContext().traceId) + expect(secondChild?.spanContext().traceId).toBe(secondParent?.spanContext().traceId) + expect(firstParent?.spanContext().traceId).not.toBe(secondParent?.spanContext().traceId) + expect(context.active()).toBe(ROOT_CONTEXT) + }) + + it("rethrows the original operation error, ends the span, and records no failure details", async () => { + const spans = mockTransport() + const { startTelemetry } = await import("../start-telemetry") + const { withTelemetrySpan } = await import(".") + const telemetry = startTelemetry( + { + OTEL_TRACES_EXPORTER: "otlp", + OTEL_EXPORTER_OTLP_ENDPOINT: "https://collector.example", + }, + { error: vi.fn(), warn: vi.fn() }, + ) + onTestFinished(() => telemetry.shutdown()) + const failure = new Error("secret-error-message") + + const result = withTelemetrySpan("test.failure", {}, async () => { + await Promise.resolve() + throw failure + }) + + await expect(result).rejects.toBe(failure) + await telemetry.shutdown() + + expect(spans).toHaveLength(1) + expect(spans[0]?.status).toEqual({ code: SpanStatusCode.ERROR }) + expect(spans[0]?.events).toEqual([]) + expect(spans[0]?.attributes).toEqual({}) + expect(spans[0]?.ended).toBe(true) + expect( + JSON.stringify( + spans.map(({ name, attributes, events, status }) => ({ + name, + attributes, + events, + status, + })), + ), + ).not.toContain("secret") + expect(trace.getActiveSpan()).toBeUndefined() + }) +}) diff --git a/hub/server/integrations/opentelemetry/with-telemetry-span/index.ts b/hub/server/integrations/opentelemetry/with-telemetry-span/index.ts new file mode 100644 index 00000000..ce589732 --- /dev/null +++ b/hub/server/integrations/opentelemetry/with-telemetry-span/index.ts @@ -0,0 +1,42 @@ +import { SpanStatusCode, trace } from "@opentelemetry/api" +import { z } from "zod" + +import type { TelemetryAttributes } from "./types.ts" + +const attributesSchema = z.strictObject({ + "replica.repository.id": z.uuid().optional(), + "replica.prototype.id": z.uuid().optional(), + "replica.build.id": z.uuid().optional(), + "replica.thread.id": z.uuid().optional(), + "replica.comment.id": z.uuid().optional(), +}) + +export async function withTelemetrySpan( + name: string, + attributes: TelemetryAttributes, + operation: () => T | Promise, +): Promise { + if (name.length > 100 || !/^[a-z][a-z0-9]*(?:[._-][a-z0-9]+)*$/.test(name)) { + throw new Error( + "Telemetry span names must be static lowercase identifiers of 1 through 100 characters.", + ) + } + const parsed = attributesSchema.safeParse(attributes) + if (!parsed.success) { + throw new Error("Telemetry attributes must contain only approved UUID identifiers.") + } + return trace + .getTracer("replica-hub") + .startActiveSpan(name, { attributes: parsed.data }, async (span) => { + try { + const result = await operation() + span.setStatus({ code: SpanStatusCode.OK }) + return result + } catch (error) { + span.setStatus({ code: SpanStatusCode.ERROR }) + throw error + } finally { + span.end() + } + }) +} diff --git a/hub/server/integrations/opentelemetry/with-telemetry-span/types.ts b/hub/server/integrations/opentelemetry/with-telemetry-span/types.ts new file mode 100644 index 00000000..5afe951e --- /dev/null +++ b/hub/server/integrations/opentelemetry/with-telemetry-span/types.ts @@ -0,0 +1,7 @@ +export type TelemetryAttributes = { + "replica.repository.id"?: string + "replica.prototype.id"?: string + "replica.build.id"?: string + "replica.thread.id"?: string + "replica.comment.id"?: string +} diff --git a/hub/server/run-api/index.test.ts b/hub/server/run-api/index.test.ts new file mode 100644 index 00000000..b564915e --- /dev/null +++ b/hub/server/run-api/index.test.ts @@ -0,0 +1,277 @@ +import { once } from "node:events" +import type * as Http from "node:http" +import { createServer, request, Server } from "node:http" +import type { IncomingMessage } from "node:http" +import type { AddressInfo } from "node:net" +import { describe, expect, it, onTestFinished, vi } from "vitest" +import WebSocket from "ws" + +import { deferred } from "../../test/ws/deferred" +import { rejected } from "../../test/ws/rejected" +import { createApp } from "../create-app" +import { createLogger } from "../integrations/pino/create-logger" +import { attachRealtimeServer } from "../integrations/ws/attach-realtime-server" +import type * as RealtimeModule from "../integrations/ws/attach-realtime-server" +import type { RealtimeServer } from "../integrations/ws/types" + +vi.mock("../create-app", { spy: true }) +vi.mock("../integrations/ws/attach-realtime-server", { spy: true }) + +vi.mock("node:http", async (importOriginal) => { + const actual = await importOriginal() + return { ...actual, createServer: vi.fn(actual.createServer) } +}) + +async function startApi(options: { port: number; shutdownTimeoutMs?: number }) { + const { runApi } = await import(".") + const logger = createLogger({ level: "silent" }) + const readiness = vi.spyOn(logger, "info") + const signals = { + SIGINT: process.listeners("SIGINT"), + SIGTERM: process.listeners("SIGTERM"), + } + const running = runApi({ ...options, logger }) + void running.catch(() => {}) + const server = vi.mocked(createServer).mock.results.at(-1)?.value as Server + + function signal(name: "SIGINT" | "SIGTERM") { + const handler = process.listeners(name).find((listener) => !signals[name].includes(listener)) + if (handler) Reflect.apply(handler, process, [name]) + } + + onTestFinished(async () => { + signal("SIGTERM") + server.closeAllConnections() + await running.catch(() => {}) + if (server.listening) { + await new Promise((resolve) => { + server.close(() => resolve()) + }) + } + }) + + return { server, running, signal, signals, readiness } +} + +describe("runApi()", () => { + it("attaches realtime to the HTTP server, injects broadcast, and forbids upgrades without access policies", async () => { + const { server, running, signal } = await startApi({ port: 0 }) + await once(server, "listening") + const { port } = server.address() as AddressInfo + const options = vi.mocked(attachRealtimeServer).mock.calls.at(-1)?.[0] + const realtime = vi.mocked(attachRealtimeServer).mock.results.at(-1)?.value as RealtimeServer + expect(options).toMatchObject({ server, path: "/api/realtime" }) + expect(options?.authenticate).toBeUndefined() + expect(options?.authorizeSubscription).toBeUndefined() + expect(createApp).toHaveBeenCalledWith( + expect.objectContaining({ broadcast: realtime.broadcast }), + ) + const socket = new WebSocket(`ws://127.0.0.1:${port}/api/realtime`) + socket.on("error", () => {}) + onTestFinished(() => { + socket.terminate() + }) + + expect(await rejected(socket)).toBe(403) + signal("SIGTERM") + await running + + expect(server.listenerCount("upgrade")).toBe(0) + }) + + it("stops upgrades and waits for realtime cleanup before resolving", async () => { + const finished = deferred() + const { attachRealtimeServer: attach } = await vi.importActual( + "../integrations/ws/attach-realtime-server", + ) + const close = vi.fn() + vi.mocked(attachRealtimeServer).mockImplementationOnce((options) => { + const realtime = attach(options) + close.mockImplementation(async () => { + await realtime.close() + await finished.promise + }) + return { broadcast: realtime.broadcast, close } + }) + const { server, running, signal } = await startApi({ port: 0 }) + await once(server, "listening") + let settled = false + void running.then( + () => { + settled = true + }, + () => { + settled = true + }, + ) + + try { + signal("SIGTERM") + await vi.waitFor(() => expect(close).toHaveBeenCalledTimes(1)) + await new Promise((resolve) => setImmediate(resolve)) + + expect(settled).toBe(false) + expect(server.listenerCount("upgrade")).toBe(0) + expect(server.listening).toBe(false) + } finally { + finished.resolve() + await running + } + + expect(settled).toBe(true) + }) + + it("serves health and drains an active request before resolving on SIGTERM", async () => { + const { server, running, signal, signals, readiness } = await startApi({ port: 0 }) + await once(server, "listening") + const { port, address } = server.address() as AddressInfo + expect(address).toBe("127.0.0.1") + expect(readiness).toHaveBeenCalledWith( + expect.objectContaining({ host: "127.0.0.1", port }), + expect.any(String), + ) + const health = await fetch(`http://127.0.0.1:${port}/api/health`, { + headers: { connection: "close" }, + }) + expect(health.status).toBe(200) + expect(await health.json()).toEqual({ status: "ok" }) + + const incoming = once(server, "request") + const client = request({ + host: "127.0.0.1", + port, + path: "/api/health", + method: "POST", + headers: { + connection: "close", + "content-type": "application/json", + "content-length": 2, + }, + }) + client.on("error", () => {}) + onTestFinished(() => { + client.destroy() + }) + const response = new Promise((resolve, reject) => { + client.once("response", resolve) + client.once("error", reject) + }) + void response.catch(() => {}) + client.write("{") + await incoming + let settled = false + void running.then(() => { + settled = true + }) + + signal("SIGTERM") + await new Promise((resolve) => setImmediate(resolve)) + expect(settled).toBe(false) + client.end("}") + const received = await response + expect(received.statusCode).toBe(404) + received.resume() + await once(received, "end") + await running + + expect(settled).toBe(true) + expect(server.listening).toBe(false) + expect(process.listeners("SIGINT")).toEqual(signals.SIGINT) + expect(process.listeners("SIGTERM")).toEqual(signals.SIGTERM) + }) + + it("rejects bind failure and removes its signal listeners", async () => { + const occupied = createServer() + onTestFinished( + () => + new Promise((resolve) => { + occupied.close(() => resolve()) + }), + ) + occupied.listen(0, "127.0.0.1") + await once(occupied, "listening") + const { port } = occupied.address() as AddressInfo + const { running, server, signals } = await startApi({ port }) + + await expect(running).rejects.toMatchObject({ code: "EADDRINUSE" }) + + expect(server.listening).toBe(false) + expect(server.listenerCount("upgrade")).toBe(0) + expect(process.listeners("SIGINT")).toEqual(signals.SIGINT) + expect(process.listeners("SIGTERM")).toEqual(signals.SIGTERM) + }) + + it("bounds draining, rejects the timeout, and closes unfinished connections", async () => { + const { running, server, signal, signals } = await startApi({ + port: 0, + shutdownTimeoutMs: 30, + }) + await once(server, "listening") + const { port } = server.address() as AddressInfo + const incoming = once(server, "request") + const client = request({ + host: "127.0.0.1", + port, + path: "/api/health", + method: "POST", + headers: { connection: "close", "content-type": "application/json", "content-length": 2 }, + }) + client.on("error", () => {}) + onTestFinished(() => { + client.destroy() + }) + const disconnected = new Promise((resolve) => { + client.once("close", () => resolve()) + }) + const serverClosed = once(server, "close") + client.write("{") + await incoming + + signal("SIGTERM") + await expect(running).rejects.toThrow(/shutdown|deadline|timeout/i) + await Promise.all([disconnected, serverClosed]) + + expect(server.listening).toBe(false) + expect(process.listeners("SIGINT")).toEqual(signals.SIGINT) + expect(process.listeners("SIGTERM")).toEqual(signals.SIGTERM) + }) + + it("does not leave a server listening when a signal arrives before binding completes", async () => { + const server = new Server() + let listening = false + vi.spyOn(server, "listening", "get").mockImplementation(() => listening) + const listen = vi.spyOn(server, "listen").mockImplementation(() => server) + const close = vi.spyOn(server, "close").mockImplementation((callback) => { + listening = false + queueMicrotask(() => { + server.emit("close") + callback?.() + }) + return server + }) + vi.mocked(createServer).mockReturnValueOnce(server) + const { running, signal, signals } = await startApi({ port: 0 }) + let settled = false + void running.then(() => { + settled = true + }) + + signal("SIGINT") + await new Promise((resolve) => setImmediate(resolve)) + const returnedWithoutCancelling = settled && close.mock.calls.length === 0 + if (close.mock.calls.length === 0) { + listening = true + server.emit("listening") + const callback: unknown = listen.mock.calls[0]?.find( + (argument) => typeof argument === "function", + ) + if (typeof callback === "function") Reflect.apply(callback, server, []) + } + await running + + expect(returnedWithoutCancelling).toBe(false) + expect(server.listening).toBe(false) + expect(process.listeners("SIGINT")).toEqual(signals.SIGINT) + expect(process.listeners("SIGTERM")).toEqual(signals.SIGTERM) + }) +}) diff --git a/hub/server/run-api/index.ts b/hub/server/run-api/index.ts new file mode 100644 index 00000000..c41852e6 --- /dev/null +++ b/hub/server/run-api/index.ts @@ -0,0 +1,73 @@ +import { createServer } from "node:http" +import type { Logger } from "pino" + +import { createApp } from "../create-app/index.ts" +import { attachRealtimeServer } from "../integrations/ws/attach-realtime-server/index.ts" +import { shutdownServer } from "../shutdown-server/index.ts" + +export async function runApi({ + port, + logger, + shutdownTimeoutMs = 30_000, +}: { + port: number + logger: Logger + shutdownTimeoutMs?: number +}) { + if (!Number.isInteger(port) || port < 0 || port > 65535) { + throw new Error("API port must be an integer between 0 and 65535.") + } + if (!Number.isInteger(shutdownTimeoutMs) || shutdownTimeoutMs < 1) { + throw new Error("API shutdown timeout must be a positive integer.") + } + const server = createServer() + const realtime = attachRealtimeServer({ server, path: "/api/realtime", logger }) + server.on("request", createApp({ logger, broadcast: realtime.broadcast })) + let finishBinding!: () => void + const binding = new Promise((resolve) => { + finishBinding = resolve + }) + server.once("listening", finishBinding) + server.once("error", finishBinding) + let stop!: () => void + const stopping = new Promise((resolve, reject) => { + stop = resolve + server.once("error", reject) + }) + process.once("SIGINT", stop) + process.once("SIGTERM", stop) + try { + server.listen(port, "127.0.0.1", () => { + const address = server.address() + logger.info( + { host: "127.0.0.1", port: typeof address === "object" ? address?.port : port }, + "Hub API listening", + ) + }) + await stopping + } finally { + process.removeListener("SIGINT", stop) + process.removeListener("SIGTERM", stop) + let timer: ReturnType | undefined + async function drain() { + // A stop signal can arrive before listen() has opened its socket. + await binding + await shutdownServer(server, realtime) + } + try { + await Promise.race([ + drain(), + new Promise((_, reject) => { + timer = setTimeout(() => { + server.closeAllConnections() + reject(new Error("API shutdown exceeded its deadline.")) + }, shutdownTimeoutMs) + }), + ]) + } finally { + clearTimeout(timer) + server.removeListener("listening", finishBinding) + server.removeListener("error", finishBinding) + } + } +} diff --git a/hub/server/shutdown-server/index.test.ts b/hub/server/shutdown-server/index.test.ts new file mode 100644 index 00000000..eb986f3b --- /dev/null +++ b/hub/server/shutdown-server/index.test.ts @@ -0,0 +1,64 @@ +import { Server } from "node:http" +import { describe, expect, it, vi } from "vitest" + +import type { RealtimeServer } from "../integrations/ws/types" + +import { shutdownServer } from "." + +describe("shutdownServer()", () => { + it.each(["HTTP", "realtime"])( + "waits for both cleanup operations before rejecting a safe error after %s fails", + async (failingService) => { + const server = new Server() + vi.spyOn(server, "listening", "get").mockReturnValue(true) + const http: { finish(error?: Error): void } = { finish() {} } + const closeHttp = vi.spyOn(server, "close").mockImplementation((callback) => { + if (callback) http.finish = callback + return server + }) + const websocket: { finish(): void; fail(error: Error): void } = { + finish() {}, + fail() {}, + } + const realtime: RealtimeServer = { + broadcast: vi.fn(), + close: vi.fn( + () => + new Promise((resolve, reject) => { + websocket.finish = resolve + websocket.fail = reject + }), + ), + } + const shuttingDown = shutdownServer(server, realtime) + void shuttingDown.catch(() => {}) + let settled = false + void shuttingDown.then( + () => { + settled = true + }, + () => { + settled = true + }, + ) + const failure = new Error("secret-transport-error https://private.example") + await Promise.resolve() + + try { + expect(closeHttp).toHaveBeenCalledTimes(1) + expect(realtime.close).toHaveBeenCalledTimes(1) + if (failingService === "HTTP") http.finish(failure) + else websocket.fail(failure) + await new Promise((resolve) => setImmediate(resolve)) + + expect(settled).toBe(false) + } finally { + if (failingService === "HTTP") websocket.finish() + else http.finish() + } + + await expect(shuttingDown).rejects.toBeInstanceOf(Error) + await expect(shuttingDown).rejects.not.toThrow(/secret|https:|private\.example/) + }, + ) +}) diff --git a/hub/server/shutdown-server/index.ts b/hub/server/shutdown-server/index.ts index f24d441c..da72ecac 100644 --- a/hub/server/shutdown-server/index.ts +++ b/hub/server/shutdown-server/index.ts @@ -1,16 +1,20 @@ import type { Server } from "node:http" -import { log } from "../integrations/pino/log/index.ts" import type { RealtimeServer } from "../integrations/ws/types.ts" export async function shutdownServer(server: Server, realtime: RealtimeServer) { const httpClosed = new Promise((resolve, reject) => { + if (!server.listening) { + resolve() + return + } server.close((error) => (error ? reject(error) : resolve())) }) - try { - await Promise.all([realtime.close(), httpClosed]) - } catch { - log("error", "Hub API shutdown failed") - process.exitCode = 1 + const results = await Promise.allSettled([ + httpClosed, + Promise.resolve().then(() => realtime.close()), + ]) + if (results.some((result) => result.status === "rejected")) { + throw new Error("Hub API shutdown failed.") } } diff --git a/hub/server/workers/index.ts b/hub/server/workers/index.ts index 743cc450..5f0cefd8 100644 --- a/hub/server/workers/index.ts +++ b/hub/server/workers/index.ts @@ -1,6 +1,7 @@ import "dotenv/config" import { z } from "zod" +import { startTelemetry } from "../integrations/opentelemetry/start-telemetry/index.ts" import { logger } from "../integrations/pino/index.ts" import { jobProcessors } from "./configuration.ts" import { runJobWorker } from "./run-job-worker/index.ts" @@ -15,11 +16,16 @@ try { .min(1) .max(100) .parse(process.env[variable] ?? 1) - await runJobWorker({ name, databaseUrl, concurrency, logger, processors: jobProcessors[name] }) + const telemetry = startTelemetry(process.env, logger) + try { + await runJobWorker({ name, databaseUrl, concurrency, logger, processors: jobProcessors[name] }) + } finally { + await telemetry.shutdown() + } } catch (error) { logger.fatal( { errorType: error instanceof Error ? error.name : "UnknownError" }, - "Job worker failed; check database access, worker configuration, and shutdown time", + "Job worker failed; check database access, worker configuration, telemetry, and shutdown time", ) process.exit(1) }