Add support for different versions of CAPI bootstrap and infrastructure specs, passthrough

Signed-off-by: Erick Bourgeois <erick@jeb.ca>

Author: ebourgeois

src/constants.rs

@@ -17,6 +17,8 @@
 //! - Operator instance environment variables
 //! - Security constraints (`MAX_DURATION_SECS`, `RESERVED_LABEL_PREFIXES`, …)
 
+use std::sync::OnceLock;
+
 // ============================================================================
 // Kubernetes API Constants
 // ============================================================================
@@ -623,11 +625,52 @@ pub const CAPI_MACHINE_API_GROUP: &str = "cluster.x-k8s.io";
 /// CAPI API group
 pub const CAPI_GROUP: &str = "cluster.x-k8s.io";
 
-/// CAPI Machine API version
-pub const CAPI_MACHINE_API_VERSION: &str = "v1beta1";
-
-/// Full CAPI Machine API version string
-pub const CAPI_MACHINE_API_VERSION_FULL: &str = "cluster.x-k8s.io/v1beta1";
+/// Runtime-resolved CAPI Machine API version (`"v1beta1"` / `"v1beta2"`), recorded
+/// once at startup by `main` via **mandatory discovery** (or the
+/// `CAPI_MACHINE_API_VERSION` env override). There is deliberately NO compile-time
+/// default version: the controller refuses to start if it can't determine which
+/// version the cluster serves, rather than silently guessing one. The Machine
+/// *group* and *kind* (`cluster.x-k8s.io` / `Machine`) are fixed by the CAPI
+/// contract; only the version varies, and it is never hardcoded for production.
+static CAPI_MACHINE_API_VERSION_RUNTIME: OnceLock<String> = OnceLock::new();
+
+/// Record the CAPI Machine API version resolved at startup (discovery or the
+/// `CAPI_MACHINE_API_VERSION` env override). Idempotent — only the first call
+/// takes effect; later calls are ignored (the served version is fixed for the
+/// controller's lifetime).
+pub fn set_capi_machine_api_version(version: String) {
+    let _ = CAPI_MACHINE_API_VERSION_RUNTIME.set(version);
+}
+
+/// The CAPI Machine API version the controller should create/watch Machines at.
+///
+/// Panics if called before [`set_capi_machine_api_version`] has run — that is an
+/// invariant violation: `main` resolves the version (mandatory discovery) before
+/// any reconcile runs, so the value is always present in production. Unit tests,
+/// which never perform real discovery, get the `v1beta1` their fixtures assume.
+#[must_use]
+pub fn capi_machine_api_version() -> &'static str {
+    match CAPI_MACHINE_API_VERSION_RUNTIME.get() {
+        Some(version) => version.as_str(),
+        None => {
+            #[cfg(test)]
+            {
+                "v1beta1"
+            }
+            #[cfg(not(test))]
+            panic!(
+                "CAPI Machine API version not resolved — set_capi_machine_api_version() \
+                 must run (after mandatory discovery) before any reconcile"
+            )
+        }
+    }
+}
+
+/// Full `cluster.x-k8s.io/<version>` string for the active CAPI Machine version.
+#[must_use]
+pub fn capi_machine_api_version_full() -> String {
+    format!("{CAPI_GROUP}/{}", capi_machine_api_version())
+}
 
 /// CAPI cluster name label
 pub const CAPI_CLUSTER_NAME_LABEL: &str = "cluster.x-k8s.io/cluster-name";

src/main.rs

@@ -18,10 +18,11 @@ use std::sync::Arc;
 use anyhow::Result;
 use clap::Parser;
 use five_spot::constants::{
-    CAPI_GROUP, CAPI_MACHINE_API_VERSION, CAPI_RESOURCE_MACHINES, CHILD_NODE_EVENT_CHANNEL_CAP,
-    DEFAULT_LEASE_DURATION_SECS, DEFAULT_LEASE_GRACE_SECS, DEFAULT_LEASE_NAME,
-    DEFAULT_LEASE_NAMESPACE, DEFAULT_LEASE_RENEW_DEADLINE_SECS, DEFAULT_LEASE_RETRY_PERIOD_SECS,
-    HEALTH_PORT, K8S_API_TIMEOUT_SECS, METRICS_PORT, SPOT_SCHEDULE_EVENT_CHANNEL_CAP,
+    capi_machine_api_version, set_capi_machine_api_version, CAPI_GROUP, CAPI_RESOURCE_MACHINES,
+    CHILD_NODE_EVENT_CHANNEL_CAP, DEFAULT_LEASE_DURATION_SECS, DEFAULT_LEASE_GRACE_SECS,
+    DEFAULT_LEASE_NAME, DEFAULT_LEASE_NAMESPACE, DEFAULT_LEASE_RENEW_DEADLINE_SECS,
+    DEFAULT_LEASE_RETRY_PERIOD_SECS, HEALTH_PORT, K8S_API_TIMEOUT_SECS, METRICS_PORT,
+    SPOT_SCHEDULE_EVENT_CHANNEL_CAP,
 };
 use five_spot::crd::ScheduledMachine;
 use five_spot::health::{start_health_server, HealthState};
