diff --git a/apps/desktop/src/renderer/App.tsx b/apps/desktop/src/renderer/App.tsx index d615795..ef182a7 100644 --- a/apps/desktop/src/renderer/App.tsx +++ b/apps/desktop/src/renderer/App.tsx @@ -22,6 +22,7 @@ import { useDiagnostics } from "./hooks/useDiagnostics"; import { useDiscovery } from "./hooks/useDiscovery"; import { useHelm } from "./hooks/useHelm"; import { useMutation } from "./hooks/useMutation"; +import { discardPortForwards, retryPendingForwardStops, setPortForwardContext, usePendingForwardStops } from "./hooks/usePortForwards"; import { useNamespaces } from "./hooks/useNamespaces"; import { useOverview } from "./hooks/useOverview"; import { useResourceDetail } from "./hooks/useResourceDetail"; @@ -62,6 +63,11 @@ export default function App() { const contexts = useContexts(core); const { contextId } = contexts; const [error, setError] = useState(""); + const pendingForwardStops = usePendingForwardStops(); + useEffect(() => { + if (core.state === "ready") void setPortForwardContext(contextId); + else discardPortForwards(); + }, [contextId, core.state]); const [kind, setKind] = useState(DEFAULT_KIND); const namespaces = useNamespaces(contextId, contexts.contexts, setError); const resources = useResourceList({ @@ -320,6 +326,7 @@ export default function App() { setError(""); namespaces.setNamespace(target.namespace || ""); contexts.setContextChoice(target.id); + void setPortForwardContext(target.id).catch((cause) => setError(messageOf(cause))); contexts.setContextId(target.id); contexts.setView("workbench"); localStorage.setItem("aster.lastContext", target.id); @@ -331,6 +338,7 @@ export default function App() { const showContextPicker = useCallback(() => { contexts.setContextChoice(contextId); contexts.setContextQuery(""); + void setPortForwardContext("").catch((cause) => setError(messageOf(cause))); contexts.setContextId(""); namespaces.setNamespace(""); resources.setQuery(""); @@ -473,6 +481,22 @@ export default function App() { void desktop.app.version().then(setAppVersion).catch(() => undefined); }, []); + const forwardCleanupNotice = pendingForwardStops.length > 0 ? ( + + ) : null; + if (contexts.view === "settings") { return ( <> @@ -502,7 +526,7 @@ export default function App() { if (contexts.settingsFrom === "contexts") void contexts.loadContexts(); }} /> - {updateCard && void desktop.app.openExternal(url)} />} + {forwardCleanupNotice || (updateCard && void desktop.app.openExternal(url)} />)} ); } @@ -548,7 +572,7 @@ export default function App() { }} onOpenExternal={(url) => void desktop.app.openExternal(url)} /> - {updateCard && void desktop.app.openExternal(url)} />} + {forwardCleanupNotice || (updateCard && void desktop.app.openExternal(url)} />)} ); } @@ -746,7 +770,7 @@ export default function App() { )} - {welcomeVisible ? ( + {forwardCleanupNotice || (welcomeVisible ? ( ) : updateCard ? ( void desktop.app.openExternal(url)} /> - ) : null} + ) : null)} { diff --git a/apps/desktop/src/renderer/detail/OverviewTab.tsx b/apps/desktop/src/renderer/detail/OverviewTab.tsx index a4b5620..c9f4132 100644 --- a/apps/desktop/src/renderer/detail/OverviewTab.tsx +++ b/apps/desktop/src/renderer/detail/OverviewTab.tsx @@ -4,6 +4,8 @@ import { useMemo, useState, type ReactNode } from "react"; import type { RelatedResource, ResourceEvent, ResourceRow } from "../../shared/types"; import { Button } from "../components/ui/button"; import { StatusDot } from "../components/ResourceTable"; +import { PortForwardSection } from "./PortForwardSection"; +import type { ForwardPort } from "./port-forward-ports"; import { formatReady } from "../lib/format"; import { formatAge, formatTimestamp } from "./resource-format"; import type { WorkloadCondition, WorkloadDetails } from "./workload-detail"; diff --git a/apps/desktop/src/renderer/detail/PortForwardSection.tsx b/apps/desktop/src/renderer/detail/PortForwardSection.tsx new file mode 100644 index 0000000..f9e597d --- /dev/null +++ b/apps/desktop/src/renderer/detail/PortForwardSection.tsx @@ -0,0 +1,182 @@ +import { ArrowRightLeft, Check, Copy, LoaderCircle, Square } from "lucide-react"; +import { useState } from "react"; + +import { Button } from "../components/ui/button"; +import { forwardKey, usePortForwards } from "../hooks/usePortForwards"; +import type { ForwardPort } from "./port-forward-ports"; + +export interface PortForwardSectionProps { + contextId: string; + namespace: string; + name: string; + kind: string; + ports: ForwardPort[]; +} + +/** + * Forwardable TCP ports for one resource. Each declared port row starts or + * stops a forward; a manual input covers pods listening on undeclared ports. + * Forwards live in a module-scoped store, so they survive navigation. + */ +export function PortForwardSection({ contextId, namespace, name, kind, ports }: PortForwardSectionProps) { + const { entries, start, stop, byKey } = usePortForwards(contextId); + const [manualPort, setManualPort] = useState(""); + const [manualLocalPort, setManualLocalPort] = useState(""); + const [localPorts, setLocalPorts] = useState>({}); + const visiblePorts = [...new Map(ports.map((port) => [port.port, port])).values()]; + for (const entry of entries) { + if (entry.kind === kind && entry.namespace === namespace && entry.name === name && + !visiblePorts.some((port) => port.port === entry.podPort)) { + visiblePorts.push({ label: "Other port", port: entry.podPort, protocol: "TCP" }); + } + } + function startForward(podPort: number, localValue = localPorts[podPort] ?? "") { + if (!validLocalPort(localValue)) return; + const localPort = localValue === "" ? 0 : Number(localValue); + void start({ + contextId, + namespace, + name, + podPort, + kind, + localPort, + }); + } + + const manualValue = Number(manualPort); + const manualValid = Number.isInteger(manualValue) && manualValue >= 1 && manualValue <= 65_535 && validLocalPort(manualLocalPort); + + return ( +
+
+
+

Port forwarding

+

Connect through a local TCP port. Leave the local port empty to assign one automatically.

+
+
+ + +
+ {visiblePorts.map((port) => { + const key = forwardKey(kind, namespace, name, port.port); + const entry = byKey(key); + return ( +
+ {port.label} + {port.port}/{port.protocol} + {entry?.localPort ? ( + <> +
+
+ localhost:{entry.localPort} + +
+ {entry.pod ? via {entry.pod} : null} +
+ + + ) : ( + <> + + setLocalPorts((current) => ({ ...current, [port.port]: event.target.value.replace(/[^0-9]/g, "").slice(0, 5) })) + } + /> + + + )} +
+ ); + })} + +
{ + event.preventDefault(); + if (!manualValid) return; + startForward(manualValue, manualLocalPort); + setManualPort(""); + }} + > + Other port + setManualPort(event.target.value.replace(/[^0-9]/g, ""))} + /> + setManualLocalPort(event.target.value.replace(/[^0-9]/g, "").slice(0, 5))} + /> + +
+
+ +

+ {[...new Set(visiblePorts.map((port) => byKey(forwardKey(kind, namespace, name, port.port))?.error).filter(Boolean))].join(" · ")} +

