From 2e59434ac12e170d5836df6eee1361d2d75f0a38 Mon Sep 17 00:00:00 2001 From: Copilot <223556219+Copilot@users.noreply.github.com> Date: Fri, 30 Oct 2026 16:21:25 +0000 Subject: [PATCH 1/2] feat(hub): add explicit telemetry foundation Co-authored-by: Nishiki <1121087+nshki@users.noreply.github.com> Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- hub/.env.example | 4 + hub/AGENTS.md | 1 + hub/docs/README.md | 1 + hub/docs/architecture.md | 2 + hub/docs/development.md | 2 + hub/docs/telemetry.md | 99 +++++ hub/package-lock.json | 220 ++++++++++++ hub/package.json | 5 + hub/server/index.ts | 19 +- .../start-telemetry/index.test.ts | 337 ++++++++++++++++++ .../opentelemetry/start-telemetry/index.ts | 191 ++++++++++ .../opentelemetry/start-telemetry/types.ts | 7 + .../with-telemetry-span/index.test.ts | 223 ++++++++++++ .../with-telemetry-span/index.ts | 42 +++ .../with-telemetry-span/types.ts | 7 + hub/server/run-api/index.test.ts | 202 +++++++++++ hub/server/run-api/index.ts | 72 ++++ hub/server/workers/index.ts | 10 +- 18 files changed, 1437 insertions(+), 7 deletions(-) create mode 100644 hub/docs/telemetry.md create mode 100644 hub/server/integrations/opentelemetry/start-telemetry/index.test.ts create mode 100644 hub/server/integrations/opentelemetry/start-telemetry/index.ts create mode 100644 hub/server/integrations/opentelemetry/start-telemetry/types.ts create mode 100644 hub/server/integrations/opentelemetry/with-telemetry-span/index.test.ts create mode 100644 hub/server/integrations/opentelemetry/with-telemetry-span/index.ts create mode 100644 hub/server/integrations/opentelemetry/with-telemetry-span/types.ts create mode 100644 hub/server/run-api/index.test.ts create mode 100644 hub/server/run-api/index.ts 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 5d471ed3..24ee4c74 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. - [Testing and operations](testing-and-operations.md): baseline checks and their limits. - [Storybook](storybook.md): retained presentation, fixtures, and screenshot evidence. diff --git a/hub/docs/architecture.md b/hub/docs/architecture.md index bf72e320..70ffae10 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. + ## Frontend ```mermaid diff --git a/hub/docs/development.md b/hub/docs/development.md index 452682d3..0cc359bd 100644 --- a/hub/docs/development.md +++ b/hub/docs/development.md @@ -40,6 +40,8 @@ Open . Vite serves the frontend and forwards `/api` reque 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..64276b7c --- /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 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/package-lock.json b/hub/package-lock.json index 25ceab39..b25fb44e 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", @@ -1941,6 +1946,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 259f8e00..d4c712bd 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 64bafe13..e1fb35ce 100644 --- a/hub/server/index.ts +++ b/hub/server/index.ts @@ -1,13 +1,22 @@ import "dotenv/config" -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 { 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.") } -createApp().listen(port, "127.0.0.1", () => { - log("info", "Hub API listening", { host: "127.0.0.1", port }) -}) +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) +} 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..aa150394 --- /dev/null +++ b/hub/server/integrations/opentelemetry/start-telemetry/index.test.ts @@ -0,0 +1,337 @@ +import { context, diag, propagation, ROOT_CONTEXT, trace, TraceFlags } 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_TRACES_EXPORTER: "none", OTEL_EXPORTER_OTLP_ENDPOINT: "invalid" }, + { 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.each([ + ["https://collector.example", "https://collector.example/v1/traces"], + ["http://collector.example/base/", "http://collector.example/base/v1/traces"], + ])("appends the trace path to a common endpoint %s", async (endpoint, expected) => { + 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: endpoint, + 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: expected, 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", {}], + ["relative endpoint", { OTEL_EXPORTER_OTLP_ENDPOINT: "/secret-relative" }], + ["non-HTTP endpoint", { OTEL_EXPORTER_OTLP_ENDPOINT: "file:///secret-path" }], + ["credentials", { OTEL_EXPORTER_OTLP_ENDPOINT: "https://secret-user:secret-pass@example.com" }], + ["query", { OTEL_EXPORTER_OTLP_ENDPOINT: "https://example.com?token=secret-query" }], + ["fragment", { OTEL_EXPORTER_OTLP_ENDPOINT: "https://example.com/#secret-fragment" }], + ["trace endpoint without path", { OTEL_EXPORTER_OTLP_TRACES_ENDPOINT: "https://example.com" }], + ["unsupported protocol", { OTEL_EXPORTER_OTLP_PROTOCOL: "secret-protocol" }], + ["invalid trace protocol", { OTEL_EXPORTER_OTLP_TRACES_PROTOCOL: "secret-trace-protocol" }], + ["missing header assignment", { OTEL_EXPORTER_OTLP_HEADERS: "secret-header" }], + ["blank header name", { OTEL_EXPORTER_OTLP_HEADERS: "=secret-value" }], + ["blank header value", { OTEL_EXPORTER_OTLP_HEADERS: "secret-header=" }], + ["malformed percent encoding", { OTEL_EXPORTER_OTLP_HEADERS: "x-token=%secret" }], + ["encoded header newline", { OTEL_EXPORTER_OTLP_HEADERS: "x-token=secret%0D%0Avalue" }], + ["literal header newline", { OTEL_EXPORTER_OTLP_HEADERS: "x-token=secret\r\nvalue" }], + ["unsupported sampler", { OTEL_TRACES_SAMPLER: "secret-sampler" }], + ["nonfinite ratio", { OTEL_TRACES_SAMPLER_ARG: "Infinity" }], + ["negative ratio", { OTEL_TRACES_SAMPLER_ARG: "-0.1" }], + ["ratio greater than one", { OTEL_TRACES_SAMPLER_ARG: "1.1" }], + ["nonnumeric ratio", { OTEL_TRACES_SAMPLER_ARG: "secret-ratio" }], + ])("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:|file:|Infinity|-0\.1|1\.1/) + expect(JSON.stringify([logger.error.mock.calls, logger.warn.mock.calls])).not.toContain( + "secret", + ) + expect(OTLPTraceExporter).not.toHaveBeenCalled() + }) + + it.each([ + [undefined, undefined, true], + ["parentbased_traceidratio", "0", false], + ["parentbased_traceidratio", "1", true], + ["always_on", undefined, true], + ["always_off", undefined, false], + ])("applies sampler %s with ratio %s", async (sampler, ratio, sampled) => { + 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(sampled ? 1 : 0) + }) + + it("honors an unsampled parent rather than applying the root ratio again", 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: "https://collector.example", + }, + { error: vi.fn(), warn: vi.fn() }, + ) + onTestFinished(() => telemetry.shutdown()) + const parent = trace.wrapSpanContext({ + traceId: "12345678901234567890123456789012", + spanId: "1234567890123456", + traceFlags: TraceFlags.NONE, + isRemote: true, + }) + + await context.with(trace.setSpan(context.active(), parent), () => + withTelemetrySpan("test.child", {}, () => {}), + ) + await telemetry.shutdown() + + expect(transport.spans).toHaveLength(0) + }) + + it("rejects duplicate initialization and releases its globals without adding signal handlers", 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 signals = [process.listenerCount("SIGINT"), process.listenerCount("SIGTERM")] + const telemetry = startTelemetry(env, logger) + onTestFinished(() => telemetry.shutdown()) + + expect(() => startTelemetry(env, logger)).toThrow() + expect([process.listenerCount("SIGINT"), process.listenerCount("SIGTERM")]).toEqual(signals) + expect(propagation.fields()).toEqual([]) + expect(transport.spans).toHaveLength(0) + await withTelemetrySpan("test.first", {}, () => {}) + await Promise.all([telemetry.shutdown(), telemetry.shutdown()]) + expect(transport.shutdown).toHaveBeenCalledTimes(1) + await withTelemetrySpan("test.disabled", {}, () => {}) + + const restarted = startTelemetry(env, logger) + onTestFinished(() => restarted.shutdown()) + await withTelemetrySpan("test.restarted", {}, () => {}) + await restarted.shutdown() + + expect(transport.spans.map((span) => span.name)).toEqual(["test.first", "test.restarted"]) + 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 safely and permits a later initialization", 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]) + + transport.shutdown.mockResolvedValue() + const restarted = startTelemetry(env, logger) + onTestFinished(() => restarted.shutdown()) + await restarted.shutdown() + }) + + 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..07bb3d71 --- /dev/null +++ b/hub/server/integrations/opentelemetry/with-telemetry-span/index.test.ts @@ -0,0 +1,223 @@ +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([ + ["empty name", "", {}], + ["uppercase name", "Test.operation", {}], + ["name containing private data", "test.secret@example.com", {}], + ["repeated separators", "test..operation", {}], + ["trailing separator", "test.operation-", {}], + ["overlong name", "a".repeat(101), {}], + ["unknown attribute", "test.operation", { "http.url": "https://secret.example" }], + ["invalid UUID", "test.operation", { "replica.repository.id": "secret-repository" }], + ["non-string ID", "test.operation", { "replica.build.id": 42 }], + ])("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:|Test\.operation|a{101}|42/) + expect(operation).not.toHaveBeenCalled() + expect(OTLPTraceExporter).not.toHaveBeenCalled() + }) + + it("exports only allowed IDs, accepts boundary names, 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("a".repeat(100), attributes, () => Promise.resolve(result)), + ).resolves.toBe(result) + await withTelemetrySpan("test.operation_part-2", {}, () => {}) + await telemetry.shutdown() + + expect(spans).toHaveLength(2) + expect(spans[0]?.attributes).toEqual(attributes) + expect(spans[1]?.attributes).toEqual({}) + for (const span of spans) { + expect(span.status).toEqual({ code: SpanStatusCode.OK }) + expect(span.ended).toBe(true) + expect(span.events).toEqual([]) + } + 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.each(["synchronous Error", "asynchronous non-Error"])( + "rethrows an identical %s, ends the span, and records no failure details", + async (failureKind) => { + 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 = + failureKind === "synchronous Error" + ? Object.assign(new Error("secret-error-message"), { token: "secret-error-token" }) + : { private: "secret-rejection" } + + const result = withTelemetrySpan("test.failure", {}, () => { + if (failure instanceof Error) throw failure + // The public wrapper must preserve arbitrary JavaScript rejection values. + // eslint-disable-next-line @typescript-eslint/prefer-promise-reject-errors + return Promise.reject(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..9b01570a --- /dev/null +++ b/hub/server/run-api/index.test.ts @@ -0,0 +1,202 @@ +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 { createLogger } from "../integrations/pino/create-logger" + +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.each(["SIGINT", "SIGTERM"] as const)( + "serves health and drains an active request before resolving on %s", + async (signalName) => { + 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(signalName) + 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(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..2c257392 --- /dev/null +++ b/hub/server/run-api/index.ts @@ -0,0 +1,72 @@ +import { createServer } from "node:http" +import type { Logger } from "pino" + +import { createApp } from "../create-app/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(createApp({ logger })) + 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 + if (!server.listening) return + await new Promise((resolve, reject) => { + server.close((error) => (error ? reject(error) : resolve())) + }) + } + 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/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) } From a3fdc18deb8acd6ba97dd7ad6b6b4f2a394cdf41 Mon Sep 17 00:00:00 2001 From: Copilot <223556219+Copilot@users.noreply.github.com> Date: Fri, 30 Oct 2026 17:13:05 +0000 Subject: [PATCH 2/2] test(hub): focus telemetry coverage on critical paths Co-authored-by: Nishiki <1121087+nshki@users.noreply.github.com> Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../start-telemetry/index.test.ts | 117 +++++------------- .../with-telemetry-span/index.test.ts | 112 +++++++---------- hub/server/run-api/index.test.ts | 113 ++++++++--------- 3 files changed, 132 insertions(+), 210 deletions(-) diff --git a/hub/server/integrations/opentelemetry/start-telemetry/index.test.ts b/hub/server/integrations/opentelemetry/start-telemetry/index.test.ts index aa150394..d71bf838 100644 --- a/hub/server/integrations/opentelemetry/start-telemetry/index.test.ts +++ b/hub/server/integrations/opentelemetry/start-telemetry/index.test.ts @@ -1,4 +1,4 @@ -import { context, diag, propagation, ROOT_CONTEXT, trace, TraceFlags } from "@opentelemetry/api" +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" @@ -50,23 +50,22 @@ beforeEach(resetGlobals) afterEach(resetGlobals) describe("startTelemetry()", () => { - it.each([ - {}, - { OTEL_TRACES_EXPORTER: "none", OTEL_EXPORTER_OTLP_ENDPOINT: "invalid" }, - { 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()) + 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() - }) + 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() @@ -103,17 +102,14 @@ describe("startTelemetry()", () => { expect(JSON.stringify(transport.spans[0]?.resource.attributes)).not.toContain("resource-secret") }) - it.each([ - ["https://collector.example", "https://collector.example/v1/traces"], - ["http://collector.example/base/", "http://collector.example/base/v1/traces"], - ])("appends the trace path to a common endpoint %s", async (endpoint, expected) => { + 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: endpoint, + OTEL_EXPORTER_OTLP_ENDPOINT: "http://collector.example/base/", OTEL_EXPORTER_OTLP_HEADERS: "x-token=test%20value", }, { error: vi.fn(), warn: vi.fn() }, @@ -124,7 +120,10 @@ describe("startTelemetry()", () => { await telemetry.shutdown() expect(OTLPTraceExporter).toHaveBeenCalledWith( - expect.objectContaining({ url: expected, headers: { "x-token": "test value" } }), + 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") }) @@ -132,25 +131,11 @@ describe("startTelemetry()", () => { it.each([ ["unsupported exporter", { OTEL_TRACES_EXPORTER: "secret-exporter" }], ["missing endpoint", {}], - ["relative endpoint", { OTEL_EXPORTER_OTLP_ENDPOINT: "/secret-relative" }], - ["non-HTTP endpoint", { OTEL_EXPORTER_OTLP_ENDPOINT: "file:///secret-path" }], ["credentials", { OTEL_EXPORTER_OTLP_ENDPOINT: "https://secret-user:secret-pass@example.com" }], ["query", { OTEL_EXPORTER_OTLP_ENDPOINT: "https://example.com?token=secret-query" }], - ["fragment", { OTEL_EXPORTER_OTLP_ENDPOINT: "https://example.com/#secret-fragment" }], - ["trace endpoint without path", { OTEL_EXPORTER_OTLP_TRACES_ENDPOINT: "https://example.com" }], ["unsupported protocol", { OTEL_EXPORTER_OTLP_PROTOCOL: "secret-protocol" }], - ["invalid trace protocol", { OTEL_EXPORTER_OTLP_TRACES_PROTOCOL: "secret-trace-protocol" }], - ["missing header assignment", { OTEL_EXPORTER_OTLP_HEADERS: "secret-header" }], - ["blank header name", { OTEL_EXPORTER_OTLP_HEADERS: "=secret-value" }], - ["blank header value", { OTEL_EXPORTER_OTLP_HEADERS: "secret-header=" }], - ["malformed percent encoding", { OTEL_EXPORTER_OTLP_HEADERS: "x-token=%secret" }], ["encoded header newline", { OTEL_EXPORTER_OTLP_HEADERS: "x-token=secret%0D%0Avalue" }], - ["literal header newline", { OTEL_EXPORTER_OTLP_HEADERS: "x-token=secret\r\nvalue" }], - ["unsupported sampler", { OTEL_TRACES_SAMPLER: "secret-sampler" }], - ["nonfinite ratio", { OTEL_TRACES_SAMPLER_ARG: "Infinity" }], - ["negative ratio", { OTEL_TRACES_SAMPLER_ARG: "-0.1" }], ["ratio greater than one", { OTEL_TRACES_SAMPLER_ARG: "1.1" }], - ["nonnumeric ratio", { OTEL_TRACES_SAMPLER_ARG: "secret-ratio" }], ])("rejects %s synchronously without leaking supplied configuration", async (_, invalid) => { const logger = { error: vi.fn(), warn: vi.fn() } const { startTelemetry } = await import(".") @@ -170,7 +155,7 @@ describe("startTelemetry()", () => { } expect(thrown).toBeInstanceOf(Error) - expect(String(thrown)).not.toMatch(/secret|https:|file:|Infinity|-0\.1|1\.1/) + expect(String(thrown)).not.toMatch(/secret|https:|1\.1/) expect(JSON.stringify([logger.error.mock.calls, logger.warn.mock.calls])).not.toContain( "secret", ) @@ -178,12 +163,9 @@ describe("startTelemetry()", () => { }) it.each([ - [undefined, undefined, true], - ["parentbased_traceidratio", "0", false], - ["parentbased_traceidratio", "1", true], - ["always_on", undefined, true], - ["always_off", undefined, false], - ])("applies sampler %s with ratio %s", async (sampler, ratio, sampled) => { + ["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") @@ -201,37 +183,10 @@ describe("startTelemetry()", () => { await withTelemetrySpan("test.sample", {}, () => {}) await telemetry.shutdown() - expect(transport.spans).toHaveLength(sampled ? 1 : 0) - }) - - it("honors an unsampled parent rather than applying the root ratio again", 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: "https://collector.example", - }, - { error: vi.fn(), warn: vi.fn() }, - ) - onTestFinished(() => telemetry.shutdown()) - const parent = trace.wrapSpanContext({ - traceId: "12345678901234567890123456789012", - spanId: "1234567890123456", - traceFlags: TraceFlags.NONE, - isRemote: true, - }) - - await context.with(trace.setSpan(context.active(), parent), () => - withTelemetrySpan("test.child", {}, () => {}), - ) - await telemetry.shutdown() - expect(transport.spans).toHaveLength(0) }) - it("rejects duplicate initialization and releases its globals without adding signal handlers", async () => { + it("rejects duplicate initialization and shuts down idempotently", async () => { const transport = mockTransport() const { startTelemetry } = await import(".") const { withTelemetrySpan } = await import("../with-telemetry-span") @@ -240,25 +195,16 @@ describe("startTelemetry()", () => { OTEL_EXPORTER_OTLP_ENDPOINT: "https://collector.example", } const logger = { error: vi.fn(), warn: vi.fn() } - const signals = [process.listenerCount("SIGINT"), process.listenerCount("SIGTERM")] const telemetry = startTelemetry(env, logger) onTestFinished(() => telemetry.shutdown()) expect(() => startTelemetry(env, logger)).toThrow() - expect([process.listenerCount("SIGINT"), process.listenerCount("SIGTERM")]).toEqual(signals) - expect(propagation.fields()).toEqual([]) - expect(transport.spans).toHaveLength(0) await withTelemetrySpan("test.first", {}, () => {}) await Promise.all([telemetry.shutdown(), telemetry.shutdown()]) expect(transport.shutdown).toHaveBeenCalledTimes(1) await withTelemetrySpan("test.disabled", {}, () => {}) - const restarted = startTelemetry(env, logger) - onTestFinished(() => restarted.shutdown()) - await withTelemetrySpan("test.restarted", {}, () => {}) - await restarted.shutdown() - - expect(transport.spans.map((span) => span.name)).toEqual(["test.first", "test.restarted"]) + expect(transport.spans.map((span) => span.name)).toEqual(["test.first"]) expect(context.active()).toBe(ROOT_CONTEXT) }) @@ -287,7 +233,7 @@ describe("startTelemetry()", () => { expectSafeLogs([...logger.error.mock.calls, ...logger.warn.mock.calls]) }) - it("reports shutdown rejection safely and permits a later initialization", async () => { + 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() } @@ -303,11 +249,6 @@ describe("startTelemetry()", () => { 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]) - - transport.shutdown.mockResolvedValue() - const restarted = startTelemetry(env, logger) - onTestFinished(() => restarted.shutdown()) - await restarted.shutdown() }) it("bounds shutdown at five seconds when an exporter never settles", async () => { diff --git a/hub/server/integrations/opentelemetry/with-telemetry-span/index.test.ts b/hub/server/integrations/opentelemetry/with-telemetry-span/index.test.ts index 07bb3d71..da9919b1 100644 --- a/hub/server/integrations/opentelemetry/with-telemetry-span/index.test.ts +++ b/hub/server/integrations/opentelemetry/with-telemetry-span/index.test.ts @@ -46,15 +46,9 @@ describe("withTelemetrySpan()", () => { }) it.each([ - ["empty name", "", {}], - ["uppercase name", "Test.operation", {}], ["name containing private data", "test.secret@example.com", {}], - ["repeated separators", "test..operation", {}], - ["trailing separator", "test.operation-", {}], - ["overlong name", "a".repeat(101), {}], ["unknown attribute", "test.operation", { "http.url": "https://secret.example" }], ["invalid UUID", "test.operation", { "replica.repository.id": "secret-repository" }], - ["non-string ID", "test.operation", { "replica.build.id": 42 }], ])("rejects %s before running an operation, even when disabled", async (_, name, attributes) => { const { withTelemetrySpan } = await import(".") const operation = vi.fn() @@ -71,12 +65,12 @@ describe("withTelemetrySpan()", () => { } expect(thrown).toBeInstanceOf(Error) - expect(String(thrown)).not.toMatch(/secret|https:|Test\.operation|a{101}|42/) + expect(String(thrown)).not.toMatch(/secret|https:/) expect(operation).not.toHaveBeenCalled() expect(OTLPTraceExporter).not.toHaveBeenCalled() }) - it("exports only allowed IDs, accepts boundary names, and ends successful spans", async () => { + 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(".") @@ -98,19 +92,17 @@ describe("withTelemetrySpan()", () => { const result = { preserved: true } await expect( - withTelemetrySpan("a".repeat(100), attributes, () => Promise.resolve(result)), + withTelemetrySpan("build.import", attributes, () => Promise.resolve(result)), ).resolves.toBe(result) - await withTelemetrySpan("test.operation_part-2", {}, () => {}) await telemetry.shutdown() - expect(spans).toHaveLength(2) - expect(spans[0]?.attributes).toEqual(attributes) - expect(spans[1]?.attributes).toEqual({}) - for (const span of spans) { - expect(span.status).toEqual({ code: SpanStatusCode.OK }) - expect(span.ended).toBe(true) - expect(span.events).toEqual([]) - } + expect(spans).toHaveLength(1) + expect(spans[0]).toMatchObject({ + attributes, + status: { code: SpanStatusCode.OK }, + ended: true, + events: [], + }) expect(trace.getActiveSpan()).toBeUndefined() }) @@ -173,51 +165,43 @@ describe("withTelemetrySpan()", () => { expect(context.active()).toBe(ROOT_CONTEXT) }) - it.each(["synchronous Error", "asynchronous non-Error"])( - "rethrows an identical %s, ends the span, and records no failure details", - async (failureKind) => { - 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 = - failureKind === "synchronous Error" - ? Object.assign(new Error("secret-error-message"), { token: "secret-error-token" }) - : { private: "secret-rejection" } - - const result = withTelemetrySpan("test.failure", {}, () => { - if (failure instanceof Error) throw failure - // The public wrapper must preserve arbitrary JavaScript rejection values. - // eslint-disable-next-line @typescript-eslint/prefer-promise-reject-errors - return Promise.reject(failure) - }) + 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") - 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() - }, - ) + 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/run-api/index.test.ts b/hub/server/run-api/index.test.ts index 5a9a1530..b564915e 100644 --- a/hub/server/run-api/index.test.ts +++ b/hub/server/run-api/index.test.ts @@ -121,67 +121,64 @@ describe("runApi()", () => { expect(settled).toBe(true) }) - it.each(["SIGINT", "SIGTERM"] as const)( - "serves health and drains an active request before resolving on %s", - async (signalName) => { - 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" }) + 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 - }) + 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(signalName) - 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 + 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) - }, - ) + 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()