From 762dd0d86be57d77201ac117e24f270c814cd374 Mon Sep 17 00:00:00 2001 From: yansun1996 Date: Wed, 5 Aug 2026 19:44:49 +0000 Subject: [PATCH 1/3] feat(spurctld): scope k0s cluster to a subset of nodes spur k8s up provisioned k0s over the controller's entire registered inventory, with no way to target a subset. Add node selection via --nodes (hostlist), --partition, and --selector key=val, combined as a union and persisted as the cluster's member scope. The reconcile loop assigns k0s roles only to member nodes; un-scoped nodes get no role and stay schedulable for Spur, which also gives a clean per-node opt-out. Empty selection enrolls the whole inventory (back-compat). --- crates/spur-cli/src/k8s.rs | 80 ++++++++ crates/spur-core/src/k0s.rs | 30 ++- crates/spur-core/src/wal.rs | 33 ++++ crates/spurctld/src/cluster.rs | 79 +++++++- crates/spurctld/src/cluster_k8s.rs | 181 ++++++++++++++++++- crates/spurctld/src/server.rs | 154 +++++++++++++++- proto/slurm.proto | 7 + tests/native_host/e2e/cluster.py | 7 + tests/native_host/e2e/test_k8s_scheduling.py | 17 ++ 9 files changed, 572 insertions(+), 16 deletions(-) diff --git a/crates/spur-cli/src/k8s.rs b/crates/spur-cli/src/k8s.rs index 387a25f7..3dadc389 100644 --- a/crates/spur-cli/src/k8s.rs +++ b/crates/spur-cli/src/k8s.rs @@ -42,6 +42,16 @@ pub enum K8sCommand { /// Overrides --replicas. #[arg(long = "control-plane-nodes", value_delimiter = ',')] control_plane_nodes: Vec, + /// Scope the cluster to a subset of nodes (hostlist, e.g. "gpu[01-08]"). Combined with + /// --partition/--selector as a union; empty = enroll the whole inventory. + #[arg(long)] + nodes: Option, + /// Scope the cluster to a partition's nodes. + #[arg(long)] + partition: Option, + /// Scope the cluster to nodes matching every key=value label (repeatable). + #[arg(long = "selector", value_parser = parse_key_val)] + selector: Vec<(String, String)>, }, /// Tear the k0s cluster down. Down { @@ -88,12 +98,18 @@ pub async fn main_with_args(args: Vec) -> Result<()> { control_plane_node, replicas, control_plane_nodes, + nodes, + partition, + selector, } => { cmd_up( &controller, control_plane_node, replicas, control_plane_nodes, + nodes, + partition, + selector, ) .await } @@ -112,6 +128,16 @@ fn effective_user() -> String { whoami::username().unwrap_or_else(|_| "unknown".into()) } +fn parse_key_val(s: &str) -> Result<(String, String), String> { + let (k, v) = s + .split_once('=') + .ok_or_else(|| format!("expected key=value, got {s}"))?; + if k.is_empty() { + return Err(format!("empty selector key in {s}")); + } + Ok((k.to_string(), v.to_string())) +} + async fn cmd_install_k0s(version: &str, path: &str, force: bool) -> Result<()> { let dest = std::path::Path::new(path); if dest.exists() && !force { @@ -130,11 +156,15 @@ async fn cmd_install_k0s(version: &str, path: &str, force: bool) -> Result<()> { Ok(()) } +#[allow(clippy::too_many_arguments)] async fn cmd_up( controller: &str, control_plane_node: Option, replicas: Option, control_plane_nodes: Vec, + nodes: Option, + partition: Option, + selector: Vec<(String, String)>, ) -> Result<()> { let mut client = SlurmControllerClient::new(spur_client::connect_channel(controller).await?); let resp = client @@ -143,6 +173,9 @@ async fn cmd_up( control_plane_replicas: replicas, control_plane_nodes, caller: effective_user(), + nodes: nodes.unwrap_or_default(), + partition: partition.unwrap_or_default(), + selector: selector.into_iter().collect(), }) .await? .into_inner(); @@ -186,6 +219,11 @@ async fn cmd_status(controller: &str) -> Result<()> { } else if !resp.control_plane_node.is_empty() { println!("control-plane: {}", resp.control_plane_node); } + if resp.member_nodes.is_empty() { + println!("members: all nodes"); + } else { + println!("members: {}", resp.member_nodes.join(", ")); + } for n in resp.nodes { println!( " {:<24} {:<11} {:<11} enabled={}", @@ -223,6 +261,7 @@ mod tests { control_plane_node, replicas, control_plane_nodes, + .. } => { assert_eq!(control_plane_node.as_deref(), Some("head-node")); assert_eq!(replicas, None); @@ -232,6 +271,47 @@ mod tests { } } + #[test] + fn parses_up_with_node_scope_flags() { + let args = K8sArgs::try_parse_from([ + "k8s", + "up", + "--nodes", + "gpu[01-08]", + "--partition", + "batch", + "--selector", + "zone=z1", + "--selector", + "gpu=mi300", + ]) + .unwrap(); + match args.command { + K8sCommand::Up { + nodes, + partition, + selector, + .. + } => { + assert_eq!(nodes.as_deref(), Some("gpu[01-08]")); + assert_eq!(partition.as_deref(), Some("batch")); + assert_eq!( + selector, + vec![ + ("zone".to_string(), "z1".to_string()), + ("gpu".to_string(), "mi300".to_string()) + ] + ); + } + _ => panic!("wrong command"), + } + } + + #[test] + fn selector_without_equals_is_rejected() { + assert!(K8sArgs::try_parse_from(["k8s", "up", "--selector", "bogus"]).is_err()); + } + #[test] fn parses_up_with_replicas_and_node_set() { let args = K8sArgs::try_parse_from(["k8s", "up", "--replicas", "3"]).unwrap(); diff --git a/crates/spur-core/src/k0s.rs b/crates/spur-core/src/k0s.rs index 38e8a433..ee836769 100644 --- a/crates/spur-core/src/k0s.rs +++ b/crates/spur-core/src/k0s.rs @@ -125,7 +125,7 @@ mod cluster_state_tests { phase: K0sPhase::Ready, control_plane_node: Some("cp-1".into()), control_plane_nodes: vec!["cp-1".into(), "cp-2".into(), "cp-3".into()], - reset_requested: false, + ..Default::default() }; assert_eq!(st.controllers(), vec!["cp-1", "cp-2", "cp-3"]); assert_eq!(st.bootstrap().as_deref(), Some("cp-1")); @@ -135,9 +135,8 @@ mod cluster_state_tests { fn bootstrap_falls_back_to_first_of_set_when_singular_absent() { let st = K0sClusterState { phase: K0sPhase::Ready, - control_plane_node: None, control_plane_nodes: vec!["cp-1".into(), "cp-2".into(), "cp-3".into()], - reset_requested: false, + ..Default::default() }; assert_eq!(st.bootstrap().as_deref(), Some("cp-1")); } @@ -146,6 +145,23 @@ mod cluster_state_tests { fn controllers_empty_when_down() { assert!(K0sClusterState::default().controllers().is_empty()); } + + #[test] + fn is_member_empty_scope_matches_all() { + let st = K0sClusterState::default(); + assert!(st.is_member("anything"), "empty scope = whole inventory"); + } + + #[test] + fn is_member_respects_recorded_scope() { + let st = K0sClusterState { + member_nodes: vec!["a".into(), "b".into()], + ..Default::default() + }; + assert!(st.is_member("a")); + assert!(st.is_member("b")); + assert!(!st.is_member("c"), "out-of-scope node excluded"); + } } #[cfg(test)] @@ -233,6 +249,9 @@ pub struct K0sClusterState { /// All control-plane nodes (1/3/5). Empty on pre-multi-CP state — read via [`Self::controllers`]. #[serde(default)] pub control_plane_nodes: Vec, + /// Nodes the cluster is scoped to. Empty = enroll the whole inventory (back-compat). + #[serde(default)] + pub member_nodes: Vec, #[serde(default)] pub reset_requested: bool, } @@ -254,4 +273,9 @@ impl K0sClusterState { .clone() .or_else(|| self.control_plane_nodes.first().cloned()) } + + /// Whether `name` is in scope for enrollment. An empty `member_nodes` means the whole inventory. + pub fn is_member(&self, name: &str) -> bool { + self.member_nodes.is_empty() || self.member_nodes.iter().any(|n| n == name) + } } diff --git a/crates/spur-core/src/wal.rs b/crates/spur-core/src/wal.rs index c52442b0..759710c7 100644 --- a/crates/spur-core/src/wal.rs +++ b/crates/spur-core/src/wal.rs @@ -257,6 +257,8 @@ pub enum WalOperation { #[serde(default)] control_plane_nodes: Vec, #[serde(default)] + member_nodes: Vec, + #[serde(default)] reset_requested: bool, }, NodeK0sClear { @@ -765,6 +767,12 @@ mod deregistration_wal_tests { phase: K0sPhase::Ready, control_plane_node: Some("head-node".into()), control_plane_nodes: vec!["head-node".into(), "cp-2".into(), "cp-3".into()], + member_nodes: vec![ + "head-node".into(), + "cp-2".into(), + "cp-3".into(), + "w-4".into(), + ], reset_requested: false, }; let back: WalOperation = @@ -774,11 +782,13 @@ mod deregistration_wal_tests { phase, control_plane_node, control_plane_nodes, + member_nodes, reset_requested, } => { assert_eq!(phase, K0sPhase::Ready); assert_eq!(control_plane_node.as_deref(), Some("head-node")); assert_eq!(control_plane_nodes, vec!["head-node", "cp-2", "cp-3"]); + assert_eq!(member_nodes, vec!["head-node", "cp-2", "cp-3", "w-4"]); assert!(!reset_requested); } _ => panic!("wrong variant"), @@ -798,17 +808,40 @@ mod deregistration_wal_tests { phase, control_plane_node, control_plane_nodes, + member_nodes, reset_requested, } => { assert_eq!(phase, K0sPhase::Ready); assert_eq!(control_plane_node.as_deref(), Some("head-node")); assert!(control_plane_nodes.is_empty()); + assert!(member_nodes.is_empty()); assert!(!reset_requested); } _ => panic!("wrong variant"), } } + // Frozen pre-member-scope K0sSetPhase entry (has control_plane_nodes, no member_nodes); must + // still deserialize with member_nodes defaulting empty (= whole inventory). Never regenerate. + #[test] + fn k0s_set_phase_pre_member_scope_payload_still_deserializes() { + const K0S_SET_PHASE_PRE_MEMBER_SCOPE: &str = r#"{"K0sSetPhase":{"phase":"ready","control_plane_node":"head-node","control_plane_nodes":["head-node","cp-2","cp-3"],"reset_requested":false}}"#; + let op: WalOperation = serde_json::from_str(K0S_SET_PHASE_PRE_MEMBER_SCOPE).expect( + "pre-member-scope K0sSetPhase must deserialize; member_nodes needs #[serde(default)]", + ); + match op { + WalOperation::K0sSetPhase { + control_plane_nodes, + member_nodes, + .. + } => { + assert_eq!(control_plane_nodes, vec!["head-node", "cp-2", "cp-3"]); + assert!(member_nodes.is_empty()); + } + _ => panic!("wrong variant"), + } + } + #[test] fn node_remove_none_reason_round_trips() { let op = WalOperation::NodeRemove { diff --git a/crates/spurctld/src/cluster.rs b/crates/spurctld/src/cluster.rs index 1d372ba0..bfd114e2 100644 --- a/crates/spurctld/src/cluster.rs +++ b/crates/spurctld/src/cluster.rs @@ -2041,19 +2041,21 @@ impl ClusterManager { Ok(()) } - /// set the cluster-wide k0s phase (+ optional control-plane node/set / reset flag). A `None` - /// `control_plane_node` or empty `control_plane_nodes` leaves the persisted value untouched. + /// set the cluster-wide k0s phase. A `None`/empty control-plane or `member_nodes` leaves the + /// persisted value untouched; a `reset_requested` teardown clears the member scope. pub fn set_k0s_phase( &self, phase: spur_core::k0s::K0sPhase, control_plane_node: Option, control_plane_nodes: Vec, + member_nodes: Vec, reset_requested: bool, ) -> anyhow::Result<()> { self.propose(WalOperation::K0sSetPhase { phase, control_plane_node, control_plane_nodes, + member_nodes, reset_requested, })?; info!(?phase, "k0s cluster phase set"); @@ -4755,6 +4757,7 @@ impl ClusterManager { phase, control_plane_node, control_plane_nodes, + member_nodes, reset_requested, } => { let mut k0s = self.k0s.write(); @@ -4765,7 +4768,13 @@ impl ClusterManager { if !control_plane_nodes.is_empty() { k0s.control_plane_nodes = control_plane_nodes.clone(); } + if !member_nodes.is_empty() { + k0s.member_nodes = member_nodes.clone(); + } k0s.reset_requested = *reset_requested; + if *reset_requested { + k0s.member_nodes.clear(); + } } } self.next_job_id.store(next_id, Ordering::Relaxed); @@ -11908,8 +11917,14 @@ mod tests { register_node(&cm, "n1", 4, 8000); cm.assign_node_k0s("n1", K0sRole::Worker, "10.44.0.2", "10.42.2.0/24") .unwrap(); - cm.set_k0s_phase(K0sPhase::Ready, Some("head-node".into()), Vec::new(), false) - .unwrap(); + cm.set_k0s_phase( + K0sPhase::Ready, + Some("head-node".into()), + Vec::new(), + Vec::new(), + false, + ) + .unwrap(); wait_for("k0s state applied", || { cm.k0s_state().phase == K0sPhase::Ready && cm.get_node("n1").and_then(|n| n.k0s_role).is_some() @@ -12036,8 +12051,14 @@ mod tests { register_node(&cm, "node-a", 4, 8000); wait_for("registered", || cm.get_node("node-a").is_some()); // Cluster is Ready with the control plane already recorded, but node-a has no k0s role. - cm.set_k0s_phase(K0sPhase::Ready, Some("node-a".into()), Vec::new(), false) - .unwrap(); + cm.set_k0s_phase( + K0sPhase::Ready, + Some("node-a".into()), + Vec::new(), + Vec::new(), + false, + ) + .unwrap(); wait_for("phase ready", || cm.k0s_state().phase == K0sPhase::Ready); assert!(cm.get_node("node-a").and_then(|n| n.k0s_role).is_none()); @@ -12096,6 +12117,50 @@ mod tests { assert_eq!(cm.k0s_state().control_plane_node.as_deref(), Some("node-a")); } + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn provision_only_roles_scoped_member_nodes() { + use spur_core::k0s::K0sPhase; + let dir = TempDir::new().unwrap(); + let cm = test_cluster(&dir).await; + for name in ["node-a", "node-b", "node-c"] { + register_node(&cm, name, 4, 8000); + } + wait_for("all registered", || { + ["node-a", "node-b", "node-c"] + .iter() + .all(|n| cm.get_node(n).is_some()) + }); + // Scope to a/b only. node-c is out of scope and must NEVER get a role, so it stays + // schedulable for Spur — the SPUR-112 fix (whole-inventory enrollment was the bug). + cm.set_k0s_phase( + K0sPhase::Provisioning, + Some("node-a".into()), + vec!["node-a".into()], + vec!["node-a".into(), "node-b".into()], + false, + ) + .unwrap(); + wait_for("scope recorded", || cm.k0s_state().member_nodes.len() == 2); + + let net = crate::cluster_k8s::ClusterNetworking { + mesh_cidr: "10.44.0.0/16".into(), + pod_cidr: "10.42.0.0/16".into(), + service_cidr: "10.43.0.0/16".into(), + cni_mtu: 1450, + cni: "kuberouter".into(), + control_plane_node: Some("node-a".into()), + }; + crate::cluster_k8s::provision_assignments(&cm, &net, &cm.k0s_state()).unwrap(); + wait_for("scoped nodes assigned", || { + cm.get_node("node-a").and_then(|n| n.k0s_role).is_some() + && cm.get_node("node-b").and_then(|n| n.k0s_role).is_some() + }); + assert!( + cm.get_node("node-c").and_then(|n| n.k0s_role).is_none(), + "out-of-scope node must stay un-roled and schedulable" + ); + } + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn provision_assigns_three_controllers_for_ha_set() { use spur_core::k0s::{K0sPhase, K0sRole}; @@ -12114,6 +12179,7 @@ mod tests { K0sPhase::Provisioning, Some("cp-a".into()), vec!["cp-a".into(), "cp-b".into(), "cp-c".into()], + Vec::new(), false, ) .unwrap(); @@ -12177,6 +12243,7 @@ mod tests { K0sPhase::Provisioning, None, vec!["cp-b".into(), "cp-a".into(), "cp-c".into()], + Vec::new(), false, ) .unwrap(); diff --git a/crates/spurctld/src/cluster_k8s.rs b/crates/spurctld/src/cluster_k8s.rs index cdad724e..bc39e3e4 100644 --- a/crates/spurctld/src/cluster_k8s.rs +++ b/crates/spurctld/src/cluster_k8s.rs @@ -129,6 +129,7 @@ pub(crate) fn provision_assignments( state: &K0sClusterState, ) -> anyhow::Result<()> { let mut nodes = cluster.get_nodes(); + nodes.retain(|n| state.is_member(&n.name)); if nodes.is_empty() { return Ok(()); } @@ -211,11 +212,68 @@ pub(crate) fn provision_assignments( // Persist the bootstrap choice if not already recorded (legacy single-CP path; `cluster_up` // records the full set up front for HA). if state.control_plane_node.as_deref() != Some(bootstrap.as_str()) { - cluster.set_k0s_phase(K0sPhase::Provisioning, Some(bootstrap), Vec::new(), false)?; + cluster.set_k0s_phase( + K0sPhase::Provisioning, + Some(bootstrap), + Vec::new(), + Vec::new(), + false, + )?; } Ok(()) } +/// Resolve the member scope for `spur k8s up` fail-closed: the UNION of a `nodes` hostlist, a +/// `partition`'s members, and a label `selector` (all pairs match). Empty (nothing given) = whole inventory. +pub(crate) fn resolve_member_nodes( + all_nodes: &[spur_core::node::Node], + nodes_hostlist: &str, + partition: &str, + selector: &HashMap, +) -> Result, String> { + if nodes_hostlist.is_empty() && partition.is_empty() && selector.is_empty() { + return Ok(Vec::new()); + } + let registered: HashSet<&str> = all_nodes.iter().map(|n| n.name.as_str()).collect(); + let mut members: HashSet = HashSet::new(); + + if !nodes_hostlist.is_empty() { + let expanded = spur_core::hostlist::expand(nodes_hostlist) + .map_err(|e| format!("invalid --nodes hostlist {nodes_hostlist}: {e}"))?; + for name in expanded { + if !registered.contains(name.as_str()) { + return Err(format!("node {name} is not a registered node")); + } + members.insert(name); + } + } + if !partition.is_empty() { + let mut any = false; + for n in all_nodes { + if n.partitions.iter().any(|p| p == partition) { + members.insert(n.name.clone()); + any = true; + } + } + if !any { + return Err(format!("partition {partition} has no registered nodes")); + } + } + if !selector.is_empty() { + for n in all_nodes { + if selector.iter().all(|(k, v)| n.labels.get(k) == Some(v)) { + members.insert(n.name.clone()); + } + } + } + if members.is_empty() { + return Err("node selection matched no registered nodes".to_string()); + } + let mut out: Vec = members.into_iter().collect(); + out.sort(); + Ok(out) +} + /// Resolve the control-plane set for `spur k8s up`, fail-closed, bootstrap node first: an explicit /// `nodes` list wins, else the lowest `replicas` candidates. Count must be 1/3/5 and fit the nodes. pub(crate) fn resolve_control_plane_set( @@ -259,12 +317,19 @@ pub(crate) fn resolve_control_plane_set( candidates.len() )); } + // Fail closed on a pinned bootstrap outside the candidate set (e.g. a --control-plane-node not in + // the requested node scope) — else `.1`/etcd-seed silently lands on a different, in-scope node. + if let Some(boot) = pinned_bootstrap { + if !candidates.iter().any(|c| c == boot) { + return Err(format!( + "control-plane node {boot} is not among the selected cluster nodes" + )); + } + } // Pin the bootstrap into the set first so `.1` lands on it, then fill from the lowest names. let mut set: Vec = Vec::new(); if let Some(boot) = pinned_bootstrap { - if candidates.iter().any(|c| c == boot) { - set.push(boot.to_string()); - } + set.push(boot.to_string()); } for c in candidates { if set.len() >= replicas as usize { @@ -646,7 +711,7 @@ async fn converge_provisioning( // Only transition on the edge — this reconcile also runs every tick while already Ready (to // heal re-added nodes), so an unconditional set would churn a WAL write + log line each tick. if all_active && cluster.k0s_state().phase != K0sPhase::Ready { - match cluster.set_k0s_phase(K0sPhase::Ready, None, Vec::new(), false) { + match cluster.set_k0s_phase(K0sPhase::Ready, None, Vec::new(), Vec::new(), false) { Ok(()) => info!("k0s cluster converged: all components active -> Ready"), Err(e) => warn!(error = %e, "failed to mark k0s cluster Ready"), } @@ -937,6 +1002,17 @@ mod tests { assert_eq!(set, names(&["a"])); } + #[test] + fn resolve_cp_set_rejects_singular_pin_outside_candidates() { + // With candidates narrowed to the member scope, a --control-plane-node outside it must error + // rather than be silently dropped and a different in-scope node elected. + let err = resolve_control_plane_set(names(&["a", "b"]), &[], Some("z"), 1).unwrap_err(); + assert!( + err.contains("control-plane node z is not among the selected"), + "got: {err}" + ); + } + #[test] fn first_host_is_dot_one() { assert_eq!( @@ -945,6 +1021,101 @@ mod tests { ); } + fn scope_node(name: &str, parts: &[&str], labels: &[(&str, &str)]) -> spur_core::node::Node { + let mut n = spur_core::node::Node::new(name.to_string(), Default::default()); + n.partitions = parts.iter().map(|s| s.to_string()).collect(); + n.labels = labels + .iter() + .map(|(k, v)| (k.to_string(), v.to_string())) + .collect(); + n + } + + fn sel(pairs: &[(&str, &str)]) -> HashMap { + pairs + .iter() + .map(|(k, v)| (k.to_string(), v.to_string())) + .collect() + } + + #[test] + fn resolve_members_empty_selection_is_whole_inventory() { + let nodes = vec![scope_node("a", &[], &[]), scope_node("b", &[], &[])]; + assert_eq!( + resolve_member_nodes(&nodes, "", "", &HashMap::new()).unwrap(), + Vec::::new(), + "no selection = empty = whole inventory" + ); + } + + #[test] + fn resolve_members_hostlist_expands_and_sorts() { + let nodes = vec![ + scope_node("gpu01", &[], &[]), + scope_node("gpu02", &[], &[]), + scope_node("gpu03", &[], &[]), + ]; + let out = resolve_member_nodes(&nodes, "gpu[01-02]", "", &HashMap::new()).unwrap(); + assert_eq!(out, names(&["gpu01", "gpu02"])); + } + + #[test] + fn resolve_members_hostlist_rejects_unregistered() { + let nodes = vec![scope_node("a", &[], &[])]; + let err = resolve_member_nodes(&nodes, "a,ghost", "", &HashMap::new()).unwrap_err(); + assert!(err.contains("ghost is not a registered node"), "got: {err}"); + } + + #[test] + fn resolve_members_partition_selects_members() { + let nodes = vec![ + scope_node("a", &["gpu"], &[]), + scope_node("b", &["cpu"], &[]), + scope_node("c", &["gpu"], &[]), + ]; + let out = resolve_member_nodes(&nodes, "", "gpu", &HashMap::new()).unwrap(); + assert_eq!(out, names(&["a", "c"])); + } + + #[test] + fn resolve_members_empty_partition_rejected() { + let nodes = vec![scope_node("a", &["gpu"], &[])]; + let err = resolve_member_nodes(&nodes, "", "nope", &HashMap::new()).unwrap_err(); + assert!(err.contains("partition nope has no"), "got: {err}"); + } + + #[test] + fn resolve_members_selector_matches_all_pairs() { + let nodes = vec![ + scope_node("a", &[], &[("zone", "z1"), ("gpu", "mi300")]), + scope_node("b", &[], &[("zone", "z1"), ("gpu", "mi200")]), + scope_node("c", &[], &[("zone", "z2"), ("gpu", "mi300")]), + ]; + let out = resolve_member_nodes(&nodes, "", "", &sel(&[("zone", "z1"), ("gpu", "mi300")])) + .unwrap(); + assert_eq!(out, names(&["a"]), "only the node matching BOTH pairs"); + } + + #[test] + fn resolve_members_union_dedups_across_surfaces() { + let nodes = vec![ + scope_node("a", &["gpu"], &[("fast", "1")]), + scope_node("b", &["gpu"], &[]), + scope_node("c", &[], &[("fast", "1")]), + scope_node("d", &[], &[]), + ]; + // hostlist {a} ∪ partition gpu {a,b} ∪ selector fast=1 {a,c} = {a,b,c}, a not duplicated. + let out = resolve_member_nodes(&nodes, "a", "gpu", &sel(&[("fast", "1")])).unwrap(); + assert_eq!(out, names(&["a", "b", "c"])); + } + + #[test] + fn resolve_members_selector_no_match_rejected() { + let nodes = vec![scope_node("a", &[], &[("zone", "z1")])]; + let err = resolve_member_nodes(&nodes, "", "", &sel(&[("zone", "z9")])).unwrap_err(); + assert!(err.contains("matched no registered nodes"), "got: {err}"); + } + fn mesh_node( name: &str, mesh_ip: Option<&str>, diff --git a/crates/spurctld/src/server.rs b/crates/spurctld/src/server.rs index 3f842873..6e9e9f68 100644 --- a/crates/spurctld/src/server.rs +++ b/crates/spurctld/src/server.rs @@ -2159,9 +2159,29 @@ impl SlurmController for ControllerService { let nodes = self.cluster.get_nodes(); let assigned = nodes.iter().any(|n| n.k0s_role.is_some()); + // Resolve the node scope fail-closed; a bare re-up of an assigned cluster keeps the recorded + // scope, a fresh up with no selection = whole inventory. CP candidates are the in-scope members. + let scope_requested = + !req.nodes.is_empty() || !req.partition.is_empty() || !req.selector.is_empty(); + let member_nodes = if assigned && !scope_requested { + state.member_nodes.clone() + } else { + crate::cluster_k8s::resolve_member_nodes( + &nodes, + &req.nodes, + &req.partition, + &req.selector, + ) + .map_err(Status::invalid_argument)? + }; + let candidates: Vec = if member_nodes.is_empty() { + nodes.iter().map(|n| n.name.clone()).collect() + } else { + member_nodes.clone() + }; + // Resolve the HA control-plane set fail-closed BEFORE recording intent: an explicit node // list wins, else `--replicas` (or the config default) picks the lowest-named nodes. - let candidates: Vec = nodes.into_iter().map(|n| n.name).collect(); let explicit_override = !req.control_plane_nodes.is_empty() || req.control_plane_replicas.is_some(); let replicas = req @@ -2201,6 +2221,12 @@ impl SlurmController for ControllerService { cp_set.join(", "), ))); } + if scope_requested && member_nodes != state.member_nodes { + return Err(Status::failed_precondition( + "cluster membership is already assigned; tear the cluster down \ + (spur k8s down --reset) before changing the node scope", + )); + } } let bootstrap = cp_set.first().cloned(); @@ -2209,6 +2235,7 @@ impl SlurmController for ControllerService { spur_core::k0s::K0sPhase::Provisioning, bootstrap, cp_set, + member_nodes, false, ) .map_err(|e| Status::internal(format!("set k0s phase: {e}")))?; @@ -2243,7 +2270,13 @@ impl SlurmController for ControllerService { )); } self.cluster - .set_k0s_phase(spur_core::k0s::K0sPhase::Down, None, Vec::new(), req.reset) + .set_k0s_phase( + spur_core::k0s::K0sPhase::Down, + None, + Vec::new(), + Vec::new(), + req.reset, + ) .map_err(|e| Status::internal(format!("set k0s phase: {e}")))?; Ok(Response::new(ClusterDownResponse { accepted: true, @@ -2267,6 +2300,7 @@ impl SlurmController for ControllerService { phase: crate::cluster_k8s::phase_str(state.phase), control_plane_node: state.control_plane_node.unwrap_or_default(), control_plane_nodes, + member_nodes: state.member_nodes, nodes: crate::cluster_k8s::live_node_statuses(&self.cluster).await, })) } @@ -3692,6 +3726,7 @@ mod tests { K0sPhase::Ready, Some("cp-a".into()), vec!["cp-a".into(), "cp-b".into(), "cp-c".into()], + Vec::new(), false, ) .unwrap(); @@ -3800,6 +3835,121 @@ mod tests { assert!(resp.accepted); } + async fn register_plain_node(svc: &ControllerService, name: &str, port: u16) { + svc.cluster + .register_node( + name.into(), + name.into(), + spur_core::resource::ResourceSet { + cpus: 4, + memory_mb: 8000, + ..Default::default() + }, + "127.0.0.1".into(), + port, + String::new(), + String::new(), + spur_core::node::NodeSource::NativeHost, + std::collections::HashMap::new(), + ) + .unwrap(); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn cluster_up_scopes_membership_to_selected_nodes() { + let dir = tempfile::TempDir::new().unwrap(); + let svc = test_service(&dir).await; + for (i, n) in ["node-a", "node-b", "node-c"].iter().enumerate() { + register_plain_node(&svc, n, 6818 + i as u16).await; + } + let resp = svc + .cluster_up(Request::new(ClusterUpRequest { + nodes: "node-a,node-b".into(), + ..Default::default() + })) + .await + .expect("scoped up accepted") + .into_inner(); + assert!(resp.accepted); + assert_eq!( + svc.cluster.k0s_state().member_nodes, + vec!["node-a", "node-b"] + ); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn cluster_up_rejects_control_plane_outside_scope() { + let dir = tempfile::TempDir::new().unwrap(); + let svc = test_service(&dir).await; + for (i, n) in ["node-a", "node-b", "node-c"].iter().enumerate() { + register_plain_node(&svc, n, 6818 + i as u16).await; + } + let err = svc + .cluster_up(Request::new(ClusterUpRequest { + nodes: "node-a,node-b".into(), + control_plane_nodes: vec!["node-c".into()], + ..Default::default() + })) + .await + .expect_err("a control plane outside the node scope must be rejected"); + assert_eq!(err.code(), Code::InvalidArgument); + } + + async fn scoped_assigned_cluster(svc: &ControllerService) { + for (i, n) in ["node-a", "node-b", "node-c"].iter().enumerate() { + register_plain_node(svc, n, 6818 + i as u16).await; + } + svc.cluster + .set_k0s_phase( + spur_core::k0s::K0sPhase::Provisioning, + Some("node-a".into()), + vec!["node-a".into()], + vec!["node-a".into(), "node-b".into()], + false, + ) + .unwrap(); + svc.cluster + .assign_node_k0s( + "node-a", + spur_core::k0s::K0sRole::Single, + "10.44.0.1", + "10.42.0.0/24", + ) + .unwrap(); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn cluster_up_bare_reup_preserves_recorded_scope() { + let dir = tempfile::TempDir::new().unwrap(); + let svc = test_service(&dir).await; + scoped_assigned_cluster(&svc).await; + let resp = svc + .cluster_up(Request::new(ClusterUpRequest::default())) + .await + .expect("bare re-up accepted") + .into_inner(); + assert!(resp.accepted); + assert_eq!( + svc.cluster.k0s_state().member_nodes, + vec!["node-a", "node-b"] + ); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn cluster_up_rejects_scope_change_on_assigned_cluster() { + let dir = tempfile::TempDir::new().unwrap(); + let svc = test_service(&dir).await; + scoped_assigned_cluster(&svc).await; + let err = svc + .cluster_up(Request::new(ClusterUpRequest { + nodes: "node-a,node-c".into(), + ..Default::default() + })) + .await + .expect_err("changing scope on an assigned cluster must be rejected"); + assert_eq!(err.code(), Code::FailedPrecondition); + } + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn cluster_kubeconfig_admin_flag_denied_for_non_admin() { let dir = tempfile::TempDir::new().unwrap(); diff --git a/proto/slurm.proto b/proto/slurm.proto index 0056a20e..b4cfa069 100644 --- a/proto/slurm.proto +++ b/proto/slurm.proto @@ -805,6 +805,11 @@ message ClusterUpRequest { // Explicit control-plane node names (1, 3, or 5). First is the etcd bootstrap node. repeated string control_plane_nodes = 4; string caller = 5; // For authorization (cluster admin required). + // Scope the cluster to a subset of the inventory: UNION of a nodes hostlist (e.g. "gpu[01-08]"), + // a partition's members, and a label selector (all key=val match). Empty = whole inventory. + string nodes = 6; + string partition = 7; + map selector = 8; } message ClusterUpResponse { bool accepted = 1; @@ -830,6 +835,8 @@ message ClusterStatusResponse { repeated ClusterNodeStatus nodes = 3; // All control-plane nodes (HA). Bootstrap node first; empty pre-multi-CP (see control_plane_node). repeated string control_plane_nodes = 4; + // Member nodes the cluster is scoped to. Empty = the whole inventory is enrolled. + repeated string member_nodes = 5; } message ClusterNodeStatus { diff --git a/tests/native_host/e2e/cluster.py b/tests/native_host/e2e/cluster.py index 63f5f9a9..075a6961 100644 --- a/tests/native_host/e2e/cluster.py +++ b/tests/native_host/e2e/cluster.py @@ -463,6 +463,13 @@ def k8s_control_planes(self) -> list[str]: return [n.strip() for n in names.split(",") if n.strip()] return [] + def k8s_members(self) -> str: + """Parse the `members:` line from `spur k8s status` ("all nodes" or a name list).""" + for line in self.k8s_status().splitlines(): + if line.startswith("members:"): + return line.split(":", 1)[1].strip() + return "" + def k8s_active_controllers(self) -> list[str]: """Node names whose `spur k8s status` row is role=controller/single and active.""" out = [] diff --git a/tests/native_host/e2e/test_k8s_scheduling.py b/tests/native_host/e2e/test_k8s_scheduling.py index 22d192fe..3b216a14 100644 --- a/tests/native_host/e2e/test_k8s_scheduling.py +++ b/tests/native_host/e2e/test_k8s_scheduling.py @@ -73,3 +73,20 @@ def test_k8s_reserved_node_excluded_from_scheduling(self, k8s_enabled_cluster): f"job should stay pending on a k8s-reserved node:\n{cluster.squeue_all()}" ) cluster.scancel(str(held)) + + def test_k8s_up_scopes_membership_to_selected_node(self, k8s_enabled_cluster): + # `spur k8s up --nodes ` records exactly that node as the member scope + # (SPUR-112); an empty scope would report "all nodes" instead. + cluster = k8s_enabled_cluster + target = cluster.node_names[0] + cluster.k8s_up(["--nodes", target]) + deadline = time.time() + 60 + members = "" + while time.time() < deadline: + members = cluster.k8s_members() + if members == target: + break + time.sleep(3) + assert members == target, ( + f"members scope should be exactly [{target}], got [{members}]:\n{cluster.k8s_status()}" + ) From b102e3966f382779905c362157a4da62767fd98c Mon Sep 17 00:00:00 2001 From: yansun1996 Date: Wed, 5 Aug 2026 21:34:37 +0000 Subject: [PATCH 2/3] fix(spurctld): harden k0s node-scope handling Reject duplicate --selector keys in the CLI instead of silently dropping earlier values (last-wins would change the intended label AND-match). Error when a supplied --selector matches no node, mirroring the --partition guard, so a typo isn't silently ignored. Clear the recorded member scope and control-plane set on any k8s down so the next up starts from a clean cluster identity rather than reusing stale state. Drop internal ticket refs from comments. --- crates/spur-cli/src/k8s.rs | 37 +++++++++++++++++-- crates/spurctld/src/cluster.rs | 38 +++++++++++++++++--- crates/spurctld/src/cluster_k8s.rs | 14 ++++++++ tests/native_host/e2e/test_k8s_scheduling.py | 4 +-- 4 files changed, 84 insertions(+), 9 deletions(-) diff --git a/crates/spur-cli/src/k8s.rs b/crates/spur-cli/src/k8s.rs index 3dadc389..771b3f4d 100644 --- a/crates/spur-cli/src/k8s.rs +++ b/crates/spur-cli/src/k8s.rs @@ -42,8 +42,8 @@ pub enum K8sCommand { /// Overrides --replicas. #[arg(long = "control-plane-nodes", value_delimiter = ',')] control_plane_nodes: Vec, - /// Scope the cluster to a subset of nodes (hostlist, e.g. "gpu[01-08]"). Combined with - /// --partition/--selector as a union; empty = enroll the whole inventory. + /// Scope the cluster to a subset of nodes (hostlist, e.g. "gpu[01-08]"), unioned with + /// --partition/--selector; empty = whole inventory. Resolved once here (not re-evaluated). #[arg(long)] nodes: Option, /// Scope the cluster to a partition's nodes. @@ -138,6 +138,20 @@ fn parse_key_val(s: &str) -> Result<(String, String), String> { Ok((k.to_string(), v.to_string())) } +/// Fold repeated `--selector key=val` into a map, rejecting a duplicate key rather than silently +/// dropping the earlier value (last-wins would change the intended AND scope). +fn selector_map( + selector: Vec<(String, String)>, +) -> Result, anyhow::Error> { + let mut map = std::collections::HashMap::new(); + for (k, v) in selector { + if map.insert(k.clone(), v).is_some() { + anyhow::bail!("duplicate --selector key {k}"); + } + } + Ok(map) +} + async fn cmd_install_k0s(version: &str, path: &str, force: bool) -> Result<()> { let dest = std::path::Path::new(path); if dest.exists() && !force { @@ -166,6 +180,7 @@ async fn cmd_up( partition: Option, selector: Vec<(String, String)>, ) -> Result<()> { + let selector = selector_map(selector)?; let mut client = SlurmControllerClient::new(spur_client::connect_channel(controller).await?); let resp = client .cluster_up(ClusterUpRequest { @@ -175,7 +190,7 @@ async fn cmd_up( caller: effective_user(), nodes: nodes.unwrap_or_default(), partition: partition.unwrap_or_default(), - selector: selector.into_iter().collect(), + selector, }) .await? .into_inner(); @@ -312,6 +327,22 @@ mod tests { assert!(K8sArgs::try_parse_from(["k8s", "up", "--selector", "bogus"]).is_err()); } + #[test] + fn selector_map_rejects_duplicate_key() { + let dup = vec![ + ("zone".to_string(), "z1".to_string()), + ("zone".to_string(), "z2".to_string()), + ]; + let err = selector_map(dup).unwrap_err().to_string(); + assert!(err.contains("duplicate --selector key zone"), "got: {err}"); + let ok = selector_map(vec![ + ("zone".to_string(), "z1".to_string()), + ("gpu".to_string(), "mi300".to_string()), + ]) + .unwrap(); + assert_eq!(ok.len(), 2); + } + #[test] fn parses_up_with_replicas_and_node_set() { let args = K8sArgs::try_parse_from(["k8s", "up", "--replicas", "3"]).unwrap(); diff --git a/crates/spurctld/src/cluster.rs b/crates/spurctld/src/cluster.rs index bfd114e2..39ed410e 100644 --- a/crates/spurctld/src/cluster.rs +++ b/crates/spurctld/src/cluster.rs @@ -2042,7 +2042,7 @@ impl ClusterManager { } /// set the cluster-wide k0s phase. A `None`/empty control-plane or `member_nodes` leaves the - /// persisted value untouched; a `reset_requested` teardown clears the member scope. + /// persisted value untouched; the `Down` phase clears the member scope + control-plane set. pub fn set_k0s_phase( &self, phase: spur_core::k0s::K0sPhase, @@ -4772,8 +4772,11 @@ impl ClusterManager { k0s.member_nodes = member_nodes.clone(); } k0s.reset_requested = *reset_requested; - if *reset_requested { + // Teardown resets cluster identity so the next `up` starts clean (no stale scope/CP). + if *phase == spur_core::k0s::K0sPhase::Down { k0s.member_nodes.clear(); + k0s.control_plane_node = None; + k0s.control_plane_nodes.clear(); } } } @@ -12130,8 +12133,7 @@ mod tests { .iter() .all(|n| cm.get_node(n).is_some()) }); - // Scope to a/b only. node-c is out of scope and must NEVER get a role, so it stays - // schedulable for Spur — the SPUR-112 fix (whole-inventory enrollment was the bug). + // Scope to a/b only; node-c is out of scope and must never get a role (stays schedulable). cm.set_k0s_phase( K0sPhase::Provisioning, Some("node-a".into()), @@ -12161,6 +12163,34 @@ mod tests { ); } + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn down_clears_member_scope_and_control_plane() { + use spur_core::k0s::K0sPhase; + let dir = TempDir::new().unwrap(); + let cm = test_cluster(&dir).await; + cm.set_k0s_phase( + K0sPhase::Provisioning, + Some("node-a".into()), + vec!["node-a".into()], + vec!["node-a".into(), "node-b".into()], + false, + ) + .unwrap(); + wait_for("scope recorded", || { + cm.k0s_state().member_nodes.len() == 2 && !cm.k0s_state().controllers().is_empty() + }); + // A plain down (reset=false) must clear the recorded scope + control plane so the next up + // starts clean rather than silently reusing stale identity. + cm.set_k0s_phase(K0sPhase::Down, None, Vec::new(), Vec::new(), false) + .unwrap(); + wait_for("state cleared on down", || { + let s = cm.k0s_state(); + s.member_nodes.is_empty() + && s.control_plane_node.is_none() + && s.controllers().is_empty() + }); + } + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn provision_assigns_three_controllers_for_ha_set() { use spur_core::k0s::{K0sPhase, K0sRole}; diff --git a/crates/spurctld/src/cluster_k8s.rs b/crates/spurctld/src/cluster_k8s.rs index bc39e3e4..743a5b38 100644 --- a/crates/spurctld/src/cluster_k8s.rs +++ b/crates/spurctld/src/cluster_k8s.rs @@ -260,11 +260,16 @@ pub(crate) fn resolve_member_nodes( } } if !selector.is_empty() { + let mut any = false; for n in all_nodes { if selector.iter().all(|(k, v)| n.labels.get(k) == Some(v)) { members.insert(n.name.clone()); + any = true; } } + if !any { + return Err("--selector matched no registered nodes".to_string()); + } } if members.is_empty() { return Err("node selection matched no registered nodes".to_string()); @@ -1116,6 +1121,15 @@ mod tests { assert!(err.contains("matched no registered nodes"), "got: {err}"); } + #[test] + fn resolve_members_bogus_selector_rejected_even_when_other_surface_matches() { + // A supplied selector that matches nothing must error even if --nodes/--partition matched, + // so a typo'd selector isn't silently ignored. + let nodes = vec![scope_node("a", &["gpu"], &[("zone", "z1")])]; + let err = resolve_member_nodes(&nodes, "a", "", &sel(&[("zone", "z9")])).unwrap_err(); + assert!(err.contains("--selector matched no"), "got: {err}"); + } + fn mesh_node( name: &str, mesh_ip: Option<&str>, diff --git a/tests/native_host/e2e/test_k8s_scheduling.py b/tests/native_host/e2e/test_k8s_scheduling.py index 3b216a14..aa889330 100644 --- a/tests/native_host/e2e/test_k8s_scheduling.py +++ b/tests/native_host/e2e/test_k8s_scheduling.py @@ -75,8 +75,8 @@ def test_k8s_reserved_node_excluded_from_scheduling(self, k8s_enabled_cluster): cluster.scancel(str(held)) def test_k8s_up_scopes_membership_to_selected_node(self, k8s_enabled_cluster): - # `spur k8s up --nodes ` records exactly that node as the member scope - # (SPUR-112); an empty scope would report "all nodes" instead. + # `spur k8s up --nodes ` records exactly that node as the member scope; + # an empty scope would report "all nodes" instead. cluster = k8s_enabled_cluster target = cluster.node_names[0] cluster.k8s_up(["--nodes", target]) From 22196ba19b8d008d8e5c006be2d3dbe1f54dc2c3 Mon Sep 17 00:00:00 2001 From: yansun1996 Date: Wed, 5 Aug 2026 22:10:25 +0000 Subject: [PATCH 3/3] fix(spurctld): reject k0s re-up while teardown drains roles Clearing the member scope and control-plane set on `down` takes effect immediately, but node roles drain on later reconcile ticks. A bare `k8s up` landing in that window saw roles still present (assigned) yet an emptied scope, so it reused the empty scope and silently enrolled the whole inventory while re-electing the control plane. Reject an up while the cluster is tearing down (phase Down with roles still present) so it can't reuse half-cleared state. --- crates/spur-cli/src/k8s.rs | 2 +- crates/spurctld/src/server.rs | 31 +++++++++++++++++++++++++++++++ 2 files changed, 32 insertions(+), 1 deletion(-) diff --git a/crates/spur-cli/src/k8s.rs b/crates/spur-cli/src/k8s.rs index 771b3f4d..6a329226 100644 --- a/crates/spur-cli/src/k8s.rs +++ b/crates/spur-cli/src/k8s.rs @@ -43,7 +43,7 @@ pub enum K8sCommand { #[arg(long = "control-plane-nodes", value_delimiter = ',')] control_plane_nodes: Vec, /// Scope the cluster to a subset of nodes (hostlist, e.g. "gpu[01-08]"), unioned with - /// --partition/--selector; empty = whole inventory. Resolved once here (not re-evaluated). + /// --partition/--selector; empty = whole inventory. Resolved once at up time (not re-evaluated). #[arg(long)] nodes: Option, /// Scope the cluster to a partition's nodes. diff --git a/crates/spurctld/src/server.rs b/crates/spurctld/src/server.rs index 6e9e9f68..02badef8 100644 --- a/crates/spurctld/src/server.rs +++ b/crates/spurctld/src/server.rs @@ -2159,6 +2159,14 @@ impl SlurmController for ControllerService { let nodes = self.cluster.get_nodes(); let assigned = nodes.iter().any(|n| n.k0s_role.is_some()); + // Teardown clears the recorded scope/CP but roles drain on later reconcile ticks; block a + // re-up in that window so it can't reuse the emptied scope and silently enroll every node. + if state.phase == spur_core::k0s::K0sPhase::Down && assigned { + return Err(Status::failed_precondition( + "cluster teardown is in progress; wait for it to finish before bringing it back up", + )); + } + // Resolve the node scope fail-closed; a bare re-up of an assigned cluster keeps the recorded // scope, a fresh up with no selection = whole inventory. CP candidates are the in-scope members. let scope_requested = @@ -3950,6 +3958,29 @@ mod tests { assert_eq!(err.code(), Code::FailedPrecondition); } + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn cluster_up_rejected_while_teardown_drains_roles() { + let dir = tempfile::TempDir::new().unwrap(); + let svc = test_service(&dir).await; + scoped_assigned_cluster(&svc).await; + // down clears the recorded scope/CP immediately; node-a's role drains on a later tick. A + // re-up in that window must be rejected, not silently widen to the whole inventory. + svc.cluster + .set_k0s_phase( + spur_core::k0s::K0sPhase::Down, + None, + Vec::new(), + Vec::new(), + false, + ) + .unwrap(); + let err = svc + .cluster_up(Request::new(ClusterUpRequest::default())) + .await + .expect_err("re-up during teardown must be rejected"); + assert_eq!(err.code(), Code::FailedPrecondition); + } + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn cluster_kubeconfig_admin_flag_denied_for_non_admin() { let dir = tempfile::TempDir::new().unwrap();