@@ -179,6 +180,15 @@ async fn main() -> Result<()> {
     // Mark Kubernetes as connected
     health_state.set_k8s_connected(true);
 
+    // Resolve the CAPI Machine API version the cluster actually serves (v1beta1 on
+    // CAPI ≤ v1.10, v1beta2 on v1.11+ / modern k0smotron), so we create and watch
+    // Machines at the version downstream controllers expect. Env override wins;
+    // discovery otherwise. MANDATORY — if it can't be resolved the controller errors
+    // out (and is restarted by Kubernetes) rather than guessing a version.
+    let capi_machine_version = resolve_capi_machine_version(&client).await?;
+    info!(version = %capi_machine_version, "Using CAPI Machine API version");
+    set_capi_machine_api_version(capi_machine_version);
+
     // Create shared context
     let context = Arc::new(
         Context::new(client.clone(), cli.instance_id, cli.instance_count)
@@ -296,7 +306,7 @@ async fn main() -> Result<()> {
     //    `SM.status.nodeRef.name` closes the watch-amplification vector
     //    documented in Phase 3 of the 2026-04-25 security audit roadmap.
     let machine_ar = ApiResource::from_gvk_with_plural(
-        &GroupVersionKind::gvk(CAPI_GROUP, CAPI_MACHINE_API_VERSION, "Machine"),
+        &GroupVersionKind::gvk(CAPI_GROUP, capi_machine_api_version(), "Machine"),
         CAPI_RESOURCE_MACHINES,
     );
     let machines_api: Api<DynamicObject> = Api::all_with(client.clone(), &machine_ar);
@@ -422,6 +432,57 @@ async fn main() -> Result<()> {
     Ok(())
 }
 
+/// Resolve which `cluster.x-k8s.io` Machine API version this cluster serves.
+///
+/// Precedence: the `CAPI_MACHINE_API_VERSION` env var (explicit operator pin) →
+/// API discovery of the group's recommended/served version. Discovery is
+/// **mandatory** — there is no compile-time default. If the version can't be
+/// determined (Cluster API not installed / not yet ready) this returns an error and
+/// the controller exits, to be restarted by Kubernetes — we never guess a version.
+/// One binary works against CAPI lines serving `v1beta1` (≤ v1.10) or `v1beta2`
+/// (v1.11+ / modern k0smotron) with no config.
+async fn resolve_capi_machine_version(client: &Client) -> Result<String> {
+    if let Ok(pinned) = std::env::var("CAPI_MACHINE_API_VERSION") {
+        let pinned = pinned.trim();
+        if !pinned.is_empty() {
+            info!(version = %pinned, "CAPI Machine API version pinned via env");
+            return Ok(pinned.to_string());
+        }
+    }
+    // Retry briefly so a controller that starts moments before CAPI finishes
+    // installing converges instead of crash-looping on the first probe.
+    const ATTEMPTS: u32 = 6;
+    let mut last_reason = String::from("discovery not attempted");
+    for attempt in 1..=ATTEMPTS {
+        match kube::discovery::group(client, CAPI_GROUP).await {
+            Ok(apigroup) => match apigroup.recommended_kind("Machine") {
+                Some((ar, _caps)) => {
+                    info!(version = %ar.version, "Discovered CAPI Machine API version");
+                    return Ok(ar.version);
+                }
+                None => {
+                    last_reason =
+                        format!("group '{CAPI_GROUP}' is served but exposes no Machine kind");
+                }
+            },
+            Err(error) => {
+                last_reason = format!("discovery of '{CAPI_GROUP}' failed: {error}");
+            }
+        }
+        warn!(
+            attempt,
+            max = ATTEMPTS,
+            reason = %last_reason,
+            "CAPI Machine API version not resolved yet; retrying"
+        );
+        tokio::time::sleep(std::time::Duration::from_secs(5)).await;
+    }
+    Err(anyhow::anyhow!(
+        "could not resolve the CAPI Machine API version after {ATTEMPTS} attempts ({last_reason}); \
+         is Cluster API installed? set the CAPI_MACHINE_API_VERSION env var to override discovery"
+    ))
+}
+
 /// Check if the `ScheduledMachine` CRD is installed in the cluster
 async fn check_crd_installed(api: &Api<ScheduledMachine>) -> Result<()> {
     // Try to list resources with limit 0 - this will fail if CRD doesn't exist

src/reconcilers/helpers.rs

@@ -42,16 +42,15 @@ use tracing::{debug, error, info, warn};
 
 use super::{Context, ReconcilerError};
 use crate::constants::{
-    ALLOWED_BOOTSTRAP_API_GROUPS, ALLOWED_INFRASTRUCTURE_API_GROUPS, API_VERSION_FULL,
-    CAPI_CLUSTER_NAME_LABEL, CAPI_GROUP, CAPI_MACHINE_API_VERSION, CAPI_MACHINE_API_VERSION_FULL,
-    CAPI_RESOURCE_MACHINES, CONDITION_STATUS_TRUE, CONDITION_TYPE_READY, DEFAULT_INSTANCE_ID,
-    ENV_OPERATOR_INSTANCE_ID, ERROR_REQUEUE_SECS, FINALIZER_CLEANUP_TIMEOUT_SECS,
-    FINALIZER_SCHEDULED_MACHINE, MAX_BACKOFF_SECS, MAX_CLUSTER_NAME_LEN, MAX_DURATION_SECS,
-    MAX_KILL_IF_COMMANDS_COUNT, MAX_KILL_IF_COMMAND_LEN, MAX_RECONCILE_RETRIES, PHASE_ACTIVE,
-    PHASE_ERROR, PHASE_INACTIVE, PHASE_SHUTTING_DOWN, PHASE_TERMINATED,
-    POD_EVICTION_GRACE_PERIOD_SECS, RBAC_VERB_CREATE, REASON_GRACE_PERIOD, REASON_KILL_SWITCH,
-    REASON_RECONCILE_SUCCEEDED, RESERVED_LABEL_PREFIXES, SCHEDULED_MACHINE_LABEL,
-    TIMER_REQUEUE_SECS,
+    capi_machine_api_version, ALLOWED_BOOTSTRAP_API_GROUPS, ALLOWED_INFRASTRUCTURE_API_GROUPS,
+    API_VERSION_FULL, CAPI_CLUSTER_NAME_LABEL, CAPI_GROUP, CAPI_RESOURCE_MACHINES,
+    CONDITION_STATUS_TRUE, CONDITION_TYPE_READY, DEFAULT_INSTANCE_ID, ENV_OPERATOR_INSTANCE_ID,
+    ERROR_REQUEUE_SECS, FINALIZER_CLEANUP_TIMEOUT_SECS, FINALIZER_SCHEDULED_MACHINE,
+    MAX_BACKOFF_SECS, MAX_CLUSTER_NAME_LEN, MAX_DURATION_SECS, MAX_KILL_IF_COMMANDS_COUNT,
+    MAX_KILL_IF_COMMAND_LEN, MAX_RECONCILE_RETRIES, PHASE_ACTIVE, PHASE_ERROR, PHASE_INACTIVE,
+    PHASE_SHUTTING_DOWN, PHASE_TERMINATED, POD_EVICTION_GRACE_PERIOD_SECS, RBAC_VERB_CREATE,
+    REASON_GRACE_PERIOD, REASON_KILL_SWITCH, REASON_RECONCILE_SUCCEEDED, RESERVED_LABEL_PREFIXES,
+    SCHEDULED_MACHINE_LABEL, TIMER_REQUEUE_SECS,
 };
 use crate::crd::{Condition, NodeRef, ScheduledMachine, ScheduledMachineStatus};
 use crate::metrics::{
@@ -1113,6 +1112,23 @@ pub fn validate_schedule_ref(
 // CAPI Resource Creation
 // ============================================================================
 
+/// Derive the CAPI `Machine` API version (`v1beta1` / `v1beta2`) for a
+/// `ScheduledMachine` from the version component of its bootstrapSpec `apiVersion`
+/// (passthrough — the Machine shares the one CAPI contract version the user
+/// declared for bootstrap + infra). Falls back to the runtime-resolved *served*
+/// version ([`capi_machine_api_version`]) when bootstrap carries no `/<version>`
+/// segment. The Machine group/kind (`cluster.x-k8s.io`/`Machine`) are fixed by the
+/// CAPI contract; only the version is data-driven, never hardcoded for production.
+fn machine_api_version_for(bootstrap_api_version: Option<&str>) -> String {
+    bootstrap_api_version
+        .and_then(|bv| bv.rsplit_once('/').map(|(_, ver)| ver))
+        .filter(|ver| !ver.is_empty())
+        .map_or_else(
+            || capi_machine_api_version().to_string(),
+            |ver| ver.to_string(),
+        )
+}
+
 /// Add machine to cluster by creating bootstrap, infrastructure, and Machine resources
 ///
 /// This function:
@@ -1166,6 +1182,12 @@ pub async fn add_machine_to_cluster(
             "bootstrapSpec missing required field 'apiVersion'".to_string(),
         )
     })?;
+    // The created CAPI Machine shares the contract version the user declared on
+    // bootstrapSpec (passthrough); see machine_api_version_for.
+    let machine_api_version_full = format!(
+        "{CAPI_GROUP}/{}",
+        machine_api_version_for(Some(bootstrap_api_version))
+    );
     let bootstrap_kind = resource.spec.bootstrap_spec.kind().ok_or_else(|| {
         ReconcilerError::InvalidConfig("bootstrapSpec missing required field 'kind'".to_string())
     })?;
@@ -1245,7 +1267,7 @@ pub async fn add_machine_to_cluster(
     ensure_can_create(
         client,
         namespace,
-        CAPI_MACHINE_API_VERSION_FULL,
+        &machine_api_version_full,
         "Machine",
         "machine",
     )
@@ -1370,7 +1392,7 @@ pub async fn add_machine_to_cluster(
     }
 
     let machine_obj = json!({
-        "apiVersion": CAPI_MACHINE_API_VERSION_FULL,
+        "apiVersion": machine_api_version_full.as_str(),
         "kind": "Machine",
         "metadata": {
             "name": name,
@@ -1401,7 +1423,7 @@ pub async fn add_machine_to_cluster(
     create_dynamic_resource(
         client,
         namespace,
-        CAPI_MACHINE_API_VERSION_FULL,
+        &machine_api_version_full,
         "Machine",
         machine_obj,
     )
@@ -1625,9 +1647,13 @@ pub async fn remove_machine_from_cluster(
         "Deleting CAPI resources"
     );
 
-    // 1. Delete the Machine resource first (this triggers CAPI cleanup)
+    // 1. Delete the Machine resource first (this triggers CAPI cleanup). Use the
+    //    CAPI contract version the user declared on bootstrapSpec (passthrough,
+    //    matching create); fall back to the runtime-discovered served version when
+    //    it has none. CAPI conversion bridges either way.
+    let machine_api_version = machine_api_version_for(resource.spec.bootstrap_spec.api_version());
     let ar = kube::api::ApiResource::from_gvk_with_plural(
-        &kube::api::GroupVersionKind::gvk(CAPI_GROUP, CAPI_MACHINE_API_VERSION, "Machine"),
+        &kube::api::GroupVersionKind::gvk(CAPI_GROUP, &machine_api_version, "Machine"),
         CAPI_RESOURCE_MACHINES,
     );
     let machines: Api<kube::core::DynamicObject> =
@@ -1696,9 +1722,21 @@ pub fn extract_machine_refs(
         .get("status")
         .and_then(|s| s.get("nodeRef"))
         .and_then(|nr| {
-            let api_version = nr.get("apiVersion").and_then(serde_json::Value::as_str)?;
-            let kind = nr.get("kind").and_then(serde_json::Value::as_str)?;
+            // `name` is the only field guaranteed across CAPI versions: v1beta1's
+            // nodeRef is a corev1 ObjectReference (apiVersion/kind/uid present),
+            // but v1beta2 simplified it to a `MachineNodeReference` carrying just
+            // `name` (it is always a core/v1 Node). Require only `name`; default
+            // the rest so v1beta2 nodeRef enrichment (node taints, reclaim-agent)
+            // still fires.
             let name = nr.get("name").and_then(serde_json::Value::as_str)?;
+            let api_version = nr
+                .get("apiVersion")
+                .and_then(serde_json::Value::as_str)
+                .unwrap_or("v1");
+            let kind = nr
+                .get("kind")
+                .and_then(serde_json::Value::as_str)
+                .unwrap_or("Node");
             let uid = nr
                 .get("uid")
                 .and_then(serde_json::Value::as_str)
@@ -1727,7 +1765,7 @@ pub async fn fetch_capi_machine(
     machine_name: &str,
 ) -> Result<Option<kube::core::DynamicObject>, ReconcilerError> {
     let ar = kube::api::ApiResource::from_gvk_with_plural(
-        &kube::api::GroupVersionKind::gvk(CAPI_GROUP, CAPI_MACHINE_API_VERSION, "Machine"),
+        &kube::api::GroupVersionKind::gvk(CAPI_GROUP, capi_machine_api_version(), "Machine"),
         CAPI_RESOURCE_MACHINES,
     );
     let machines: Api<kube::core::DynamicObject> =