Integration tests: check reboot with boot ID, test Ingress resolvability, wait for image committed

test_utils/src/lib.rs

@@ -10,6 +10,7 @@ use k8s_openapi::api::core::v1::{
     ConfigMap, LoadBalancerStatus, Namespace, Secret, Service, ServicePort, ServiceSpec,
     ServiceStatus,
 };
+use k8s_openapi::apimachinery::pkg::apis::meta::v1::Condition;
 use kube::api::{DeleteParams, ObjectMeta, Patch};
 use kube::runtime::wait::await_condition;
 use kube::{Api, Client};
@@ -21,9 +22,10 @@ use tokio::time::timeout;
 use trusted_cluster_operator_lib::certificates::{
     Certificate, CertificateIssuerRef, CertificateSpec, CertificateStatus,
 };
+use trusted_cluster_operator_lib::conditions::COMMITTED_CONDITION;
 use trusted_cluster_operator_lib::issuers::{Issuer, IssuerCa, IssuerSpec};
 
-use trusted_cluster_operator_lib::{ApprovedImage, AttestationKey, Machine};
+use trusted_cluster_operator_lib::{ApprovedImage, ApprovedImageStatus, AttestationKey, Machine};
 use trusted_cluster_operator_lib::{TrustedExecutionCluster, endpoints::*, images::*};
 
 pub mod timer;
@@ -36,6 +38,7 @@ pub mod virt;
 use compute_pcrs_lib::Pcr;
 
 const TEST_TIMEOUT_MULTIPLIER_ENV: &str = "TEST_TIMEOUT_MULTIPLIER";
+const EXPOSE_MAX_ATTEMPTS: u32 = 3;
 
 const PLATFORM_ENV: &str = "PLATFORM";
 const CLUSTER_URL_ENV: &str = "CLUSTER_URL";
@@ -54,6 +57,7 @@ const ATT_REG_SECRET: &str = "att-reg-secret";
 const REG_CERT: &str = "reg-srv-cert";
 const TRUSTEE_CERT: &str = "trustee-cert";
 const ATT_REG_CERT: &str = "att-reg-cert";
