From afb26251e546e74c3a404e321cec17ba5a4dcf5a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sun, 13 Sep 2026 13:57:33 +0200 Subject: [PATCH] fix(logging): ship to a private Loki and report failed pushes LOKI_ENDPOINT is read only from the server environment, so the push now dials its exact origin through the pinned operator-approved resolver instead of the public-host pin that refused a Loki on the operator's own network. Link-local, metadata and unspecified addresses stay refused. A thrown push or a non-2xx answer used to vanish in an empty catch. It now writes one line to stderr per reason at most every five minutes, naming the Loki origin, the reason and the dropped event count, without credentials. The endpoint accepts both the base URL and the full push URL; the second form used to double the path into a silent 404. Refs #947 --- .env.production.example | 14 +- docker-compose.yml | 9 +- .../logging/__tests__/transports-loki.test.ts | 352 ++++++++++++++++++ src/lib/logging/transports.ts | 172 ++++++++- 4 files changed, 530 insertions(+), 17 deletions(-) create mode 100644 src/lib/logging/__tests__/transports-loki.test.ts diff --git a/.env.production.example b/.env.production.example index 0d5f7a2c5..06ff00f4b 100644 --- a/.env.production.example +++ b/.env.production.example @@ -521,10 +521,16 @@ API_TOKEN_HMAC_KEY="" # LOG_SLOW_THRESHOLD_MS="3000" # LOG_INCLUDE_STACK="false" -# Optional log shipping to a Loki instance. LOKI_ENDPOINT is the push URL and -# is what turns shipping on; the username/password pair is only needed when -# your Loki sits behind basic auth. Leave all three unset and logs stay on -# stdout for the Docker log driver to rotate. +# Optional log shipping to a Loki instance. LOKI_ENDPOINT turns shipping on. +# Either form works: the base URL (http://loki:3100) or the full push URL +# (http://loki:3100/loki/api/v1/push). A Loki on your own network is fine, +# private addresses and internal hostnames included; only link-local and +# cloud-metadata addresses are refused. The username/password pair is only +# needed when your Loki sits behind basic auth. When a push fails (Loki down, +# wrong path, rejected credentials) the app writes one line to stderr naming +# the Loki origin, the reason and how many events were dropped, at most once +# every five minutes per reason; the password never appears in it. Leave all +# three unset and logs stay on stdout for the Docker log driver to rotate. # LOKI_ENDPOINT="" # LOKI_USERNAME="" # LOKI_PASSWORD="" diff --git a/docker-compose.yml b/docker-compose.yml index e55a024af..2ad99006b 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -345,10 +345,11 @@ x-healthlog-env: &healthlog_env LOG_SAMPLE_RATE: "${LOG_SAMPLE_RATE:-}" LOG_SLOW_THRESHOLD_MS: "${LOG_SLOW_THRESHOLD_MS:-}" LOG_INCLUDE_STACK: "${LOG_INCLUDE_STACK:-}" - # Optional log shipping to a Loki instance. Set LOKI_ENDPOINT to the push - # URL to turn it on; the username/password pair is only needed when your - # Loki is behind basic auth. Leave all three unset and logs stay on stdout - # for the Docker log driver to rotate. + # Optional log shipping to a Loki instance. Set LOKI_ENDPOINT to the base + # URL or the full push URL to turn it on; a Loki on your own network works. + # The username/password pair is only needed when your Loki is behind basic + # auth. Push failures show up as a rate-limited line on stderr. Leave all + # three unset and logs stay on stdout for the Docker log driver to rotate. LOKI_ENDPOINT: "${LOKI_ENDPOINT:-}" LOKI_USERNAME: "${LOKI_USERNAME:-}" LOKI_PASSWORD: "${LOKI_PASSWORD:-}" diff --git a/src/lib/logging/__tests__/transports-loki.test.ts b/src/lib/logging/__tests__/transports-loki.test.ts new file mode 100644 index 000000000..a234d8aa4 --- /dev/null +++ b/src/lib/logging/__tests__/transports-loki.test.ts @@ -0,0 +1,352 @@ +/** + * Loki shipping (#947 follow-up): the endpoint forms an operator types, the + * dial path a private Loki needs, the floor no configuration opens, and the + * stderr notice that replaces the silent catch. + * + * Two layers. The mocked-`safeFetch` block pins what the transport asks for + * and what it says when that fails. The loopback block runs the real + * `safeFetch` against a real HTTP server on 127.0.0.1, which is the case the + * public-host pin used to refuse: an address that is not public at all. + */ +import { createServer, type Server } from "node:http"; +import type { AddressInfo } from "node:net"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { resetLoggingConfig } from "../config"; +import type { WideEvent } from "../types"; + +const safeFetchMock = vi.hoisted(() => vi.fn()); +const useRealSafeFetch = vi.hoisted(() => ({ value: false })); + +vi.mock("@/lib/safe-fetch", async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + safeFetch: (...args: Parameters) => + useRealSafeFetch.value + ? actual.safeFetch(...args) + : safeFetchMock(...args), + }; +}); + +import { SafeFetchError } from "@/lib/safe-fetch"; +import { + _resetLokiTransportForTests, + emitEvent, + flushLokiBuffer, + LOKI_FAILURE_NOTICE_INTERVAL_MS, + resolveLokiPushTarget, +} from "../transports"; + +const USERNAME = "loki-user"; +const PASSWORD = "s3cret-loki-password"; +const BASIC = Buffer.from(`${USERNAME}:${PASSWORD}`).toString("base64"); + +function makeEvent(): WideEvent { + return { + timestamp: new Date().toISOString(), + duration_ms: 1, + request_id: "req-loki", + trace_id: "trace-loki", + level: "info", + kind: "http", + service: "healthlog", + environment: "test", + }; +} + +let stderrLines: string[]; +let stderrSpy: ReturnType; +let stdoutSpy: ReturnType; + +function configure(endpoint: string, withAuth = true): void { + vi.stubEnv("LOKI_ENDPOINT", endpoint); + vi.stubEnv("LOKI_USERNAME", withAuth ? USERNAME : ""); + vi.stubEnv("LOKI_PASSWORD", withAuth ? PASSWORD : ""); + resetLoggingConfig(); +} + +function buffer(count: number): void { + for (let i = 0; i < count; i++) emitEvent(makeEvent()); +} + +beforeEach(() => { + _resetLokiTransportForTests(); + safeFetchMock.mockReset(); + useRealSafeFetch.value = false; + stderrLines = []; + stderrSpy = vi + .spyOn(process.stderr, "write") + .mockImplementation((chunk: string | Uint8Array) => { + stderrLines.push(String(chunk)); + return true; + }); + stdoutSpy = vi.spyOn(process.stdout, "write").mockReturnValue(true); +}); + +afterEach(() => { + stderrSpy.mockRestore(); + stdoutSpy.mockRestore(); + vi.unstubAllEnvs(); + vi.useRealTimers(); + resetLoggingConfig(); + _resetLokiTransportForTests(); +}); + +describe("resolveLokiPushTarget", () => { + it.each([ + ["http://loki:3100", "http://loki:3100/loki/api/v1/push"], + ["http://loki:3100/", "http://loki:3100/loki/api/v1/push"], + ["http://loki:3100/loki/api/v1/push", "http://loki:3100/loki/api/v1/push"], + ["http://loki:3100/loki/api/v1/push/", "http://loki:3100/loki/api/v1/push"], + [ + "https://logs.example.com/tenant-a", + "https://logs.example.com/tenant-a/loki/api/v1/push", + ], + [ + "https://logs.example.com/tenant-a/loki/api/v1/push", + "https://logs.example.com/tenant-a/loki/api/v1/push", + ], + [" http://10.0.0.5:3100 ", "http://10.0.0.5:3100/loki/api/v1/push"], + ])("%s pushes to %s", (endpoint, expected) => { + const resolved = resolveLokiPushTarget(endpoint); + expect(resolved.ok).toBe(true); + if (resolved.ok) expect(resolved.target.url).toBe(expected); + }); + + it("approves exactly the endpoint origin, private addresses included", () => { + const resolved = resolveLokiPushTarget("http://192.168.1.20:3100/"); + expect(resolved).toEqual({ + ok: true, + target: { + url: "http://192.168.1.20:3100/loki/api/v1/push", + origin: "http://192.168.1.20:3100", + }, + }); + }); + + it.each([ + "http://169.254.169.254", + "http://169.254.10.1:3100", + "http://0.0.0.0:3100", + "http://[fe80::1]:3100", + ])("refuses the never-grantable host %s", (endpoint) => { + const resolved = resolveLokiPushTarget(endpoint); + expect(resolved.ok).toBe(false); + if (!resolved.ok) expect(resolved.reason).toBe("never_grantable"); + }); + + it.each([ + "loki:3100", + "ftp://loki:3100", + "http://user:pw@loki:3100", + "http://loki:3100?token=abc", + ])("refuses the malformed endpoint %s", (endpoint) => { + const resolved = resolveLokiPushTarget(endpoint); + expect(resolved.ok).toBe(false); + if (!resolved.ok) { + expect(resolved.reason).toBe("invalid_endpoint"); + expect(resolved.shown).not.toContain("pw"); + expect(resolved.shown).not.toContain("abc"); + } + }); +}); + +describe("flushLokiBuffer — dial path", () => { + it("dials a private endpoint as an operator-approved origin, not through the public pin", async () => { + configure("http://10.0.0.5:3100/loki/api/v1/push"); + safeFetchMock.mockResolvedValue(new Response(null, { status: 204 })); + buffer(2); + + await flushLokiBuffer(); + + expect(safeFetchMock).toHaveBeenCalledTimes(1); + const [url, init, opts] = safeFetchMock.mock.calls[0]; + expect(url).toBe("http://10.0.0.5:3100/loki/api/v1/push"); + expect(init.headers.Authorization).toBe(`Basic ${BASIC}`); + expect(opts.operatorApprovedPrivateOrigin).toBe("http://10.0.0.5:3100"); + expect(opts.requirePublicHost).toBeUndefined(); + expect(JSON.parse(init.body).streams[0].values).toHaveLength(2); + expect(stderrLines).toEqual([]); + }); + + it("never dials a metadata endpoint and says why", async () => { + configure("http://169.254.169.254:3100"); + buffer(3); + + await flushLokiBuffer(); + + expect(safeFetchMock).not.toHaveBeenCalled(); + expect(stderrLines).toHaveLength(1); + expect(stderrLines[0]).toContain("never_grantable"); + expect(stderrLines[0]).toContain("3 events dropped"); + }); +}); + +describe("flushLokiBuffer — failures reach stderr, once per window", () => { + it("reports a non-2xx answer with origin, status and dropped count", async () => { + configure("https://logs.example.com/loki/api/v1/push/loki/api/v1/push"); + safeFetchMock.mockResolvedValue( + new Response("404 page not found", { status: 404 }), + ); + buffer(4); + + await flushLokiBuffer(); + + expect(stderrLines).toHaveLength(1); + expect(stderrLines[0]).toContain("https://logs.example.com"); + expect(stderrLines[0]).toContain("HTTP 404"); + expect(stderrLines[0]).toContain("4 events dropped"); + }); + + it("reports a thrown error by class and socket code, never by message", async () => { + configure("http://loki.lan:3100"); + const cause = Object.assign(new Error("connect ECONNREFUSED"), { + code: "ECONNREFUSED", + }); + safeFetchMock.mockRejectedValue( + new SafeFetchError( + `safeFetch network error: http://${USERNAME}:${PASSWORD}@loki.lan`, + "network", + { cause }, + ), + ); + buffer(1); + + await flushLokiBuffer(); + + expect(stderrLines).toHaveLength(1); + expect(stderrLines[0]).toContain("http://loki.lan:3100"); + expect(stderrLines[0]).toContain("network ECONNREFUSED"); + expect(stderrLines[0]).toContain("1 event dropped"); + }); + + it("writes one line per reason per window and carries the suppressed count forward", async () => { + vi.useFakeTimers({ toFake: ["Date"] }); + configure("http://loki.lan:3100"); + safeFetchMock.mockImplementation( + async () => new Response(null, { status: 503 }), + ); + + for (let i = 0; i < 5; i++) { + buffer(2); + await flushLokiBuffer(); + vi.advanceTimersByTime(5_000); + } + expect(stderrLines).toHaveLength(1); + expect(stderrLines[0]).toContain("2 events dropped"); + + // A different reason is not muted by the first one's window. + safeFetchMock.mockImplementationOnce( + async () => new Response(null, { status: 401 }), + ); + buffer(1); + await flushLokiBuffer(); + expect(stderrLines).toHaveLength(2); + expect(stderrLines[1]).toContain("HTTP 401"); + + vi.advanceTimersByTime(LOKI_FAILURE_NOTICE_INTERVAL_MS); + buffer(2); + await flushLokiBuffer(); + expect(stderrLines).toHaveLength(3); + expect(stderrLines[2]).toContain("HTTP 503"); + // Four suppressed batches of two plus this batch of two. + expect(stderrLines[2]).toContain("10 events dropped"); + }); + + it("never prints the credentials or the basic-auth header", async () => { + configure("http://loki.lan:3100"); + safeFetchMock.mockResolvedValueOnce(new Response(null, { status: 403 })); + safeFetchMock.mockRejectedValueOnce( + new SafeFetchError(`failed with Basic ${BASIC} ${PASSWORD}`, "timeout"), + ); + buffer(1); + await flushLokiBuffer(); + buffer(1); + await flushLokiBuffer(); + + expect(stderrLines).toHaveLength(2); + const all = stderrLines.join(""); + expect(all).not.toContain(PASSWORD); + expect(all).not.toContain(USERNAME); + expect(all).not.toContain(BASIC); + }); + + it("stays silent on success", async () => { + configure("http://loki.lan:3100"); + safeFetchMock.mockResolvedValue(new Response(null, { status: 204 })); + buffer(1); + await flushLokiBuffer(); + expect(stderrLines).toEqual([]); + }); + + it("does not feed its own notice back into the Loki buffer", async () => { + configure("http://loki.lan:3100"); + safeFetchMock.mockResolvedValue(new Response(null, { status: 500 })); + buffer(1); + await flushLokiBuffer(); + expect(stderrLines).toHaveLength(1); + + // Nothing was buffered by the failure, so the next flush has nothing + // to send. + safeFetchMock.mockClear(); + await flushLokiBuffer(); + expect(safeFetchMock).not.toHaveBeenCalled(); + }); +}); + +describe("flushLokiBuffer — real dial to a loopback Loki", () => { + let server: Server; + let received: Array<{ url: string; auth: string | undefined }>; + let status: number; + + beforeEach(async () => { + useRealSafeFetch.value = true; + received = []; + status = 204; + server = createServer((req, res) => { + received.push({ url: req.url ?? "", auth: req.headers.authorization }); + req.resume(); + req.on("end", () => { + res.statusCode = status; + res.end(); + }); + }); + await new Promise((resolve) => + server.listen(0, "127.0.0.1", resolve), + ); + }); + + afterEach(async () => { + await new Promise((resolve) => server.close(() => resolve())); + }); + + function endpoint(path = ""): string { + const { port } = server.address() as AddressInfo; + return `http://127.0.0.1:${port}${path}`; + } + + it("delivers to a non-public address given as the full push URL", async () => { + configure(endpoint("/loki/api/v1/push")); + buffer(1); + + await flushLokiBuffer(); + + expect(received).toEqual([ + { url: "/loki/api/v1/push", auth: `Basic ${BASIC}` }, + ]); + expect(stderrLines).toEqual([]); + }); + + it("reports the 404 a wrong path produces instead of swallowing it", async () => { + status = 404; + configure(endpoint("/wrong")); + buffer(2); + + await flushLokiBuffer(); + + expect(received).toHaveLength(1); + expect(received[0].url).toBe("/wrong/loki/api/v1/push"); + expect(stderrLines).toHaveLength(1); + expect(stderrLines[0]).toContain(`${endpoint()} failed (HTTP 404)`); + }); +}); diff --git a/src/lib/logging/transports.ts b/src/lib/logging/transports.ts index 4bbddb090..0e1b4cb02 100644 --- a/src/lib/logging/transports.ts +++ b/src/lib/logging/transports.ts @@ -2,7 +2,8 @@ import type { WideEvent } from "./types"; import { getLoggingConfig } from "./config"; import { shouldEmit } from "./sampler"; import { appendLogEvent } from "./in-memory-buffer"; -import { safeFetch } from "@/lib/safe-fetch"; +import { safeFetch, SafeFetchError } from "@/lib/safe-fetch"; +import { canonicalOrigin } from "@/lib/private-origin-policy"; /** Event auf stdout als einzelne JSON-Zeile schreiben */ function emitToStdout(event: WideEvent): void { @@ -31,7 +32,133 @@ function initLokiTransport(): void { if (lokiFlushTimer.unref) lokiFlushTimer.unref(); } -async function flushLokiBuffer(): Promise { +const LOKI_PUSH_PATH = "/loki/api/v1/push"; +const LOKI_PUSH_TIMEOUT_MS = 10_000; + +export interface LokiPushTarget { + /** The URL the batch is POSTed to. */ + url: string; + /** + * The exact origin handed to `safeFetch` as the operator approval. Also the + * only part of the endpoint that ever appears in a failure line: it carries + * no userinfo, path or query by construction. + */ + origin: string; +} + +export type LokiTargetResolution = + | { ok: true; target: LokiPushTarget } + | { + ok: false; + reason: "invalid_endpoint" | "never_grantable"; + shown: string; + }; + +/** + * Turn `LOKI_ENDPOINT` into the push URL and the origin it may dial. + * + * Both spellings an operator reaches for work: the base URL + * (`http://loki:3100`, the push path is appended) and the full push URL + * (`http://loki:3100/loki/api/v1/push`, used as is, trailing slash or not). + * Appending unconditionally turned the second form into + * `/loki/api/v1/push/loki/api/v1/push` and a 404. + * + * Why the endpoint dials as an operator-approved origin rather than through + * the public-host pin: `LOKI_ENDPOINT` is read from the server environment in + * `config.ts` and nowhere else. No request field, settings row or admin form + * can set it, so there is no user-supplied host to defend against, and the + * public-only pin was refusing the common deployment (a Loki on the same LAN + * or Docker network) without a word. The exact origin still goes through the + * pinned operator resolver in `safeFetch` with redirects forbidden, and the + * never-grantable floor holds: an endpoint whose literal host is the + * unspecified address, link-local or the metadata range is refused here + * (`canonicalOrigin`), and a name that resolves there is dropped at dial time. + */ +export function resolveLokiPushTarget(endpoint: string): LokiTargetResolution { + let url: URL; + try { + url = new URL(endpoint.trim()); + } catch { + return { ok: false, reason: "invalid_endpoint", shown: "(unparseable)" }; + } + if (url.protocol !== "http:" && url.protocol !== "https:") { + return { ok: false, reason: "invalid_endpoint", shown: url.protocol }; + } + // `url.origin` never carries userinfo, so it is safe to print even when + // the endpoint itself is refused for carrying credentials. + if (url.username || url.password || url.search || url.hash) { + return { ok: false, reason: "invalid_endpoint", shown: url.origin }; + } + const origin = canonicalOrigin(url.origin); + if (!origin) { + return { ok: false, reason: "never_grantable", shown: url.origin }; + } + const basePath = url.pathname.replace(/\/+$/, ""); + const pushPath = basePath.endsWith(LOKI_PUSH_PATH) + ? basePath + : `${basePath}${LOKI_PUSH_PATH}`; + return { ok: true, target: { url: `${origin}${pushPath}`, origin } }; +} + +/** + * Failure notices for the Loki push, written straight to stderr. + * + * Never through `emitEvent`: a failing Loki push that reported itself as a + * wide event would buffer that event for the same failing push. At most one + * line per reason per window, so a Loki that is down for an hour costs twelve + * lines, not seven hundred. Events dropped while a reason is quiet are counted + * and reported on its next line, so the numbers add up. + */ +export const LOKI_FAILURE_NOTICE_INTERVAL_MS = 5 * 60 * 1000; +const lokiFailureNotices = new Map< + string, + { lastAt: number; droppedSinceLast: number } +>(); + +function reportLokiFailure( + shownEndpoint: string, + reason: string, + droppedEvents: number, +): void { + const now = Date.now(); + const entry = lokiFailureNotices.get(reason); + if (entry && now - entry.lastAt < LOKI_FAILURE_NOTICE_INTERVAL_MS) { + entry.droppedSinceLast += droppedEvents; + return; + } + const dropped = droppedEvents + (entry?.droppedSinceLast ?? 0); + lokiFailureNotices.set(reason, { lastAt: now, droppedSinceLast: 0 }); + process.stderr.write( + `[logging] Loki push to ${shownEndpoint} failed (${reason}); ` + + `${dropped} event${dropped === 1 ? "" : "s"} dropped. ` + + `Repeats of this failure are reported at most every 5 minutes.\n`, + ); +} + +/** + * The failure reason as a short class, never the error message: messages + * from the fetch stack can embed the target URL, and this line goes to stderr + * unredacted. A socket error code (`ECONNREFUSED`, `ENOTFOUND`) is kept + * because it is the part that tells an operator what to fix. + */ +function classifyLokiError(err: unknown): string { + const kind = err instanceof SafeFetchError ? err.kind : "error"; + let cause: unknown = err instanceof SafeFetchError ? err.cause : err; + for ( + let depth = 0; + depth < 4 && cause && typeof cause === "object"; + depth++ + ) { + const code = (cause as { code?: unknown }).code; + if (typeof code === "string" && /^[A-Z][A-Z0-9_]{1,40}$/.test(code)) { + return `${kind} ${code}`; + } + cause = (cause as { cause?: unknown }).cause; + } + return kind; +} + +export async function flushLokiBuffer(): Promise { if (lokiBuffer.length === 0) return; const config = getLoggingConfig(); if (!config.lokiEndpoint) return; @@ -39,6 +166,13 @@ async function flushLokiBuffer(): Promise { const batch = lokiBuffer; lokiBuffer = []; + const resolved = resolveLokiPushTarget(config.lokiEndpoint); + if (!resolved.ok) { + reportLokiFailure(resolved.shown, resolved.reason, batch.length); + return; + } + const { target } = resolved; + const streams = [ { stream: { @@ -64,21 +198,41 @@ async function flushLokiBuffer(): Promise { ); } + let res: Response; try { - await safeFetch( - `${config.lokiEndpoint}/loki/api/v1/push`, + res = await safeFetch( + target.url, { method: "POST", headers, body: JSON.stringify({ streams }), }, - // Operator-configured Loki endpoint — pin the connect-time IP - // against DNS rebinding. - { timeoutMs: 10_000, requirePublicHost: true }, + // Operator-configured endpoint: dial exactly its origin through the + // pinned operator resolver (see `resolveLokiPushTarget`). + { + timeoutMs: LOKI_PUSH_TIMEOUT_MS, + operatorApprovedPrivateOrigin: target.origin, + }, ); - } catch { - // Events gehen verloren — akzeptabel fuer Logging + } catch (err) { + // The batch is already detached from the buffer; say so rather than + // losing it without a trace. + reportLokiFailure(target.origin, classifyLokiError(err), batch.length); + return; } + + // Loki answers 204. Anything else, including a 3xx the forbidden redirect + // left unfollowed, means the batch did not land. + if (!res.ok) { + reportLokiFailure(target.origin, `HTTP ${res.status}`, batch.length); + } + await res.body?.cancel().catch(() => {}); +} + +/** Test helper: clear the buffer and the failure-notice windows. */ +export function _resetLokiTransportForTests(): void { + lokiBuffer = []; + lokiFailureNotices.clear(); } /**