diff --git a/src/lib/inference/serving/managed-cluster-discovery-production.ts b/src/lib/inference/serving/managed-cluster-discovery-production.ts index a5c2ab47d32..ff874dafb0b 100644 --- a/src/lib/inference/serving/managed-cluster-discovery-production.ts +++ b/src/lib/inference/serving/managed-cluster-discovery-production.ts @@ -17,6 +17,7 @@ import { managedVllmStateDir } from "../vllm-api-key.js"; import { buildLocalManagedVllmDockerEnv, buildVllmSshTransportEnv } from "../vllm-docker-env.js"; import type { ManagedClusterCommandResult, + ManagedClusterConnectivityFailure, ManagedClusterConnectivityRequest, ManagedClusterDiscoveryDeps, ManagedClusterHostObservation, @@ -960,7 +961,11 @@ function isRecordArray(value: unknown): value is Record[] { function connectivityCheck( transport: ManagedClusterReadOnlyHostTransport, request: ManagedClusterConnectivityRequest, -): boolean { +): ManagedClusterConnectivityFailure | null { + const failed = (check: "route" | "jumbo" | "neighbor"): ManagedClusterConnectivityFailure => ({ + check, + netdev: request.netdev, + }); const routeResult = transport.execute([ "ip", "-j", @@ -976,9 +981,9 @@ function connectivityCheck( try { routeValue = parseJsonCommandResult(routeResult, "DGX Spark direct route probe"); } catch { - return false; + return failed("route"); } - if (!isRecordArray(routeValue) || routeValue.length !== 1) return false; + if (!isRecordArray(routeValue) || routeValue.length !== 1) return failed("route"); const route = routeValue[0]!; const routeSource = route.prefsrc ?? route.src; if ( @@ -987,7 +992,7 @@ function connectivityCheck( route.gateway !== undefined || (route.scope !== undefined && String(route.scope).toLowerCase() !== "link") ) { - return false; + return failed("route"); } const ping = transport.execute([ "ping", @@ -1003,7 +1008,7 @@ function connectivityCheck( request.sourceAddress, request.peerAddress, ]); - if (!commandSucceeded(ping)) return false; + if (!commandSucceeded(ping)) return failed("jumbo"); const neighborResult = transport.execute([ "ip", "-j", @@ -1018,28 +1023,30 @@ function connectivityCheck( try { neighborValue = parseJsonCommandResult(neighborResult, "DGX Spark neighbor probe"); } catch { - return false; + return failed("neighbor"); } - if (!isRecordArray(neighborValue) || neighborValue.length !== 1) return false; + if (!isRecordArray(neighborValue) || neighborValue.length !== 1) return failed("neighbor"); const neighbor = neighborValue[0]!; const states = Array.isArray(neighbor.state) ? neighbor.state : [neighbor.state]; - return ( + // `ip` applies the `dev` filter itself and then omits `dev` from the JSON, so an + // absent `dev` already means the entry belongs to the requested netdev. + const matched = String(neighbor.dst ?? "") === request.peerAddress && - String(neighbor.dev ?? "") === request.netdev && + String(neighbor.dev ?? request.netdev) === request.netdev && String(neighbor.lladdr ?? "").toLowerCase() === request.expectedPeerMac && states.length > 0 && states.every( (state) => typeof state === "string" && /^(?:REACHABLE|STALE|DELAY|PROBE|PERMANENT|NOARP)$/i.test(state), - ) - ); + ); + return matched ? null : failed("neighbor"); } function probeConnectivity( transport: ManagedClusterReadOnlyHostTransport, requests: readonly ManagedClusterConnectivityRequest[], -): boolean { +): ManagedClusterConnectivityFailure | null { if ( requests.length !== 2 || new Set(requests.map(({ netdev }) => netdev)).size !== 2 || @@ -1051,9 +1058,13 @@ function probeConnectivity( !MAC_PATTERN.test(expectedPeerMac), ) ) { - return false; + return { check: "rails" }; + } + for (const request of requests) { + const failure = connectivityCheck(transport, request); + if (failure) return failure; } - return requests.every((request) => connectivityCheck(transport, request)); + return null; } function createCanonicalReadiness( diff --git a/src/lib/inference/serving/managed-cluster-discovery.test.ts b/src/lib/inference/serving/managed-cluster-discovery.test.ts index cb18b8047f2..58a33b4401f 100644 --- a/src/lib/inference/serving/managed-cluster-discovery.test.ts +++ b/src/lib/inference/serving/managed-cluster-discovery.test.ts @@ -12,6 +12,7 @@ import type { SystemReadinessReport } from "../../readiness/types.js"; import { confirmManagedClusterManagedServingCapability, createManagedClusterDiscoveryDeps, + type ManagedClusterConnectivityRequest, type ManagedClusterDetectedManagedServingCapability, type ManagedClusterDiscoveryDeps, type ManagedClusterHostObservation, @@ -250,7 +251,7 @@ function fixture(overrides: Partial = {}) { }, probeConnectivity: (_candidate, requests) => { events.push(`connectivity:${requests[0]?.sourceAddress ?? "missing"}`); - return true; + return null; }, claimBinding: () => true, writeBinding: (_statePath, peerIdentity) => { @@ -720,6 +721,72 @@ describe("managed DGX Spark cluster discovery", () => { }); }); +function connectivityRequests(): ManagedClusterConnectivityRequest[] { + return [ + { + netdev: "enp1s0f0np0", + sourceAddress: "192.168.100.1", + peerAddress: "192.168.100.2", + expectedPeerMac: "02:00:00:00:01:01", + }, + { + netdev: "enp2s0f0np0", + sourceAddress: "192.168.101.1", + peerAddress: "192.168.101.2", + expectedPeerMac: "02:00:00:00:01:02", + }, + ]; +} + +/** + * Serves healthy route, ping, and neighbor output for every rail, degraded only where + * the options ask for it. `neighborKeys` selects which keys the neighbor JSON carries, + * so a test can reproduce the real `ip` output that omits the filtered-on `dev`. + */ +function connectivityTransport( + requests: readonly ManagedClusterConnectivityRequest[], + options: { + neighborKeys?: readonly ("dst" | "dev" | "lladdr" | "state")[]; + jumboFailsOn?: string; + neighborMissingOn?: string; + }, +): ManagedClusterReadOnlyHostTransport { + const keys = options.neighborKeys ?? ["dst", "dev", "lladdr", "state"]; + return { + execute: (argv) => { + const routeResponse = () => ({ + status: 0, + stdout: JSON.stringify([{ dev: argv.at(-1), prefsrc: argv[6], scope: "link" }]), + stderr: "", + }); + const pingResponse = () => ({ + status: Number(argv.at(-1) === options.jumboFailsOn), + stdout: "", + stderr: "", + }); + const neighborResponse = () => { + const request = requests.find(({ peerAddress }) => peerAddress === argv[5])!; + const entry = { + dst: request.peerAddress, + dev: request.netdev, + lladdr: request.expectedPeerMac, + state: ["REACHABLE"], + }; + const emitted = Object.fromEntries(keys.map((key) => [key, entry[key]])); + const rows = [emitted].filter(() => request.peerAddress !== options.neighborMissingOn); + return { status: 0, stdout: JSON.stringify(rows), stderr: "" }; + }; + return argv[1] === "-j" && argv[2] === "route" + ? routeResponse() + : argv[0] === "ping" + ? pingResponse() + : neighborResponse(); + }, + readFile: () => "", + readdir: () => [], + }; +} + describe("production pinned peer transport", () => { it("atomically preserves an existing binding-root owner", () => { const parent = fs.mkdtempSync(path.join(os.tmpdir(), "nemoclaw-spark-binding-")); @@ -869,7 +936,6 @@ describe("production pinned peer transport", () => { stdout: JSON.stringify([ { dst: request.peerAddress, - dev: request.netdev, lladdr: request.expectedPeerMac, state: ["REACHABLE"], }, @@ -887,9 +953,45 @@ describe("production pinned peer transport", () => { readdir: () => [], }; - expect(deps.probeConnectivity(directTransport, requests)).toBe(true); + expect(deps.probeConnectivity(directTransport, requests)).toBeNull(); routedThroughGateway = true; - expect(deps.probeConnectivity(directTransport, requests)).toBe(false); + expect(deps.probeConnectivity(directTransport, requests)).toEqual({ + check: "route", + netdev: "enp1s0f0np0", + }); + }); + + it("accepts a neighbor entry that omits the dev key filtered out by ip (#8519)", () => { + const deps = createManagedClusterDiscoveryDeps(() => ({ status: 0, stdout: "", stderr: "" })); + const requests = connectivityRequests(); + // `ip -j neigh show to dev ` filters on `dev` and then drops it + // from the JSON, so the healthy fabric reports no `dev` at all. + const transport = connectivityTransport(requests, { + neighborKeys: ["dst", "lladdr", "state"], + }); + + expect(deps.probeConnectivity(transport, requests)).toBeNull(); + }); + + it("names the rail and the sub-check that rejected the fabric (#8519)", () => { + const deps = createManagedClusterDiscoveryDeps(() => ({ status: 0, stdout: "", stderr: "" })); + const requests = connectivityRequests(); + + expect( + deps.probeConnectivity( + connectivityTransport(requests, { jumboFailsOn: "192.168.101.2" }), + requests, + ), + ).toEqual({ check: "jumbo", netdev: "enp2s0f0np0" }); + expect( + deps.probeConnectivity( + connectivityTransport(requests, { neighborMissingOn: "192.168.100.2" }), + requests, + ), + ).toEqual({ check: "neighbor", netdev: "enp1s0f0np0" }); + expect( + deps.probeConnectivity(connectivityTransport(requests, {}), requests.slice(0, 1)), + ).toEqual({ check: "rails" }); }); it("uses strict SSH and a fixed argv executor without interpolated shell", () => { diff --git a/src/lib/inference/serving/managed-cluster-discovery.ts b/src/lib/inference/serving/managed-cluster-discovery.ts index 0dbd3e25423..02b5457361d 100644 --- a/src/lib/inference/serving/managed-cluster-discovery.ts +++ b/src/lib/inference/serving/managed-cluster-discovery.ts @@ -153,6 +153,14 @@ export interface ManagedClusterConnectivityRequest { readonly expectedPeerMac: string; } +/** + * Which connectivity probe rejected the fabric. `rails` means the candidate rail + * set itself was unusable, so no per-rail probe ran. + */ +export type ManagedClusterConnectivityFailure = + | { readonly check: "rails" } + | { readonly check: "route" | "jumbo" | "neighbor"; readonly netdev: string }; + export interface ManagedClusterDiscoveryDeps { now(): Date; currentUid(): number | null; @@ -169,10 +177,11 @@ export interface ManagedClusterDiscoveryDeps { buildIdentity: BuildIdentity, now: Date, ): SystemReadinessReport; + /** Null means every rail passed every probe. */ probeConnectivity( transport: ManagedClusterReadOnlyHostTransport, requests: readonly ManagedClusterConnectivityRequest[], - ): boolean; + ): ManagedClusterConnectivityFailure | null; /** Atomically claim a new binding root. False means an existing owner won. */ claimBinding(statePath: string): boolean; writeBinding(statePath: string, identity: QualifiedManagedVllmSshIdentity): ManagedVllmSshBinding; @@ -928,6 +937,21 @@ function topologyFailureReason(result: ReturnType [host.nodeId, transport] as const), ]); for (const node of cluster.nodes) { - if ( - !deps.probeConnectivity( - transportByNodeId.get(node.host.nodeId)!, - cluster.connectivity.get(node.host.nodeId)!, - ) - ) { + const connectivityFailure = deps.probeConnectivity( + transportByNodeId.get(node.host.nodeId)!, + cluster.connectivity.get(node.host.nodeId)!, + ); + if (connectivityFailure) { return disposition(selection, { code: "connectivity-unavailable", - reason: `Direct route, neighbor, or jumbo connectivity failed on ${node.host.hostname}.`, + reason: connectivityFailureReason(node.host.hostname, connectivityFailure), }); } }