diff --git a/apps/api/package.json b/apps/api/package.json index 6ec29cf..f6bd0a6 100644 --- a/apps/api/package.json +++ b/apps/api/package.json @@ -31,6 +31,10 @@ "dependencies": { "@flakemetry/contracts": "workspace:*", "@flakemetry/db": "workspace:*", + "@opentelemetry/api": "^1.9.0", + "@opentelemetry/exporter-metrics-otlp-http": "^0.220.0", + "@opentelemetry/resources": "^2.0.0", + "@opentelemetry/sdk-metrics": "^2.0.0", "@trpc/server": "^11.0.0", "fastify": "^5.2.0" } diff --git a/apps/api/src/__tests__/hardening.test.ts b/apps/api/src/__tests__/hardening.test.ts new file mode 100644 index 0000000..626ae32 --- /dev/null +++ b/apps/api/src/__tests__/hardening.test.ts @@ -0,0 +1,157 @@ +import { gzipSync } from 'node:zlib' + +import { RESOURCE_ATTR, SPAN_ATTR, SPAN_NAMES } from '@flakemetry/contracts' +import { generateToken, hashToken, PrismaClient } from '@flakemetry/db' +import { afterAll, beforeEach, describe, expect, it } from 'vitest' + +import { buildApp } from '../app' + +const hasDb = Boolean(process.env.DATABASE_URL) +const prisma = new PrismaClient() + +const validBatch = (idempotencyKey: string) => ({ + contractVersion: '0.1.0', + idempotencyKey, + resource: { ciProvider: 'github_actions', commitSha: 'abc1234', branch: 'main', trigger: 'push' }, + run: { status: 'passed', startedAt: '2026-07-16T10:00:00Z' }, + executions: [ + { + filePath: 'e2e/login.spec.ts', + suite: 'auth', + title: 'logs in', + status: 'pass', + attempt: 1, + startedAt: '2026-07-16T10:00:01Z', + durationMs: 1200, + }, + ], +}) + +const s = (value: string) => ({ stringValue: value }) +const i = (value: number) => ({ intValue: String(value) }) +const RUN_START = Date.parse('2026-07-16T10:00:00Z') * 1_000_000 +const nano = (ms: number) => String(RUN_START + ms * 1_000_000) + +const otlpRequest = (idempotencyKey: string) => ({ + resourceSpans: [ + { + resource: { + attributes: [ + { key: RESOURCE_ATTR.project, value: s('web') }, + { key: RESOURCE_ATTR.commitSha, value: s('abc1234') }, + { key: RESOURCE_ATTR.branch, value: s('main') }, + { key: RESOURCE_ATTR.ciProvider, value: s('github_actions') }, + { key: RESOURCE_ATTR.trigger, value: s('push') }, + { key: RESOURCE_ATTR.idempotencyKey, value: s(idempotencyKey) }, + ], + }, + scopeSpans: [ + { + spans: [ + { + traceId: 'a'.repeat(32), + spanId: 'b'.repeat(16), + name: SPAN_NAMES.run, + startTimeUnixNano: nano(0), + endTimeUnixNano: nano(5000), + status: { code: 1 }, + }, + { + traceId: 'a'.repeat(32), + spanId: 'c'.repeat(16), + parentSpanId: 'b'.repeat(16), + name: SPAN_NAMES.case, + startTimeUnixNano: nano(10), + endTimeUnixNano: nano(1810), + attributes: [ + { key: SPAN_ATTR.fingerprint, value: s('fp-login') }, + { key: SPAN_ATTR.title, value: s('logs in') }, + { key: SPAN_ATTR.filePath, value: s('e2e/login.spec.ts') }, + { key: SPAN_ATTR.status, value: s('pass') }, + { key: SPAN_ATTR.attempt, value: i(1) }, + { key: SPAN_ATTR.durationMs, value: i(1800) }, + ], + }, + ], + }, + ], + }, + ], +}) + +const seedToken = async () => { + const org = await prisma.org.create({ data: { name: 'Acme', slug: `acme-${Date.now()}` } }) + const project = await prisma.project.create({ data: { orgId: org.id, name: 'Web', slug: 'web' } }) + const raw = generateToken() + await prisma.ingestToken.create({ + data: { orgId: org.id, projectId: project.id, name: 'ci', tokenHash: hashToken(raw) }, + }) + return { raw } +} + +describe.skipIf(!hasDb)('api hardening', () => { + beforeEach(async () => { + await prisma.ingestionJob.deleteMany() + await prisma.ingestToken.deleteMany() + await prisma.project.deleteMany() + await prisma.org.deleteMany() + }) + + afterAll(async () => { + await prisma.$disconnect() + }) + + it('rate limits a token beyond the configured window', async () => { + const { raw } = await seedToken() + const app = buildApp({ prisma, rateLimit: { max: 1, windowMs: 60_000 } }) + const send = (key: string) => + app.inject({ + method: 'POST', + url: '/v1/ingest', + headers: { authorization: `Bearer ${raw}` }, + payload: validBatch(key), + }) + + expect((await send('run-000001')).statusCode).toBe(202) + const limited = await send('run-000002') + expect(limited.statusCode).toBe(429) + expect(limited.headers['retry-after']).toBeDefined() + }) + + it('sheds load with 503 when the queue depth exceeds the limit', async () => { + const { raw } = await seedToken() + const app = buildApp({ prisma, maxQueueDepth: 1 }) + const send = (key: string) => + app.inject({ + method: 'POST', + url: '/v1/ingest', + headers: { authorization: `Bearer ${raw}` }, + payload: validBatch(key), + }) + + expect((await send('run-000001')).statusCode).toBe(202) + const shed = await send('run-000002') + expect(shed.statusCode).toBe(503) + expect(shed.headers['retry-after']).toBeDefined() + }) + + it('decompresses gzip-encoded OTLP request bodies', async () => { + const { raw } = await seedToken() + const app = buildApp({ prisma }) + const body = gzipSync(Buffer.from(JSON.stringify(otlpRequest('run-000001')))) + + const res = await app.inject({ + method: 'POST', + url: '/v1/traces', + headers: { + authorization: `Bearer ${raw}`, + 'content-type': 'application/json', + 'content-encoding': 'gzip', + }, + payload: body, + }) + + expect(res.statusCode).toBe(200) + expect(await prisma.ingestionJob.count()).toBe(1) + }) +}) diff --git a/apps/api/src/__tests__/rate-limit.test.ts b/apps/api/src/__tests__/rate-limit.test.ts new file mode 100644 index 0000000..c08e52d --- /dev/null +++ b/apps/api/src/__tests__/rate-limit.test.ts @@ -0,0 +1,32 @@ +import { describe, expect, it } from 'vitest' + +import { createRateLimiter } from '../rate-limit' + +describe('createRateLimiter', () => { + it('allows up to max requests per window then rejects', () => { + const limiter = createRateLimiter({ max: 2, windowMs: 1_000, now: () => 1_000 }) + + expect(limiter.check('a').allowed).toBe(true) + expect(limiter.check('a').allowed).toBe(true) + const blocked = limiter.check('a') + expect(blocked.allowed).toBe(false) + expect(blocked.retryAfterMs).toBeGreaterThan(0) + }) + + it('resets the counter when the window elapses', () => { + let clock = 1_000 + const limiter = createRateLimiter({ max: 1, windowMs: 1_000, now: () => clock }) + + expect(limiter.check('a').allowed).toBe(true) + expect(limiter.check('a').allowed).toBe(false) + clock += 1_000 + expect(limiter.check('a').allowed).toBe(true) + }) + + it('tracks keys independently', () => { + const limiter = createRateLimiter({ max: 1, windowMs: 1_000, now: () => 0 }) + expect(limiter.check('a').allowed).toBe(true) + expect(limiter.check('b').allowed).toBe(true) + expect(limiter.check('a').allowed).toBe(false) + }) +}) diff --git a/apps/api/src/app.ts b/apps/api/src/app.ts index 281cdb8..dab2f9c 100644 --- a/apps/api/src/app.ts +++ b/apps/api/src/app.ts @@ -1,3 +1,5 @@ +import { createGunzip } from 'node:zlib' + import { ingestRunBatchSchema, otlpToIngestBatch, @@ -5,9 +7,15 @@ import { } from '@flakemetry/contracts' import { IngestionQueue, type PrismaClient } from '@flakemetry/db' import { fastifyTRPCPlugin } from '@trpc/server/adapters/fastify' -import Fastify, { type FastifyInstance } from 'fastify' +import Fastify, { + type FastifyInstance, + type FastifyReply, + type FastifyServerOptions, +} from 'fastify' import { authenticateProject } from './auth' +import { createRateLimiter } from './rate-limit' +import { apiMetrics, observeQueueDepth } from './telemetry' import { createContextFactory } from './trpc/context' import { appRouter } from './trpc/router' @@ -15,16 +23,61 @@ export interface AppOptions { prisma: PrismaClient queue?: IngestionQueue bodyLimitBytes?: number + logger?: FastifyServerOptions['logger'] + maxQueueDepth?: number + rateLimit?: { max: number; windowMs: number } + rateLimitNow?: () => number } +type Admission = { ok: true } | { ok: false; status: number; reason: string; retryAfterMs: number } + export const buildApp = (options: AppOptions): FastifyInstance => { const { prisma } = options const queue = options.queue ?? new IngestionQueue(prisma) const app = Fastify({ - logger: false, + logger: options.logger ?? false, bodyLimit: options.bodyLimitBytes ?? 8 * 1024 * 1024, }) + const limiter = createRateLimiter({ + max: options.rateLimit?.max ?? 600, + windowMs: options.rateLimit?.windowMs ?? 60_000, + now: options.rateLimitNow, + }) + + observeQueueDepth(() => queue.depth()) + + app.addHook('preParsing', async (request, _reply, payload) => { + if (request.headers['content-encoding'] !== 'gzip') return payload + delete request.headers['content-encoding'] + delete request.headers['content-length'] + return payload.pipe(createGunzip()) + }) + + app.addHook('onResponse', async (request, reply) => { + apiMetrics.requestDuration.record(reply.elapsedTime, { + route: request.routeOptions.url ?? request.url, + status: reply.statusCode, + }) + }) + + const admit = async (projectId: string): Promise => { + const decision = limiter.check(projectId) + if (!decision.allowed) { + apiMetrics.rateLimited.add(1) + return { ok: false, status: 429, reason: 'rate_limited', retryAfterMs: decision.retryAfterMs } + } + if (options.maxQueueDepth != null && (await queue.depth()) >= options.maxQueueDepth) { + apiMetrics.backpressured.add(1) + return { ok: false, status: 503, reason: 'backpressure', retryAfterMs: 1_000 } + } + return { ok: true } + } + + const setRetryAfter = (reply: FastifyReply, ms: number): void => { + reply.header('retry-after', Math.max(1, Math.ceil(ms / 1_000))) + } + app.get('/health', async () => ({ status: 'ok', service: 'api' })) void app.register(fastifyTRPCPlugin, { @@ -38,6 +91,12 @@ export const buildApp = (options: AppOptions): FastifyInstance => { return reply.code(401).send({ error: 'unauthorized', message: 'missing or invalid token' }) } + const admission = await admit(project.projectId) + if (!admission.ok) { + setRetryAfter(reply, admission.retryAfterMs) + return reply.code(admission.status).send({ error: admission.reason }) + } + const parsed = ingestRunBatchSchema.safeParse(request.body) if (!parsed.success) { return reply.code(400).send({ @@ -57,6 +116,9 @@ export const buildApp = (options: AppOptions): FastifyInstance => { payload: JSON.parse(JSON.stringify(batch)), }) + apiMetrics.runsAccepted.add(1) + apiMetrics.executionsAccepted.add(batch.executions.length) + return reply.code(202).send({ receiptId: jobId, acceptedExecutions: batch.executions.length, @@ -70,6 +132,14 @@ export const buildApp = (options: AppOptions): FastifyInstance => { return reply.code(401).send({ error: 'unauthorized', message: 'missing or invalid token' }) } + const admission = await admit(project.projectId) + if (!admission.ok) { + setRetryAfter(reply, admission.retryAfterMs) + return reply + .code(admission.status) + .send({ partialSuccess: { rejectedSpans: '0', errorMessage: admission.reason } }) + } + const parsedRequest = otlpTraceRequestSchema.safeParse(request.body) if (!parsedRequest.success) { return reply.code(400).send({ @@ -96,6 +166,9 @@ export const buildApp = (options: AppOptions): FastifyInstance => { payload: JSON.parse(JSON.stringify(batch)), }) + apiMetrics.runsAccepted.add(1) + apiMetrics.executionsAccepted.add(batch.executions.length) + return reply.code(200).send({}) }) diff --git a/apps/api/src/index.ts b/apps/api/src/index.ts index a5c7886..6d4dc11 100644 --- a/apps/api/src/index.ts +++ b/apps/api/src/index.ts @@ -1,13 +1,35 @@ import { getPrismaClient } from '@flakemetry/db' import { buildApp } from './app' +import { initSelfTelemetry } from './telemetry' export type { AppRouter } from './app' const port = Number(process.env.PORT ?? 4000) const host = process.env.HOST ?? '0.0.0.0' -const app = buildApp({ prisma: getPrismaClient() }) +const selfOtelEndpoint = process.env.FLAKEMETRY_SELF_OTEL_ENDPOINT +if (selfOtelEndpoint) { + const shutdown = initSelfTelemetry({ endpoint: selfOtelEndpoint }) + for (const signal of ['SIGINT', 'SIGTERM'] as const) { + process.on(signal, () => { + void shutdown() + }) + } +} + +const maxQueueDepth = process.env.FLAKEMETRY_MAX_QUEUE_DEPTH + ? Number(process.env.FLAKEMETRY_MAX_QUEUE_DEPTH) + : undefined + +const app = buildApp({ + prisma: getPrismaClient(), + logger: { + level: process.env.LOG_LEVEL ?? 'info', + redact: ['req.headers.authorization'], + }, + maxQueueDepth, +}) app .listen({ port, host }) diff --git a/apps/api/src/rate-limit.ts b/apps/api/src/rate-limit.ts new file mode 100644 index 0000000..28e83f5 --- /dev/null +++ b/apps/api/src/rate-limit.ts @@ -0,0 +1,47 @@ +export interface RateLimitOptions { + max: number + windowMs: number + now?: () => number +} + +export interface RateLimitDecision { + allowed: boolean + remaining: number + retryAfterMs: number +} + +export interface RateLimiter { + check: (key: string) => RateLimitDecision +} + +interface Bucket { + count: number + windowStart: number +} + +export const createRateLimiter = (options: RateLimitOptions): RateLimiter => { + const now = options.now ?? (() => Date.now()) + const buckets = new Map() + + return { + check(key: string): RateLimitDecision { + const at = now() + const bucket = buckets.get(key) + if (!bucket || at - bucket.windowStart >= options.windowMs) { + buckets.set(key, { count: 1, windowStart: at }) + return { allowed: true, remaining: options.max - 1, retryAfterMs: 0 } + } + + if (bucket.count >= options.max) { + return { + allowed: false, + remaining: 0, + retryAfterMs: bucket.windowStart + options.windowMs - at, + } + } + + bucket.count += 1 + return { allowed: true, remaining: options.max - bucket.count, retryAfterMs: 0 } + }, + } +} diff --git a/apps/api/src/telemetry.ts b/apps/api/src/telemetry.ts new file mode 100644 index 0000000..e40fb41 --- /dev/null +++ b/apps/api/src/telemetry.ts @@ -0,0 +1,58 @@ +import { metrics } from '@opentelemetry/api' +import { OTLPMetricExporter } from '@opentelemetry/exporter-metrics-otlp-http' +import { resourceFromAttributes } from '@opentelemetry/resources' +import { MeterProvider, PeriodicExportingMetricReader } from '@opentelemetry/sdk-metrics' + +const meter = metrics.getMeter('flakemetry-api') + +export const apiMetrics = { + runsAccepted: meter.createCounter('flakemetry.ingest.runs_accepted', { + description: 'accepted ingest runs', + }), + executionsAccepted: meter.createCounter('flakemetry.ingest.executions_accepted', { + description: 'accepted test executions', + }), + rateLimited: meter.createCounter('flakemetry.ingest.rate_limited', { + description: 'requests rejected by the rate limiter', + }), + backpressured: meter.createCounter('flakemetry.ingest.backpressured', { + description: 'requests rejected due to queue backpressure', + }), + requestDuration: meter.createHistogram('flakemetry.http.server.duration', { + description: 'request duration in milliseconds', + unit: 'ms', + }), +} + +export const observeQueueDepth = (getDepth: () => Promise): void => { + meter + .createObservableGauge('flakemetry.queue.depth', { description: 'pending ingestion jobs' }) + .addCallback(async (result) => { + result.observe(await getDepth()) + }) +} + +export interface SelfTelemetryOptions { + endpoint: string + headers?: Record + exportIntervalMs?: number + serviceName?: string +} + +export const initSelfTelemetry = (options: SelfTelemetryOptions): (() => Promise) => { + const exporter = new OTLPMetricExporter({ + url: `${options.endpoint.replace(/\/+$/, '')}/v1/metrics`, + headers: options.headers, + }) + const provider = new MeterProvider({ + resource: resourceFromAttributes({ 'service.name': options.serviceName ?? 'flakemetry-api' }), + readers: [ + new PeriodicExportingMetricReader({ + exporter, + exportIntervalMillis: options.exportIntervalMs ?? 30_000, + }), + ], + }) + metrics.setGlobalMeterProvider(provider) + return () => provider.shutdown() +} diff --git a/docs/configuration.md b/docs/configuration.md index 490468a..f381b8b 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -73,6 +73,17 @@ Unknown keys are rejected with an error naming the offending path — typos fail | `FLAKEMETRY_TRANSPORT` | `otlp` (default) or `json` | | `FLAKEMETRY_BUFFER_DIR` | Directory to buffer runs to when delivery fails; replayed on the next run | | `FLAKEMETRY_SAMPLE_RATE` | Fraction (0–1) of **passing** runs to deliver; runs containing a failure or flake are always delivered | +| `FLAKEMETRY_COMPRESSION` | `gzip` to compress OTLP export (the ingestion API decompresses gzip request bodies) | + +### Ingestion API service + +| Variable | Effect | +|---|---| +| `LOG_LEVEL` | Structured (pino) log level; `authorization` header is redacted | +| `FLAKEMETRY_MAX_QUEUE_DEPTH` | Backpressure threshold — return `503` once pending jobs reach it | +| `FLAKEMETRY_SELF_OTEL_ENDPOINT` | OTLP endpoint to export the API's own metrics to (dogfooding); metrics are no-ops when unset | + +The API also rate-limits per project token (fixed window) and returns `429` with `Retry-After` when exceeded. ## Inspecting the resolved configuration diff --git a/packages/reporter/src/reporter.ts b/packages/reporter/src/reporter.ts index 26d42bb..22a80f5 100644 --- a/packages/reporter/src/reporter.ts +++ b/packages/reporter/src/reporter.ts @@ -148,7 +148,11 @@ export default class FlakemetryReporter implements Reporter { this.options.transport ?? (this.env.FLAKEMETRY_TRANSPORT === 'json' ? 'json' : 'otlp') if (transport === 'otlp') { try { - await exportRunOverOtlp(recorder, idempotencyKey, { endpoint, token }) + await exportRunOverOtlp(recorder, idempotencyKey, { + endpoint, + token, + compression: this.env.FLAKEMETRY_COMPRESSION === 'gzip', + }) return } catch (error) { process.stderr.write( diff --git a/packages/sdk/package.json b/packages/sdk/package.json index 5411a44..b9bbef7 100644 --- a/packages/sdk/package.json +++ b/packages/sdk/package.json @@ -48,6 +48,7 @@ "@opentelemetry/api": "^1.9.0", "@opentelemetry/core": "^2.0.0", "@opentelemetry/exporter-trace-otlp-http": "^0.220.0", + "@opentelemetry/otlp-exporter-base": "^0.220.0", "@opentelemetry/resources": "^2.0.0", "@opentelemetry/sdk-trace-base": "^2.0.0", "@opentelemetry/semantic-conventions": "^1.28.0" diff --git a/packages/sdk/src/exporter.ts b/packages/sdk/src/exporter.ts index 11552cf..33236c3 100644 --- a/packages/sdk/src/exporter.ts +++ b/packages/sdk/src/exporter.ts @@ -1,4 +1,5 @@ import { OTLPTraceExporter } from '@opentelemetry/exporter-trace-otlp-http' +import { CompressionAlgorithm } from '@opentelemetry/otlp-exporter-base' import { resourceFromAttributes } from '@opentelemetry/resources' import { BasicTracerProvider, @@ -14,6 +15,7 @@ export interface OtlpExporterOptions { endpoint: string token: string timeoutMillis?: number + compression?: boolean } export const OTLP_TRACES_PATH = '/v1/traces' @@ -49,6 +51,7 @@ export const exportRunOverOtlp = async ( url: `${endpoint}${OTLP_TRACES_PATH}`, headers: { authorization: `Bearer ${options.token}` }, timeoutMillis: options.timeoutMillis ?? 10_000, + compression: options.compression ? CompressionAlgorithm.GZIP : CompressionAlgorithm.NONE, }) try { diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 8c8d1de..1f5f746 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -35,6 +35,18 @@ importers: '@flakemetry/db': specifier: workspace:* version: link:../../packages/db + '@opentelemetry/api': + specifier: ^1.9.0 + version: 1.9.1 + '@opentelemetry/exporter-metrics-otlp-http': + specifier: ^0.220.0 + version: 0.220.0(@opentelemetry/api@1.9.1) + '@opentelemetry/resources': + specifier: ^2.0.0 + version: 2.9.0(@opentelemetry/api@1.9.1) + '@opentelemetry/sdk-metrics': + specifier: ^2.0.0 + version: 2.9.0(@opentelemetry/api@1.9.1) '@trpc/server': specifier: ^11.0.0 version: 11.18.0(typescript@5.9.3) @@ -277,6 +289,9 @@ importers: '@opentelemetry/exporter-trace-otlp-http': specifier: ^0.220.0 version: 0.220.0(@opentelemetry/api@1.9.1) + '@opentelemetry/otlp-exporter-base': + specifier: ^0.220.0 + version: 0.220.0(@opentelemetry/api@1.9.1) '@opentelemetry/resources': specifier: ^2.0.0 version: 2.9.0(@opentelemetry/api@1.9.1) @@ -814,6 +829,12 @@ packages: peerDependencies: '@opentelemetry/api': '>=1.0.0 <1.10.0' + '@opentelemetry/exporter-metrics-otlp-http@0.220.0': + resolution: {integrity: sha512-Yqt3RBw/bRVncaE9qIIhk4WfjbAQqXuP9FgAaU+IKPndnLEp/cUqZlSC324+bpmduRz7DoTjig8Ub0PeILWXUA==} + engines: {node: ^18.19.0 || >=20.6.0} + peerDependencies: + '@opentelemetry/api': ^1.3.0 + '@opentelemetry/exporter-trace-otlp-http@0.220.0': resolution: {integrity: sha512-/+ExB3lRkf+erv4PnoywyL7RHKITidxtUpUTS55k7OQ0dB42S7gEF1gry7swb9MSm1hYLUhJg4QQh9W8SpwwqA==} engines: {node: ^18.19.0 || >=20.6.0} @@ -2777,6 +2798,15 @@ snapshots: '@opentelemetry/api': 1.9.1 '@opentelemetry/semantic-conventions': 1.43.0 + '@opentelemetry/exporter-metrics-otlp-http@0.220.0(@opentelemetry/api@1.9.1)': + dependencies: + '@opentelemetry/api': 1.9.1 + '@opentelemetry/core': 2.9.0(@opentelemetry/api@1.9.1) + '@opentelemetry/otlp-exporter-base': 0.220.0(@opentelemetry/api@1.9.1) + '@opentelemetry/otlp-transformer': 0.220.0(@opentelemetry/api@1.9.1) + '@opentelemetry/resources': 2.9.0(@opentelemetry/api@1.9.1) + '@opentelemetry/sdk-metrics': 2.9.0(@opentelemetry/api@1.9.1) + '@opentelemetry/exporter-trace-otlp-http@0.220.0(@opentelemetry/api@1.9.1)': dependencies: '@opentelemetry/api': 1.9.1