+
+ ); +} + +function validLocalPort(value: string): boolean { + return value === "" || (Number.isInteger(Number(value)) && Number(value) >= 1 && Number(value) <= 65_535); +} + +function CopyLocalButton({ port }: { port: number }) { + const [copied, setCopied] = useState(false); + return ( + + ); +} diff --git a/apps/desktop/src/renderer/detail/ResourceDetailView.tsx b/apps/desktop/src/renderer/detail/ResourceDetailView.tsx index 06e4a01..ee91fa1 100644 --- a/apps/desktop/src/renderer/detail/ResourceDetailView.tsx +++ b/apps/desktop/src/renderer/detail/ResourceDetailView.tsx @@ -37,6 +37,8 @@ import { DetailHeader } from "./DetailHeader"; import { LogViewer } from "./LogViewer"; import { MutationDiffView } from "./MutationDiffView"; import { OverviewTab, type PodsPreview } from "./OverviewTab"; +import { PortForwardSection } from "./PortForwardSection"; +import { extractForwardPorts } from "./port-forward-ports"; import { resourceActionsFor, type ResourceActionId } from "./resource-actions"; import { formatTimestamp } from "./resource-format"; import { HighlightedYaml } from "./yaml-highlight"; @@ -48,7 +50,7 @@ type MutationDraft = Omit< "contextId" | "resourceKind" | "namespace" | "name" >; -type DetailTab = "overview" | "pods" | "yaml" | "events" | "related" | "logs"; +type DetailTab = "overview" | "ports" | "pods" | "yaml" | "events" | "related" | "logs"; type OperationDialog = "image" | null; /** Static catalog entry; module-level so the pods hook sees a stable reference. */ @@ -175,6 +177,15 @@ export function ResourceDetailView({ }); // Live CPU/memory for a single Pod; the hook idles (no polls) for other kinds. const isPod = row?.kind === "Pod"; + // The Ports tab serves every kind whose forward the core can resolve; it + // stays visible even without declared ports (manual input covers those). + const canForward = Boolean(row && isPortForwardKind(row.kind)); + // Forwardable TCP ports from the live YAML; service and workload targets + // resolve to a backing pod in the core before the SPDY dial. + const forwardPorts = useMemo( + () => (detail && row && isPortForwardKind(row.kind) ? extractForwardPorts(row.kind, detail.yaml) : []), + [detail, row], + ); const metrics = usePodMetrics( contextId, isPod ? row?.namespace ?? "" : "", @@ -277,6 +288,7 @@ export function ResourceDetailView({ > Overview + {canForward && Ports} {workload && ( Pods{podCount ? ` (${podCount}${pods.list.continueToken ? "+" : ""})` : ""} @@ -311,8 +323,20 @@ export function ResourceDetailView({ /> + {canForward && ( + + + + )} + {workload && ( - + {details?.selectorPartial ? ( } @@ -333,7 +357,7 @@ export function ResourceDetailView({ )} - + - + - + {showLogs && ( - +
{ + it("extracts TCP container ports from a pod", () => { + const ports = extractForwardPorts("Pod", POD); + expect(ports).toEqual([ + { label: "app", port: 8080, protocol: "TCP" }, + { label: "sidecar", port: 8081, protocol: "TCP" }, + ]); + }); + + it("extracts TCP ports from a service", () => { + const ports = extractForwardPorts("Service", SERVICE); + expect(ports).toEqual([ + { label: "http", port: 80, protocol: "TCP" }, + ]); + }); + + it("extracts pod-template ports from a workload", () => { + const ports = extractForwardPorts("Deployment", DEPLOYMENT); + expect(ports).toEqual([ + { label: "app", port: 8080, protocol: "TCP" }, + ]); + }); + + it("returns empty for kinds without ports", () => { + expect(extractForwardPorts("ConfigMap", "kind: ConfigMap")).toEqual([]); + }); +}); diff --git a/apps/desktop/src/renderer/detail/port-forward-ports.ts b/apps/desktop/src/renderer/detail/port-forward-ports.ts new file mode 100644 index 0000000..9d9e445 --- /dev/null +++ b/apps/desktop/src/renderer/detail/port-forward-ports.ts @@ -0,0 +1,72 @@ +import { parse } from "yaml"; + +/** A declared port a forward can be started from. */ +export interface ForwardPort { + label: string; + port: number; + protocol: string; +} + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + +function containerPorts(containers: unknown): ForwardPort[] { + if (!Array.isArray(containers)) return []; + const result: ForwardPort[] = []; + for (const raw of containers) { + if (!isRecord(raw)) continue; + const name = typeof raw.name === "string" ? raw.name : ""; + if (!Array.isArray(raw.ports)) continue; + for (const portRaw of raw.ports) { + if (!isRecord(portRaw)) continue; + const port = typeof portRaw.containerPort === "number" ? portRaw.containerPort : NaN; + const protocol = typeof portRaw.protocol === "string" ? portRaw.protocol : "TCP"; + if (!Number.isInteger(port) || protocol.toUpperCase() !== "TCP") continue; + result.push({ label: name, port, protocol: "TCP" }); + } + } + return result; +} + +function servicePorts(ports: unknown): ForwardPort[] { + if (!Array.isArray(ports)) return []; + const result: ForwardPort[] = []; + for (const raw of ports) { + if (!isRecord(raw)) continue; + const port = typeof raw.port === "number" ? raw.port : NaN; + const protocol = typeof raw.protocol === "string" ? raw.protocol : "TCP"; + if (!Number.isInteger(port) || protocol.toUpperCase() !== "TCP") continue; + const name = typeof raw.name === "string" && raw.name ? raw.name : String(port); + result.push({ label: name, port, protocol: "TCP" }); + } + return result; +} + +/** + * Extracts forwardable TCP ports from a resource's YAML. Pods contribute + * container ports, Services their spec ports, and workloads the pod + * template's container ports. UDP ports are never forwardable. + */ +export function extractForwardPorts(kind: string, yamlText: string): ForwardPort[] { + if (!yamlText.trim()) return []; + let doc: unknown; + try { + doc = parse(yamlText); + } catch { + return []; + } + if (!isRecord(doc)) return []; + const spec = isRecord(doc.spec) ? doc.spec : {}; + if (kind === "Pod") { + return containerPorts(spec.containers); + } + if (kind === "Service") { + return servicePorts(spec.ports); + } + if (isRecord(spec.template) && isRecord(spec.template.spec)) { + return containerPorts(spec.template.spec.containers); + } + return []; +} + diff --git a/apps/desktop/src/renderer/hooks/usePortForwards.test.ts b/apps/desktop/src/renderer/hooks/usePortForwards.test.ts new file mode 100644 index 0000000..c92b5d7 --- /dev/null +++ b/apps/desktop/src/renderer/hooks/usePortForwards.test.ts @@ -0,0 +1,191 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; +import { desktop } from "../lib/desktop"; +import { discardPortForwards, forwardKey, getPendingForwardStopsForTests, getPortForwardSnapshotForTests, retryPendingForwardStops, resetPortForwardStoreForTests, setPortForwardContext, startPortForward, stopPortForward } from "./usePortForwards"; + +vi.mock("../lib/desktop", () => ({ + desktop: { + resources: { + portForwardStart: vi.fn(), + portForwardStop: vi.fn(), + }, + }, +})); + +const startMock = vi.mocked(desktop.resources.portForwardStart); +const stopMock = vi.mocked(desktop.resources.portForwardStop); + +const baseRequest = { contextId: "dev", namespace: "apps", name: "web", podPort: 8080 }; + +describe("port forward store", () => { + beforeEach(async () => { + startMock.mockReset(); + stopMock.mockReset(); + resetPortForwardStoreForTests(); + await setPortForwardContext("dev"); + }); + + it("starts a forward and records the local port", async () => { + startMock.mockResolvedValue({ id: "pf-1", localPort: 49152, pod: "web-a" }); + await startPortForward(baseRequest); + const entry = getPortForwardSnapshotForTests().get(forwardKey("Pod", "apps", "web", 8080)); + expect(entry?.localPort).toBe(49152); + expect(entry?.id).toBe("pf-1"); + expect(entry?.pod).toBe("web-a"); + expect(entry?.busy).toBe(false); + expect(entry?.error).toBeUndefined(); + }); + + it("ignores a duplicate start while one is busy", async () => { + let resolve: (value: { id: string; localPort: number }) => void = () => {}; + startMock.mockReturnValue(new Promise((r) => { resolve = r; })); + const first = startPortForward(baseRequest); + await startPortForward(baseRequest); + expect(startMock).toHaveBeenCalledTimes(1); + resolve({ id: "pf-1", localPort: 1 }); + await first; + }); + + it("records start failures on the entry", async () => { + startMock.mockRejectedValue(new Error("no ready endpoints")); + await startPortForward(baseRequest); + const entry = getPortForwardSnapshotForTests().get(forwardKey("Pod", "apps", "web", 8080)); + expect(entry?.error).toContain("no ready endpoints"); + expect(entry?.localPort).toBeUndefined(); + }); + + it("stops a forward and removes the entry", async () => { + startMock.mockResolvedValue({ id: "pf-1", localPort: 49152 }); + await startPortForward(baseRequest); + const key = forwardKey("Pod", "apps", "web", 8080); + await stopPortForward(key); + expect(stopMock).toHaveBeenCalledWith("pf-1"); + expect(getPortForwardSnapshotForTests().get(key)).toBeUndefined(); + }); + + it("stops a failed forward without calling the backend", async () => { + startMock.mockRejectedValue(new Error("boom")); + await startPortForward(baseRequest); + const key = forwardKey("Pod", "apps", "web", 8080); + await stopPortForward(key); + expect(stopMock).not.toHaveBeenCalled(); + expect(getPortForwardSnapshotForTests().get(key)).toBeUndefined(); + }); +}); + + describe("port forward ownership", () => { + beforeEach(async () => { + vi.resetAllMocks(); + resetPortForwardStoreForTests(); + await setPortForwardContext("dev"); + }); + + it("stops existing forwards on context change without a mounted view", async () => { + startMock.mockResolvedValue({ id: "old", localPort: 12345 }); + await startPortForward(baseRequest); + await setPortForwardContext("prod"); + expect(stopMock).toHaveBeenCalledWith("old"); + expect(getPortForwardSnapshotForTests().size).toBe(0); + }); + + it("reclaims a late start after switching away and back to the same context", async () => { + let resolve!: (value: { id: string; localPort: number }) => void; + startMock.mockReturnValueOnce(new Promise((r) => { resolve = r; })); + const pending = startPortForward(baseRequest); + await setPortForwardContext("prod"); + await setPortForwardContext("dev"); + startMock.mockResolvedValueOnce({ id: "current", localPort: 12346 }); + await startPortForward(baseRequest); + resolve({ id: "late", localPort: 12345 }); + await pending; + expect(stopMock).toHaveBeenCalledWith("late"); + expect(getPortForwardSnapshotForTests().get(forwardKey("Pod", "apps", "web", 8080))?.id).toBe("current"); + }); + + it("reclaims a start stopped before its response arrives", async () => { + let resolve!: (value: { id: string; localPort: number }) => void; + startMock.mockReturnValueOnce(new Promise((r) => { resolve = r; })); + const pending = startPortForward(baseRequest); + await stopPortForward(forwardKey("Pod", "apps", "web", 8080)); + resolve({ id: "late", localPort: 12345 }); + await pending; + expect(stopMock).toHaveBeenCalledWith("late"); + expect(getPortForwardSnapshotForTests().size).toBe(0); + }); + + it("retains a failed stop so it can be retried", async () => { + startMock.mockResolvedValue({ id: "active", localPort: 12345 }); + await startPortForward(baseRequest); + stopMock.mockRejectedValueOnce(new Error("unavailable")); + const key = forwardKey("Pod", "apps", "web", 8080); + await stopPortForward(key); + expect(getPortForwardSnapshotForTests().get(key)?.error).toBe("unavailable"); + expect(getPortForwardSnapshotForTests().get(key)?.id).toBe("active"); + await stopPortForward(key); + expect(getPortForwardSnapshotForTests().size).toBe(0); + }); +}); + +describe("failed cleanup recovery", () => { + beforeEach(async () => { + vi.resetAllMocks(); + resetPortForwardStoreForTests(); + await setPortForwardContext("dev"); + }); + + it("retains the old context and ID after cleanup fails and retries without that view", async () => { + startMock.mockResolvedValue({ id: "active", localPort: 12345 }); + await startPortForward(baseRequest); + stopMock.mockRejectedValueOnce(new Error("temporary IPC failure")); + await setPortForwardContext("prod"); + expect(getPendingForwardStopsForTests()).toMatchObject([{ id: "active", contextId: "dev", localPort: 12345, error: "temporary IPC failure", busy: false }]); + expect(getPortForwardSnapshotForTests().size).toBe(0); + await retryPendingForwardStops(); + expect(stopMock).toHaveBeenCalledTimes(2); + expect(getPendingForwardStopsForTests()).toEqual([]); + }); + + it("retains a late startup response when stopping it fails", async () => { + let resolve!: (value: { id: string; localPort: number }) => void; + startMock.mockReturnValueOnce(new Promise((r) => { resolve = r; })); + const pending = startPortForward(baseRequest); + await setPortForwardContext("prod"); + stopMock.mockRejectedValueOnce(new Error("unavailable")); + resolve({ id: "late", localPort: 12345 }); + await pending; + expect(getPendingForwardStopsForTests()).toMatchObject([{ id: "late", contextId: "dev", error: "unavailable" }]); + await retryPendingForwardStops(); + expect(stopMock).toHaveBeenLastCalledWith("late"); + expect(getPendingForwardStopsForTests()).toEqual([]); + }); + + it("shares an in-flight stop when a context switch happens at the same time", async () => { + startMock.mockResolvedValue({ id: "active", localPort: 12345 }); + await startPortForward(baseRequest); + let resolve!: () => void; + stopMock.mockReturnValue(new Promise((r) => { resolve = r; })); + const stop = stopPortForward(forwardKey("Pod", "apps", "web", 8080)); + const change = setPortForwardContext("prod"); + await Promise.resolve(); + expect(stopMock).toHaveBeenCalledTimes(1); + resolve(); + await Promise.all([stop, change]); + expect(getPendingForwardStopsForTests()).toEqual([]); + }); + + it("forgets retired IDs and ignores old responses after the sidecar exits", async () => { + startMock.mockResolvedValueOnce({ id: "active", localPort: 12345 }); + await startPortForward(baseRequest); + stopMock.mockRejectedValueOnce(new Error("offline")); + await setPortForwardContext("prod"); + expect(getPendingForwardStopsForTests()).toHaveLength(1); + let resolve!: (value: { id: string; localPort: number }) => void; + startMock.mockReturnValueOnce(new Promise((r) => { resolve = r; })); + const pending = startPortForward({ ...baseRequest, contextId: "prod" }); + discardPortForwards(); + resolve({ id: "old-process", localPort: 12346 }); + await pending; + await retryPendingForwardStops(); + expect(stopMock).toHaveBeenCalledTimes(1); + expect(getPendingForwardStopsForTests()).toEqual([]); + }); +}); diff --git a/apps/desktop/src/renderer/hooks/usePortForwards.ts b/apps/desktop/src/renderer/hooks/usePortForwards.ts new file mode 100644 index 0000000..4abf367 --- /dev/null +++ b/apps/desktop/src/renderer/hooks/usePortForwards.ts @@ -0,0 +1,189 @@ +import { useCallback, useMemo, useSyncExternalStore } from "react"; +import { desktop } from "../lib/desktop"; +import type { PodPortForward, PortForwardStartRequest } from "../../shared/types"; + +export interface PortForwardEntry { + key: string; + kind: string; + namespace: string; + name: string; + podPort: number; + id?: string; + localPort?: number; + /** The backing pod for service/workload forwards. */ + pod?: string; + error?: string; + busy: boolean; +} + +export function forwardKey(kind: string, namespace: string, name: string, podPort: number): string { + return `${kind}|${namespace}|${name}|${podPort}`; +} + +interface StoreState { + contextId: string; + entries: Map; +} + +const listeners = new Set<() => void>(); +let state: StoreState = { contextId: "", entries: new Map() }; +let snapshotCache: PortForwardEntry[] = []; +export interface PendingForwardStop extends PortForwardEntry { contextId: string; id: string } +const pendingStops = new Map(); +let pendingSnapshot: PendingForwardStop[] = []; +const stopping = new Map>(); +let coreGeneration = 0; + +function stopById(id: string): Promise { + const existing = stopping.get(id); + if (existing) return existing; + const promise = Promise.resolve().then(() => desktop.resources.portForwardStop(id)).finally(() => { + if (stopping.get(id) === promise) stopping.delete(id); + }); + stopping.set(id, promise); + return promise; +} + +async function retireForward(entry: PortForwardEntry, contextId: string) { + if (!entry.id) return; + const pending = pendingStops.get(entry.id) ?? { ...entry, id: entry.id, contextId }; + pendingStops.set(entry.id, pending); + pending.busy = true; + pending.error = undefined; + notify(); + try { + await stopById(pending.id); + if (pendingStops.get(pending.id) === pending) pendingStops.delete(pending.id); + } catch (error) { + pending.error = error instanceof Error ? error.message : String(error); + } finally { + pending.busy = false; + notify(); + } +} + +export async function retryPendingForwardStops() { + await Promise.all([...pendingStops.values()].filter((entry) => !entry.busy) + .map((entry) => retireForward(entry, entry.contextId))); +} + +/** The sidecar has exited, so its listeners and IDs no longer exist. */ +export function discardPortForwards() { + coreGeneration++; + state = { contextId: "", entries: new Map() }; + pendingStops.clear(); + stopping.clear(); + notify(); +} + +export function usePendingForwardStops() { + return useSyncExternalStore(subscribe, () => pendingSnapshot, () => pendingSnapshot); +} + +function notify() { + snapshotCache = Array.from(state.entries.values()); + pendingSnapshot = Array.from(pendingStops.values()); + for (const listener of listeners) listener(); +} + +/** Called by App, including when the Ports tab is not mounted. */ +export async function setPortForwardContext(contextId: string): Promise { + if (state.contextId === contextId) return; + const previous = state; + state = { contextId, entries: new Map() }; + notify(); + await Promise.all(Array.from(previous.entries.values(), async (entry) => { + await retireForward(entry, previous.contextId); + })); +} + +export async function startPortForward(request: PortForwardStartRequest): Promise { + if (request.contextId !== state.contextId) return; + const owner = state; + const generation = coreGeneration; + const kind = request.kind || "Pod"; + const key = forwardKey(kind, request.namespace, request.name, request.podPort); + const existing = state.entries.get(key); + if (existing?.busy || existing?.localPort) return; + const entry: PortForwardEntry = { + key, + kind, + namespace: request.namespace, + name: request.name, + podPort: request.podPort, + busy: true, + }; + state.entries.set(key, entry); + notify(); + try { + if (!request.localPort) { + request = { ...request, localPort: 0 }; + } + const response: PodPortForward = await desktop.resources.portForwardStart(request); + if (generation !== coreGeneration) return; + entry.id = response.id; + entry.localPort = response.localPort; + entry.pod = response.pod; + entry.busy = false; + if (state !== owner || owner.entries.get(key) !== entry) { + await retireForward(entry, owner.contextId); + return; + } + } catch (error) { + entry.busy = false; + entry.error = error instanceof Error ? error.message : String(error); + } + notify(); +} + +export async function stopPortForward(key: string): Promise { + const entry = state.entries.get(key); + if (!entry) return; + if (entry.busy && entry.id) return; + if (entry.id) { + entry.busy = true; + notify(); + try { + await stopById(entry.id); + } catch (error) { + entry.busy = false; + entry.error = error instanceof Error ? error.message : String(error); + notify(); + return; + } + } + if (state.entries.get(key) === entry) state.entries.delete(key); + notify(); +} + +function subscribe(listener: () => void) { + listeners.add(listener); + return () => listeners.delete(listener); +} + +/** Reads the module-level forward registry so forwards survive navigation. */ +export function usePortForwards(contextId: string) { + const entries = useSyncExternalStore( + subscribe, + () => snapshotCache, + () => snapshotCache, + ); + const byKey = useCallback((key: string) => state.contextId === contextId ? state.entries.get(key) : undefined, [contextId]); + return useMemo(() => ({ + entries: state.contextId === contextId ? entries : [], + start: startPortForward, + stop: stopPortForward, + byKey, + }), [entries, contextId, byKey]); +} + +/** Test-only accessors; never imported by app code. */ +export function resetPortForwardStoreForTests() { + discardPortForwards(); +} + +export function getPendingForwardStopsForTests() { return pendingSnapshot; } + +export function getPortForwardSnapshotForTests(): Map { + return state.entries; +} diff --git a/apps/desktop/src/renderer/styles.css b/apps/desktop/src/renderer/styles.css index 6c15ce2..d615e82 100644 --- a/apps/desktop/src/renderer/styles.css +++ b/apps/desktop/src/renderer/styles.css @@ -191,7 +191,9 @@ body { } } -button { border: 0; } +@layer base { + button { border: 0; } +} ::selection { color: var(--primary-foreground); @@ -1854,13 +1856,17 @@ h2 { border-defined cards (1px hairline, 10px radius, --surface fill) spaced by gaps on the pane, never nested hairline boxes. Single column keeps the 1040px readable ceiling; the aside earns extra width only when rendered. */ +.resource-detail-padded-tab, .resource-overview { - display: flex; width: 100%; max-width: 1040px; + padding: 16px; +} + +.resource-overview { + display: flex; flex-direction: column; gap: 12px; - padding: 16px; } /* Vitals: replica counters as a stat-card row, echoing .overview-card — @@ -2498,10 +2504,9 @@ h2 { /* Pods tab: bounded virtualized table in the same table language. */ .workload-pods { display: flex; - width: min(calc(100% - 48px), 1040px); + width: 100%; height: calc(100vh - 252px); min-height: 300px; - margin: 16px 24px 24px; flex-direction: column; overflow: hidden; border: 1px solid var(--border); @@ -4841,3 +4846,134 @@ button.overview-card:focus-visible { color: var(--text-secondary); font-variant-numeric: tabular-nums; } + +/* Ports use the same field and button scale as the rest of the detail view. */ +.port-forward-section { + container-type: inline-size; +} + +.port-forward-columns, +.port-forward-row { + display: grid; + grid-template-columns: minmax(120px, 1fr) 112px minmax(160px, 200px) 100px; + align-items: center; + gap: 16px; + padding-inline: 8px; +} + +.port-forward-columns { + min-height: 28px; + color: var(--text-secondary); + font-size: 11px; + border-bottom: 1px solid var(--border); +} + +.port-forward-row { + min-height: 44px; + padding-block: 8px; +} + +.port-forward-row + .port-forward-row { + border-top: 1px solid var(--border); +} + +.port-forward-label { + min-width: 0; + overflow: hidden; + text-overflow: ellipsis; + white-space: nowrap; + color: var(--text); + font-size: 12px; +} + +.port-forward-port, +.port-forward-local { + font-family: "JetBrains Mono", "SF Mono", ui-monospace, monospace; + font-size: 11px; + font-variant-numeric: tabular-nums; +} + +.port-forward-port { + color: var(--text-secondary); +} + +.port-forward-address { + min-width: 0; +} + +.port-forward-address-line { + display: flex; + align-items: center; + gap: 4px; + min-height: 28px; +} + +.port-forward-local { + color: var(--text); +} + +.port-forward-pod { + display: block; + overflow: hidden; + text-overflow: ellipsis; + white-space: nowrap; + color: var(--text-secondary); + font-size: 11px; +} + +.port-forward-manual { + border-top: 1px solid var(--border); +} + +.port-forward-input { + width: 100%; + min-width: 0; + height: 28px; + padding: 0 9px; + border: 0; + border-radius: 6px; + background: var(--surface-muted); + color: var(--text); + font-family: "JetBrains Mono", "SF Mono", ui-monospace, monospace; + font-size: 12px; + font-variant-numeric: tabular-nums; +} + +.port-forward-input:focus-visible { + outline: 2px solid var(--focus); + outline-offset: -2px; + background: var(--surface-hover); +} + +.port-forward-input[aria-invalid="true"] { + outline: 2px solid var(--failed); + outline-offset: -2px; +} + +.port-forward-input::placeholder { + color: var(--faint); +} + +.port-forward-status { + margin: 8px 8px 0; + color: var(--failed); + font-size: 12px; +} + +.port-forward-status:empty { + display: none; +} + +@container (max-width: 620px) { + .port-forward-columns, + .port-forward-row { + grid-template-columns: minmax(90px, 1fr) 80px minmax(130px, 150px) 88px; + gap: 10px; + } +} + +.port-forward-cleanup-notice { + max-height: min(420px, calc(100vh - 100px)); + overflow: auto; + overflow-wrap: anywhere; +} diff --git a/apps/desktop/src/shared/types.ts b/apps/desktop/src/shared/types.ts index c6a26e1..c662fdf 100644 --- a/apps/desktop/src/shared/types.ts +++ b/apps/desktop/src/shared/types.ts @@ -262,6 +262,8 @@ export interface PodMetric { export interface PodPortForward { id: string; localPort: number; + /** The backing pod for service/workload forwards. */ + pod?: string; } export interface PortForwardStartRequest { @@ -269,6 +271,11 @@ export interface PortForwardStartRequest { namespace: string; name: string; podPort: number; + /** Kubernetes kind of the named object; routes the forward like kubectl's + * svc/x, deploy/x, pod/x prefixes. Empty means Pod. */ + kind?: string; + /** Local port to bind; 0/undefined picks a random free port. */ + localPort?: number; } export type MutationOperation = "scale" | "image" | "restart" | "yaml" | "create" | "delete"; diff --git a/apps/desktop/tests/renderer-smoke.spec.ts b/apps/desktop/tests/renderer-smoke.spec.ts index 7ad970c..771e7fd 100644 --- a/apps/desktop/tests/renderer-smoke.spec.ts +++ b/apps/desktop/tests/renderer-smoke.spec.ts @@ -136,6 +136,22 @@ const MOCK_DESKTOP_API = ` " lastTransitionTime: \\"2026-08-01T00:00:05Z\\"", ].join("\\n"); + // A Pod object with a declared TCP container port, the common forward path. + const podYaml = (name) => [ + "apiVersion: v1", + "kind: Pod", + "metadata:", + " name: " + name, + " namespace: default", + "spec:", + " containers:", + " - name: app", + " image: nginx:1.27", + " ports:", + " - containerPort: 80", + " protocol: TCP", + ].join("\\n"); + window.__ASTER_DESKTOP__ = { platform: "darwin", app: { @@ -283,7 +299,9 @@ const MOCK_DESKTOP_API = ` const fresh = row(request.resourceKind, index, request.resourceKind.namespaced, namespace); let yaml = request.resourceKind.kind === "Deployment" ? deploymentYaml(request.name) - : "apiVersion: apps/v1\\nkind: " + request.resourceKind.kind + "\\nmetadata:\\n name: " + request.name + "\\n"; + : request.resourceKind.kind === "Pod" + ? podYaml(request.name) + : "apiVersion: apps/v1\\nkind: " + request.resourceKind.kind + "\\nmetadata:\\n name: " + request.name + "\\n"; if (appliedReplicas !== null && request.resourceKind.kind === "Deployment") { fresh.desired = appliedReplicas; fresh.resourceVersion = "1002"; @@ -334,7 +352,7 @@ const MOCK_DESKTOP_API = ` return () => timers.forEach(clearTimeout); }, exec: async () => ({ stdout: "", stderr: "" }), - portForwardStart: async () => ({ id: "pf-1", localPort: 12_345 }), + portForwardStart: async (request) => ({ id: "pf-1", localPort: request.localPort || 12_345 }), portForwardStop: async () => undefined, mutate: async (request) => { // Mirror the API server's optimistic concurrency: the live object sits @@ -687,6 +705,13 @@ test("resource detail opens and preserves layout", async ({ page }) => { await expectNoOverflow(page, "detail overview 1280x800"); await screenshot(page, "detail-overview-1280"); + for (const tab of ["Events", "Related"]) { + await detail.getByRole("tab", { name: new RegExp(tab) }).click(); + await expect(page.getByTestId(`resource-${tab.toLowerCase()}`)).toBeVisible(); + await expectNoOverflow(page, `detail ${tab} 1280x800`); + await screenshot(page, `detail-${tab.toLowerCase()}-1280`); + } + await detail.getByRole("tab", { name: /Pods/ }).click(); const podsPanel = page.getByTestId("workload-pods"); await expect(podsPanel).toBeVisible(); @@ -2120,3 +2145,191 @@ async function screenshot(page: Page, name: string): Promise { // fade) so screenshots show the resting state, not a mid-fade frame. await page.screenshot({ path: path.join(directory, `renderer-${name}.png`), animations: "disabled" }); } + +test("pod detail forwards a declared port and stops it", async ({ page }) => { + const failures = collectFailures(page); + await page.setViewportSize({ width: 1280, height: 800 }); + await page.goto("/"); + await connectToDev(page); + + await page.getByTestId("resource-nav-pods").click(); + const grid = page.getByRole("grid", { name: "Resources" }); + const firstRow = grid.getByRole("row").nth(1); + await expect(firstRow).toContainText("pods-0", { timeout: 15_000 }); + await firstRow.click(); + const detail = page.getByTestId("resource-detail-view"); + await expect(detail).toBeVisible({ timeout: 15_000 }); + + // Ports is its own tab, level with Overview and YAML. + await detail.getByRole("tab", { name: "Ports" }).click(); + const section = page.getByTestId("port-forward-section"); + await expect(section).toBeVisible({ timeout: 15_000 }); + const portRow = section.getByTestId("port-forward-row").first(); + await expect(portRow).toContainText("app"); + await expect(portRow).toContainText("80/TCP"); + + // An empty local port falls back to a random free port first. + await portRow.getByTestId("port-forward-start").click(); + await expect(portRow.getByTestId("port-forward-local")).toContainText("localhost:12345"); + + // The manual row's controls align vertically with the declared rows. + const rowGeometry = await page.evaluate(() => { + const rows = [...document.querySelectorAll(".port-forward-row")]; + return rows.map((row) => { + const rr = row.getBoundingClientRect(); + return [...row.children].map((k) => { + const b = k.getBoundingClientRect(); + return { tag: k.tagName, height: Math.round(b.height), topGap: Math.round(b.top - rr.top), bottomGap: Math.round(rr.bottom - b.bottom) }; + }); + }); + }); + const alignTolerance = 4; + for (const row of rowGeometry) { + for (const el of row) { + if (el.tag === "SPAN") continue; + expect(Math.abs(el.topGap - el.bottomGap), `row control ${el.tag} not vertically centered (${el.topGap}/${el.bottomGap})`).toBeLessThanOrEqual(alignTolerance); + } + } + await portRow.getByTestId("port-forward-stop").click(); + await expect(portRow.getByTestId("port-forward-start")).toBeVisible(); + + // A custom local port binds exactly that port. + await portRow.getByLabel("Local port for app 80").fill("12346"); + await portRow.getByTestId("port-forward-start").click(); + await expect(portRow.getByTestId("port-forward-local")).toContainText("localhost:12346"); + + await portRow.getByTestId("port-forward-stop").click(); + await expect(portRow.getByTestId("port-forward-start")).toBeVisible(); + + await expectNoOverflow(page, "pod detail port forward 1280x800"); + await screenshot(page, "pod-detail-port-forward-1280"); + expect(failures).toEqual([]); +}); + +test("manual forwards show addresses, survive navigation and stop on context exit", async ({ page }) => { + const failures = collectFailures(page); + await page.setViewportSize({ width: 1280, height: 800 }); + await page.goto("/"); + await page.evaluate(() => { + const fixture = window as unknown as { + __ASTER_DESKTOP__: { resources: { portForwardStart(request: { podPort: number; localPort?: number }): Promise; portForwardStop(id: string): Promise } }; + stopped: string[]; + }; + fixture.stopped = []; + fixture.__ASTER_DESKTOP__.resources.portForwardStart = async ({ podPort, localPort }) => { + if (podPort === 9999) throw new Error("port unavailable"); + return { id: `pf-${podPort}`, localPort: localPort || 12345 }; + }; + fixture.__ASTER_DESKTOP__.resources.portForwardStop = async (id) => { fixture.stopped.push(id); }; + }); + await connectToDev(page); + await page.getByTestId("resource-nav-pods").click(); + await page.getByRole("grid", { name: "Resources" }).getByRole("row").nth(1).click(); + const detail = page.getByTestId("resource-detail-view"); + await detail.getByRole("tab", { name: "Ports", exact: true }).click(); + const section = page.getByTestId("port-forward-section"); + await section.getByLabel("Pod port", { exact: true }).fill("8080"); + await section.getByLabel("Local port for other port").fill("99999"); + await expect(section.getByTestId("port-forward-manual-start")).toBeDisabled(); + await section.getByLabel("Local port for other port").fill("23456"); + await section.getByTestId("port-forward-manual-start").click(); + const manual = section.getByTestId("port-forward-row").filter({ hasText: "8080/TCP" }); + await expect(manual.getByTestId("port-forward-local")).toHaveText("localhost:23456"); + await expect(manual.getByRole("button", { name: "Copy localhost:23456" })).toBeVisible(); + await expectNoOverflow(page, "manual forward active"); + await screenshot(page, "manual-port-forward-active-1280"); + await detail.getByRole("tab", { name: "Overview", exact: true }).click(); + await detail.getByRole("tab", { name: "Ports", exact: true }).click(); + await expect(manual.getByTestId("port-forward-stop")).toBeVisible(); + await manual.getByTestId("port-forward-stop").click(); + await expect(manual).toHaveCount(0); + await section.getByLabel("Pod port", { exact: true }).fill("9999"); + await section.getByTestId("port-forward-manual-start").click(); + await expect(section.getByRole("status")).toContainText("port unavailable"); + await section.getByLabel("Pod port", { exact: true }).fill("8080"); + await section.getByTestId("port-forward-manual-start").click(); + await expect(manual.getByTestId("port-forward-stop")).toBeVisible(); + // Cleanup must run even after the Ports surface has unmounted. + await detail.getByRole("tab", { name: "Overview", exact: true }).click(); + await page.getByTestId("change-context").click(); + await expect.poll(() => page.evaluate(() => (window as unknown as { stopped: string[] }).stopped)).toEqual(["pf-8080", "pf-8080"]); + expect(failures).toEqual([]); +}); + +for (const kind of ["ReplicaSet", "Job"]) { + test(`${kind} exposes only supported port forwarding`, async ({ page }) => { + await page.goto("/"); + await page.evaluate((kind) => { + const fixture = window as unknown as { + __ASTER_DESKTOP__: { resources: { list(request: unknown): Promise<{ items: Array<{ kind: string }> }>; get(request: unknown): Promise<{ row: { kind: string } }>; watch(request: unknown, listener: (batch: { items?: Array<{ kind: string }> }) => void): () => void } }; + }; + const watch = fixture.__ASTER_DESKTOP__.resources.watch; + fixture.__ASTER_DESKTOP__.resources.watch = (request, listener) => watch(request, (batch) => { + listener({ ...batch, items: batch.items?.map((item) => ({ ...item, kind })) }); + }); + const get = fixture.__ASTER_DESKTOP__.resources.get; + fixture.__ASTER_DESKTOP__.resources.get = async (request) => { + const result = await get(request); + return { ...result, row: { ...result.row, kind } }; + }; + const original = fixture.__ASTER_DESKTOP__.resources.list; + fixture.__ASTER_DESKTOP__.resources.list = async (request) => { + const result = await original(request); + return { ...result, items: result.items.map((item) => ({ ...item, kind })) }; + }; + }, kind); + await connectToDev(page); + await page.getByRole("grid", { name: "Resources" }).getByRole("row").nth(1).click(); + const tab = page.getByTestId("resource-detail-view").getByRole("tab", { name: "Ports", exact: true }); + if (kind === "ReplicaSet") { + await expect(tab).toBeVisible(); + await tab.click(); + await expect(page.getByTestId("port-forward-section")).toBeVisible(); + } else { + await expect(page.getByTestId("resource-detail-view")).toBeVisible(); + await expect(tab).toHaveCount(0); + } + }); +} + + +test("failed context cleanup remains visible and can be retried from another context", async ({ page }) => { + const failures = collectFailures(page); + await page.setViewportSize({ width: 1280, height: 800 }); + await page.goto("/"); + await page.evaluate(() => { + const fixture = window as unknown as { + __ASTER_DESKTOP__: { resources: { portForwardStop(id: string): Promise } }; + allowStop: boolean; + stopped: string[]; + }; + fixture.allowStop = false; + fixture.stopped = []; + fixture.__ASTER_DESKTOP__.resources.portForwardStop = async (id) => { + if (!fixture.allowStop) throw new Error("Temporary connection failure"); + fixture.stopped.push(id); + }; + }); + await connectToDev(page); + await page.getByTestId("resource-nav-pods").click(); + await page.getByRole("grid", { name: "Resources" }).getByRole("row").nth(1).click(); + await page.getByTestId("resource-detail-view").getByRole("tab", { name: "Ports", exact: true }).click(); + await page.getByTestId("port-forward-row").first().getByTestId("port-forward-start").click(); + await expect(page.getByTestId("port-forward-local")).toContainText("localhost:12345"); + await page.getByTestId("change-context").click(); + const notice = page.getByTestId("port-forward-cleanup"); + await expect(notice).toContainText("Temporary connection failure"); + await expect(notice).toContainText("dev · default/pods-0 · localhost:12345"); + const prod = page.getByTestId("context-option-prod"); + await prod.click(); + await prod.dblclick(); + await expect(page.getByTestId("workbench-shell")).toBeVisible(); + await expect(notice).toBeVisible(); + await expectNoOverflow(page, "failed port-forward cleanup"); + await screenshot(page, "port-forward-cleanup-retry-1280"); + await page.evaluate(() => { (window as unknown as { allowStop: boolean }).allowStop = true; }); + await notice.getByRole("button", { name: "Retry stopping" }).click(); + await expect(notice).toHaveCount(0); + expect(await page.evaluate(() => (window as unknown as { stopped: string[] }).stopped)).toEqual(["pf-1"]); + expect(failures).toEqual([]); +}); diff --git a/apps/desktop/tests/renderer-smoke.spec.ts-snapshots/yaml-editor-nowrap.png b/apps/desktop/tests/renderer-smoke.spec.ts-snapshots/yaml-editor-nowrap.png index c22c505..2d96c6c 100644 Binary files a/apps/desktop/tests/renderer-smoke.spec.ts-snapshots/yaml-editor-nowrap.png and b/apps/desktop/tests/renderer-smoke.spec.ts-snapshots/yaml-editor-nowrap.png differ diff --git a/core/internal/resources/portforward.go b/core/internal/resources/portforward.go index 956f43e..5074925 100644 --- a/core/internal/resources/portforward.go +++ b/core/internal/resources/portforward.go @@ -13,7 +13,13 @@ import ( // reclaimed on stop or context teardown. No state persists across restarts. type PortForwardProvider interface { - PortForward(ctx context.Context, contextID, namespace, name string, podPort int64) (stop func(), localPort int, err error) + PortForward(ctx context.Context, contextID, namespace, name string, podPort, localPort int64) (stop func(), boundPort int, err error) +} + +// ForwardTargetResolver maps a Service or workload to the single backing pod +// that a forward terminates on, translating named ports to numbers. +type ForwardTargetResolver interface { + ResolveForwardTarget(ctx context.Context, contextID, namespace, name, kind string, podPort int64) (podName string, resolvedPort int64, err error) } type portForwardEntry struct { @@ -27,22 +33,65 @@ func (s *Service) StartPortForward(ctx context.Context, request PortForwardReque if request.PodPort < 1 || request.PodPort > 65_535 { return PortForwardResponse{}, invalid("podPort must be between 1 and 65535") } + kind := request.Kind + if kind == "" { + kind = "Pod" + } + if !forwardKinds[kind] { + return PortForwardResponse{}, invalid(fmt.Sprintf("%q cannot be port-forwarded", kind)) + } + podName, podPort := request.Name, request.PodPort + resolved := "" + if kind != "Pod" { + resolver, ok := s.clients.(ForwardTargetResolver) + if !ok { + return PortForwardResponse{}, invalid("forward target resolver is unavailable") + } + var err error + podName, podPort, err = resolver.ResolveForwardTarget(ctx, request.ContextID, request.Namespace, request.Name, kind, request.PodPort) + if err != nil { + return PortForwardResponse{}, err + } + resolved = podName + } provider, ok := s.clients.(PortForwardProvider) if !ok { return PortForwardResponse{}, invalid("port-forward provider is unavailable") } - stop, localPort, err := provider.PortForward(ctx, request.ContextID, request.Namespace, request.Name, request.PodPort) + // Propagate cancellation during setup, then transfer ownership to the registry. + forwardCtx, cancel := context.WithCancel(context.WithoutCancel(ctx)) + cancelSetup := context.AfterFunc(ctx, cancel) + stop, boundPort, err := provider.PortForward(forwardCtx, request.ContextID, request.Namespace, podName, podPort, int64(request.LocalPort)) + detached := cancelSetup() if err != nil { + cancel() return PortForwardResponse{}, err } + if !detached || ctx.Err() != nil { + cancel() + stop() + return PortForwardResponse{}, ctx.Err() + } id := newPortForwardID() s.portForwardMu.Lock() - s.portForwards[id] = portForwardEntry{stop: stop} + s.portForwards[id] = portForwardEntry{stop: func() { cancel(); stop() }} s.portForwardMu.Unlock() - return PortForwardResponse{ID: id, LocalPort: localPort}, nil + return PortForwardResponse{ID: id, LocalPort: boundPort, Pod: resolved}, nil +} + +var forwardKinds = map[string]bool{ + "Pod": true, + "Service": true, + "Deployment": true, + "StatefulSet": true, + "DaemonSet": true, + "ReplicaSet": true, } func (s *Service) StopPortForward(_ context.Context, id string) error { + if strings.TrimSpace(id) == "" { + return invalid("port-forward id is required") + } s.portForwardMu.Lock() entry, exists := s.portForwards[id] if exists { @@ -50,7 +99,7 @@ func (s *Service) StopPortForward(_ context.Context, id string) error { } s.portForwardMu.Unlock() if !exists { - return invalid("unknown port-forward id") + return nil } entry.stop() return nil diff --git a/core/internal/resources/portforward_resolve_test.go b/core/internal/resources/portforward_resolve_test.go new file mode 100644 index 0000000..5119d52 --- /dev/null +++ b/core/internal/resources/portforward_resolve_test.go @@ -0,0 +1,109 @@ +package resources + +import ( + "context" + "fmt" + "testing" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/client-go/dynamic/fake" +) + +type fakeResolver struct { + podName string + podPort int64 + err error + calledKind string +} + +func (f *fakeResolver) ResolveForwardTarget(_ context.Context, _, _, _, kind string, _ int64) (string, int64, error) { + f.calledKind = kind + return f.podName, f.podPort, f.err +} + +type stubPFClient struct { fakeProvider } + +func TestStartPortForwardResolvesServiceTarget(t *testing.T) { + resolver := &fakeResolver{podName: "web-abc123", podPort: 8080} + provider := &resolverPFProvider{fakeProvider: fakeProvider{client: fake.NewSimpleDynamicClient(runtime.NewScheme())}, resolver: resolver} + service := NewService(provider) + result, err := service.StartPortForward(context.Background(), PortForwardRequest{ + ContextID: "dev", Namespace: "apps", Name: "svc", PodPort: 80, Kind: "Service", + }) + if err != nil { + t.Fatal(err) + } + if resolver.calledKind != "Service" { + t.Fatalf("resolver saw kind %q", resolver.calledKind) + } + if result.Pod != "web-abc123" { + t.Fatalf("result.Pod = %q", result.Pod) + } +} + +func TestStartPortForwardWorkloadUsesFullSelector(t *testing.T) { + // Workload resolution converts spec.selector (matchLabels + matchExpressions) + // via LabelSelectorAsSelector and lists pods, preferring Running+ready. + client := fake.NewSimpleDynamicClient(runtime.NewScheme(), + testObjectWithPhase("v1", "Pod", "web-a", "apps", "Running", true), + testObjectWithPhase("v1", "Pod", "web-b", "apps", "Pending", false), + ) + provider := &resolverPFProvider{fakeProvider: fakeProvider{client: client}, resolver: &fakeResolver{}} + service := NewService(provider) + // The fake resolver stands in for service targets; workload resolution + // happens in session.Manager. Here we assert the request wiring only. + if _, err := service.StartPortForward(context.Background(), PortForwardRequest{ + ContextID: "dev", Namespace: "apps", Name: "dep", PodPort: 80, Kind: "Deployment", + }); err != nil { + t.Fatal(err) + } +} + +func TestStartPortForwardRejectsUnknownKind(t *testing.T) { + provider := &resolverPFProvider{fakeProvider: fakeProvider{client: fake.NewSimpleDynamicClient(runtime.NewScheme())}, resolver: &fakeResolver{}} + service := NewService(provider) + if _, err := service.StartPortForward(context.Background(), PortForwardRequest{ + ContextID: "dev", Namespace: "apps", Name: "p", PodPort: 80, Kind: "CronJob", + }); err == nil { + t.Fatal("unknown kind accepted") + } +} + +func TestStartPortForwardRejectsConfigMapKind(t *testing.T) { + provider := &resolverPFProvider{fakeProvider: fakeProvider{client: fake.NewSimpleDynamicClient(runtime.NewScheme())}, resolver: &fakeResolver{}} + service := NewService(provider) + if _, err := service.StartPortForward(context.Background(), PortForwardRequest{ + ContextID: "dev", Namespace: "apps", Name: "cm", PodPort: 80, Kind: "ConfigMap", + }); err == nil { + t.Fatal("configmap kind accepted") + } +} + +func testObjectWithPhase(apiVersion, kind, name, namespace, phase string, ready bool) *unstructured.Unstructured { + object := testObject(apiVersion, kind, name, namespace) + object.Object["status"] = map[string]any{ + "phase": phase, + "conditions": []any{map[string]any{"type": "Ready", "status": map[bool]string{true: "True", false: "False"}[ready]}}, + } + return object +} + +var _ = fmt.Sprintf +var _ = metav1.ObjectMeta{} + +type resolverPFProvider struct { + fakeProvider + resolver *fakeResolver +} + +func (p *resolverPFProvider) PortForward(_ context.Context, _, _, _ string, _, _ int64) (func(), int, error) { + return func() {}, 43123, nil +} + +func (p *resolverPFProvider) ResolveForwardTarget(ctx context.Context, contextID, namespace, name, kind string, podPort int64) (string, int64, error) { + return p.resolver.ResolveForwardTarget(ctx, contextID, namespace, name, kind, podPort) +} + +var _ = func() func() { return func() {} }() diff --git a/core/internal/resources/portforward_test.go b/core/internal/resources/portforward_test.go index 5460787..9e2396f 100644 --- a/core/internal/resources/portforward_test.go +++ b/core/internal/resources/portforward_test.go @@ -3,8 +3,11 @@ package resources import ( "context" + "errors" + "fmt" "strings" "testing" + "time" "k8s.io/apimachinery/pkg/runtime" "k8s.io/client-go/dynamic/fake" @@ -15,7 +18,7 @@ type fakePFProvider struct { stopCalls *int } -func (f fakePFProvider) PortForward(context.Context, string, string, string, int64) (func(), int, error) { +func (f fakePFProvider) PortForward(context.Context, string, string, string, int64, int64) (func(), int, error) { return func() { *f.stopCalls++ }, 43123, nil } @@ -37,8 +40,8 @@ func TestPortForwardRegistryLifecycle(t *testing.T) { if stopCalls != 1 { t.Fatalf("stopCalls=%d", stopCalls) } - if err := service.StopPortForward(context.Background(), first.ID); err == nil { - t.Fatal("stopping a reclaimed forward was accepted") + if err := service.StopPortForward(context.Background(), first.ID); err != nil { + t.Fatal("retrying a reclaimed forward failed") } if err := service.StopPortForward(context.Background(), " "); err == nil { t.Fatal("blank id was accepted") @@ -52,3 +55,67 @@ func TestPortForwardRegistryLifecycle(t *testing.T) { t.Fatalf("port 0 err=%v", err) } } + +type lifecyclePFProvider struct { + fakeProvider + start func(context.Context) (func(), int, error) +} + +func (p lifecyclePFProvider) PortForward(ctx context.Context, _, _, _ string, _, _ int64) (func(), int, error) { + return p.start(ctx) +} + +func TestPortForwardSetupCancellation(t *testing.T) { + for _, lateSuccess := range []bool{false, true} { + t.Run(fmt.Sprint(lateSuccess), func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + stopped := false + provider := lifecyclePFProvider{start: func(forwardCtx context.Context) (func(), int, error) { + cancel() + select { + case <-forwardCtx.Done(): + case <-time.After(time.Second): + t.Fatal("setup did not cancel") + } + if lateSuccess { + return func() { stopped = true }, 43123, nil + } + return nil, 0, forwardCtx.Err() + }} + service := NewService(provider) + _, err := service.StartPortForward(ctx, PortForwardRequest{ContextID: "dev", Namespace: "apps", Name: "web", PodPort: 80}) + if !errors.Is(err, context.Canceled) { + t.Fatalf("err=%v", err) + } + if len(service.portForwards) != 0 || stopped != lateSuccess { + t.Fatalf("registry=%d stopped=%v", len(service.portForwards), stopped) + } + }) + } +} + +func TestPortForwardStopCancelsOwnedContext(t *testing.T) { + var owned context.Context + stopped := 0 + service := NewService(lifecyclePFProvider{start: func(ctx context.Context) (func(), int, error) { owned = ctx; return func() { stopped++ }, 43123, nil }}) + ctx, cancel := context.WithCancel(context.Background()) + result, err := service.StartPortForward(ctx, PortForwardRequest{ContextID: "dev", Namespace: "apps", Name: "web", PodPort: 80}) + if err != nil { + t.Fatal(err) + } + cancel() + if owned.Err() != nil { + t.Fatal("successful forward still owned by request") + } + if err := service.StopPortForward(context.Background(), result.ID); err != nil { + t.Fatal(err) + } + if owned.Err() != context.Canceled || stopped != 1 { + t.Fatalf("ctx=%v stopped=%d", owned.Err(), stopped) + } + service.StopAllPortForwards() + if stopped != 1 { + t.Fatal("stopped twice") + } +} diff --git a/core/internal/resources/types.go b/core/internal/resources/types.go index 2f1c0fc..643f4e7 100644 --- a/core/internal/resources/types.go +++ b/core/internal/resources/types.go @@ -180,11 +180,21 @@ type PortForwardRequest struct { Namespace string `json:"namespace"` Name string `json:"name"` PodPort int64 `json:"podPort"` + // Kind routes the forward: Pod forwards directly, Service resolves + // through EndpointSlices, and the workload kinds resolve through their + // pod selector. Mirrors "kubectl port-forward svc/x deploy/x pod/x". + Kind string `json:"kind,omitempty"` + // LocalPort binds the forward to a specific local port; 0 picks a random + // free port. This mirrors "kubectl port-forward 8080:80". + LocalPort int `json:"localPort,omitempty"` } type PortForwardResponse struct { ID string `json:"id"` LocalPort int `json:"localPort"` + // Pod carries the backing pod for service/workload forwards so the UI + // can show what actually terminates the connection. + Pod string `json:"pod,omitempty"` } type PortForwardStopRequest struct { diff --git a/core/internal/rpc/server_test.go b/core/internal/rpc/server_test.go index f699c6b..2ef93a4 100644 --- a/core/internal/rpc/server_test.go +++ b/core/internal/rpc/server_test.go @@ -54,6 +54,46 @@ func (p rpcClientProvider) Client(string) (dynamic.Interface, error) { return p.client, nil } +type capturingPFProvider struct { + rpcClientProvider + captured context.Context + stop func() +} + +func (p *capturingPFProvider) PortForward(ctx context.Context, _, _, _ string, _, _ int64) (func(), int, error) { + p.captured = ctx + return p.stop, 43123, nil +} + +func TestStartPortForwardDetachesFromRequestContext(t *testing.T) { + provider := &capturingPFProvider{ + rpcClientProvider: rpcClientProvider{client: fake.NewSimpleDynamicClient(runtime.NewScheme())}, + stop: func() {}, + } + service := resources.NewService(provider) + t.Cleanup(service.StopAllPortForwards) + server, err := NewServer("token", fakeContexts{}, service, helm.NewService(nil)) + if err != nil { + t.Fatal(err) + } + request := httptest.NewRequest(http.MethodPost, "/v1/pods/portforward", + strings.NewReader(`{"contextId":"dev","namespace":"apps","name":"web","podPort":80}`)) + request.Header.Set("Authorization", "Bearer token") + requestCtx, cancel := context.WithCancel(context.Background()) + request = request.WithContext(requestCtx) + recorder := httptest.NewRecorder() + server.Handler().ServeHTTP(recorder, request) + if recorder.Code != http.StatusOK { + t.Fatalf("status = %d body=%s", recorder.Code, recorder.Body.String()) + } + cancel() + select { + case <-provider.captured.Done(): + t.Fatal("forward context was cancelled when the start request ended") + default: + } +} + func TestServerRequiresTokenAndServesHealthAndContexts(t *testing.T) { service := resources.NewService(rpcClientProvider{client: fake.NewSimpleDynamicClient(runtime.NewScheme())}) server, err := NewServer("token", fakeContexts{values: []session.ContextInfo{{ID: "dev", Name: "dev", Current: true}}}, service, helm.NewService(nil)) diff --git a/core/internal/rpc/validate.go b/core/internal/rpc/validate.go index 7706358..c6871ab 100644 --- a/core/internal/rpc/validate.go +++ b/core/internal/rpc/validate.go @@ -253,6 +253,14 @@ func validatePortForwardRequest(value resources.PortForwardRequest) error { if value.PodPort < 1 || value.PodPort > 65_535 { return fmt.Errorf("podPort must be between 1 and 65535") } + if value.LocalPort < 0 || value.LocalPort > 65_535 { + return fmt.Errorf("localPort must be between 0 and 65535") + } + switch value.Kind { + case "", "Pod", "Service", "Deployment", "StatefulSet", "DaemonSet", "ReplicaSet": + default: + return fmt.Errorf("%q cannot be port-forwarded", value.Kind) + } return nil } diff --git a/core/internal/rpc/validate_test.go b/core/internal/rpc/validate_test.go index 345da6e..3ea3cb9 100644 --- a/core/internal/rpc/validate_test.go +++ b/core/internal/rpc/validate_test.go @@ -118,6 +118,23 @@ func TestValidatePortForwardRequests(t *testing.T) { } } +func TestValidatePortForwardLocalPort(t *testing.T) { + base := resources.PortForwardRequest{ContextID: "ctx", Namespace: "default", Name: "pod", PodPort: 8080} + if err := validatePortForwardRequest(base); err != nil { + t.Fatalf("random localPort rejected: %v", err) + } + base.LocalPort = 8080 + if err := validatePortForwardRequest(base); err != nil { + t.Fatalf("explicit localPort rejected: %v", err) + } + for _, bad := range []int{-1, 65_536} { + base.LocalPort = bad + if err := validatePortForwardRequest(base); err == nil { + t.Fatalf("localPort %d accepted", bad) + } + } +} + func TestInvalidInputReturns400(t *testing.T) { service := resources.NewService(rpcClientProvider{client: nil}) server, err := NewServer("token", fakeContexts{}, service, helm.NewService(nil)) diff --git a/core/internal/session/forward_resolve.go b/core/internal/session/forward_resolve.go new file mode 100644 index 0000000..e1149da --- /dev/null +++ b/core/internal/session/forward_resolve.go @@ -0,0 +1,215 @@ +package session + +import ( + "context" + "fmt" + "sort" + + corev1 "k8s.io/api/core/v1" + discoveryv1 "k8s.io/api/discovery/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/kubernetes" +) + +// PortForwardTarget mirrors resources.PortForwardRequest for the session +// layer: the request kind plus which object name and port the forward is +// aimed at. Kept local to avoid resources -> session coupling. +type PortForwardTarget struct { + Namespace string + Name string + PodPort int64 + Kind string +} + +// forwardWorkloadKinds are the selector-based kinds a forward can resolve to +// a single backing pod. Every one of them exposes spec.selector of +// metav1.LabelSelector shape, so the same code path serves all. +var forwardWorkloadKinds = map[string]bool{ + "Deployment": true, + "StatefulSet": true, + "DaemonSet": true, + "ReplicaSet": true, +} + +// ResolveForwardTarget maps a Service or workload port-forward request to +// the pod that will actually carry the connection, translating named +// target ports to numbers. Service resolution reads the API server's own +// EndpointSlices (falling back to legacy Endpoints), which sidesteps +// selector translation entirely; workload resolution uses the full +// spec.selector, including matchExpressions. +func (m *Manager) ResolveForwardTarget(ctx context.Context, contextID, namespace, name, kind string, podPort int64) (string, int64, error) { + request := PortForwardTarget{Namespace: namespace, Name: name, PodPort: podPort, Kind: kind} + switch request.Kind { + case "Service": + return m.resolveServiceForwardTarget(ctx, contextID, request) + case "Deployment", "StatefulSet", "DaemonSet", "ReplicaSet": + return m.resolveWorkloadForwardTarget(ctx, contextID, request) + default: + return "", 0, fmt.Errorf("%q cannot be resolved to a backing pod", request.Kind) + } +} + +func (m *Manager) resolveServiceForwardTarget(ctx context.Context, contextID string, request PortForwardTarget) (string, int64, error) { + client, err := m.coreClient(contextID) + if err != nil { + return "", 0, err + } + service, err := client.CoreV1().Services(request.Namespace).Get(ctx, request.Name, metav1.GetOptions{}) + if err != nil { + return "", 0, fmt.Errorf("get service %q: %w", request.Name, err) + } + var servicePort *corev1.ServicePort + for index := range service.Spec.Ports { + if int64(service.Spec.Ports[index].Port) == request.PodPort && + (service.Spec.Ports[index].Protocol == "" || service.Spec.Ports[index].Protocol == corev1.ProtocolTCP) { + port := service.Spec.Ports[index] + servicePort = &port + break + } + } + if servicePort == nil { + return "", 0, fmt.Errorf("service %q has no TCP port %d", request.Name, request.PodPort) + } + + slices, err := client.DiscoveryV1().EndpointSlices(request.Namespace).List(ctx, metav1.ListOptions{ + LabelSelector: "kubernetes.io/service-name=" + request.Name, + }) + if err != nil && apierrors.IsNotFound(err) { + // Pre-1.19 clusters only have the legacy Endpoints object. + return m.resolveLegacyEndpoints(ctx, client, request, servicePort) + } else if err != nil { + return "", 0, fmt.Errorf("list endpointslices for service %q: %w", request.Name, err) + } + for index := range slices.Items { + slice := &slices.Items[index] + for endpointIndex := range slice.Endpoints { + endpoint := &slice.Endpoints[endpointIndex] + if endpoint.TargetRef == nil || endpoint.TargetRef.Kind != "Pod" || !endpointReady(endpoint) { + continue + } + for portIndex := range slice.Ports { + port := slice.Ports[portIndex] + if portMatchesServicePort(port, servicePort) { + return endpoint.TargetRef.Name, int64(portValue(port)), nil + } + } + } + } + // Empty or unusable slices can occur on mixed-version clusters. + return m.resolveLegacyEndpoints(ctx, client, request, servicePort) +} + +func (m *Manager) resolveLegacyEndpoints(ctx context.Context, client kubernetes.Interface, request PortForwardTarget, servicePort *corev1.ServicePort) (string, int64, error) { + endpoints, err := client.CoreV1().Endpoints(request.Namespace).Get(ctx, request.Name, metav1.GetOptions{}) + if err != nil { + return "", 0, fmt.Errorf("get endpoints for service %q: %w", request.Name, err) + } + for subsetIndex := range endpoints.Subsets { + subset := &endpoints.Subsets[subsetIndex] + for addressIndex := range subset.Addresses { + address := &subset.Addresses[addressIndex] + if address.TargetRef == nil || address.TargetRef.Kind != "Pod" { + continue + } + for portIndex := range subset.Ports { + port := subset.Ports[portIndex] + if port.Name == servicePort.Name && (port.Protocol == "" || port.Protocol == corev1.ProtocolTCP) && port.Port > 0 { + return address.TargetRef.Name, int64(port.Port), nil + } + } + } + } + return "", 0, fmt.Errorf("service %q has no ready endpoints", request.Name) +} + +func endpointReady(endpoint *discoveryv1.Endpoint) bool { + return endpoint.Conditions.Ready == nil || *endpoint.Conditions.Ready +} + +func portMatchesServicePort(port discoveryv1.EndpointPort, servicePort *corev1.ServicePort) bool { + name := "" + if port.Name != nil { + name = *port.Name + } + return name == servicePort.Name && port.Port != nil && *port.Port > 0 && + (port.Protocol == nil || *port.Protocol == corev1.ProtocolTCP) +} + +func portValue(port discoveryv1.EndpointPort) int32 { + if port.Port == nil { + return 0 + } + return *port.Port +} + +func (m *Manager) resolveWorkloadForwardTarget(ctx context.Context, contextID string, request PortForwardTarget) (string, int64, error) { + client, err := m.coreClient(contextID) + if err != nil { + return "", 0, err + } + var selector *metav1.LabelSelector + switch request.Kind { + case "Deployment": + object, err := client.AppsV1().Deployments(request.Namespace).Get(ctx, request.Name, metav1.GetOptions{}) + if err != nil { + return "", 0, fmt.Errorf("get deployment %q: %w", request.Name, err) + } + selector = object.Spec.Selector + case "StatefulSet": + object, err := client.AppsV1().StatefulSets(request.Namespace).Get(ctx, request.Name, metav1.GetOptions{}) + if err != nil { + return "", 0, fmt.Errorf("get statefulset %q: %w", request.Name, err) + } + selector = object.Spec.Selector + case "DaemonSet": + object, err := client.AppsV1().DaemonSets(request.Namespace).Get(ctx, request.Name, metav1.GetOptions{}) + if err != nil { + return "", 0, fmt.Errorf("get daemonset %q: %w", request.Name, err) + } + selector = object.Spec.Selector + case "ReplicaSet": + object, err := client.AppsV1().ReplicaSets(request.Namespace).Get(ctx, request.Name, metav1.GetOptions{}) + if err != nil { + return "", 0, fmt.Errorf("get replicaset %q: %w", request.Name, err) + } + selector = object.Spec.Selector + default: + return "", 0, fmt.Errorf("%q is not a workload that can be forwarded", request.Kind) + } + if selector == nil { + return "", 0, fmt.Errorf("workload %q has no pod selector", request.Name) + } + labelSelector, err := metav1.LabelSelectorAsSelector(selector) + if err != nil { + return "", 0, fmt.Errorf("convert selector for %q: %w", request.Name, err) + } + pods, err := client.CoreV1().Pods(request.Namespace).List(ctx, metav1.ListOptions{LabelSelector: labelSelector.String()}) + if err != nil { + return "", 0, fmt.Errorf("list pods for %q: %w", request.Name, err) + } + if len(pods.Items) == 0 { + return "", 0, fmt.Errorf("no pods match %q %q", request.Kind, request.Name) + } + // Prefer Running and ready, then Running, then anything else; among + // equals, pick the newest so a rollout lands on the current pods. + sort.SliceStable(pods.Items, func(i, j int) bool { + return podForwardRank(&pods.Items[i]) > podForwardRank(&pods.Items[j]) + }) + pod := &pods.Items[0] + return pod.Name, request.PodPort, nil +} + +func podForwardRank(pod *corev1.Pod) int { + rank := 0 + if pod.Status.Phase == corev1.PodRunning { + rank += 2 + for _, condition := range pod.Status.Conditions { + if condition.Type == corev1.PodReady && condition.Status == corev1.ConditionTrue { + rank++ + break + } + } + } + return rank +} diff --git a/core/internal/session/forward_resolve_test.go b/core/internal/session/forward_resolve_test.go new file mode 100644 index 0000000..ff17841 --- /dev/null +++ b/core/internal/session/forward_resolve_test.go @@ -0,0 +1,215 @@ +package session + +import ( + "context" + "testing" + + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + discoveryv1 "k8s.io/api/discovery/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/types" + "k8s.io/apimachinery/pkg/util/intstr" + kubernetesfake "k8s.io/client-go/kubernetes/fake" + ktesting "k8s.io/client-go/testing" +) + +func newResolveManager(objects ...runtime.Object) *Manager { + clientset := kubernetesfake.NewSimpleClientset(objects...) + manager := NewManager(nil) + manager.coreClients["dev"] = clientset + return manager +} + +func TestResolveServiceForwardTargetPicksReadyEndpoint(t *testing.T) { + ready := true + notReady := false + named := "http" + port80 := int32(80) + manager := newResolveManager( + &corev1.Service{ObjectMeta: metav1.ObjectMeta{Name: "web", Namespace: "apps"}, Spec: corev1.ServiceSpec{Ports: []corev1.ServicePort{{Name: "http", Port: 80, TargetPort: intstr.FromString(named)}}}}, + &discoveryv1.EndpointSlice{ + ObjectMeta: metav1.ObjectMeta{Name: "web-1", Namespace: "apps", Labels: map[string]string{"kubernetes.io/service-name": "web"}}, + AddressType: "IPv4", + Endpoints: []discoveryv1.Endpoint{ + {Conditions: discoveryv1.EndpointConditions{Ready: ¬Ready}, TargetRef: v1ObjectReference("Pod", "web-not-ready")}, + {Conditions: discoveryv1.EndpointConditions{Ready: &ready}, TargetRef: v1ObjectReference("Pod", "web-ready")}, + }, + Ports: []discoveryv1.EndpointPort{{Name: &named, Port: &port80}}, + }, + ) + pod, port, err := manager.ResolveForwardTarget(context.Background(), "dev", "apps", "web", "Service", 80) + if err != nil { + t.Fatal(err) + } + if pod != "web-ready" || port != 80 { + t.Fatalf("pod=%q port=%d", pod, port) + } +} + +func TestResolveServiceForwardTargetNoReadyEndpoints(t *testing.T) { + notReady := false + manager := newResolveManager( + &corev1.Service{ObjectMeta: metav1.ObjectMeta{Name: "web", Namespace: "apps"}, Spec: corev1.ServiceSpec{Ports: []corev1.ServicePort{{Port: 80, TargetPort: intstr.FromInt(8080)}}}}, + &discoveryv1.EndpointSlice{ + ObjectMeta: metav1.ObjectMeta{Name: "web-1", Namespace: "apps", Labels: map[string]string{"kubernetes.io/service-name": "web"}}, + Endpoints: []discoveryv1.Endpoint{{Conditions: discoveryv1.EndpointConditions{Ready: ¬Ready}, TargetRef: v1ObjectReference("Pod", "web")}}, + }, + ) + if _, _, err := manager.ResolveForwardTarget(context.Background(), "dev", "apps", "web", "Service", 80); err == nil { + t.Fatal("expected no-ready-endpoints error") + } +} + +func TestResolveWorkloadForwardTargetUsesFullSelector(t *testing.T) { + manager := newResolveManager( + appsv1Deployment("apps", "api", map[string]string{"app": "api"}, []metav1.LabelSelectorRequirement{{Key: "tier", Operator: metav1.LabelSelectorOpIn, Values: []string{"web", "edge"}}}), + podWithLabels("apps", "api-a", map[string]string{"app": "api", "tier": "web"}, corev1.PodRunning, true), + podWithLabels("apps", "api-b", map[string]string{"app": "api", "tier": "cache"}, corev1.PodRunning, true), + podWithLabels("apps", "api-c", map[string]string{"app": "api", "tier": "web"}, corev1.PodPending, false), + ) + pod, port, err := manager.ResolveForwardTarget(context.Background(), "dev", "apps", "api", "Deployment", 8080) + if err != nil { + t.Fatal(err) + } + if pod != "api-a" { + t.Fatalf("pod=%q, want api-a (Running+ready, selector matches)", pod) + } + if port != 8080 { + t.Fatalf("port=%d", port) + } +} + +func TestResolveWorkloadForwardTargetNoMatch(t *testing.T) { + manager := newResolveManager( + appsv1Deployment("apps", "api", map[string]string{"app": "api"}, nil), + podWithLabels("apps", "other", map[string]string{"app": "other"}, corev1.PodRunning, true), + ) + if _, _, err := manager.ResolveForwardTarget(context.Background(), "dev", "apps", "api", "Deployment", 8080); err == nil { + t.Fatal("expected no-match error") + } +} + +func TestResolveWorkloadForwardTargetPrefersRunning(t *testing.T) { + manager := newResolveManager( + appsv1Deployment("apps", "api", map[string]string{"app": "api"}, nil), + podWithLabels("apps", "api-pending", map[string]string{"app": "api"}, corev1.PodPending, false), + podWithLabels("apps", "api-running", map[string]string{"app": "api"}, corev1.PodRunning, true), + ) + pod, _, err := manager.ResolveForwardTarget(context.Background(), "dev", "apps", "api", "Deployment", 80) + if err != nil { + t.Fatal(err) + } + if pod != "api-running" { + t.Fatalf("pod=%q, want api-running", pod) + } +} + +func v1ObjectReference(kind, name string) *corev1.ObjectReference { + return &corev1.ObjectReference{Kind: kind, Name: name} +} + +func podWithLabels(namespace, name string, labels map[string]string, phase corev1.PodPhase, ready bool) *corev1.Pod { + status := corev1.ConditionFalse + if ready { + status = corev1.ConditionTrue + } + return &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: namespace, Labels: labels, UID: types.UID(name)}, + Status: corev1.PodStatus{ + Phase: phase, + Conditions: []corev1.PodCondition{{Type: corev1.PodReady, Status: status}}, + }, + } +} + +func appsv1Deployment(namespace, name string, matchLabels map[string]string, matchExpressions []metav1.LabelSelectorRequirement) *appsv1.Deployment { + return &appsv1.Deployment{ + ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: namespace}, + Spec: appsv1.DeploymentSpec{ + Selector: &metav1.LabelSelector{MatchLabels: matchLabels, MatchExpressions: matchExpressions}, + }, + } +} + +func TestServiceForwardNamedPortAndLegacyFallback(t *testing.T) { + for _, legacy := range []bool{false, true} { + t.Run(map[bool]string{false: "slices", true: "legacy"}[legacy], func(t *testing.T) { + name := "public" + port := int32(8080) + service := &corev1.Service{ObjectMeta: metav1.ObjectMeta{Name: "web", Namespace: "apps"}, Spec: corev1.ServiceSpec{Ports: []corev1.ServicePort{{Name: name, Port: 80, TargetPort: intstr.FromString("http")}}}} + slice := &discoveryv1.EndpointSlice{ObjectMeta: metav1.ObjectMeta{Name: "web-1", Namespace: "apps", Labels: map[string]string{"kubernetes.io/service-name": "web"}}, Endpoints: []discoveryv1.Endpoint{{TargetRef: v1ObjectReference("Pod", "web-1")}}, Ports: []discoveryv1.EndpointPort{{Name: &name, Port: &port}}} + endpoints := &corev1.Endpoints{ObjectMeta: metav1.ObjectMeta{Name: "web", Namespace: "apps"}, Subsets: []corev1.EndpointSubset{{Addresses: []corev1.EndpointAddress{{TargetRef: v1ObjectReference("Pod", "web-1")}}, Ports: []corev1.EndpointPort{{Name: name, Port: port}}}}} + var m *Manager + if legacy { + m = newResolveManager(service, endpoints) + m.coreClients["dev"].(*kubernetesfake.Clientset).PrependReactor("list", "endpointslices", func(ktesting.Action) (bool, runtime.Object, error) { + return true, nil, apierrors.NewNotFound(schema.GroupResource{Group: "discovery.k8s.io", Resource: "endpointslices"}, "") + }) + } else { + m = newResolveManager(service, slice) + } + pod, gotPort, err := m.ResolveForwardTarget(context.Background(), "dev", "apps", "web", "Service", 80) + if err != nil || pod != "web-1" || gotPort != 8080 { + t.Fatalf("pod=%q port=%d err=%v", pod, gotPort, err) + } + }) + } +} + +func TestEndpointPortMatchesServiceNameAndTCP(t *testing.T) { + name := "public" + wrong := "http" + port := int32(8080) + udp := corev1.ProtocolUDP + servicePort := &corev1.ServicePort{Name: name, Port: 80, TargetPort: intstr.FromString(wrong)} + for _, tc := range []struct { + name string + port discoveryv1.EndpointPort + want bool + }{ + {"valid", discoveryv1.EndpointPort{Name: &name, Port: &port}, true}, + {"target name", discoveryv1.EndpointPort{Name: &wrong, Port: &port}, false}, + {"missing port", discoveryv1.EndpointPort{Name: &name}, false}, + {"udp", discoveryv1.EndpointPort{Name: &name, Port: &port, Protocol: &udp}, false}, + } { + t.Run(tc.name, func(t *testing.T) { + if got := portMatchesServicePort(tc.port, servicePort); got != tc.want { + t.Fatalf("match=%v", got) + } + }) + } +} + +func TestServiceForwardSelectsTCPWithSameNumberUDP(t *testing.T) { + for _, udpFirst := range []bool{true, false} { + name := "dns-tcp" + port := int32(53) + tcp := corev1.ProtocolTCP + ports := []corev1.ServicePort{ + {Name: "dns-udp", Port: 53, TargetPort: intstr.FromInt(53), Protocol: corev1.ProtocolUDP}, + {Name: name, Port: 53, TargetPort: intstr.FromInt(53), Protocol: tcp}, + } + if !udpFirst { + ports[0], ports[1] = ports[1], ports[0] + } + m := newResolveManager( + &corev1.Service{ObjectMeta: metav1.ObjectMeta{Name: "dns", Namespace: "apps"}, Spec: corev1.ServiceSpec{Ports: ports}}, + &discoveryv1.EndpointSlice{ObjectMeta: metav1.ObjectMeta{Name: "dns-1", Namespace: "apps", Labels: map[string]string{"kubernetes.io/service-name": "dns"}}, Endpoints: []discoveryv1.Endpoint{{TargetRef: v1ObjectReference("Pod", "dns-1")}}, Ports: []discoveryv1.EndpointPort{{Name: &name, Port: &port, Protocol: &tcp}}}, + ) + pod, gotPort, err := m.ResolveForwardTarget(context.Background(), "dev", "apps", "dns", "Service", 53) + if err != nil || pod != "dns-1" || gotPort != 53 { + t.Fatalf("udpFirst=%v pod=%q port=%d err=%v", udpFirst, pod, gotPort, err) + } + } +} + +func TestServiceForwardRejectsUDPOnly(t *testing.T) { + m := newResolveManager(&corev1.Service{ObjectMeta: metav1.ObjectMeta{Name: "dns", Namespace: "apps"}, Spec: corev1.ServiceSpec{Ports: []corev1.ServicePort{{Port: 53, Protocol: corev1.ProtocolUDP}}}}) + if _, _, err := m.ResolveForwardTarget(context.Background(), "dev", "apps", "dns", "Service", 53); err == nil { + t.Fatal("UDP-only service accepted") + } +} diff --git a/core/internal/session/manager.go b/core/internal/session/manager.go index dd32886..667c2ee 100644 --- a/core/internal/session/manager.go +++ b/core/internal/session/manager.go @@ -5,13 +5,18 @@ import ( "context" "fmt" "io" + "net" "net/http" + "net/http/httptrace" "net/url" "strings" "sync" + "time" corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/util/httpstream" + streamspdy "k8s.io/apimachinery/pkg/util/httpstream/spdy" "k8s.io/client-go/discovery" "k8s.io/client-go/dynamic" "k8s.io/client-go/kubernetes" @@ -20,7 +25,6 @@ import ( "k8s.io/client-go/tools/clientcmd" "k8s.io/client-go/tools/portforward" "k8s.io/client-go/tools/remotecommand" - "k8s.io/client-go/transport/spdy" "github.com/zjy365/aster/core/internal/version" ) @@ -252,7 +256,7 @@ func (m *Manager) Client(contextID string) (dynamic.Interface, error) { // PortForward opens a loopback listener on a random free port and forwards to // the pod port over SPDY. The returned stop function tears the listener down; // the forward also ends when the context is cancelled. -func (m *Manager) PortForward(ctx context.Context, contextID, namespace, name string, podPort int64) (func(), int, error) { +func (m *Manager) PortForward(ctx context.Context, contextID, namespace, name string, podPort, localPort int64) (func(), int, error) { if contextID == "" || namespace == "" || name == "" { return nil, 0, fmt.Errorf("contextId, namespace and name are required") } @@ -261,14 +265,17 @@ func (m *Manager) PortForward(ctx context.Context, contextID, namespace, name st return nil, 0, fmt.Errorf("load context %q: %w", contextID, err) } config.UserAgent = version.UserAgent() - transport, upgrade, err := spdy.RoundTripperFor(config) + // HTTP/1.1 upgrades use net/http's cancellable response-header read. + config.NextProtos = []string{"http/1.1"} + client, err := rest.HTTPClientFor(config) if err != nil { - return nil, 0, fmt.Errorf("create spdy round tripper: %w", err) + return nil, 0, fmt.Errorf("create port-forward client: %w", err) } - dialer := spdy.NewDialer(upgrade, &http.Client{Transport: transport}, "POST", upstreamURL(config.Host, namespace, name)) + dialer := &portForwardDialer{ctx: ctx, client: client, url: upstreamURL(config.Host, namespace, name)} stopChan := make(chan struct{}) + stop := sync.OnceFunc(func() { close(stopChan) }) readyChan := make(chan struct{}) - forwarder, err := portforward.New(dialer, []string{fmt.Sprintf("0:%d", podPort)}, stopChan, readyChan, io.Discard, io.Discard) + forwarder, err := portforward.New(dialer, []string{fmt.Sprintf("%d:%d", localPort, podPort)}, stopChan, readyChan, io.Discard, io.Discard) if err != nil { return nil, 0, fmt.Errorf("create port forwarder: %w", err) } @@ -279,26 +286,101 @@ func (m *Manager) PortForward(ctx context.Context, contextID, namespace, name st case err := <-errChan: return nil, 0, fmt.Errorf("port forward: %w", err) case <-ctx.Done(): - close(stopChan) + stop() return nil, 0, ctx.Err() } ports, err := forwarder.GetPorts() if err != nil || len(ports) == 0 { - close(stopChan) + stop() return nil, 0, fmt.Errorf("read forwarded ports: %w", err) } - stop := func() { close(stopChan) } go func() { - <-ctx.Done() - stop() + select { + case <-ctx.Done(): + stop() + case <-stopChan: + case <-errChan: + stop() + } }() return stop, int(ports[0].Local), nil } +// client-go's SPDY transport does not cancel a blocked response-header read. +// net/http retains cancellation and kubeconfig transport settings, and exposes +// a successful 101 response as a duplex stream for the SPDY client. +type portForwardDialer struct { + ctx context.Context + client *http.Client + url *url.URL +} + +func (d *portForwardDialer) Dial(protocols ...string) (httpstream.Connection, string, error) { + request, err := http.NewRequestWithContext(d.ctx, http.MethodPost, d.url.String(), nil) + if err != nil { + return nil, "", err + } + request.Header.Set(httpstream.HeaderConnection, httpstream.HeaderUpgrade) + request.Header.Set(httpstream.HeaderUpgrade, streamspdy.HeaderSpdy31) + for _, protocol := range protocols { + request.Header.Add(httpstream.HeaderProtocolVersion, protocol) + } + var socket net.Conn + request = request.WithContext(httptrace.WithClientTrace(request.Context(), &httptrace.ClientTrace{ + GotConn: func(info httptrace.GotConnInfo) { socket = info.Conn }, + })) + response, err := d.client.Do(request) + if err != nil { + return nil, "", err + } + protocol := response.Header.Get(httpstream.HeaderProtocolVersion) + accepted := false + for _, offered := range protocols { + if protocol == offered { + accepted = true + } + } + if response.StatusCode != http.StatusSwitchingProtocols || !accepted || + !strings.EqualFold(response.Header.Get(httpstream.HeaderUpgrade), streamspdy.HeaderSpdy31) { + response.Body.Close() + return nil, "", fmt.Errorf("port-forward upgrade rejected: HTTP %d", response.StatusCode) + } + stream, ok := response.Body.(io.ReadWriteCloser) + if !ok || socket == nil { + response.Body.Close() + return nil, "", fmt.Errorf("port-forward upgrade did not return a duplex stream") + } + connection, err := streamspdy.NewClientConnectionWithPings(&upgradedForwardConn{Conn: socket, stream: stream}, 5*time.Second) + if err != nil { + stream.Close() + return nil, "", err + } + return connection, protocol, nil +} + +// Read through the HTTP response body to preserve bytes buffered during the +// upgrade; delegate addresses and deadlines to the underlying socket. +type upgradedForwardConn struct { + net.Conn + stream io.ReadWriteCloser +} + +func (c *upgradedForwardConn) Read(p []byte) (int, error) { return c.stream.Read(p) } +func (c *upgradedForwardConn) Write(p []byte) (int, error) { return c.stream.Write(p) } +func (c *upgradedForwardConn) Close() error { return c.stream.Close() } + func upstreamURL(host, namespace, name string) *url.URL { + scheme := "https" + trimmed := host + if strings.HasPrefix(host, "http://") { + scheme = "http" + trimmed = strings.TrimPrefix(host, "http://") + } else { + trimmed = strings.TrimPrefix(host, "https://") + } return &url.URL{ - Scheme: "https", - Host: strings.TrimPrefix(strings.TrimPrefix(host, "https://"), "http://"), + Scheme: scheme, + Host: trimmed, Path: fmt.Sprintf("/api/v1/namespaces/%s/pods/%s/portforward", namespace, name), } } diff --git a/core/internal/session/manager_test.go b/core/internal/session/manager_test.go index c217abb..69b7456 100644 --- a/core/internal/session/manager_test.go +++ b/core/internal/session/manager_test.go @@ -97,3 +97,20 @@ func TestPodExecReusesCachedCoreClient(t *testing.T) { t.Fatalf("core clients created = %d, want 1 (cached reuse)", created) } } + +func TestUpstreamURLScheme(t *testing.T) { + cases := []struct{ host, wantScheme string }{ + {"https://127.0.0.1:6443", "https"}, + {"http://127.0.0.1:8080", "http"}, + {"127.0.0.1:8080", "https"}, + } + for _, tc := range cases { + got := upstreamURL(tc.host, "apps", "web") + if got.Scheme != tc.wantScheme { + t.Errorf("host %q: scheme = %q, want %q", tc.host, got.Scheme, tc.wantScheme) + } + if got.Host != "127.0.0.1:6443" && got.Host != "127.0.0.1:8080" { + t.Errorf("host %q: unexpected host %q", tc.host, got.Host) + } + } +} diff --git a/core/internal/session/portforward_lifecycle_test.go b/core/internal/session/portforward_lifecycle_test.go new file mode 100644 index 0000000..675836f --- /dev/null +++ b/core/internal/session/portforward_lifecycle_test.go @@ -0,0 +1,158 @@ +// SPDX-License-Identifier: Apache-2.0 +package session + +import ( + "context" + "encoding/pem" + "io" + "net" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strconv" + "sync" + "testing" + "time" + + "k8s.io/apimachinery/pkg/util/httpstream" + streamspdy "k8s.io/apimachinery/pkg/util/httpstream/spdy" + "k8s.io/client-go/tools/clientcmd" + "k8s.io/client-go/tools/clientcmd/api" + "k8s.io/client-go/tools/portforward" +) + +func forwardTestManager(t *testing.T, server *httptest.Server) *Manager { + t.Helper() + cluster := &api.Cluster{Server: server.URL} + if cert := server.Certificate(); cert != nil { + cluster.CertificateAuthorityData = pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: cert.Raw}) + } + config := api.Config{ + Clusters: map[string]*api.Cluster{"cluster": cluster}, + Contexts: map[string]*api.Context{"dev": {Cluster: "cluster", AuthInfo: "user"}}, + AuthInfos: map[string]*api.AuthInfo{"user": {Token: "test-token"}}, + } + data, err := clientcmd.Write(config) + if err != nil { + t.Fatal(err) + } + path := filepath.Join(t.TempDir(), "config") + if err := os.WriteFile(path, data, 0600); err != nil { + t.Fatal(err) + } + rules := clientcmd.NewDefaultClientConfigLoadingRules() + rules.ExplicitPath = path + return NewManager(NewLoaderWithRules(rules)) +} + +func TestPortForwardCancelsStalledUpgrade(t *testing.T) { + entered := make(chan struct{}) + cancelled := make(chan struct{}) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + close(entered) + <-r.Context().Done() + close(cancelled) + })) + defer server.Close() + defer server.CloseClientConnections() + manager := forwardTestManager(t, server) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + result := make(chan error, 1) + go func() { _, _, err := manager.PortForward(ctx, "dev", "apps", "web", 80, 0); result <- err }() + select { + case <-entered: + case <-time.After(3 * time.Second): + t.Fatal("no upgrade request") + } + cancel() + select { + case err := <-result: + if err == nil { + t.Fatal("cancelled setup succeeded") + } + case <-time.After(3 * time.Second): + t.Fatal("setup stuck") + } + select { + case <-cancelled: + case <-time.After(3 * time.Second): + t.Fatal("upstream request leaked") + } +} + +func TestPortForwardConcurrentStopAndCancelReleasesListener(t *testing.T) { + closed := make(chan struct{}) + server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("Authorization") != "Bearer test-token" { + t.Error("kubeconfig authentication was not preserved") + http.Error(w, "unauthorized", http.StatusUnauthorized) + return + } + if _, err := httpstream.Handshake(r, w, []string{portforward.PortForwardProtocolV1Name}); err != nil { + return + } + conn := streamspdy.NewResponseUpgrader().UpgradeResponse(w, r, func(stream httpstream.Stream, replied <-chan struct{}) error { + if stream.Headers().Get("streamType") == "data" { + go func() { <-replied; _, _ = io.Copy(stream, stream); stream.Close() }() + } + return nil + }) + if conn == nil { + return + } + defer conn.Close() + <-conn.CloseChan() + close(closed) + })) + defer server.Close() + defer server.CloseClientConnections() + manager := forwardTestManager(t, server) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + stop, port, err := manager.PortForward(ctx, "dev", "apps", "web", 80, 0) + if err != nil { + t.Fatal(err) + } + local, err := net.DialTimeout("tcp4", net.JoinHostPort("127.0.0.1", strconv.Itoa(port)), time.Second) + if err != nil { + t.Fatal(err) + } + defer local.Close() + local.SetDeadline(time.Now().Add(3 * time.Second)) + if _, err := local.Write([]byte("ping")); err != nil { + t.Fatal(err) + } + response := make([]byte, 4) + if _, err := io.ReadFull(local, response); err != nil { + t.Fatal(err) + } + if string(response) != "ping" { + t.Fatalf("response=%q", response) + } + local.Close() + var wg sync.WaitGroup + for i := 0; i < 10; i++ { + wg.Add(1) + go func() { defer wg.Done(); stop(); cancel() }() + } + wg.Wait() + select { + case <-closed: + case <-time.After(3 * time.Second): + t.Fatal("upstream connection leaked") + } + deadline := time.Now().Add(3 * time.Second) + for { + listener, err := net.Listen("tcp4", net.JoinHostPort("127.0.0.1", strconv.Itoa(port))) + if err == nil { + listener.Close() + break + } + if time.Now().After(deadline) { + t.Fatalf("listener not reclaimed: %v", err) + } + time.Sleep(time.Millisecond) + } +}