From f6a17e4978e247fc52148a4354ee61c9f34cc0d0 Mon Sep 17 00:00:00 2001 From: Abhimanyu Saharan Date: Sun, 4 Oct 2026 05:56:59 +0530 Subject: [PATCH] Correct frontend telemetry metadata and SLO inputs --- apps/web/tests/browser/telemetry.mjs | 201 +++++++++++++++++++++ apps/web/tests/telemetry.test.ts | 254 ++++++++++++++++++++++++++- packages/core/OBSERVABILITY.md | 48 +++++ packages/telemetry/src/delivery.ts | 25 ++- packages/telemetry/src/receiver.ts | 44 +++-- packages/telemetry/src/server.ts | 28 ++- scripts/ci/admin-browser.sh | 1 + scripts/ci/reader-parity.sh | 1 + 8 files changed, 567 insertions(+), 35 deletions(-) create mode 100644 apps/web/tests/browser/telemetry.mjs diff --git a/apps/web/tests/browser/telemetry.mjs b/apps/web/tests/browser/telemetry.mjs new file mode 100644 index 000000000..886d4cf82 --- /dev/null +++ b/apps/web/tests/browser/telemetry.mjs @@ -0,0 +1,201 @@ +import assert from "node:assert/strict"; +import { spawn } from "node:child_process"; +import { once } from "node:events"; +import { readFile } from "node:fs/promises"; +import { createServer } from "node:http"; +import { fileURLToPath } from "node:url"; +import { chromium } from "playwright"; + +const root = fileURLToPath(new URL("../../../../", import.meta.url)); +const appName = process.argv[2] || "web"; +assert.ok(["web", "admin"].includes(appName)); +const { version } = JSON.parse(await readFile(`${root}/apps/${appName}/package.json`, "utf8")); +const serverKey = "synthetic-server-only-collector-key"; +const forwarded = []; +const fixture = createServer(async (request, response) => { + if (request.url === "/collect") { + let body = ""; + for await (const chunk of request) body += chunk; + forwarded.push({ headers: request.headers, body: JSON.parse(body) }); + response.writeHead(202); + response.end(); + return; + } + response.writeHead(401, { "Content-Type": "application/json" }); + response.end(JSON.stringify({ detail: "Not authenticated" })); +}); +await new Promise((resolve) => fixture.listen(0, "127.0.0.1", resolve)); +async function availablePort() { + const reservation = createServer(); + await new Promise((resolve) => reservation.listen(0, "127.0.0.1", resolve)); + const port = reservation.address().port; + await new Promise((resolve) => reservation.close(resolve)); + return port; +} +const port = await availablePort(); +let metricsPort = await availablePort(); +while (metricsPort === port) metricsPort = await availablePort(); +const origin = `http://127.0.0.1:${port}`; +const upstream = `http://127.0.0.1:${fixture.address().port}`; +const env = Object.fromEntries( + Object.entries(process.env).filter(([key]) => !key.startsWith("DEVFEED_")), +); +const app = spawn( + process.execPath, + [ + `${root}/node_modules/next/dist/bin/next`, + "start", + "--hostname", + "127.0.0.1", + "--port", + String(port), + ], + { + cwd: `${root}/apps/${appName}`, + env: { + ...env, + NODE_ENV: "production", + DEVFEED_VERSION: "stale-runtime-version", + DEVFEED_PUBLIC_API_URL: upstream, + DEVFEED_USER_API_URL: upstream, + DEVFEED_ADMIN_API_URL: upstream, + DEVFEED_USER_BASE_URL: origin, + DEVFEED_ADMIN_BASE_URL: origin, + DEVFEED_FARO_ENABLED: "true", + DEVFEED_FARO_COLLECTOR_URL: `${upstream}/collect`, + DEVFEED_FARO_API_KEY: serverKey, + DEVFEED_TELEMETRY_ENVIRONMENT: "production", + DEVFEED_METRICS_ENABLED: "true", + DEVFEED_METRICS_HOST: "127.0.0.1", + DEVFEED_METRICS_PORT: String(metricsPort), + DEVFEED_OTLP_ENDPOINT: "", + DEVFEED_PYROSCOPE_SERVER: "", + DEVFEED_ANALYTICS_ENABLED: "false", + DEVFEED_X_PIXEL_ENABLED: "false", + }, + stdio: "pipe", + }, +); +let logs = ""; +app.stdout.on("data", (data) => (logs += data)); +app.stderr.on("data", (data) => (logs += data)); +let browser; +try { + const path = appName === "web" ? "/legal/privacy" : "/login"; + let ready = false; + for (let i = 0; i < 100; i++) { + if (app.exitCode !== null) throw new Error(logs); + try { + const response = await fetch(`${origin}${path}`, { signal: AbortSignal.timeout(1000) }); + await response.body?.cancel(); + if (response.ok) { + ready = true; + break; + } + } catch {} + await new Promise((resolve) => setTimeout(resolve, 200)); + } + assert.ok(ready, logs); + browser = await chromium.launch({ headless: true }); + const page = await browser.newPage(); + await page.route(/^https?:\/\//, (route) => + new URL(route.request().url()).origin === origin ? route.continue() : route.abort(), + ); + page.on("pageerror", (error) => { + logs += `\nbrowser: ${error.message}`; + }); + page.on("console", (message) => { + if (message.type() === "error" || message.type() === "warn") + logs += `\nbrowser: ${message.text()}`; + }); + // Select a sampled session deterministically in this test only. Production stays at 10%. + await page.addInitScript(() => { + Math.random = () => 0.01; + const randomValues = crypto.getRandomValues.bind(crypto); + crypto.getRandomValues = (array) => { + if (array instanceof Uint32Array && array.length === 1) return array.fill(0); + return randomValues(array); + }; + }); + const delivered = page.waitForResponse( + (response) => + response.url() === `${origin}/telemetry/collect` && + response.request().method() === "POST" && + response + .request() + .postDataJSON() + .events?.some((event) => event.name === "telemetry_ready"), + { timeout: 20_000 }, + ); + await page.goto(`${origin}${path}`); + const response = await delivered; + assert.equal(response.status(), 202); + const request = response.request(); + assert.equal(await request.headerValue("origin"), origin); + assert.equal(await request.headerValue("content-encoding"), null); + assert.equal(await request.headerValue("x-api-key"), null); + const sent = request.postDataJSON(); + assert.deepEqual(sent.meta.app, { + name: `devfeed-${appName}`, + version, + environment: "production", + }); + assert.ok(sent.events?.some((event) => event.name === "telemetry_ready")); + assert.ok(!JSON.stringify(sent).includes(serverKey)); + assert.ok(forwarded.length > 0); + for (const batch of forwarded) { + assert.equal(batch.headers["x-api-key"], serverKey); + assert.equal(batch.headers.origin, origin); + assert.deepEqual(batch.body.meta.app, sent.meta.app); + } + const before = forwarded.length; + const spoofed = await page.evaluate( + async () => + ( + await fetch("/telemetry/collect", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ + meta: { app: { name: "spoofed", version: "development", environment: "development" } }, + }), + }) + ).status, + ); + assert.equal(spoofed, 202); + assert.ok(forwarded.length > before); + assert.deepEqual(forwarded.at(-1).body.meta.app, sent.meta.app); + const rejected = await fetch(`${origin}/telemetry/collect`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: "{}", + }); + assert.equal(rejected.status, 403); + const metrics = await (await fetch(`http://127.0.0.1:${metricsPort}/metrics`)).text(); + assert.match( + metrics, + /devfeed_faro_deliveries_total\{status="202",outcome="accepted",service="(?:web|admin)"\} [1-9]/, + ); + assert.match( + metrics, + /devfeed_faro_deliveries_total\{status="403",outcome="origin_missing",service="(?:web|admin)"\} 1/, + ); + assert.ok(metrics.includes('devfeed_request_kind="request"')); + assert.ok(metrics.includes('http_response_status_code="200"')); + console.log( + `${appName}: sampled real Faro browser batch accepted, build metadata authoritative, bounded rejection metrics recorded`, + ); +} catch (error) { + console.error(logs); + throw error; +} finally { + await browser?.close(); + if (app.exitCode === null) { + const exited = once(app, "exit"); + app.kill("SIGTERM"); + const timer = setTimeout(() => app.kill("SIGKILL"), 5000); + await exited; + clearTimeout(timer); + } + fixture.closeAllConnections(); + await new Promise((resolve) => fixture.close(resolve)); +} diff --git a/apps/web/tests/telemetry.test.ts b/apps/web/tests/telemetry.test.ts index 453179cc3..679ac8052 100644 --- a/apps/web/tests/telemetry.test.ts +++ b/apps/web/tests/telemetry.test.ts @@ -4,11 +4,25 @@ import { describe, expect, it, vi } from "vitest"; import { createServer } from "node:http"; import { once } from "node:events"; import { routeName, normalizeBody, normalizePayload } from "@devfeed/telemetry/privacy"; -import { receiveTelemetry } from "@devfeed/telemetry/receiver"; +import { browserSettings, receiveTelemetry } from "@devfeed/telemetry/receiver"; import { registerTelemetry } from "@devfeed/telemetry/server"; const settings = { enabled: true, app: "web" as const, version: "test", environment: "test" }; describe("operational telemetry", () => { + it.each(["web", "admin"] as const)("uses the deployed build version for %s Faro", (app) => { + vi.stubEnv("DEVFEED_BUILD_VERSION", "9.8.7"); + vi.stubEnv("DEVFEED_VERSION", "0.0.8"); + vi.stubEnv("DEVFEED_TELEMETRY_ENVIRONMENT", "production"); + try { + expect(browserSettings(app)).toMatchObject({ version: "9.8.7", environment: "production" }); + vi.stubEnv("DEVFEED_BUILD_VERSION", ""); + expect(browserSettings(app).version).toBe("0.0.8"); + vi.stubEnv("DEVFEED_VERSION", ""); + expect(browserSettings(app).version).toBe("development"); + } finally { + vi.unstubAllEnvs(); + } + }); it("bounds unknown routes, slugs, query strings and dynamic API paths", () => { const routes = new Set( Array.from({ length: 1000 }, (_, i) => routeName(`/articles/private-${i}?token=secret`)), @@ -135,6 +149,11 @@ describe("operational telemetry", () => { vi.stubEnv("DEVFEED_FARO_API_KEY", "server-key"); vi.stubEnv("DEVFEED_USER_BASE_URL", "https://devfeed.tech"); vi.stubEnv("DEVFEED_ADMIN_BASE_URL", "https://devfeed.tech"); + vi.stubEnv("DEVFEED_BUILD_VERSION", "9.8.7"); + vi.stubEnv("DEVFEED_VERSION", "0.0.8"); + vi.stubEnv("DEVFEED_TELEMETRY_ENVIRONMENT", "production"); + const outcome = vi.fn(); + vi.stubGlobal("devfeedFaroDelivery", outcome); const upstream = vi.fn().mockResolvedValue(new Response(null, { status: 202 })); vi.stubGlobal("fetch", upstream); const request = (body: string, origin = "https://devfeed.tech") => @@ -149,12 +168,19 @@ describe("operational telemetry", () => { ); expect((await receiveTelemetry(request("x".repeat(65537)), app)).status).toBe(413); expect(upstream).not.toHaveBeenCalled(); + expect(outcome.mock.calls).toEqual([ + [403, "origin_mismatch"], + [413, "too_large"], + ]); expect( ( await receiveTelemetry( request( JSON.stringify({ - meta: { user: { email: "secret" } }, + meta: { + app: { version: "development", environment: "development" }, + user: { email: "secret" }, + }, logs: [{ message: "secret" }], }), ), @@ -165,15 +191,144 @@ describe("operational telemetry", () => { const options = upstream.mock.calls[0][1]; expect(options.headers["X-API-Key"]).toBe("server-key"); expect(JSON.parse(options.body).logs).toEqual([{ message: "secret" }]); + expect(JSON.parse(options.body).meta.app).toMatchObject({ + name: `devfeed-${app}`, + version: "9.8.7", + environment: "production", + }); expect(JSON.stringify(options.headers)).not.toContain("secret-cookie"); + expect(outcome).toHaveBeenLastCalledWith(202, "accepted"); + upstream.mockResolvedValueOnce(new Response(null, { status: 403 })); + expect((await receiveTelemetry(request("{}"), app)).status).toBe(503); + expect(outcome).toHaveBeenLastCalledWith(503, "upstream_rejected"); upstream.mockRejectedValueOnce(new Error("private-collector-error")); expect((await receiveTelemetry(request("{}"), app)).status).toBe(503); + expect(outcome).toHaveBeenLastCalledWith(503, "upstream_error"); + } finally { + vi.unstubAllGlobals(); + vi.unstubAllEnvs(); + } + }, + ); + it.each(["web", "admin"] as const)( + "distinguishes bounded %s receiver rejection reasons", + async (app) => { + vi.stubEnv("DEVFEED_FARO_ENABLED", "true"); + vi.stubEnv("DEVFEED_FARO_COLLECTOR_URL", "http://collector.invalid/collect"); + vi.stubEnv("DEVFEED_FARO_API_KEY", "server-key"); + const originKey = app === "web" ? "DEVFEED_USER_BASE_URL" : "DEVFEED_ADMIN_BASE_URL"; + vi.stubEnv(originKey, "https://devfeed.tech"); + const outcome = vi.fn(); + const upstream = vi.fn(); + vi.stubGlobal("devfeedFaroDelivery", outcome); + vi.stubGlobal("fetch", upstream); + const reject = async ( + headers: Record, + body: string | undefined, + status: number, + reason: string, + ) => { + const response = await receiveTelemetry( + new Request("https://devfeed.tech/telemetry/collect", { + method: "POST", + headers, + body, + }), + app, + ); + expect(response.status).toBe(status); + expect(response.headers.get("Cache-Control")).toBe("no-store"); + expect(outcome).toHaveBeenLastCalledWith(status, reason); + }; + const headers = { Origin: "https://devfeed.tech", "Content-Type": "application/json" }; + try { + await reject({ "Content-Type": "application/json" }, "{}", 403, "origin_missing"); + await reject({ ...headers, "Content-Encoding": "gzip" }, "{}", 403, "content_encoding"); + await reject({ ...headers, "Content-Type": "text/plain" }, "{}", 415, "unsupported_type"); + await reject({ ...headers, "Content-Length": "65537" }, "{}", 413, "too_large"); + await reject(headers, undefined, 400, "missing_body"); + await reject(headers, "invalid-json", 400, "invalid_body"); + vi.stubEnv(originKey, ""); + await reject(headers, "{}", 403, "origin_unconfigured"); + vi.stubEnv("DEVFEED_FARO_API_KEY", ""); + await reject(headers, "{}", 404, "collector_unconfigured"); + vi.stubEnv("DEVFEED_FARO_ENABLED", "false"); + await reject(headers, "{}", 404, "disabled"); + expect(upstream).not.toHaveBeenCalled(); + expect(outcome).toHaveBeenCalledTimes(9); + expect(JSON.stringify(outcome.mock.calls)).not.toMatch( + /devfeed\.tech|server-key|invalid-json|gzip/, + ); } finally { vi.unstubAllGlobals(); vi.unstubAllEnvs(); } }, ); + it("labels rate limits and body timeouts without forwarding", async () => { + vi.stubEnv("DEVFEED_FARO_ENABLED", "true"); + vi.stubEnv("DEVFEED_FARO_COLLECTOR_URL", "http://collector.invalid/collect"); + vi.stubEnv("DEVFEED_FARO_API_KEY", "server-key"); + vi.stubEnv("DEVFEED_USER_BASE_URL", "https://devfeed.tech"); + const now = Date.now(); + const clock = vi.spyOn(Date, "now").mockReturnValue(now); + const outcome = vi.fn(); + const upstream = vi.fn(); + vi.stubGlobal("devfeedFaroDelivery", outcome); + vi.stubGlobal("fetch", upstream); + vi.resetModules(); + const { receiveTelemetry: isolatedReceiver } = await import("@devfeed/telemetry/receiver"); + const headers = { Origin: "https://devfeed.tech", "Content-Type": "application/json" }; + try { + for (let i = 0; i < 40; i++) { + expect( + ( + await isolatedReceiver( + new Request("https://devfeed.tech/telemetry/collect", { + method: "POST", + headers, + body: "invalid-json", + }), + "web", + ) + ).status, + ).toBe(400); + } + expect( + ( + await isolatedReceiver( + new Request("https://devfeed.tech/telemetry/collect", { + method: "POST", + headers, + body: "{}", + }), + "web", + ) + ).status, + ).toBe(429); + expect(outcome).toHaveBeenLastCalledWith(429, "rate_limited"); + clock.mockReturnValue(now + 1000); + vi.useFakeTimers(); + const stalled = isolatedReceiver( + new Request("https://devfeed.tech/telemetry/collect", { + method: "POST", + headers, + body: new ReadableStream(), + duplex: "half", + } as RequestInit), + "web", + ); + await vi.advanceTimersByTimeAsync(2000); + expect((await stalled).status).toBe(408); + expect(outcome).toHaveBeenLastCalledWith(408, "body_timeout"); + expect(upstream).not.toHaveBeenCalled(); + } finally { + vi.useRealTimers(); + clock.mockRestore(); + vi.unstubAllGlobals(); + vi.unstubAllEnvs(); + } + }); it("serves metrics only on the distinct listener and records real HTTP status", async () => { const reservation = createServer(); reservation.listen(0, "127.0.0.1"); @@ -189,6 +344,28 @@ describe("operational telemetry", () => { vi.stubEnv("DEVFEED_PYROSCOPE_SERVER", ""); const log = vi.spyOn(console, "info").mockImplementation(() => {}); const app = createServer((request, response) => { + if (request.url === "/api/v1/user/notifications/chimely/v1/inbox/stream?unauthorized=1") { + response.statusCode = 401; + response.end(); + return; + } + if (request.url?.includes("/stream") || request.url === "/search?sse=1") { + response.setHeader("Content-Type", "text/event-stream; charset=utf-8"); + response.write("data: ready\n\n"); + if (!request.url.includes("abort")) response.end(); + return; + } + if (request.url === "/search?html=1") { + response.setHeader("Content-Type", "text/html"); + response.write("

chunked

"); + response.end(); + return; + } + if (request.url === "/search?error=1") { + response.statusCode = 500; + response.end(); + return; + } response.statusCode = request.url === "/" ? 200 : 404; response.end(); }); @@ -199,6 +376,33 @@ describe("operational telemetry", () => { const appPort = (app.address() as { port: number }).port; expect((await fetch(`http://127.0.0.1:${appPort}/metrics`)).status).toBe(404); expect((await fetch(`http://127.0.0.1:${appPort}/`)).status).toBe(200); + expect((await fetch(`http://127.0.0.1:${appPort}/search?error=1`)).status).toBe(500); + expect( + ( + await fetch( + `http://127.0.0.1:${appPort}/api/v1/user/notifications/chimely/v1/inbox/stream?unauthorized=1`, + ) + ).status, + ).toBe(401); + for (const path of [ + "/api/v1/user/notifications/chimely/v1/inbox/stream?session=private", + "/v1/admin/notifications/chimely/v1/inbox/stream/", + "/search?sse=1", + "/search?html=1", + ]) { + await (await fetch(`http://127.0.0.1:${appPort}${path}`)).text(); + } + const abort = new AbortController(); + const streaming = await fetch(`http://127.0.0.1:${appPort}/search/stream?abort=1`, { + signal: abort.signal, + }); + await streaming.body!.getReader().read(); + abort.abort(); + await vi.waitFor(() => { + expect(log.mock.calls.map(([line]) => JSON.parse(line))).toContainEqual( + expect.objectContaining({ route: "/search/stream?abort=1", status_code: 499 }), + ); + }); await fetch(`http://127.0.0.1:${appPort}/articles/item-123?token=original-value`); expect(log.mock.calls.map(([line]) => JSON.parse(line))).toContainEqual( expect.objectContaining({ route: "/articles/item-123?token=original-value" }), @@ -210,6 +414,52 @@ describe("operational telemetry", () => { expect(metrics).toContain('version="9.8.7"'); expect(metrics).not.toContain('version="0.0.8"'); expect(metrics).toContain('route="/",status="200",service="web"} 1'); + const histogram = metrics + .split("\n") + .filter((line) => line.startsWith("http_server_request_duration_seconds_count{")); + const series = histogram.map((line) => ({ + ...Object.fromEntries( + [...line.matchAll(/([a-z_]+)="([^"]*)"/g)].map((match) => [match[1], match[2]]), + ), + count: Number(line.slice(line.lastIndexOf(" ") + 1)), + })); + for (const [route, status, kind] of [ + ["/", "200", "request"], + ["/search", "500", "request"], + ["/search", "200", "request"], + ["/v1/user/notifications/:path", "200", "stream"], + ["/v1/user/notifications/:path", "401", "stream"], + ["/v1/admin/notifications/:path", "200", "stream"], + ["/search", "200", "stream"], + ["unmatched", "499", "stream"], + ]) { + expect(series).toContainEqual({ + http_request_method: "GET", + http_route: route, + http_response_status_code: status, + devfeed_request_kind: kind, + service: "web", + count: 1, + }); + } + expect(metrics).not.toContain("devfeed_http_request_duration_seconds"); + expect(histogram.join("\n")).not.toMatch(/private|session=|abort=/); + vi.stubEnv("DEVFEED_FARO_ENABLED", "true"); + vi.stubEnv("DEVFEED_FARO_COLLECTOR_URL", "http://collector.invalid/collect"); + vi.stubEnv("DEVFEED_FARO_API_KEY", "server-key"); + vi.stubEnv("DEVFEED_USER_BASE_URL", "https://devfeed.tech"); + expect( + ( + await receiveTelemetry( + new Request("https://devfeed.tech/telemetry/collect", { method: "POST", body: "{}" }), + "web", + ) + ).status, + ).toBe(403); + const deliveries = await (await fetch(`http://127.0.0.1:${port}/metrics`)).text(); + expect(deliveries).toContain( + 'devfeed_faro_deliveries_total{status="403",outcome="origin_missing",service="web"} 1', + ); expect(metrics).toContain('devfeed_http_requests_in_progress{service="web"} 0'); } finally { app.close(); diff --git a/packages/core/OBSERVABILITY.md b/packages/core/OBSERVABILITY.md index fa635b0ff..6d13d490a 100644 --- a/packages/core/OBSERVABILITY.md +++ b/packages/core/OBSERVABILITY.md @@ -141,6 +141,54 @@ Metric routes remain grouped to limit cardinality; Node request logs use concret Restart running services to load the updated logging code. Previously masked values in retained records cannot be reconstructed. This change needs no database migration. +## Frontend telemetry + +Web and admin Faro metadata uses the version embedded in the Next build +(`DEVFEED_BUILD_VERSION`), then `DEVFEED_VERSION`, then `development`. +`DEVFEED_TELEMETRY_ENVIRONMENT` supplies the environment independently. The receiver +assigns these values again when forwarding a payload; client metadata is not authoritative. + +`devfeed_faro_deliveries_total{service,status,outcome}` counts receiver outcomes once. +`accepted` / 202 means the upstream collector accepted the batch, not that every +event was stored. The bounded outcomes distinguish missing, mismatched or unconfigured +origins, content encoding, invalid bodies, size/rate/time limits, disabled or +unconfigured collectors, upstream rejection, and upstream transport errors. They +contain no origin values, URLs, payloads or user/session identities. + +```promql +sum by (service, status, outcome) ( + increase(devfeed_faro_deliveries_total{namespace="devfeed"}[24h]) +) +``` + +A high 403 fraction alone cannot establish genuine browser data loss. Compare the +rejection reasons with a sampled browser's network response and collector ingestion +in the same window. Faro samples 10% of sessions, honors Do Not Track, and batches +events for up to five seconds. Unsampled sessions and unfinished page-lifecycle vitals +are not evidence of delivery failure. Do not weaken origin validation to improve the +acceptance ratio. + +Node emits `http_server_request_duration_seconds` with `http_request_method`, +`http_route`, `http_response_status_code`, and `devfeed_request_kind` labels. Known +Chimely inbox stream paths (including the `/api` proxy prefix) and responses with +`Content-Type: text/event-stream` have kind `stream`; ordinary chunked HTML has kind +`request`. Disconnections are recorded once as 499. Measure request latency and error +ratios with `devfeed_request_kind="request"` to exclude stream lifetimes. + +This replaces the Node `devfeed_http_request_duration_seconds` histogram. The +existing GitOps `devfeed:http_duration_seconds_*` rules already accept the new +histogram and exclude streams; update any direct queries of the old metric. +`devfeed_http_requests_total` remains available. During a rolling update, older +replicas still expose the previous histogram without status or stream classification; +confirm all replicas have updated before judging the corrected SLOs. New outcome +labels also start new counter series; older events have no reason and cannot be +classified retroactively. Let the selected rate windows age past the rollout. + +After building both Next applications, run the real Faro browser/receiver regression +with `node apps/web/tests/browser/telemetry.mjs` and +`node apps/web/tests/browser/telemetry.mjs admin`. These use a local collector fixture +and force a sampled session only inside the test; they do not send production events. + ## Migration The old Python `devfeed_http_*`, database, dependency, worker, and search Prometheus diff --git a/packages/telemetry/src/delivery.ts b/packages/telemetry/src/delivery.ts index acc372e4d..1a6fb7587 100644 --- a/packages/telemetry/src/delivery.ts +++ b/packages/telemetry/src/delivery.ts @@ -1,8 +1,25 @@ // Shared across Next's server bundles without exporting any browser state. -type DeliveryGlobals = typeof globalThis & { devfeedFaroDelivery?: (status: number) => void }; -export function recordDelivery(status: number) { - (globalThis as DeliveryGlobals).devfeedFaroDelivery?.(status); +export type DeliveryOutcome = + | "accepted" + | "disabled" + | "collector_unconfigured" + | "origin_unconfigured" + | "origin_missing" + | "origin_mismatch" + | "content_encoding" + | "unsupported_type" + | "too_large" + | "rate_limited" + | "missing_body" + | "body_timeout" + | "invalid_body" + | "upstream_rejected" + | "upstream_error"; +type DeliveryObserver = (status: number, outcome: DeliveryOutcome) => void; +type DeliveryGlobals = typeof globalThis & { devfeedFaroDelivery?: DeliveryObserver }; +export function recordDelivery(status: number, outcome: DeliveryOutcome) { + (globalThis as DeliveryGlobals).devfeedFaroDelivery?.(status, outcome); } -export function observeDeliveries(callback: (status: number) => void) { +export function observeDeliveries(callback: DeliveryObserver) { (globalThis as DeliveryGlobals).devfeedFaroDelivery = callback; } diff --git a/packages/telemetry/src/receiver.ts b/packages/telemetry/src/receiver.ts index 8cc5f32cd..0a165b5b4 100644 --- a/packages/telemetry/src/receiver.ts +++ b/packages/telemetry/src/receiver.ts @@ -1,11 +1,11 @@ -import { recordDelivery } from "./delivery"; +import { recordDelivery, type DeliveryOutcome } from "./delivery"; import { normalizeBody, type BrowserSettings } from "./privacy"; export function browserSettings(app: "web" | "admin"): BrowserSettings { return { app, enabled: process.env.DEVFEED_FARO_ENABLED === "true", - version: process.env.DEVFEED_VERSION || "development", + version: process.env.DEVFEED_BUILD_VERSION || process.env.DEVFEED_VERSION || "development", environment: process.env.DEVFEED_TELEMETRY_ENVIRONMENT || "development", }; } @@ -20,29 +20,25 @@ function accept() { return true; } export async function receiveTelemetry(request: Request, app: "web" | "admin"): Promise { - const reply = (status: number) => { - recordDelivery(status); + const reply = (status: number, outcome: DeliveryOutcome) => { + recordDelivery(status, outcome); return new Response(null, { status, headers: { "Cache-Control": "no-store" } }); }; const settings = browserSettings(app); - if ( - !settings.enabled || - !process.env.DEVFEED_FARO_COLLECTOR_URL || - !process.env.DEVFEED_FARO_API_KEY - ) - return reply(404); + if (!settings.enabled) return reply(404, "disabled"); + if (!process.env.DEVFEED_FARO_COLLECTOR_URL || !process.env.DEVFEED_FARO_API_KEY) + return reply(404, "collector_unconfigured"); const origin = process.env[app === "web" ? "DEVFEED_USER_BASE_URL" : "DEVFEED_ADMIN_BASE_URL"]; - if ( - !origin || - request.headers.get("origin") !== origin || - request.headers.get("content-encoding") - ) - return reply(403); - if (!request.headers.get("content-type")?.startsWith("application/json")) return reply(415); - if (Number(request.headers.get("content-length")) > 65_536) return reply(413); - if (!accept()) return reply(429); + if (!origin) return reply(403, "origin_unconfigured"); + if (!request.headers.get("origin")) return reply(403, "origin_missing"); + if (request.headers.get("origin") !== origin) return reply(403, "origin_mismatch"); + if (request.headers.get("content-encoding")) return reply(403, "content_encoding"); + if (!request.headers.get("content-type")?.startsWith("application/json")) + return reply(415, "unsupported_type"); + if (Number(request.headers.get("content-length")) > 65_536) return reply(413, "too_large"); + if (!accept()) return reply(429, "rate_limited"); const reader = request.body?.getReader(); - if (!reader) return reply(400); + if (!reader) return reply(400, "missing_body"); const chunks: Uint8Array[] = []; let size = 0; let forwarding = false; @@ -58,11 +54,11 @@ export async function receiveTelemetry(request: Request, app: "web" | "admin"): size += value.byteLength; if (size > 65_536) { await reader.cancel(); - return reply(413); + return reply(413, "too_large"); } chunks.push(value); } - if (timedOut) return reply(408); + if (timedOut) return reply(408, "body_timeout"); clearTimeout(timer); const raw = new Uint8Array(size); let offset = 0; @@ -84,9 +80,9 @@ export async function receiveTelemetry(request: Request, app: "web" | "admin"): body: JSON.stringify(body), }); await response.body?.cancel(); - return reply(response.ok ? 202 : 503); + return response.ok ? reply(202, "accepted") : reply(503, "upstream_rejected"); } catch { - return reply(forwarding ? 503 : 400); + return forwarding ? reply(503, "upstream_error") : reply(400, "invalid_body"); } finally { clearTimeout(timer); reader.releaseLock(); diff --git a/packages/telemetry/src/server.ts b/packages/telemetry/src/server.ts index adb9621df..f519704d0 100644 --- a/packages/telemetry/src/server.ts +++ b/packages/telemetry/src/server.ts @@ -37,10 +37,10 @@ export async function registerTelemetry(app: "web" | "admin") { const deliveries = new Counter({ name: "devfeed_faro_deliveries_total", help: "Faro receiver outcomes; status 202 means collector accepted", - labelNames: ["status"], + labelNames: ["status", "outcome"], registers: [registry], }); - observeDeliveries((status) => deliveries.labels(String(status)).inc()); + observeDeliveries((status, outcome) => deliveries.labels(String(status), outcome).inc()); const requests = new Counter({ name: "devfeed_http_requests_total", help: "Completed application HTTP requests", @@ -48,9 +48,14 @@ export async function registerTelemetry(app: "web" | "admin") { registers: [registry], }); const duration = new Histogram({ - name: "devfeed_http_request_duration_seconds", + name: "http_server_request_duration_seconds", help: "Application HTTP response duration", - labelNames: ["method", "route"], + labelNames: [ + "http_request_method", + "http_route", + "http_response_status_code", + "devfeed_request_kind", + ], buckets: [0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2, 5, 10, 30], registers: [registry], }); @@ -78,6 +83,11 @@ export async function registerTelemetry(app: "web" | "admin") { time: performance.now(), method: methodName(request.method), route: routeName(request.url), + // Match the Python admission policy before routeName bounds dynamic suffixes. + stream: + /^\/(?:api\/)?v1\/(?:user|admin)\/notifications\/chimely\/v1\/inbox\/stream\/?(?:\?|$)/.test( + request.url || "", + ), }; starts.set(response, state); active.inc(); @@ -87,7 +97,15 @@ export async function registerTelemetry(app: "web" | "admin") { finished = true; active.dec(); requests.labels(state.method, state.route, String(status)).inc(); - duration.labels(state.method, state.route).observe((performance.now() - state.time) / 1000); + const stream = + state.stream || + String(response.getHeader("content-type") || "") + .split(";")[0] + .trim() + .toLowerCase() === "text/event-stream"; + duration + .labels(state.method, state.route, String(status), stream ? "stream" : "request") + .observe((performance.now() - state.time) / 1000); starts.delete(response); const span = trace.getActiveSpan()?.spanContext(); console.info( diff --git a/scripts/ci/admin-browser.sh b/scripts/ci/admin-browser.sh index e49b5e975..4d78aa9d9 100644 --- a/scripts/ci/admin-browser.sh +++ b/scripts/ci/admin-browser.sh @@ -7,3 +7,4 @@ for test in apps/admin/tests/browser/*.mjs; do echo "Running $test" node "$test" done +node apps/web/tests/browser/telemetry.mjs admin diff --git a/scripts/ci/reader-parity.sh b/scripts/ci/reader-parity.sh index 7e492eedc..31c8c947d 100644 --- a/scripts/ci/reader-parity.sh +++ b/scripts/ci/reader-parity.sh @@ -3,5 +3,6 @@ set -euo pipefail # Edge is available on Linux x86_64; this CI job uses the matching runner. npx playwright install --with-deps chromium msedge npm run reader:test:parity +node apps/web/tests/browser/telemetry.mjs node apps/web/tests/browser/x-pixel.mjs node apps/web/tests/browser/profile.mjs