fix: use RWO for single-replica Longhorn volumes

changelogs/2.5.2.md

@@ -1,4 +1,4 @@
-# 2.5.2-rc.1
+# 2.5.2
 
 ## Fixed
 

command/main.ts

@@ -37,7 +37,7 @@ function cloneCommand<T extends object>(command: T): T {
 export const main = defineCommand({
   meta: {
     name: "kuber",
-    version: "2.5.2-rc.1",
+    version: "2.5.2",
     description: "Docker Compose -> K8s translation layer",
   },
   args: {

lib/convert.ts

@@ -1286,7 +1286,9 @@ export function volumesToPvc(
         labels: LABELS,
       },
       spec: {
-        accessModes: ["ReadWriteMany"],
+        accessModes: [
+          policy?.replicaCount === 1 ? "ReadWriteOnce" : "ReadWriteMany",
+        ],
         storageClassName: policy
           ? toLonghornStorageClassName(policy)
           : undefined,

lib/request-log.ts

@@ -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,33 +224,38 @@ export async function logServerRequest<T>(
     headers: headers(request.headers),
   };
   safeLog(logger, { event: "kuber.server.request.start", ...base });
-  try {
+  const context = { active: true };
-    const response = await handler();
+  return requestContext.run(context, async () => {
-    safeLog(logger, {
+    try {
-      event: "kuber.server.request.end",
+      const response = await handler();
-      ...base,
+      safeLog(logger, {
-      status: response instanceof Response ? response.status : 101,
+        event: "kuber.server.request.end",
-      durationMs: Date.now() - startedAt,
-      ...(response instanceof Response && {
-        responseHeaders: headers(response.headers),
-      }),
-    });
-    if (response instanceof Response)
-      logResponseBody(response, logger, {
-        event: "kuber.server.response.body",
         ...base,
-        status: response.status,
+        status: response instanceof Response ? response.status : 101,
+        durationMs: Date.now() - startedAt,
+        ...(response instanceof Response && {
+          responseHeaders: headers(response.headers),
+        }),
       });
-    return response;
+      if (response instanceof Response)
-  } catch (error) {
+        logResponseBody(response, logger, {
-    safeLog(logger, {
+          event: "kuber.server.response.body",
-      event: "kuber.server.request.failed",
+          ...base,
-      ...base,
+          status: response.status,
-      durationMs: Date.now() - startedAt,
+        });
-      error: logValue(error),
+      return response;
-    });
+    } catch (error) {
-    throw error;
+      safeLog(logger, {
-  }
+        event: "kuber.server.request.failed",
+        ...base,
+        durationMs: Date.now() - startedAt,
+        error: logValue(error),
+      });
+      throw error;
+    } finally {
+      context.active = false;
+    }
+  });
 }
 
 export async function logKubernetesRequest<T>(

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@dmgnr/kuber",
-  "version": "2.5.2-rc.1",
+  "version": "2.5.2",
   "description": "Docker Compose to Kubernetes translation layer",
   "bin": {
     "kuber": "dist/index.js"

tests/lib/convert-storage-env.test.ts

@@ -38,6 +38,7 @@ describe("volume conversion", () => {
     expect(claims[0]).toMatchObject({
       metadata: { name: "project-data", namespace: "project" },
       spec: {
+        accessModes: ["ReadWriteMany"],
         resources: { requests: { storage: "1Gi" } },
         storageClassName: expect.stringMatching(
           /^kuber-longhorn-[a-f0-9]{12}$/,
@@ -64,6 +65,16 @@ describe("volume conversion", () => {
     ]);
   });
 
+  test("renders single-replica Longhorn volumes as ReadWriteOnce", () => {
+    const service = { volumes: ["data(20Gi on 1 fast):/data"] } as Service;
+    const [claim] = volumesToPvc("project", service);
+    const [storageClass] = volumesToStorageClasses(service);
+
+    expect(claim?.spec?.accessModes).toEqual(["ReadWriteOnce"]);
+    expect(claim?.spec?.storageClassName).toBe(storageClass?.metadata?.name);
+    expect(storageClass?.parameters?.numberOfReplicas).toBe("1");
+  });
+
   test("parses inline storage sizing and placement", () => {
     const [claim] = volumesToPvc("project", {
       volumes: ["data(20Gi on 3 fast,archive):/data"],
@@ -71,6 +82,7 @@ describe("volume conversion", () => {
     expect(claim).toMatchObject({
       metadata: { name: "project-data" },
       spec: {
+        accessModes: ["ReadWriteMany"],
         resources: { requests: { storage: "20Gi" } },
         storageClassName: expect.stringMatching(/^kuber-longhorn-/),
       },
@@ -103,11 +115,29 @@ describe("volume conversion", () => {
       } as ComposeSpecification["volumes"],
     );
     expect(claim?.spec).toMatchObject({
+      accessModes: ["ReadWriteMany"],
       resources: { requests: { storage: "50Gi" } },
       storageClassName: expect.stringMatching(/^kuber-longhorn-/),
     });
   });
 
+  test("uses top-level single-replica policy for the PVC and StorageClass", () => {
+    const service = { volumes: ["data:/data"] } as Service;
+    const volumes = {
+      data: { "x-replicaCount": 1 },
+    } as ComposeSpecification["volumes"];
+    const [claim] = volumesToPvc("project", service, process.cwd(), volumes);
+    const [storageClass] = volumesToStorageClasses(
+      service,
+      process.cwd(),
+      volumes,
+    );
+
+    expect(claim?.spec?.accessModes).toEqual(["ReadWriteOnce"]);
+    expect(claim?.spec?.storageClassName).toBe(storageClass?.metadata?.name);
+    expect(storageClass?.parameters?.numberOfReplicas).toBe("1");
+  });
+
   test("does not create PVCs for anonymous, tmpfs, or managed claims", () => {
     expect(
       volumesToPvc("project", {

tests/lib/request-log.test.ts

@@ -1,5 +1,299 @@
-import { describe, expect, test } from "bun:test";
+import { describe, expect, spyOn, test } from "bun:test";
-import { logServerRequest } from "../../lib/request-log";
+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 () => {