astral/kuber
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: scope request logs to incoming traffic

Browse Source
This commit is contained in:
dmgnr
2026-10-02 13:44:14 +00:00 Unverified
parent d3e9753c4b
commit 00fe2bb2a6
2 changed files with 383 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/request-log.ts
+62 -1
View File
@@ -1,3 +1,5 @@
import { AsyncLocalStorage } from "node:async_hooks";
export type ProcessLogEntry = Record<string, unknown>;
export type ProcessLogger = {
@@ -5,6 +7,7 @@ export type ProcessLogger = {
};
const MAX_RESPONSE_LOG_BYTES = 64 * 1024;
const requestContext = new AsyncLocalStorage<{ active: boolean }>();
export function logValue(
value: unknown,
@@ -42,10 +45,63 @@ export function logValue(
}
}
const METHOD_LABELS: Record<string, string> = {
GET: "GET",
POST: "PST",
PUT: "PUT",
PATCH: "PTC",
DELETE: "DEL",
HEAD: "HED",
OPTIONS: "OPT",
CONNECT: "CON",
TRACE: "TRC",
};
function requestLine(entry: ProcessLogEntry): string | undefined {
const event = entry.event;
if (
event === "kuber.server.request.start" ||
event === "kuber.k8s.request.start" ||
event === "kuber.server.response.body"
)
return "";
const kubernetes =
event === "kuber.k8s.request.end" ||
event === "kuber.k8s.request.failed";
if (kubernetes && !requestContext.getStore()?.active) return "";
if (
!kubernetes &&
event !== "kuber.server.request.end" &&
event !== "kuber.server.request.failed"
)
return;
if (typeof entry.method !== "string" || typeof entry.pathname !== "string")
return;
if (
!kubernetes &&
entry.method.toUpperCase() === "GET" &&
entry.pathname === "/api/v2/health"
)
return "";
const method = entry.method.toUpperCase();
const label = METHOD_LABELS[method] ?? method.slice(0, 3).padEnd(3, "_");
const status =
event === "kuber.k8s.request.failed" ||
event === "kuber.server.request.failed"
? "ERR"
: entry.status;
if (typeof status !== "number" && status !== "ERR") return;
const indent = kubernetes ? " " : "";
return `${indent}${label} ${entry.pathname} ${status}`;
}
export const processLogger: ProcessLogger = {
log(entry) {
try {
console.log(`KUBER_REQUEST ${JSON.stringify(logValue(entry))}`);
const line = requestLine(entry);
if (line === "") return;
console.log(line ?? `KUBER_REQUEST ${JSON.stringify(logValue(entry))}`);
} catch {
// Process logging must never change request behavior.
}
@@ -168,6 +224,8 @@ export async function logServerRequest<T>(
headers: headers(request.headers),
};
safeLog(logger, { event: "kuber.server.request.start", ...base });
const context = { active: true };
return requestContext.run(context, async () => {
try {
const response = await handler();
safeLog(logger, {
@@ -194,7 +252,10 @@ export async function logServerRequest<T>(
error: logValue(error),
});
throw error;
} finally {
context.active = false;
}
});
}
export async function logKubernetesRequest<T>(
tests/lib/request-log.test.ts
+296 -2
View File
@@ -1,5 +1,299 @@
import { describe, expect, test } from "bun:test";
import { logServerRequest } from "../../lib/request-log";
import { describe, expect, spyOn, test } from "bun:test";
import {
logKubernetesRequest,
logServerRequest,
processLogger,
} from "../../lib/request-log";
describe("process request lines", () => {
test("hides health GET access lines while preserving events and in-scope Kubernetes lines", async () => {
const output = spyOn(console, "log").mockImplementation(() => {});
const logs: Record<string, unknown>[] = [];
const logger = {
log(entry: Record<string, unknown>) {
logs.push(entry);
processLogger.log(entry);
},
};
try {
await logServerRequest(
new Request("https://kuber.astrxl.dev/api/v2/health?probe=ready"),
"health-request",
async () => {
await logKubernetesRequest(
{ method: "GET", pathname: "/api/v1/health-dependency" },
async () => "ok",
);
return new Response(null, { status: 200 });
},
logger,
);
await expect(
logServerRequest(
new Request("https://kuber.astrxl.dev/api/v2/health?probe=failed"),
"failed-health-request",
() => {
throw new Error("unavailable");
},
logger,
),
).rejects.toThrow("unavailable");
await logServerRequest(
new Request("https://kuber.astrxl.dev/api/v2/health"),
"default-health-request",
() => new Response(null, { status: 200 }),
);
await logServerRequest(
new Request("https://kuber.astrxl.dev/api/v2/health", {
method: "POST",
}),
"post-health-request",
() => new Response(null, { status: 405 }),
);
await logServerRequest(
new Request("https://kuber.astrxl.dev/api/v2/other"),
"other-request",
() => new Response(null, { status: 200 }),
);
expect(output.mock.calls.map(([line]) => line)).toEqual([
" GET /api/v1/health-dependency 200",
"PST /api/v2/health 405",
"GET /api/v2/other 200",
]);
expect(logs).toContainEqual(
expect.objectContaining({
event: "kuber.server.request.end",
requestId: "health-request",
method: "GET",
pathname: "/api/v2/health",
status: 200,
}),
);
expect(logs).toContainEqual(
expect.objectContaining({
event: "kuber.server.request.failed",
requestId: "failed-health-request",
pathname: "/api/v2/health",
}),
);
} finally {
output.mockRestore();
}
});
test("prints one application line with pathname and completion status", async () => {
const output = spyOn(console, "log").mockImplementation(() => {});
try {
await logServerRequest(
new Request("https://kuber.astrxl.dev/path/to/endpoint?token=secret"),
"get-request",
() => new Response(null, { status: 404 }),
);
await logServerRequest(
new Request("https://kuber.astrxl.dev/path/to/another?key=value", {
method: "POST",
}),
"post-request",
() => new Response(null, { status: 200 }),
);
expect(output.mock.calls.map(([line]) => line)).toEqual([
"GET /path/to/endpoint 404",
"PST /path/to/another 200",
]);
} finally {
output.mockRestore();
}
});
test("indents Kubernetes lines for incoming requests without a CLI marker", async () => {
const output = spyOn(console, "log").mockImplementation(() => {});
try {
await logServerRequest(
new Request("https://kuber.astrxl.dev/workloads", {
headers: { "user-agent": "external-client" },
}),
"workloads-request",
async () => {
await logKubernetesRequest(
{ method: "GET", pathname: "/path/to/kubernetes/api" },
async () => "ok",
);
await logKubernetesRequest(
{ method: "POST", pathname: "/api/v1/pods" },
async () => "created",
);
return new Response(null, { status: 200 });
},
);
expect(output.mock.calls.map(([line]) => line)).toEqual([
" GET /path/to/kubernetes/api 200",
" PST /api/v1/pods 200",
"GET /workloads 200",
]);
} finally {
output.mockRestore();
}
});
test("keeps full methods in structured events and marks failures", async () => {
const logs: Record<string, unknown>[] = [];
await logServerRequest(
new Request("https://kuber.astrxl.dev/items?q=private", {
method: "DELETE",
}),
"delete-request",
() => new Response(null, { status: 204 }),
{ log: (entry) => logs.push(entry) },
);
expect(logs).toContainEqual(
expect.objectContaining({
event: "kuber.server.request.end",
method: "DELETE",
pathname: "/items",
status: 204,
}),
);
const output = spyOn(console, "log").mockImplementation(() => {});
try {
await expect(
logServerRequest(
new Request("https://kuber.astrxl.dev/failing"),
"failing-request",
async () => {
await logKubernetesRequest(
{ method: "CUSTOM", pathname: "/api/v1/fail" },
async () => {
throw new Error("unavailable");
},
);
},
),
).rejects.toThrow("unavailable");
expect(output.mock.calls.map(([line]) => line)).toEqual([
" CUS /api/v1/fail ERR",
"GET /failing ERR",
]);
processLogger.log({
event: "kuber.server.request.end",
method: "X",
pathname: "/short",
status: 200,
});
expect(output.mock.calls[2]?.[0]).toBe("X__ /short 200");
} finally {
output.mockRestore();
}
});
test("isolates concurrent requests from background work and closes completed contexts", async () => {
const output = spyOn(console, "log").mockImplementation(() => {});
const events: Record<string, unknown>[] = [];
let releaseFirst!: () => void;
const firstGate = new Promise<void>((resolve) => (releaseFirst = resolve));
let releaseSecond!: () => void;
const secondGate = new Promise<void>((resolve) => (releaseSecond = resolve));
let startedFirst!: () => void;
const firstStarted = new Promise<void>((resolve) => (startedFirst = resolve));
let startedSecond!: () => void;
const secondStarted = new Promise<void>(
(resolve) => (startedSecond = resolve),
);
let releaseDetached!: () => void;
const detachedGate = new Promise<void>(
(resolve) => (releaseDetached = resolve),
);
let detached!: Promise<string>;
const customLogger = {
log: (entry: Record<string, unknown>) => events.push(entry),
};
try {
const first = logServerRequest(
new Request("https://kuber.astrxl.dev/first"),
"first",
async () => {
detached = (async () => {
await detachedGate;
return logKubernetesRequest(
{ method: "GET", pathname: "/detached" },
async () => "done",
);
})();
startedFirst();
await firstGate;
await logKubernetesRequest(
{ method: "PUT", pathname: "/first-k8s" },
async () => "done",
);
return new Response(null, { status: 200 });
},
);
await firstStarted;
const second = logServerRequest(
new Request("https://kuber.astrxl.dev/second"),
"second",
async () => {
startedSecond();
await secondGate;
await logKubernetesRequest(
{ method: "GET", pathname: "/second-k8s" },
async () => "done",
);
return new Response(null, { status: 200 });
},
);
await secondStarted;
await logKubernetesRequest(
{ method: "GET", pathname: "/background" },
async () => "done",
customLogger,
);
await logKubernetesRequest(
{ method: "GET", pathname: "/background-default" },
async () => "done",
);
await expect(
logKubernetesRequest(
{ method: "DELETE", pathname: "/background-failed" },
async () => {
throw new Error("background failure");
},
),
).rejects.toThrow("background failure");
expect(events.map((entry) => entry.event)).toEqual([
"kuber.k8s.request.start",
"kuber.k8s.request.end",
]);
releaseFirst();
await first;
releaseDetached();
await detached;
releaseSecond();
await second;
await logKubernetesRequest(
{ method: "GET", pathname: "/after" },
async () => "done",
);
expect(output.mock.calls.map(([line]) => line)).toEqual([
" PUT /first-k8s 200",
"GET /first 200",
" GET /second-k8s 200",
"GET /second 200",
]);
} finally {
releaseFirst();
releaseSecond();
releaseDetached();
output.mockRestore();
}
});
});
describe("server response logging", () => {
test("does not consume streaming response bodies", async () => {