Separate concerns for compute-pcrs

compute-pcrs/Cargo.toml

@@ -15,7 +15,9 @@ anyhow.workspace = true
 clap.workspace = true
 trusted-cluster-operator-lib = { path = "../lib" }
 compute-pcrs-lib.workspace = true
+env_logger.workspace = true
 k8s-openapi.workspace = true
 kube.workspace = true
+log.workspace = true
 serde_json.workspace = true
 tokio.workspace = true

compute-pcrs/src/main.rs

@@ -6,11 +6,13 @@
 use anyhow::{Context, Result, anyhow};
 use clap::Parser;
 use compute_pcrs_lib::*;
+use env_logger::Env;
 use k8s_openapi::{api::core::v1::ConfigMap, jiff::Timestamp};
 use kube::{Api, Client};
+use log::info;
 use std::{fs::File, io::Read};
 
-use trusted_cluster_operator_lib::{conditions::INSTALLED_REASON, reference_values::*, *};
+use trusted_cluster_operator_lib::reference_values::*;
 
 #[derive(Parser)]
 #[command(version, about)]
@@ -25,6 +27,7 @@ struct Args {
 
 #[tokio::main]
 async fn main() -> Result<()> {
+    env_logger::Builder::from_env(Env::default().default_filter_or("info")).init();
     let args = Args::parse();
 
     let kernels = format!("{IMAGE_VOLUME_MOUNTPOINT}/usr/lib/modules");
@@ -53,30 +56,38 @@ async fn main() -> Result<()> {
     ];
 
     let client = Client::try_default().await?;
-    let config_maps: Api<ConfigMap> = Api::default_namespaced(client.clone());
-
-    let mut image_pcrs_map = config_maps.get(PCR_CONFIG_MAP).await?;
-    let image_pcrs_data = image_pcrs_map
-        .data
-        .context("Image PCRs map existed, but had no data")?;
-    let image_pcrs_str = image_pcrs_data
-        .get(PCR_CONFIG_FILE)
-        .context("Image PCRs data existed, but had no file")?;
-    let mut image_pcrs: ImagePcrs = serde_json::from_str(image_pcrs_str)?;
+    let config_maps: Api<ConfigMap> = Api::default_namespaced(client);
 
     let image_pcr = ImagePcr {
         first_seen: Timestamp::now(),
         reference: args.image,
         pcrs,
     };
-    image_pcrs.0.insert(args.resource_name.clone(), image_pcr);
-    update_image_pcrs!(config_maps, image_pcrs_map, image_pcrs);
-
-    let approved_images: Api<ApprovedImage> = Api::default_namespaced(client);
-    let image = approved_images.get(&args.resource_name).await?;
-    let committed = committed_condition(INSTALLED_REASON, image.metadata.generation, &None);
-    let conditions = Some(vec![committed]);
-    let status = ApprovedImageStatus { conditions };
-    update_status!(approved_images, &args.resource_name, status)?;
+    // If we see this causing performance problems, consider NoSQL
+    loop {
+        let mut image_pcrs_map = config_maps.get(PCR_CONFIG_MAP).await?;
+        let ctx = "Image PCRs map existed, but had no data";
+        let image_pcrs_data = image_pcrs_map.data.context(ctx)?;
+        let ctx = "Image PCRs data existed, but had no file";
+        let image_pcrs_str = image_pcrs_data.get(PCR_CONFIG_FILE).context(ctx)?;
+        let mut image_pcrs: ImagePcrs = serde_json::from_str(image_pcrs_str)?;
+        let resource_name = args.resource_name.clone();
+        image_pcrs.0.insert(resource_name, image_pcr.clone());
+        let image_pcrs_json = serde_json::to_string(&image_pcrs)?;
+        let map = (PCR_CONFIG_FILE.to_string(), image_pcrs_json);
+        let data = std::collections::BTreeMap::from([map]);
+        image_pcrs_map.data = Some(data);
+        match config_maps
+            .replace(PCR_CONFIG_MAP, &Default::default(), &image_pcrs_map)
+            .await
+        {
+            Ok(_) => break,
+            Err(kube::Error::Api(ae)) if ae.code == 409 => {
+                info!("ConfigMap update conflict, retrying");
+                continue;
+            }
+            Err(e) => return Err(e.into()),
+        }
+    }
     Ok(())
 }

lib/src/lib.rs

@@ -10,7 +10,6 @@ pub mod reference_values;
 mod kopium;
 #[allow(clippy::all)]
 mod vendor_kopium;
-use k8s_openapi::jiff::Timestamp;
 pub use kopium::approvedimages::*;
 pub use kopium::attestationkeys::*;
 pub use kopium::ingresses as openshift_ingresses;
@@ -26,88 +25,9 @@ pub use vendor_kopium::virtualmachineinstances;
 pub use vendor_kopium::virtualmachines;
 
 use anyhow::{Context, Result, anyhow};
-use conditions::*;
-use k8s_openapi::apimachinery::pkg::apis::meta::v1::{Condition, OwnerReference, Time};
+use k8s_openapi::apimachinery::pkg::apis::meta::v1::OwnerReference;
 use kube::{Api, Client, Resource};
 
-#[macro_export]
-macro_rules! update_status {
-    ($api:ident, $name:expr, $status:expr) => {{
-        let patch = kube::api::Patch::Merge(serde_json::json!({"status": $status}));
-        $api.patch_status($name, &Default::default(), &patch).await
-            .map_err(Into::<anyhow::Error>::into)
-    }}
-}
-
-pub fn condition_status(status: bool) -> String {
-    match status {
-        true => "True".to_string(),
-        false => "False".to_string(),
-    }
-}
-
-pub trait Conditions {
-    fn conditions(&self) -> &Option<Vec<Condition>>;
-}
-
-impl Conditions for TrustedExecutionClusterStatus {
-    fn conditions(&self) -> &Option<Vec<Condition>> {
-        &self.conditions
-    }
-}
-
-impl Conditions for AttestationKeyStatus {
-    fn conditions(&self) -> &Option<Vec<Condition>> {
-        &self.conditions
-    }
-}
-
-impl Conditions for ApprovedImageStatus {
-    fn conditions(&self) -> &Option<Vec<Condition>> {
-        &self.conditions
-    }
-}
-
-pub fn transition_time<S: Conditions>(
-    existing_status: &Option<S>,
-    type_: &str,
-    new_status: &str,
-) -> Time {
-    let get = |s: &S| s.conditions().clone();
-    let conditions = existing_status.as_ref().and_then(get);
-    let find = |c: &Condition| type_ == c.type_ && new_status == c.status;
-    let existing = conditions.and_then(|cs| cs.into_iter().find(find));
-    let time = existing.map(|c| c.last_transition_time);
-    time.unwrap_or(Time(Timestamp::now()))
-}
-
-pub fn committed_condition(
-    reason: &str,
-    generation: Option<i64>,
-    existing_status: &Option<ApprovedImageStatus>,
-) -> Condition {
-    let status = condition_status(reason == COMMITTED_REASON);
-    let type_ = COMMITTED_CONDITION;
-    Condition {
-        type_: type_.to_string(),
-        reason: reason.to_string(),
-        message: match reason {
-            NOT_COMMITTED_REASON_COMPUTING => "Computation is ongoing. Check jobs for progress.",
-            NOT_COMMITTED_REASON_NO_DIGEST => {
-                "Image did not specify a digest. \
-                 Only images with a digest are supported to avoid ambiguity."
-            }
-            NOT_COMMITTED_REASON_PENDING => "Pod is pending, check pods for details",
-            NOT_COMMITTED_REASON_FAILED => "Computation failed, check operator log for details",
-            _ => "",
-        }
-        .to_string(),
-        last_transition_time: transition_time(existing_status, type_, &status),
-        status,
-        observed_generation: generation,
-    }
-}
-
 /// Generate an OwnerReference for any Kubernetes resource
 pub fn generate_owner_reference<T: Resource<DynamicType = ()>>(
     object: &T,

lib/src/reference_values.rs

@@ -12,7 +12,7 @@ pub const PCR_CONFIG_MAP: &str = "image-pcrs";
 pub const PCR_CONFIG_FILE: &str = "image-pcrs.json";
 pub const IMAGE_VOLUME_MOUNTPOINT: &str = "/image";
 
-#[derive(Deserialize, Serialize)]
+#[derive(Clone, Deserialize, Serialize)]
 pub struct ImagePcr {
     pub first_seen: Timestamp,
     pub pcrs: Vec<Pcr>,
@@ -21,15 +21,3 @@ pub struct ImagePcr {
 
 #[derive(Default, Deserialize, Serialize)]
 pub struct ImagePcrs(pub BTreeMap<String, ImagePcr>);
-
-#[macro_export]
-macro_rules! update_image_pcrs {
-    ($api:ident, $map:ident, $pcrs:ident) => {
-        let image_pcrs_json = serde_json::to_string(&$pcrs)?;
-        let map = (PCR_CONFIG_FILE.to_string(), image_pcrs_json.to_string());
-        let data = std::collections::BTreeMap::from([map]);
-        $map.data = Some(data);
-        $api.replace(PCR_CONFIG_MAP, &Default::default(), &$map)
-            .await?
-    };
-}

operator/src/attestation_key_register.rs

@@ -31,14 +31,12 @@ use std::{collections::BTreeMap, sync::Arc, time::Duration};
 
 use trusted_cluster_operator_lib::conditions::ATTESTATION_KEY_MACHINE_APPROVE;
 use trusted_cluster_operator_lib::endpoints::*;
-use trusted_cluster_operator_lib::{AttestationKey, AttestationKeyStatus, Machine, update_status};
+use trusted_cluster_operator_lib::{AttestationKey, AttestationKeyStatus, Machine};
 
 use crate::conditions::attestation_key_approved_condition;
 use crate::trustee;
-use operator::{
-    ControllerError, TLS_DIR, controller_error_policy, create_or_info_if_exists, read_certificate,
-    upsert_condition,
-};
+use operator::{ControllerError, TLS_DIR, controller_error_policy};
+use operator::{create_or_info_if_exists, read_certificate, update_status, upsert_condition};
 
 /// Shared context for the three attestation-key controllers.
 /// Stores give local cache access to avoid repeated API-server reads.
@@ -243,7 +241,7 @@ async fn approve_ak(ak: &AttestationKey, machine: &Machine, ctx: &AkContextData)
 
     if changed {
         let status = AttestationKeyStatus { conditions };
-        update_status!(aks, &name, status)?;
+        update_status(&aks, &name, status).await?;
         info!("Approved attestation key {name}");
     }
 

operator/src/conditions.rs

@@ -3,8 +3,9 @@
 // SPDX-License-Identifier: MIT
 
 use k8s_openapi::apimachinery::pkg::apis::meta::v1::Condition;
+use operator::{condition_status, transition_time};
+use trusted_cluster_operator_lib::conditions::*;
 use trusted_cluster_operator_lib::{AttestationKeyStatus, TrustedExecutionClusterStatus};
-use trusted_cluster_operator_lib::{condition_status, conditions::*, transition_time};
 
 pub fn known_trustee_address_condition(
     known: bool,

operator/src/lib.rs

@@ -14,13 +14,17 @@ use k8s_openapi::api::core::v1::{Secret, SecretVolumeSource, Volume, VolumeMount
 use k8s_openapi::apimachinery::pkg::apis::meta::v1::{Condition, Time};
 use k8s_openapi::jiff::Timestamp;
 use kube::Resource;
+use kube::api::Patch;
 use kube::runtime::reflector::{self, Store};
 use kube::runtime::watcher::watcher;
 use kube::{Api, Client, runtime::controller::Action};
 use log::{info, warn};
+use serde::Serialize;
 use std::fmt::{Debug, Display};
 use std::{sync::Arc, time::Duration};
 use tokio::time::timeout;
+use trusted_cluster_operator_lib::{ApprovedImageStatus, AttestationKeyStatus};
+use trusted_cluster_operator_lib::{TrustedExecutionClusterStatus, conditions::*};
 
 // Re-export common functions from the lib
 pub use trusted_cluster_operator_lib::generate_owner_reference;
@@ -31,6 +35,84 @@ pub enum ControllerError {
     Anyhow(#[from] anyhow::Error),
 }
 
+pub async fn update_status<S: Serialize, K>(api: &Api<K>, name: &str, status: S) -> Result<()>
+where
+    K: Resource<DynamicType = ()> + serde::de::DeserializeOwned + Clone + Debug,
+{
+    let patch = Patch::Merge(serde_json::json!({"status": status}));
+    api.patch_status(name, &Default::default(), &patch).await?;
+    Ok(())
+}
+
+pub fn condition_status(status: bool) -> String {
+    match status {
+        true => "True".to_string(),
+        false => "False".to_string(),
+    }
+}
+
+pub trait Conditions {
+    fn conditions(&self) -> &Option<Vec<Condition>>;
+}
+
+impl Conditions for TrustedExecutionClusterStatus {
+    fn conditions(&self) -> &Option<Vec<Condition>> {
+        &self.conditions
+    }
+}
+
+impl Conditions for AttestationKeyStatus {
+    fn conditions(&self) -> &Option<Vec<Condition>> {
+        &self.conditions
+    }
+}
+
+impl Conditions for ApprovedImageStatus {
+    fn conditions(&self) -> &Option<Vec<Condition>> {
+        &self.conditions
+    }
+}
+
+pub fn transition_time<S: Conditions>(
+    existing_status: &Option<S>,
+    type_: &str,
+    new_status: &str,
+) -> Time {
+    let get = |s: &S| s.conditions().clone();
+    let conditions = existing_status.as_ref().and_then(get);
+    let find = |c: &Condition| type_ == c.type_ && new_status == c.status;
+    let existing = conditions.and_then(|cs| cs.into_iter().find(find));
+    let time = existing.map(|c| c.last_transition_time);
+    time.unwrap_or(Time(Timestamp::now()))
+}
+
+pub fn committed_condition(
+    reason: &str,
+    generation: Option<i64>,
+    existing_status: &Option<ApprovedImageStatus>,
+) -> Condition {
+    let status = condition_status(reason == COMMITTED_REASON);
+    let type_ = COMMITTED_CONDITION;
+    Condition {
+        type_: type_.to_string(),
+        reason: reason.to_string(),
+        message: match reason {
+            NOT_COMMITTED_REASON_COMPUTING => "Computation is ongoing. Check jobs for progress.",
+            NOT_COMMITTED_REASON_NO_DIGEST => {
+                "Image did not specify a digest. \
+                 Only images with a digest are supported to avoid ambiguity."
+            }
+            NOT_COMMITTED_REASON_PENDING => "Pod is pending, check pods for details",
+            NOT_COMMITTED_REASON_FAILED => "Computation failed, check operator log for details",
+            _ => "",
+        }
+        .to_string(),
+        last_transition_time: transition_time(existing_status, type_, &status),
+        status,
+        observed_generation: generation,
+    }
+}
+
 pub fn controller_error_policy<R, E: Display, C>(_obj: Arc<R>, error: &E, _ctx: Arc<C>) -> Action {
     log::error!("{error}");
     Action::requeue(Duration::from_secs(60))