+pub const APPROVED_IMAGE_NAME: &str = "coreos";
 
 pub fn compare_pcrs(actual: &[Pcr], expected: &[Pcr]) -> bool {
     if actual.len() != expected.len() {
@@ -306,27 +310,64 @@ impl K8sPlatform for OpenShift {
     ) -> Result<()> {
         let services: Api<Service> = Api::namespaced(self.client.clone(), &self.namespace);
         let pp = Default::default();
-        let json = json!({
-            "spec": {
-                "type": "LoadBalancer"
+        let duration = scaled_duration(120);
+        let lb = json!({ "spec": { "type": "LoadBalancer" } });
+        let clusterip = Patch::Merge(json!({ "spec": { "type": "ClusterIP" } }));
+
+        let mut url = OpenShiftHost::None;
+        let mut last_err: Option<anyhow::Error> = None;
+        for attempt in 1..=EXPOSE_MAX_ATTEMPTS {
+            services.patch(service, &pp, &Patch::Merge(&lb)).await?;
+
+            let has_ingress = |svc: Option<&Service>| {
+                let chk_lb = |bal: &LoadBalancerStatus| bal.ingress.is_some();
+                let chk_st = |st: &ServiceStatus| st.load_balancer.as_ref().map(chk_lb);
+                let chk_svc = |svc: &Service| svc.status.as_ref().and_then(chk_st);
+                svc.and_then(chk_svc).unwrap_or(false)
+            };
+            let ctx = format!(
+                "waiting for ingress on {service} (attempt {attempt}/{EXPOSE_MAX_ATTEMPTS})"
+            );
+            let ingress_ready = await_condition(services.clone(), service, has_ingress);
+            match timeout(duration, ingress_ready).await {
+                Err(_) => {
+                    last_err = Some(anyhow!(ctx));
+                    services.patch(service, &pp, &clusterip).await?;
+                    continue;
+                }
+                Ok(Err(e)) => {
+                    last_err = Some(anyhow::Error::from(e).context(ctx));
+                    services.patch(service, &pp, &clusterip).await?;
+                    continue;
+                }
+                Ok(Ok(_)) => {}
+            }
+
+            url = self.get_url(service).await;
+            if let OpenShiftHost::Hostname(ref name) = url {
+                let target = format!("{name}:0");
+                let msg =
+                    format!("{name} not DNS-resolvable (attempt {attempt}/{EXPOSE_MAX_ATTEMPTS})");
+                let chk = || async { tokio::net::lookup_host(&target).await.map(|_| ()) };
+                let poller = Poller::new().with_timeout(duration);
+                if let Err(e) = poller.with_error_message(msg).poll_async(chk).await {
+                    last_err = Some(e);
+                    services.patch(service, &pp, &clusterip).await?;
+                    continue;
+                }
             }
-        });
-        services.patch(service, &pp, &Patch::Merge(&json)).await?;
-        let has_ingress = |svc: Option<&Service>| {
-            let chk_lb = |bal: &LoadBalancerStatus| bal.ingress.is_some();
-            let chk_st = |st: &ServiceStatus| st.load_balancer.as_ref().map(chk_lb);
-            let chk_svc = |svc: &Service| svc.status.as_ref().and_then(chk_st);
-            svc.and_then(chk_svc).unwrap_or(false)
-        };
-        let ingress_ready = await_condition(services, service, has_ingress);
-        let ctx = format!("waiting for ingress on {service} to be ready");
-        let duration = scaled_duration(60);
-        timeout(duration, ingress_ready).await.context(ctx)??;
+
+            last_err = None;
+            break;
+        }
+        if let Some(e) = last_err {
+            return Err(e);
+        }
 
         let certs: Api<Certificate> = Api::namespaced(self.client.clone(), &self.namespace);
         let cert = certs.get(cert_name).await?;
         let old_revision = cert.status.and_then(|st| st.revision).unwrap_or(0);
-        let cert_patch = match self.get_url(service).await {
+        let cert_patch = match url {
             OpenShiftHost::Ip(ip) => json!({
                 "spec": {
                     "ipAddresses": [ip],
@@ -983,7 +1024,22 @@ impl TestContext {
             "Waiting for image-pcrs ConfigMap to be created"
         );
         let configmap_api: Api<ConfigMap> = Api::namespaced(self.client.clone(), ns);
-        wait_for_resource_created(&configmap_api, "image-pcrs", scaled_timeout(60)).await
+        wait_for_resource_created(&configmap_api, "image-pcrs", scaled_timeout(60)).await?;
+
+        let info = format!("Waiting for ApprovedImage {APPROVED_IMAGE_NAME} to be Committed");
+        test_info!(&self.test_name, "{info}");
+        let images: Api<ApprovedImage> = Api::namespaced(self.client.clone(), ns);
+        let image_ready = |img: Option<&ApprovedImage>| {
+            let chk_cond = |c: &Condition| c.type_ == COMMITTED_CONDITION && c.status == "True";
+            let chk_status =
+                |st: &ApprovedImageStatus| st.conditions.as_ref().map(|cs| cs.iter().any(chk_cond));
+            let chk = |img: &ApprovedImage| img.status.as_ref().and_then(chk_status);
+            img.and_then(chk).unwrap_or(false)
+        };
+        let done = await_condition(images.clone(), APPROVED_IMAGE_NAME, image_ready);
+        let ctx = format!("waiting for ApprovedImage {APPROVED_IMAGE_NAME} to be Committed");
+        timeout(scaled_duration(300), done).await.context(ctx)??;
+        Ok(())
     }
 }
 

test_utils/src/virt/mod.rs

@@ -185,30 +185,31 @@ pub trait VmBackend: Send + Sync {
     async fn get_root_key(&self) -> Result<Option<Vec<u8>>>;
     async fn cleanup(&self) -> Result<()>;
 
-    async fn wait_for_vm_ssh_ready(&self, timeout_secs: u64) -> Result<()> {
-        self.wait_for_vm_ssh(timeout_secs, true).await
+    async fn get_boot_id(&self) -> Result<String> {
+        let id = self.ssh_exec("cat /proc/sys/kernel/random/boot_id").await?;
+        Ok(id.trim().to_string())
     }
 
-    async fn wait_for_vm_ssh_unavail(&self, timeout_secs: u64) -> Result<()> {
-        self.wait_for_vm_ssh(timeout_secs, false).await
-    }
-
-    async fn wait_for_vm_ssh(&self, timeout_secs: u64, await_start: bool) -> Result<()> {
-        let avail_prefix = if await_start { "" } else { "un" };
+    async fn wait_for_vm_ssh_ready(
+        &self,
+        timeout_secs: u64,
+        prev_boot_id: Option<&str>,
+    ) -> Result<()> {
+        let err = format!("SSH access to VM did not become available after {timeout_secs} seconds");
         let poller = Poller::new()
             .with_timeout(Duration::from_secs(timeout_secs))
             .with_interval(Duration::from_secs(10))
-            .with_error_message(format!(
-                "SSH access to VM did not become {avail_prefix}available after {timeout_secs} seconds"
-            ));
-
-        let check_fn = || {
-            async move {
-                // Try a simple command to check if SSH is ready
-                let result = self.ssh_exec("echo ready").await;
-                let err = anyhow!("SSH not desired state yet: {result:?}");
-                (result.is_err() ^ await_start).then_some(()).ok_or(err)
+            .with_error_message(err);
+
+        let check_fn = || async move {
+            let cat = self.ssh_exec("cat /proc/sys/kernel/random/boot_id").await;
+            let boot_id = cat.map_err(|e| anyhow!("SSH not available yet: {e}"))?;
+            if let Some(prev) = prev_boot_id
+                && boot_id.trim() == prev
+            {
+                return Err(anyhow!("VM has not rebooted yet (boot ID unchanged)"));
             }
+            Ok(())
         };
         poller.poll_async(check_fn).await
     }

tests/attestation.rs

@@ -49,7 +49,7 @@ impl SingleAttestationContext {
         test_ctx.info(format!("VM {vm_name} is Running"));
 
         test_ctx.info(format!("Waiting for SSH access to VM {vm_name}"));
-        backend.wait_for_vm_ssh_ready(scaled_timeout(600)).await?;
+        backend.wait_for_vm_ssh_ready(scaled_timeout(600), None).await?;
         test_ctx.info("SSH access is ready");
 
         let root_key = backend.get_root_key().await?;
@@ -113,8 +113,8 @@ async fn test_parallel_vm_attestation() -> anyhow::Result<()> {
     // Wait for SSH access on both VMs in parallel
     test_ctx.info("Waiting for SSH access on both VMs");
     let (ssh1_ready, ssh2_ready) = tokio::join!(
-        backend1.wait_for_vm_ssh_ready(scaled_timeout(900)),
-        backend2.wait_for_vm_ssh_ready(scaled_timeout(900))
+        backend1.wait_for_vm_ssh_ready(scaled_timeout(900), None),
+        backend2.wait_for_vm_ssh_ready(scaled_timeout(900), None)
     );
     ssh1_ready?;
     ssh2_ready?;
@@ -165,14 +165,11 @@ async fn test_vm_reboot_attestation() -> anyhow::Result<()> {
     for i in 1..=num_reboots {
         test_ctx.info(format!("Performing reboot {i} of {num_reboots}"));
 
-        // Reboot the VM via SSH
+        let boot_id = att_ctx.backend.get_boot_id().await?;
         let _reboot_result = att_ctx.backend.ssh_exec("sudo systemctl reboot").await;
 
-        test_ctx.info(format!("Waiting for lack of SSH access after reboot {i}"));
-        att_ctx.backend.wait_for_vm_ssh_unavail(scaled_timeout(30)).await?;
-
         test_ctx.info(format!("Waiting for SSH access after reboot {i}"));
-        att_ctx.backend.wait_for_vm_ssh_ready(scaled_timeout(300)).await?;
+        att_ctx.backend.wait_for_vm_ssh_ready(scaled_timeout(300), Some(&boot_id)).await?;
 
         // Verify encrypted root is still present after reboot
         test_ctx.info(format!("Verifying encrypted root after reboot {i}"));
@@ -207,13 +204,11 @@ async fn test_vm_reboot_delete_machine() -> anyhow::Result<()> {
     wait_for_resource_deleted(&machines, name, scaled_timeout(120)).await?;
 
     test_ctx.info("Performing reboot, expecting missing resource");
+    let boot_id = att_ctx.backend.get_boot_id().await?;
     let _reboot_result = att_ctx.backend.ssh_exec("sudo systemctl reboot").await;
 
-    test_ctx.info("Waiting for lack of SSH access after reboot");
-    att_ctx.backend.wait_for_vm_ssh_unavail(scaled_timeout(30)).await?;
-
     test_ctx.info("Waiting for SSH access after machine removal");
-    let wait = att_ctx.backend.wait_for_vm_ssh_ready(scaled_timeout(300)).await;
+    let wait = att_ctx.backend.wait_for_vm_ssh_ready(scaled_timeout(300), Some(&boot_id)).await;
     assert!(wait.is_err());
 
     att_ctx.cleanup().await?;
@@ -233,13 +228,11 @@ async fn test_vm_restart_operator_existing() -> anyhow::Result<()> {
         Api::namespaced(test_ctx.client().clone(), test_ctx.namespace());
     deployments.restart("trusted-cluster-operator").await?;
 
+    let boot_id = att_ctx.backend.get_boot_id().await?;
     let _reboot_result = att_ctx.backend.ssh_exec("sudo systemctl reboot").await;
 
-    test_ctx.info("Waiting for lack of SSH access after reboot");
-    att_ctx.backend.wait_for_vm_ssh_unavail(scaled_timeout(30)).await?;
-
     test_ctx.info("Waiting for SSH access after operator restart & reboot");
-    let wait = att_ctx.backend.wait_for_vm_ssh_ready(scaled_timeout(300)).await;
+    let wait = att_ctx.backend.wait_for_vm_ssh_ready(scaled_timeout(300), Some(&boot_id)).await;
     assert!(wait.is_ok());
 
     att_ctx.cleanup().await?;

tests/trusted_execution_cluster.rs

@@ -23,7 +23,6 @@ use trusted_cluster_operator_test_utils::*;
 
 const EXPECTED_PCR4: &str = "ff2b357be4a4bc66be796d4e7b2f1f27077dc89b96220aae60b443bcf4672525";
 const TEC_NAME: &str = "trusted-execution-cluster";
-const APPROVED_IMAGE_NAME: &str = "coreos";
 const TRUSTEE_CONFIG_MAP: &str = "trustee-data";
 const RV_JSON_KEY: &str = "reference-values.json";