From d77f3a4992b153286e77978885ab24c7953f0eee Mon Sep 17 00:00:00 2001 From: Jakob Naucke Date: Thu, 6 Aug 2026 17:14:27 +0200 Subject: [PATCH 01/10] Use reflector for all repeated read operations Extend the use of a reflector cache from attestation keys to the entire operator. Replace AkContextData with OperatorContext to pass caches. Use server-side apply patches instead of get+replace. - Updating image-pcrs without the cache is only used in compute-pcrs now, thus inline the macro - trustee::update_reference_values receives PCRs from argument, it is sometimes available from the call site anyhow Fixes: #251 Signed-off-by: Jakob Naucke Assisted-by: AI --- compute-pcrs/src/main.rs | 7 +- lib/src/lib.rs | 2 +- lib/src/reference_values.rs | 12 - operator/src/attestation_key_register.rs | 80 ++----- operator/src/lib.rs | 49 +++- operator/src/main.rs | 160 +++++++------ operator/src/reference_values.rs | 207 ++++++++++------- operator/src/register_server.rs | 38 ++-- operator/src/trustee.rs | 273 +++++++++++------------ 9 files changed, 413 insertions(+), 415 deletions(-) diff --git a/compute-pcrs/src/main.rs b/compute-pcrs/src/main.rs index 0241f37f..58dcc4ce 100644 --- a/compute-pcrs/src/main.rs +++ b/compute-pcrs/src/main.rs @@ -69,7 +69,12 @@ async fn main() -> Result<()> { pcrs, }; image_pcrs.0.insert(args.resource_name.clone(), image_pcr); - update_image_pcrs!(config_maps, image_pcrs_map, image_pcrs); + let image_pcrs_json = serde_json::to_string(&image_pcrs)?; + let data = std::collections::BTreeMap::from([(PCR_CONFIG_FILE.to_string(), image_pcrs_json)]); + image_pcrs_map.data = Some(data); + config_maps + .replace(PCR_CONFIG_MAP, &Default::default(), &image_pcrs_map) + .await?; let approved_images: Api = Api::default_namespaced(client); let image = approved_images.get(&args.resource_name).await?; diff --git a/lib/src/lib.rs b/lib/src/lib.rs index 03adb539..59b519ee 100644 --- a/lib/src/lib.rs +++ b/lib/src/lib.rs @@ -138,7 +138,7 @@ pub async fn get_opt_trusted_execution_cluster( Ok(list.items.into_iter().next()) } -/// Get the single TrustedExecutionCluster in the namespace +/// Get the single TrustedExecutionCluster in the namespace (uncached) pub async fn get_trusted_execution_cluster(client: Client) -> Result { let namespace = client.default_namespace().to_string(); let cluster = get_opt_trusted_execution_cluster(client).await; diff --git a/lib/src/reference_values.rs b/lib/src/reference_values.rs index f262d37d..ef0a37f0 100644 --- a/lib/src/reference_values.rs +++ b/lib/src/reference_values.rs @@ -21,15 +21,3 @@ pub struct ImagePcr { #[derive(Default, Deserialize, Serialize)] pub struct ImagePcrs(pub BTreeMap); - -#[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? - }; -} diff --git a/operator/src/attestation_key_register.rs b/operator/src/attestation_key_register.rs index cd502ae4..8d4c628a 100644 --- a/operator/src/attestation_key_register.rs +++ b/operator/src/attestation_key_register.rs @@ -13,21 +13,13 @@ use k8s_openapi::apimachinery::pkg::{ apis::meta::v1::{LabelSelector, ObjectMeta, OwnerReference}, util::intstr::IntOrString, }; -use kube::{ - Api, Client, Resource, - api::{Patch, PatchParams}, - runtime::{ - Controller, - controller::Action, - finalizer, - finalizer::Event, - reflector::{self, ObjectRef, Store}, - watcher, - }, -}; +use kube::api::{Patch, PatchParams}; +use kube::runtime::{Controller, controller::Action, reflector::ObjectRef, watcher}; +use kube::runtime::{finalizer, finalizer::Event}; +use kube::{Api, Client, Resource}; use log::{info, warn}; use serde_json::json; -use std::{collections::BTreeMap, sync::Arc, time::Duration}; +use std::{collections::BTreeMap, sync::Arc}; use trusted_cluster_operator_lib::conditions::ATTESTATION_KEY_MACHINE_APPROVE; use trusted_cluster_operator_lib::endpoints::*; @@ -35,49 +27,9 @@ use trusted_cluster_operator_lib::{AttestationKey, AttestationKeyStatus, Machine use crate::conditions::attestation_key_approved_condition; use crate::trustee; -use operator::{ControllerError, LONG_REQUEUE, TLS_DIR, controller_error_policy}; +use operator::{ControllerError, LONG_REQUEUE, OperatorContext, TLS_DIR, controller_error_policy}; use operator::{create_or_info_if_exists, read_certificate, upsert_condition}; -/// Shared context for the three attestation-key controllers. -/// Stores give local cache access to avoid repeated API-server reads. -pub struct AkContextData { - pub client: Client, - pub machine_store: Store, - pub ak_store: Store, - pub secret_store: Store, - pub deployment_store: Store, -} - -impl AkContextData { - pub fn new(client: Client) -> Self { - let (machine_store, machine_writer) = reflector::store::(); - let (ak_store, ak_writer) = reflector::store::(); - let (secret_store, secret_writer) = reflector::store::(); - let (deployment_store, deployment_writer) = reflector::store::(); - - crate::spawn_reflector::(machine_writer, client.clone(), "Machine"); - crate::spawn_reflector::(ak_writer, client.clone(), "AttestationKey"); - crate::spawn_reflector::(secret_writer, client.clone(), "Secret"); - crate::spawn_reflector::(deployment_writer, client.clone(), "Deployment"); - - Self { - client, - machine_store, - ak_store, - secret_store, - deployment_store, - } - } - - pub async fn sync_caches(&self, timeout: Duration) -> Result<()> { - crate::sync_cache(&self.machine_store, "Machine", timeout).await?; - crate::sync_cache(&self.ak_store, "AttestationKey", timeout).await?; - crate::sync_cache(&self.secret_store, "Secret", timeout).await?; - crate::sync_cache(&self.deployment_store, "Deployment", timeout).await?; - Ok(()) - } -} - const INTERNAL_ATTESTATION_KEY_REGISTER_PORT: i32 = 8001; const ATTESTATION_KEY_SECRET_FINALIZER: &str = "trusted-execution-clusters.io/attestationkey-secret-finalizer"; @@ -185,7 +137,7 @@ pub async fn create_attestation_key_register_service( async fn ak_reconcile( ak: Arc, - ctx: Arc, + ctx: Arc, ) -> Result { let ak_name = ak.metadata.name.clone().unwrap_or_default(); info!("Attestation Key reconciliation for: {ak_name}"); @@ -201,7 +153,7 @@ async fn ak_reconcile( async fn machine_reconcile( machine: Arc, - ctx: Arc, + ctx: Arc, ) -> Result { info!( "Machine reconciliation for: {}", @@ -228,7 +180,7 @@ async fn machine_reconcile( Ok(LONG_REQUEUE) } -async fn approve_ak(ak: &AttestationKey, machine: &Machine, ctx: &AkContextData) -> Result<()> { +async fn approve_ak(ak: &AttestationKey, machine: &Machine, ctx: &OperatorContext) -> Result<()> { let name = ak.metadata.name.clone().unwrap_or_default(); let client = &ctx.client; let aks: Api = Api::default_namespaced(client.clone()); @@ -274,10 +226,8 @@ async fn approve_ak(ak: &AttestationKey, machine: &Machine, ctx: &AkContextData) let secret_name = name.clone(); let ns = client.default_namespace().to_string(); - let secret_exists = ctx - .secret_store - .get(&ObjectRef::new(&secret_name).within(&ns)) - .is_some(); + let obj_ref = ObjectRef::new(&secret_name).within(&ns); + let secret_exists = ctx.secret_store.get(&obj_ref).is_some(); if !secret_exists { let public_key_data = ByteString(ak.spec.public_key.as_bytes().to_vec()); @@ -305,7 +255,7 @@ async fn approve_ak(ak: &AttestationKey, machine: &Machine, ctx: &AkContextData) async fn secret_reconcile( secret: Arc, - ctx: Arc, + ctx: Arc, ) -> Result { let secret_name = secret.metadata.name.clone().unwrap_or_default(); @@ -357,7 +307,7 @@ async fn secret_reconcile( .map_err(|e| anyhow!("failed to reconcile attestation key secret: {e}").into()) } -pub async fn launch_ak_controller(ctx: Arc) { +pub async fn launch_ak_controller(ctx: Arc) { let aks: Api = Api::default_namespaced(ctx.client.clone()); tokio::spawn( Controller::new(aks, watcher::Config::default()) @@ -371,7 +321,7 @@ pub async fn launch_ak_controller(ctx: Arc) { ); } -pub async fn launch_machine_ak_controller(ctx: Arc) { +pub async fn launch_machine_ak_controller(ctx: Arc) { let machines: Api = Api::default_namespaced(ctx.client.clone()); tokio::spawn( Controller::new(machines, watcher::Config::default()) @@ -385,7 +335,7 @@ pub async fn launch_machine_ak_controller(ctx: Arc) { ); } -pub async fn launch_secret_ak_controller(ctx: Arc) { +pub async fn launch_secret_ak_controller(ctx: Arc) { let secrets: Api = Api::default_namespaced(ctx.client.clone()); tokio::spawn( Controller::new(secrets, watcher::Config::default()) diff --git a/operator/src/lib.rs b/operator/src/lib.rs index d77bc31e..26bea843 100644 --- a/operator/src/lib.rs +++ b/operator/src/lib.rs @@ -8,9 +8,10 @@ // // Use in other crates is not an intended purpose. -use anyhow::Result; +use anyhow::{Result, anyhow}; use futures_util::StreamExt; -use k8s_openapi::api::core::v1::{Secret, SecretVolumeSource, Volume, VolumeMount}; +use k8s_openapi::api::apps::v1::Deployment; +use k8s_openapi::api::core::v1::{ConfigMap, Secret, SecretVolumeSource, Volume, VolumeMount}; use k8s_openapi::apimachinery::pkg::apis::meta::v1::{Condition, Time}; use k8s_openapi::jiff::Timestamp; use kube::Resource; @@ -24,6 +25,50 @@ use tokio::time::timeout; // Re-export common functions from the lib pub use trusted_cluster_operator_lib::generate_owner_reference; +use trusted_cluster_operator_lib::{ + ApprovedImage, AttestationKey, Machine, TrustedExecutionCluster, +}; + +/// Unified context shared across all controllers. +/// Stores give local cache access to avoid repeated API-server reads. +pub struct OperatorContext { + pub client: Client, + pub tec_store: Store, + pub cm_store: Store, + pub deployment_store: Store, + pub machine_store: Store, + pub ak_store: Store, + pub secret_store: Store, + pub image_store: Store, +} + +impl OperatorContext { + pub fn new(client: Client) -> Self { + Self { + client, + tec_store: reflector::store().0, + cm_store: reflector::store().0, + deployment_store: reflector::store().0, + machine_store: reflector::store().0, + ak_store: reflector::store().0, + secret_store: reflector::store().0, + image_store: reflector::store().0, + } + } + + /// Return the single TrustedExecutionCluster from the cache, or an error if more than one exists. + pub fn get_opt_tec(&self) -> Result> { + let state = self.tec_store.state(); + if state.len() > 1 { + let ns = self.client.default_namespace(); + return Err(anyhow!( + "More than one TrustedExecutionCluster found in namespace {ns}. \ + trusted-cluster-operator does not support more than one TrustedExecutionCluster." + )); + } + Ok(state.into_iter().next().map(Arc::unwrap_or_clone)) + } +} #[derive(Debug, thiserror::Error)] pub enum ControllerError { diff --git a/operator/src/main.rs b/operator/src/main.rs index 18d59665..0397d0a9 100644 --- a/operator/src/main.rs +++ b/operator/src/main.rs @@ -10,16 +10,21 @@ use std::time::Duration; use anyhow::{Context, Result}; use env_logger::Env; use futures_util::StreamExt; +use k8s_openapi::api::apps::v1::Deployment; +use k8s_openapi::api::core::v1::{ConfigMap, Secret}; use k8s_openapi::apimachinery::pkg::apis::meta::v1::Condition; use kube::runtime::controller::{Action, Controller}; -use kube::runtime::reflector::{self, Store}; +use kube::runtime::reflector; use kube::runtime::watcher; use kube::{Api, Client}; use log::{info, warn}; -use operator::{generate_owner_reference, upsert_condition}; -use trusted_cluster_operator_lib::{TrustedExecutionCluster, TrustedExecutionClusterStatus}; -use trusted_cluster_operator_lib::{conditions::*, images::*, update_status}; +use operator::OperatorContext; +use operator::{generate_owner_reference, spawn_reflector, sync_cache, upsert_condition}; +use trusted_cluster_operator_lib::{ + ApprovedImage, AttestationKey, Machine, TrustedExecutionCluster, TrustedExecutionClusterStatus, + conditions::*, images::*, update_status, +}; mod attestation_key_register; mod conditions; @@ -44,29 +49,6 @@ const COMPONENT_VERSION: &str = match option_env!("COMPONENT_VERSION") { /// Default registry const TEC_REGISTRY: &str = "quay.io/trusted-execution-clusters"; -struct ClusterContext { - client: Client, - tec_store: Store, -} - -impl ClusterContext { - fn new(client: Client) -> Self { - let (tec_store, tec_writer) = reflector::store::(); - - spawn_reflector::( - tec_writer, - client.clone(), - "TrustedExecutionCluster", - ); - - Self { client, tec_store } - } - - async fn sync_cache_tec(&self, timeout: Duration) -> Result<()> { - sync_cache(&self.tec_store, "TrustedExecutionCluster", timeout).await - } -} - fn is_installed(status: Option) -> bool { let chk = |c: &Condition| c.type_ == INSTALLED_CONDITION && c.status == "True"; status @@ -77,7 +59,7 @@ fn is_installed(status: Option) -> bool { async fn reconcile( cluster: Arc, - ctx: Arc, + ctx: Arc, ) -> Result { let generation = cluster.metadata.generation; let known_address = cluster.spec.public_trustee_addr.is_some(); @@ -142,7 +124,7 @@ async fn reconcile( warn!("Installation of a component failed: {e:?}\nRequeueing..."); return Ok(Action::requeue(Duration::from_secs(60))); } - reference_values::adopt_approved_images(kube_client, &cluster).await?; + reference_values::adopt_approved_images(&ctx, &cluster).await?; let installed_condition = installed_condition(INSTALLED_REASON, generation, existing_status); let changed = upsert_condition(&mut conditions, installed_condition); @@ -254,38 +236,68 @@ async fn main() -> Result<()> { env_logger::Builder::from_env(Env::default().default_filter_or("info")).init(); let kube_client = Client::try_default().await?; - info!("trusted execution clusters operator",); + info!("trusted execution clusters operator"); const CACHE_SYNC_TIMEOUT: Duration = Duration::from_secs(60); - // Launch controllers that do not depend on reflector caches first. - register_server::launch_keygen_controller(kube_client.clone()).await; - - // Spawn reflectors (starts background list-watch immediately). - let ak_ctx = Arc::new(attestation_key_register::AkContextData::new( - kube_client.clone(), - )); - let ctx = Arc::new(ClusterContext::new(kube_client.clone())); + // Create all reflector stores and spawn background watchers. + let (tec_store, tec_writer) = reflector::store::(); + let (cm_store, cm_writer) = reflector::store::(); + let (deployment_store, deployment_writer) = reflector::store::(); + let (machine_store, machine_writer) = reflector::store::(); + let (ak_store, ak_writer) = reflector::store::(); + let (secret_store, secret_writer) = reflector::store::(); + let (image_store, image_writer) = reflector::store::(); + + let tec_kind = "TrustedExecutionCluster"; + spawn_reflector::(tec_writer, kube_client.clone(), tec_kind); + spawn_reflector::(cm_writer, kube_client.clone(), "ConfigMap"); + spawn_reflector::(deployment_writer, kube_client.clone(), "Deployment"); + spawn_reflector::(machine_writer, kube_client.clone(), "Machine"); + spawn_reflector::(ak_writer, kube_client.clone(), "AttestationKey"); + spawn_reflector::(secret_writer, kube_client.clone(), "Secret"); + spawn_reflector::(image_writer, kube_client.clone(), "ApprovedImage"); + + let mut ctx = OperatorContext::new(kube_client.clone()); + ctx.tec_store = tec_store; + ctx.cm_store = cm_store; + ctx.deployment_store = deployment_store; + ctx.machine_store = machine_store; + ctx.ak_store = ak_store; + ctx.secret_store = secret_store; + ctx.image_store = image_store; + let ctx = Arc::new(ctx); // Best-effort wait for caches; controllers will work with // partially-filled stores if the sync times out. - if let Err(e) = ak_ctx.sync_caches(CACHE_SYNC_TIMEOUT).await { - warn!("AK cache sync incomplete, controllers will retry: {e}"); + macro_rules! sync { + ($($name:expr => $store:expr),+ $(,)?) => {$( + if let Err(e) = sync_cache(&$store, $name, CACHE_SYNC_TIMEOUT).await { + warn!("{} cache sync incomplete, controllers will retry: {e}", $name); + } + )+}; } - if let Err(e) = ctx.sync_cache_tec(CACHE_SYNC_TIMEOUT).await { - warn!("TEC cache sync incomplete, controller will retry: {e}"); + sync! { + "TrustedExecutionCluster" => ctx.tec_store, + "ConfigMap" => ctx.cm_store, + "Deployment" => ctx.deployment_store, + "Machine" => ctx.machine_store, + "AttestationKey" => ctx.ak_store, + "Secret" => ctx.secret_store, + "ApprovedImage" => ctx.image_store, } info!("Starting controllers"); let cl: Api = Api::default_namespaced(kube_client.clone()); - attestation_key_register::launch_ak_controller(ak_ctx.clone()).await; - attestation_key_register::launch_machine_ak_controller(ak_ctx.clone()).await; - attestation_key_register::launch_secret_ak_controller(ak_ctx).await; + register_server::launch_keygen_controller(ctx.clone()).await; + attestation_key_register::launch_ak_controller(ctx.clone()).await; + attestation_key_register::launch_machine_ak_controller(ctx.clone()).await; + attestation_key_register::launch_secret_ak_controller(ctx.clone()).await; reference_values::create_pcrs_config_map(kube_client.clone()).await?; - reference_values::launch_rv_image_controller(kube_client.clone()).await; - reference_values::launch_rv_job_controller(kube_client.clone()).await; + reference_values::launch_rv_image_controller(ctx.clone()).await; + reference_values::launch_rv_job_controller(ctx.clone()).await; Controller::new(cl, watcher::Config::default()) .run(reconcile, controller_error_policy, ctx) @@ -301,30 +313,23 @@ mod tests { use k8s_openapi::api::apps::v1::Deployment; use k8s_openapi::api::core::v1::{ConfigMap, Service}; use k8s_openapi::{apimachinery::pkg::apis::meta::v1::Time, jiff::Timestamp}; - use kube::api::ObjectList; use kube::client::Body; - use trusted_cluster_operator_lib::ApprovedImage; + use kube::runtime::{reflector, watcher}; use super::*; use trusted_cluster_operator_test_utils::mock_client::*; - fn dummy_cluster_ctx(client: Client) -> ClusterContext { - ClusterContext { - client, - tec_store: reflector::store::().0, - } - } - - /// Build a Store pre-populated with two distinct TrustedExecutionCluster objects. - fn two_cluster_tec_store() -> Store { - let (store, mut writer) = reflector::store::(); + fn op_ctx_with_two_tecs(client: Client) -> OperatorContext { + let (tec_store, mut writer) = reflector::store::(); let mut second = dummy_cluster(); second.metadata.name = Some("test2".to_string()); writer.apply_watcher_event(&watcher::Event::Init); writer.apply_watcher_event(&watcher::Event::InitApply(dummy_cluster())); writer.apply_watcher_event(&watcher::Event::InitApply(second)); writer.apply_watcher_event(&watcher::Event::InitDone); - store + let mut ctx = OperatorContext::new(client); + ctx.tec_store = tec_store; + ctx } #[tokio::test] @@ -340,7 +345,7 @@ mod tests { count_check!(1, clos, |client| { let mut cluster = dummy_cluster(); cluster.metadata.deletion_timestamp = Some(Time(Timestamp::now())); - let result = reconcile(Arc::new(cluster), Arc::new(dummy_cluster_ctx(client))).await; + let result = reconcile(Arc::new(cluster), Arc::new(OperatorContext::new(client))).await; assert_eq!(result.unwrap(), LONG_REQUEUE); }); } @@ -356,13 +361,9 @@ mod tests { panic!("unexpected API interaction: {req:?}, counter {ctr}"); } }; - let store = two_cluster_tec_store(); count_check!(1, clos, |client| { let cluster = Arc::new(dummy_cluster()); - let ctx = Arc::new(ClusterContext { - client, - tec_store: store, - }); + let ctx = Arc::new(op_ctx_with_two_tecs(client)); let result = reconcile(cluster, ctx).await; assert_eq!(result.unwrap(), Action::requeue(Duration::from_secs(60))); }); @@ -374,13 +375,9 @@ mod tests { r if r.method() == Method::PATCH => Err(StatusCode::INTERNAL_SERVER_ERROR), _ => panic!("unexpected API interaction: {req:?}"), }; - let store = two_cluster_tec_store(); count_check!(1, clos, |client| { let cluster = Arc::new(dummy_cluster()); - let ctx = Arc::new(ClusterContext { - client, - tec_store: store, - }); + let ctx = Arc::new(op_ctx_with_two_tecs(client)); let result = reconcile(cluster, ctx).await; assert!(result.is_err()); }); @@ -420,7 +417,7 @@ mod tests { cluster.status = Some(TrustedExecutionClusterStatus { conditions: Some(vec![foreign_condition]), }); - let result = reconcile(Arc::new(cluster), Arc::new(dummy_cluster_ctx(client))).await; + let result = reconcile(Arc::new(cluster), Arc::new(OperatorContext::new(client))).await; assert_eq!(result.unwrap(), LONG_REQUEUE); }); } @@ -440,6 +437,8 @@ mod tests { observed_generation: None, }; + // adopt_approved_images now reads from image_store (empty) — no GET needed. + // 8 POSTs for install, then 1 PATCH for final status. let clos = async |req: Request, ctr| { if ctr < 8 && req.method() == Method::POST { use serde_json::to_string; @@ -458,14 +457,7 @@ mod tests { _ => unreachable!("unexpected counter {ctr}"), }; Ok(resp.unwrap()) - } else if ctr == 8 && req.method() == Method::GET { - let object_list = ObjectList:: { - items: Vec::new(), - types: Default::default(), - metadata: Default::default(), - }; - Ok(serde_json::to_string(&object_list).unwrap()) - } else if ctr == 9 && req.method() == Method::PATCH { + } else if ctr == 8 && req.method() == Method::PATCH { let body = req.into_body().collect_bytes().await.unwrap().to_vec(); let body = String::from_utf8_lossy(&body); assert!(body.contains("ForeignCondition"),); @@ -496,8 +488,8 @@ mod tests { cluster.status = Some(TrustedExecutionClusterStatus { conditions: Some(vec![pre_existing_installed, foreign_condition]), }); - count_check!(10, clos, |client| { - let result = reconcile(Arc::new(cluster), Arc::new(dummy_cluster_ctx(client))).await; + count_check!(9, clos, |client| { + let result = reconcile(Arc::new(cluster), Arc::new(OperatorContext::new(client))).await; assert_eq!(result.unwrap(), LONG_REQUEUE); }); } @@ -514,7 +506,7 @@ mod tests { count_check!(1, clos1, |client| { let mut cluster = dummy_cluster(); cluster.metadata.deletion_timestamp = Some(Time(Timestamp::now())); - reconcile(Arc::new(cluster), Arc::new(dummy_cluster_ctx(client))) + reconcile(Arc::new(cluster), Arc::new(OperatorContext::new(client))) .await .unwrap(); }); @@ -548,7 +540,7 @@ mod tests { let mut cluster = dummy_cluster(); cluster.metadata.deletion_timestamp = Some(Time(Timestamp::now())); cluster.status = Some(TrustedExecutionClusterStatus { conditions }); - reconcile(Arc::new(cluster), Arc::new(dummy_cluster_ctx(client))) + reconcile(Arc::new(cluster), Arc::new(OperatorContext::new(client))) .await .unwrap(); }); diff --git a/operator/src/reference_values.rs b/operator/src/reference_values.rs index 946315c2..16c8182b 100644 --- a/operator/src/reference_values.rs +++ b/operator/src/reference_values.rs @@ -14,13 +14,10 @@ use k8s_openapi::{ }, jiff::Timestamp, }; -use kube::api::{DeleteParams, ListParams, ObjectMeta, Patch}; -use kube::runtime::{ - controller::{Action, Controller}, - finalizer, - finalizer::Event, - watcher, -}; +use kube::api::{DeleteParams, ListParams, ObjectMeta, Patch, PatchParams}; +use kube::runtime::controller::{Action, Controller}; +use kube::runtime::{finalizer, finalizer::Event}; +use kube::runtime::{reflector::ObjectRef, watcher}; use kube::{Api, Client, Resource}; use log::{info, warn}; use oci_client::secrets::RegistryAuth; @@ -32,7 +29,7 @@ use std::{collections::BTreeMap, sync::Arc, time::Duration}; use crate::COMPONENT_VERSION; use crate::trustee::{self, get_image_pcrs}; -use operator::{ControllerError, LONG_REQUEUE, upsert_condition}; +use operator::{ControllerError, LONG_REQUEUE, OperatorContext, upsert_condition}; use operator::{controller_error_policy, controller_info, create_or_info_if_exists}; use trusted_cluster_operator_lib::{conditions::*, reference_values::*, *}; @@ -43,6 +40,15 @@ const PCR_LABEL: &str = "org.coreos.pcrs"; /// Finalizer name to discard reference values when an image is no longer approved const APPROVED_IMAGE_FINALIZER: &str = "finalizer.approved-image.trusted-execution-clusters.io"; +fn cached_image_pcrs(ctx: &OperatorContext) -> Result { + let ns = ctx.client.default_namespace().to_string(); + ctx.cm_store + .get(&ObjectRef::new(PCR_CONFIG_MAP).within(&ns)) + .map(|cm| get_image_pcrs((*cm).clone())) + .transpose() + .map(|opt| opt.unwrap_or_default()) +} + /// Synchronize with compute_pcrs_cli::Output #[derive(Deserialize)] struct ComputePcrsOutput { @@ -115,33 +121,36 @@ fn build_compute_pcrs_pod_spec( } } -async fn job_reconcile(job: Arc, client: Arc) -> Result { +async fn job_reconcile( + job: Arc, + ctx: Arc, +) -> Result { let err = "Job changed, but had no name"; let name = &job.metadata.name.clone().context(err)?; let err = format!("Job {name} changed, but had no status"); let status = &job.status.clone().context(err)?; - let kube_client = Arc::unwrap_or_clone(client); if status.completion_time.is_none() { info!("Job {name} changed, but had not completed"); return Ok(Action::requeue(Duration::from_secs(300))); } - let jobs: Api = Api::default_namespaced(kube_client.clone()); + let jobs: Api = Api::default_namespaced(ctx.client.clone()); // Foreground deletion: Delete the pod too let delete = jobs.delete(name, &DeleteParams::foreground()).await; delete.map_err(Into::::into)?; - trustee::update_reference_values(kube_client).await?; + let image_pcrs = cached_image_pcrs(&ctx)?; + trustee::update_reference_values(&ctx.client, image_pcrs).await?; Ok(Action::await_change()) } -pub async fn launch_rv_job_controller(client: Client) { - let jobs: Api = Api::default_namespaced(client.clone()); +pub async fn launch_rv_job_controller(ctx: Arc) { + let jobs: Api = Api::default_namespaced(ctx.client.clone()); let watcher = watcher::Config { label_selector: Some(format!("{JOB_LABEL_KEY}={PCR_COMMAND_NAME}")), ..Default::default() }; tokio::spawn( Controller::new(jobs, watcher) - .run(job_reconcile, controller_error_policy, Arc::new(client)) + .run(job_reconcile, controller_error_policy, ctx) .for_each(controller_info), ); } @@ -221,16 +230,14 @@ async fn adopt_approved_image( } pub async fn adopt_approved_images( - client: Client, + ctx: &OperatorContext, cluster: &TrustedExecutionCluster, ) -> Result<()> { - let images: Api = Api::default_namespaced(client.clone()); - let images_list = images.list(&Default::default()).await?; - for image in images_list.items.iter() { + for image in ctx.image_store.state() { if image.metadata.deletion_timestamp.is_none() && let Some(name) = image.metadata.name.as_ref() { - adopt_approved_image(client.clone(), name, cluster).await?; + adopt_approved_image(ctx.client.clone(), name, cluster).await?; } } Ok(()) @@ -238,14 +245,13 @@ pub async fn adopt_approved_images( async fn image_reconcile( image: Arc, - client: Arc, + ctx: Arc, ) -> Result { - let kube_client = Arc::::unwrap_or_clone(client); + let kube_client = ctx.client.clone(); let err = "ApprovedImage had no name"; let name = image.metadata.name.clone().context(err)?; - let cluster = get_opt_trusted_execution_cluster(kube_client.clone()) - .await - .map_err(|e| -> ControllerError { e.into() })?; + let map_ctl = |e: anyhow::Error| -> ControllerError { e.into() }; + let cluster = ctx.get_opt_tec().map_err(map_ctl)?; let uid_owns = |uid: &String| { let refs = image.metadata.owner_references.as_ref(); @@ -261,16 +267,16 @@ async fn image_reconcile( { adopt_approved_image(kube_client.clone(), &name, cluster) .await - .map_err(|e| -> ControllerError { e.into() })?; + .map_err(map_ctl)?; } let images: Api = Api::default_namespaced(kube_client.clone()); finalizer(&images, APPROVED_IMAGE_FINALIZER, image, |ev| async { match ev { - Event::Apply(image) => image_add_reconcile(kube_client, &image, cluster) + Event::Apply(image) => image_add_reconcile(&ctx, &image, cluster) .await .map_err(|e| finalizer::Error::::ApplyFailed(e.into())), - Event::Cleanup(image) => image_remove_reconcile(kube_client, image, cluster) + Event::Cleanup(image) => image_remove_reconcile(&ctx, image, cluster) .await .map_err(|e| finalizer::Error::::CleanupFailed(e.into())), } @@ -280,7 +286,7 @@ async fn image_reconcile( } async fn image_add_reconcile( - client: Client, + ctx: &OperatorContext, image: &ApprovedImage, cluster: Option, ) -> Result { @@ -294,7 +300,7 @@ async fn image_add_reconcile( info!("TrustedExecutionCluster is being deleted, deferring image processing for {name}"); return Ok(Action::requeue(Duration::from_secs(5))); } - let (action, reason) = match handle_new_image(client.clone(), image).await { + let (action, reason) = match handle_new_image(ctx, image).await { Ok(reason) => (LONG_REQUEUE, reason), Err(e) => { warn!("PCR computation for {name} failed: {e}"); @@ -308,7 +314,7 @@ async fn image_add_reconcile( let mut conditions = image.status.as_ref().and_then(|s| s.conditions.clone()); let changed = upsert_condition(&mut conditions, committed); if changed { - let images: Api = Api::default_namespaced(client); + let images: Api = Api::default_namespaced(ctx.client.clone()); update_status!(images, &name, ApprovedImageStatus { conditions }) .map_err(|e| finalizer::Error::::ApplyFailed(e.into()))?; } @@ -316,7 +322,7 @@ async fn image_add_reconcile( } async fn image_remove_reconcile( - client: Client, + ctx: &OperatorContext, image: Arc, cluster: Option, ) -> Result { @@ -335,15 +341,15 @@ async fn image_remove_reconcile( ); return Ok(LONG_REQUEUE); } - disallow_image(client, name).await?; + disallow_image(ctx, name).await?; Ok(LONG_REQUEUE) } -pub async fn launch_rv_image_controller(client: Client) { - let images: Api = Api::default_namespaced(client.clone()); +pub async fn launch_rv_image_controller(ctx: Arc) { + let images: Api = Api::default_namespaced(ctx.client.clone()); tokio::spawn( Controller::new(images, Default::default()) - .run(image_reconcile, controller_error_policy, Arc::new(client)) + .run(image_reconcile, controller_error_policy, ctx) .for_each(controller_info), ); } @@ -359,18 +365,19 @@ async fn is_pending(client: &Client, resource_name: &str) -> Result { .is_some_and(|phase| phase == "Pending")) } -pub async fn handle_new_image(client: Client, image: &ApprovedImage) -> Result<&'static str> { +pub async fn handle_new_image( + ctx: &OperatorContext, + image: &ApprovedImage, +) -> Result<&'static str> { let resource_name = image.metadata.name.as_ref().unwrap(); let boot_image = image.spec.image.as_ref(); - let config_maps: Api = Api::default_namespaced(client.clone()); - let mut image_pcrs_map = config_maps.get(PCR_CONFIG_MAP).await?; - let mut image_pcrs = get_image_pcrs(image_pcrs_map.clone())?; + let mut image_pcrs = cached_image_pcrs(ctx)?; if let Some(pcr) = image_pcrs.0.get(resource_name) && pcr.reference == boot_image { info!("Image {boot_image} was to be allowed, but already was allowed"); - let res = trustee::update_reference_values(client).await; - return res.map(|_| COMMITTED_REASON); + let res = trustee::update_reference_values(&ctx.client, image_pcrs); + return res.await.map(|_| COMMITTED_REASON); } let image_ref: oci_client::Reference = boot_image.parse()?; if image_ref.digest().is_none() { @@ -386,7 +393,7 @@ pub async fn handle_new_image(client: Client, image: &ApprovedImage) -> Result<& let should_compute_pcrs = match label { Err(ref e) => { warn!("Fetching PCR label for {image_ref} failed: {e}. Falling back to computation."); - if is_pending(&client, resource_name).await? { + if is_pending(&ctx.client, resource_name).await? { return Ok(NOT_COMMITTED_REASON_PENDING); } true @@ -398,8 +405,9 @@ pub async fn handle_new_image(client: Client, image: &ApprovedImage) -> Result<& _ => false, }; if should_compute_pcrs { - let err = NOT_COMMITTED_REASON_COMPUTING; - return compute_fresh_pcrs(client, image).await.map(|_| err); + return compute_fresh_pcrs(ctx.client.clone(), image) + .await + .map(|_| NOT_COMMITTED_REASON_COMPUTING); } let image_pcr = ImagePcr { @@ -408,21 +416,30 @@ pub async fn handle_new_image(client: Client, image: &ApprovedImage) -> Result<& reference: boot_image.to_string(), }; image_pcrs.0.insert(resource_name.to_string(), image_pcr); - update_image_pcrs!(config_maps, image_pcrs_map, image_pcrs); - trustee::update_reference_values(client) - .await - .map(|_| COMMITTED_REASON) + let err = COMMITTED_REASON; + apply_image_pcrs(ctx, image_pcrs).await.map(|_| err) } -pub async fn disallow_image(client: Client, resource_name: &str) -> Result<()> { - let config_maps: Api = Api::default_namespaced(client.clone()); - let mut image_pcrs_map = config_maps.get(PCR_CONFIG_MAP).await?; - let mut image_pcrs = get_image_pcrs(image_pcrs_map.clone())?; +pub async fn disallow_image(ctx: &OperatorContext, resource_name: &str) -> Result<()> { + let mut image_pcrs = cached_image_pcrs(ctx)?; if image_pcrs.0.remove(resource_name).is_none() { info!("Image {resource_name} was to be disallowed, but already was not allowed"); } - update_image_pcrs!(config_maps, image_pcrs_map, image_pcrs); - trustee::update_reference_values(client).await + apply_image_pcrs(ctx, image_pcrs).await +} + +async fn apply_image_pcrs(ctx: &OperatorContext, image_pcrs: ImagePcrs) -> Result<()> { + let image_pcrs_json = serde_json::to_string(&image_pcrs)?; + let patch = Patch::Apply(&json!({ + "apiVersion": "v1", + "kind": "ConfigMap", + "metadata": { "name": PCR_CONFIG_MAP }, + "data": { PCR_CONFIG_FILE: image_pcrs_json } + })); + let config_maps: Api = Api::default_namespaced(ctx.client.clone()); + let pp = PatchParams::apply("trusted-cluster-operator").force(); + config_maps.patch(PCR_CONFIG_MAP, &pp, &patch).await?; + trustee::update_reference_values(&ctx.client, image_pcrs).await } #[cfg(test)] @@ -432,11 +449,34 @@ mod tests { use http::{Method, Request, StatusCode}; use k8s_openapi::api::batch::v1::JobStatus; use k8s_openapi::apimachinery::pkg::apis::meta::v1::Time; - use kube::api::ObjectList; use kube::client::Body; + use kube::runtime::{reflector, watcher}; use trusted_cluster_operator_test_utils::mock_client::*; use trusted_cluster_operator_test_utils::test_error_method; + fn op_ctx_with_cm(client: Client, name: &str, mut cm: ConfigMap) -> OperatorContext { + cm.metadata.name = Some(name.to_string()); + let (cm_store, mut writer) = reflector::store::(); + writer.apply_watcher_event(&watcher::Event::Init); + writer.apply_watcher_event(&watcher::Event::InitApply(cm)); + writer.apply_watcher_event(&watcher::Event::InitDone); + let mut ctx = OperatorContext::new(client); + ctx.cm_store = cm_store; + ctx + } + + fn op_ctx_with_images(client: Client, images: Vec) -> OperatorContext { + let (image_store, mut writer) = reflector::store::(); + writer.apply_watcher_event(&watcher::Event::Init); + for img in images { + writer.apply_watcher_event(&watcher::Event::InitApply(img)); + } + writer.apply_watcher_event(&watcher::Event::InitDone); + let mut ctx = OperatorContext::new(client); + ctx.image_store = image_store; + ctx + } + const DUMMY_IMAGE_REF: &str = "quay.io/some-ref@sha256:e71dad00aa0e3d70540e726a0c66407e3004d96e045ab6c253186e327a2419e5"; @@ -490,19 +530,16 @@ mod tests { async fn test_job_reconcile_success() { let clos = async |req: Request<_>, ctr| match (ctr, req.method()) { (0, &Method::DELETE) => Ok(serde_json::to_string(&Job::default()).unwrap()), - (1, &Method::GET) => { - assert!(req.uri().path().contains(PCR_CONFIG_MAP)); - Ok(serde_json::to_string(&dummy_pcrs_map()).unwrap()) - } - (2, &Method::GET) | (3, &Method::PUT) => { + (1, &Method::PATCH) => { assert!(req.uri().path().contains(trustee::TRUSTEE_DATA_MAP)); Ok(serde_json::to_string(&dummy_trustee_map()).unwrap()) } _ => panic!("unexpected API interaction: {req:?}, counter {ctr}"), }; - count_check!(4, clos, |client| { + count_check!(2, clos, |client| { + let ctx = Arc::new(op_ctx_with_cm(client, PCR_CONFIG_MAP, dummy_pcrs_map())); let job = Arc::new(dummy_job()); - let result = job_reconcile(job, Arc::new(client)).await.unwrap(); + let result = job_reconcile(job, ctx).await.unwrap(); assert_eq!(result, Action::await_change()); }); } @@ -514,7 +551,8 @@ mod tests { let mut job = dummy_job(); let status = job.status.as_mut().unwrap(); status.completion_time = None; - let result = job_reconcile(Arc::new(job), Arc::new(client)).await; + let ctx = Arc::new(OperatorContext::new(client)); + let result = job_reconcile(Arc::new(job), ctx).await; assert_eq!(result.unwrap(), Action::requeue(Duration::from_secs(300))); }); } @@ -570,32 +608,35 @@ mod tests { #[tokio::test] async fn test_adopt_approved_images() { let cluster = dummy_cluster(); + let mut deleted = dummy_image(); + deleted.metadata.name = Some("deleted".to_string()); + deleted.metadata.deletion_timestamp = Some(Time(Timestamp::now())); + let mut second = dummy_image(); + second.metadata.name = Some("second".to_string()); let clos = async |req: Request<_>, ctr| { - if ctr == 0 && req.method() == Method::GET { - let mut deleted = dummy_image(); - deleted.metadata.deletion_timestamp = Some(Time(Timestamp::now())); - let list = ObjectList { - items: vec![dummy_image(), deleted, dummy_image()], - types: Default::default(), - metadata: Default::default(), - }; - Ok(serde_json::to_string(&list).unwrap()) - } else if ctr < 3 && req.method() == Method::PATCH { + if ctr < 2 && req.method() == Method::PATCH { Ok(serde_json::to_string(&dummy_image()).unwrap()) } else { panic!("unexpected API interaction: {req:?}, counter {ctr}") } }; - count_check!(3, clos, |client| { - assert!(adopt_approved_images(client, &cluster).await.is_ok()); + count_check!(2, clos, |client| { + let ctx = op_ctx_with_images(client, vec![dummy_image(), deleted, second]); + assert!(adopt_approved_images(&ctx, &cluster).await.is_ok()); }); } #[tokio::test] async fn test_adopt_approved_images_error() { let cluster = dummy_cluster(); - let clos = |client| adopt_approved_images(client, &cluster); - test_error_method!(clos, Method::GET); + let clos = async |req: Request<_>, _| match req.method() { + &Method::PATCH => Err(StatusCode::INTERNAL_SERVER_ERROR), + _ => panic!("unexpected API interaction: {req:?}"), + }; + count_check!(1, clos, |client| { + let ctx = op_ctx_with_images(client, vec![dummy_image()]); + assert!(adopt_approved_images(&ctx, &cluster).await.is_err()); + }); } // handle_new_image and its caller image_add_reconcile are @@ -606,19 +647,19 @@ mod tests { let image = Arc::new(dummy_image()); let cluster = Some(dummy_cluster()); let clos = async |req: Request<_>, ctr| match (ctr, req.method()) { - // fetched & updated for removal, then fetched for recomputation - (0, &Method::GET) | (1, &Method::PUT) | (2, &Method::GET) => { + (0, &Method::PATCH) => { assert!(req.uri().path().contains(PCR_CONFIG_MAP)); Ok(serde_json::to_string(&dummy_pcrs_map()).unwrap()) } - (3, &Method::GET) | (4, &Method::PUT) => { + (1, &Method::PATCH) => { assert!(req.uri().path().contains(trustee::TRUSTEE_DATA_MAP)); Ok(serde_json::to_string(&dummy_trustee_map()).unwrap()) } _ => panic!("unexpected API interaction: {req:?}, counter {ctr}"), }; - count_check!(5, clos, |client| { - assert!(image_remove_reconcile(client, image, cluster).await.is_ok()); + count_check!(2, clos, |client| { + let ctx = op_ctx_with_cm(client, PCR_CONFIG_MAP, dummy_pcrs_map()); + assert!(image_remove_reconcile(&ctx, image, cluster).await.is_ok()); }); } } diff --git a/operator/src/register_server.rs b/operator/src/register_server.rs index 8b733ef6..1e897540 100644 --- a/operator/src/register_server.rs +++ b/operator/src/register_server.rs @@ -17,6 +17,7 @@ use kube::runtime::{ controller::{Action, Controller}, finalizer, finalizer::Event, + reflector::ObjectRef, }; use kube::{Api, Client, Resource}; use log::info; @@ -24,7 +25,7 @@ use std::{collections::BTreeMap, sync::Arc}; use crate::trustee; use operator::*; -use trusted_cluster_operator_lib::{Machine, TrustedExecutionCluster, endpoints::*}; +use trusted_cluster_operator_lib::{Machine, endpoints::*}; /// Finalizer name to discard decryption keys when a machine is deleted const MACHINE_FINALIZER: &str = "finalizer.machine.trusted-execution-clusters.io"; @@ -127,26 +128,23 @@ pub async fn create_register_server_service( async fn keygen_reconcile( machine: Arc, - client: Arc, + ctx: Arc, ) -> Result { - let kube_client_clone = Arc::unwrap_or_clone(client.clone()); - let machines: Api = Api::default_namespaced(kube_client_clone.clone()); + let machines: Api = Api::default_namespaced(ctx.client.clone()); finalizer(&machines, MACHINE_FINALIZER, machine, |ev| async move { match ev { Event::Apply(machine) => { - let kube_client = Arc::unwrap_or_clone(client); let id = &machine.spec.id.clone(); async { let owner_reference = generate_owner_reference(&Arc::unwrap_or_clone(machine))?; - trustee::generate_secret(kube_client.clone(), id, owner_reference).await?; - trustee::mount_secret(kube_client, id).await + trustee::generate_secret(ctx.client.clone(), id, owner_reference).await?; + trustee::mount_secret(&ctx, id).await } .await .map(|_| LONG_REQUEUE) .map_err(|e| finalizer::Error::::ApplyFailed(e.into())) } Event::Cleanup(machine) => { - let kube_client = Arc::unwrap_or_clone(client); let id = &machine.spec.id; // Check if the TrustedExecutionCluster is being deleted @@ -157,24 +155,20 @@ async fn keygen_reconcile( .find(|owner| owner.kind == "TrustedExecutionCluster") { let tec_name = &tec_owner.name; - let tecs: Api = - Api::default_namespaced(kube_client.clone()); - - match tecs.get(tec_name).await { - Ok(tec) if tec.metadata.deletion_timestamp.is_some() => { - // TEC is being deleted, skip unmount_secret + let ns = ctx.client.default_namespace().to_string(); + match ctx.tec_store.get(&ObjectRef::new(tec_name).within(&ns)) { + Some(tec) if tec.metadata.deletion_timestamp.is_some() => { info!( "TrustedExecutionCluster {tec_name} is being deleted, \ - skipping unmount_secret for Machine {}", + skipping unmount_secret for Machine {}", machine.metadata.name.as_deref().unwrap_or("unknown") ); return Ok(LONG_REQUEUE); } - Err(kube::Error::Api(ae)) if ae.code == 404 => { - // TEC already deleted, skip unmount_secret + None => { info!( "TrustedExecutionCluster {tec_name} not found, \ - skipping unmount_secret for Machine {}", + skipping unmount_secret for Machine {}", machine.metadata.name.as_deref().unwrap_or("unknown") ); return Ok(LONG_REQUEUE); @@ -185,7 +179,7 @@ async fn keygen_reconcile( } } - trustee::unmount_secret(kube_client, id) + trustee::unmount_secret(&ctx, id) .await .map(|_| LONG_REQUEUE) .map_err(|e| finalizer::Error::::CleanupFailed(e.into())) @@ -196,11 +190,11 @@ async fn keygen_reconcile( .map_err(|e| anyhow!("failed to reconcile on machine: {e}").into()) } -pub async fn launch_keygen_controller(client: Client) { - let machines: Api = Api::default_namespaced(client.clone()); +pub async fn launch_keygen_controller(ctx: Arc) { + let machines: Api = Api::default_namespaced(ctx.client.clone()); tokio::spawn( Controller::new(machines, Default::default()) - .run(keygen_reconcile, controller_error_policy, Arc::new(client)) + .run(keygen_reconcile, controller_error_policy, ctx) .for_each(controller_info), ); } diff --git a/operator/src/trustee.rs b/operator/src/trustee.rs index 1000af65..47506d81 100644 --- a/operator/src/trustee.rs +++ b/operator/src/trustee.rs @@ -4,7 +4,6 @@ // // SPDX-License-Identifier: MIT -use crate::attestation_key_register::AkContextData; use anyhow::{Context, Result}; use base64::{Engine as _, engine::general_purpose}; use chrono::{DateTime, Utc}; @@ -27,10 +26,11 @@ use kube::{ runtime::reflector::ObjectRef, }; use log::info; -use operator::{TLS_DIR, create_or_info_if_exists, read_certificate}; +use operator::{OperatorContext, TLS_DIR, create_or_info_if_exists, read_certificate}; use serde::{Serialize, Serializer}; use serde_json::{Value::String as JsonString, json}; use std::collections::{BTreeMap, BTreeSet}; +use std::sync::Arc; use trusted_cluster_operator_lib::endpoints::*; use trusted_cluster_operator_lib::reference_values::*; @@ -100,21 +100,18 @@ fn recompute_reference_values(image_pcrs: ImagePcrs) -> Vec { .collect() } -pub async fn update_reference_values(client: Client) -> Result<()> { - let config_maps: Api = Api::default_namespaced(client); - - let image_pcrs_map = config_maps.get(PCR_CONFIG_MAP).await?; - let reference_values = recompute_reference_values(get_image_pcrs(image_pcrs_map)?); +pub async fn update_reference_values(client: &Client, image_pcrs: ImagePcrs) -> Result<()> { + let reference_values = recompute_reference_values(image_pcrs); let rv_json = serde_json::to_string(&reference_values)?; - - let mut trustee_map = config_maps.get(TRUSTEE_DATA_MAP).await?; - let err = format!("ConfigMap {TRUSTEE_DATA_MAP} existed, but had no data"); - let trustee_data = trustee_map.data.as_mut().context(err)?; - trustee_data.insert(REFERENCE_VALUES_FILE.to_string(), rv_json); - - config_maps - .replace(TRUSTEE_DATA_MAP, &Default::default(), &trustee_map) - .await?; + let config_maps: Api = Api::default_namespaced(client.clone()); + let patch = Patch::Apply(&json!({ + "apiVersion": "v1", + "kind": "ConfigMap", + "metadata": { "name": TRUSTEE_DATA_MAP }, + "data": { REFERENCE_VALUES_FILE: rv_json } + })); + let pp = PatchParams::apply("trusted-cluster-operator").force(); + config_maps.patch(TRUSTEE_DATA_MAP, &pp, &patch).await?; info!("Recomputed reference values"); Ok(()) } @@ -149,58 +146,76 @@ fn generate_secret_volume(id: &str) -> (Volume, VolumeMount) { ) } -pub async fn mount_secret(client: Client, id: &str) -> Result<()> { - let result = do_mount_secret(client, id, true).await; +pub async fn mount_secret(ctx: &OperatorContext, id: &str) -> Result<()> { + let result = do_mount_secret(ctx, id, true).await; info!("Mounted secret {id} to {TRUSTEE_DEPLOYMENT}"); result } -pub async fn unmount_secret(client: Client, id: &str) -> Result<()> { - let result = do_mount_secret(client, id, false).await; +pub async fn unmount_secret(ctx: &OperatorContext, id: &str) -> Result<()> { + let result = do_mount_secret(ctx, id, false).await; info!("Unmounted secret {id} from {TRUSTEE_DEPLOYMENT}"); result } -pub async fn do_mount_secret(client: Client, id: &str, add: bool) -> Result<()> { - let deployments: Api = Api::default_namespaced(client); - let mut deployment = deployments.get(TRUSTEE_DEPLOYMENT).await?; +pub async fn do_mount_secret(ctx: &OperatorContext, id: &str, add: bool) -> Result<()> { + let client = &ctx.client; + let ns = client.default_namespace().to_string(); + let obj_ref = ObjectRef::new(TRUSTEE_DEPLOYMENT).within(&ns); + let Some(deployment) = ctx.deployment_store.get(&obj_ref).map(Arc::unwrap_or_clone) else { + info!("{TRUSTEE_DEPLOYMENT} not found in cache, skipping secret mount for {id}"); + return Ok(()); + }; let err = format!("Deployment {TRUSTEE_DEPLOYMENT} existed, but had no spec"); - let depl_spec = deployment.spec.as_mut().context(err)?; + let depl_spec = deployment.spec.as_ref().context(err)?; let err = format!("Deployment {TRUSTEE_DEPLOYMENT} existed, but had no pod spec"); - let pod_spec = depl_spec.template.spec.as_mut().context(err)?; + let pod_spec = depl_spec.template.spec.as_ref().context(err)?; let err = format!("Deployment {TRUSTEE_DEPLOYMENT} existed, but had no containers"); - let container = pod_spec.containers.get_mut(0).context(err)?; - let vol_mounts = container.volume_mounts.get_or_insert_default(); + let container = pod_spec.containers.first().context(err)?; + + let mut volumes: Vec = pod_spec.volumes.clone().unwrap_or_default(); + let mut vol_mounts: Vec = container.volume_mounts.clone().unwrap_or_default(); if add { let (volume, volume_mount) = generate_secret_volume(id); - pod_spec.volumes.get_or_insert_default().push(volume); + volumes.push(volume); vol_mounts.push(volume_mount); } else { - let vol_result = pod_spec.volumes.as_mut().and_then(|vs| { - let pos = vs.iter().position(|v| v.name == id); - pos.map(|p| vs.swap_remove(p)) - }); - if vol_result.is_none() { + let pos = volumes.iter().position(|v| v.name == id); + if let Some(p) = pos { + volumes.swap_remove(p); + } else { info!("Secret {id} was to be dropped, but volume had already been removed"); } - let vol_mount_result = container.volume_mounts.as_mut().and_then(|vms| { - let pos = vms.iter().position(|v| v.name == id); - pos.map(|p| vms.swap_remove(p)) - }); - if vol_mount_result.is_none() { + let pos = vol_mounts.iter().position(|v| v.name == id); + if let Some(p) = pos { + vol_mounts.swap_remove(p); + } else { info!("Secret {id} was to be dropped, but volume mount had already been removed"); } } - deployments - .replace(TRUSTEE_DEPLOYMENT, &Default::default(), &deployment) - .await?; + let patch = Patch::Apply(json!({ + "apiVersion": "apps/v1", + "kind": "Deployment", + "metadata": { "name": TRUSTEE_DEPLOYMENT }, + "spec": { + "template": { + "spec": { + "volumes": volumes, + "containers": [{ "name": "kbs", "volumeMounts": vol_mounts }] + } + } + } + })); + let deployments: Api = Api::default_namespaced(client.clone()); + let pp = PatchParams::apply("trusted-cluster-operator").force(); + deployments.patch(TRUSTEE_DEPLOYMENT, &pp, &patch).await?; Ok(()) } -pub async fn update_attestation_keys(ctx: &AkContextData) -> Result<()> { +pub async fn update_attestation_keys(ctx: &OperatorContext) -> Result<()> { let client = &ctx.client; let ak_secrets: Vec = ctx .secret_store @@ -223,12 +238,8 @@ pub async fn update_attestation_keys(ctx: &AkContextData) -> Result<()> { .collect(); let ns = client.default_namespace().to_string(); - let Some(deployment) = ctx - .deployment_store - .get(&ObjectRef::new(TRUSTEE_DEPLOYMENT).within(&ns)) - .map(std::sync::Arc::unwrap_or_clone) - else { - // Trustee deployment is not (yet or no longer) present — nothing to patch. + let obj_ref = ObjectRef::new(TRUSTEE_DEPLOYMENT).within(&ns); + let Some(deployment) = ctx.deployment_store.get(&obj_ref).map(Arc::unwrap_or_clone) else { info!("{TRUSTEE_DEPLOYMENT} not found in cache, skipping attestation key volume update"); return Ok(()); }; @@ -308,13 +319,10 @@ pub async fn update_attestation_keys(ctx: &AkContextData) -> Result<()> { let vol_mounts_changed = container.volume_mounts.as_ref() != Some(&vol_mounts); if volumes_changed || vol_mounts_changed { - // Patch the deployment with updated volumes and volumeMounts - let patch = json!({ + let patch = Patch::Apply(&json!({ "apiVersion": "apps/v1", "kind": "Deployment", - "metadata": { - "name": TRUSTEE_DEPLOYMENT - }, + "metadata": { "name": TRUSTEE_DEPLOYMENT }, "spec": { "template": { "spec": { @@ -326,15 +334,10 @@ pub async fn update_attestation_keys(ctx: &AkContextData) -> Result<()> { } } } - }); + })); - deployments - .patch( - TRUSTEE_DEPLOYMENT, - &PatchParams::apply("trusted-cluster-operator").force(), - &Patch::Apply(&patch), - ) - .await?; + let pp = PatchParams::apply("trusted-cluster-operator").force(); + deployments.patch(TRUSTEE_DEPLOYMENT, &pp, &patch).await?; info!("Successfully patched {TRUSTEE_DEPLOYMENT} with attestation key volumes"); } else { info!("No changes to attestation key volumes, skipping deployment update"); @@ -671,60 +674,29 @@ mod tests { #[tokio::test] async fn test_update_rvs_success() { let clos = async |req: Request<_>, ctr| match (ctr, req.method()) { - (0, &Method::GET) => { - assert!(req.uri().path().contains(PCR_CONFIG_MAP)); - Ok(serde_json::to_string(&dummy_pcrs_map()).unwrap()) - } - (1, &Method::GET) | (2, &Method::PUT) => { + (0, &Method::PATCH) => { assert!(req.uri().path().contains(TRUSTEE_DATA_MAP)); Ok(serde_json::to_string(&dummy_trustee_map()).unwrap()) } _ => panic!("unexpected API interaction: {req:?}, counter {ctr}"), }; - count_check!(3, clos, |client| { - assert!(update_reference_values(client).await.is_ok()); - }); - } - - #[tokio::test] - async fn test_update_rvs_no_pcr_map() { - let clos = async |req: Request<_>, _| match (req.uri().path(), req.method()) { - (p, &Method::GET) if p.contains(PCR_CONFIG_MAP) => Err(StatusCode::NOT_FOUND), - _ => panic!("unexpected API interaction: {req:?}"), - }; count_check!(1, clos, |client| { - assert!(update_reference_values(client).await.is_err()); + assert!(update_reference_values(&client, dummy_pcrs()).await.is_ok()); }); } #[tokio::test] async fn test_update_rvs_no_trustee_map() { - let clos = async |req: Request<_>, ctr| match (ctr, req.uri().path()) { - (0, p) if p.contains(PCR_CONFIG_MAP) => { - Ok(serde_json::to_string(&dummy_pcrs_map()).unwrap()) - } - (1, p) if p.contains(TRUSTEE_DATA_MAP) => Err(StatusCode::NOT_FOUND), - _ => panic!("unexpected API interaction: {req:?}, counter {ctr}"), - }; - count_check!(2, clos, |client| { - assert!(update_reference_values(client).await.is_err()) - }); - } - - #[tokio::test] - async fn test_update_rvs_no_trustee_data() { - let clos = async |req: Request<_>, ctr| match (ctr, req.uri().path()) { - (0, p) if p.contains(PCR_CONFIG_MAP) => { - Ok(serde_json::to_string(&dummy_pcrs_map()).unwrap()) - } - (1, p) if p.contains(TRUSTEE_DATA_MAP) => { - Ok(serde_json::to_string(&ConfigMap::default()).unwrap()) - } - _ => panic!("unexpected API interaction: {req:?}, counter {ctr}"), + let clos = async |req: Request<_>, _| match req.method() { + &Method::PATCH => Err(StatusCode::NOT_FOUND), + _ => panic!("unexpected API interaction: {req:?}"), }; - count_check!(2, clos, |client| { - let err = update_reference_values(client).await.err().unwrap(); - assert!(err.to_string().contains("but had no data")); + count_check!(1, clos, |client| { + assert!( + update_reference_values(&client, dummy_pcrs()) + .await + .is_err() + ) }); } @@ -736,6 +708,10 @@ mod tests { fn dummy_deployment() -> Deployment { Deployment { + metadata: k8s_openapi::apimachinery::pkg::apis::meta::v1::ObjectMeta { + name: Some(TRUSTEE_DEPLOYMENT.to_string()), + ..Default::default() + }, spec: Some(DeploymentSpec { replicas: Some(1), template: PodTemplateSpec { @@ -751,65 +727,74 @@ mod tests { } } + fn op_ctx_with_deployment(client: Client, deployment: Deployment) -> OperatorContext { + use kube::runtime::{reflector, watcher}; + let (deployment_store, mut writer) = reflector::store::(); + writer.apply_watcher_event(&watcher::Event::Init); + writer.apply_watcher_event(&watcher::Event::InitApply(deployment)); + writer.apply_watcher_event(&watcher::Event::InitDone); + let mut ctx = OperatorContext::new(client); + ctx.deployment_store = deployment_store; + ctx + } + #[tokio::test] async fn test_mount_secret_success() { let clos = async |req: Request<_>, ctr| match (ctr, req.method()) { - (0, &Method::GET) | (1, &Method::PUT) => { - Ok(serde_json::to_string(&dummy_deployment()).unwrap()) - } + (0, &Method::PATCH) => Ok(serde_json::to_string(&dummy_deployment()).unwrap()), _ => panic!("unexpected API interaction: {req:?}, counter {ctr}"), }; - count_check!(2, clos, |client| { - assert!(mount_secret(client, "id").await.is_ok()); + count_check!(1, clos, |client| { + let ctx = op_ctx_with_deployment(client, dummy_deployment()); + assert!(mount_secret(&ctx, "id").await.is_ok()); }); } #[tokio::test] async fn test_mount_secret_no_depl() { - let clos = async |_, _| Err(StatusCode::NOT_FOUND); - count_check!(1, clos, |client| { - assert!(mount_secret(client, "id").await.is_err()); + let clos = async |req: Request<_>, _| panic!("unexpected API interaction: {req:?}"); + count_check!(0, clos, |client| { + let ctx = OperatorContext::new(client); + // No deployment in cache → silently succeeds (skip) + assert!(mount_secret(&ctx, "id").await.is_ok()); }); } #[tokio::test] async fn test_mount_secret_no_spec() { - let clos = async |_, _| { + let clos = async |req: Request<_>, _| panic!("unexpected API interaction: {req:?}"); + count_check!(0, clos, |client| { let mut depl = dummy_deployment(); depl.spec = None; - Ok(serde_json::to_string(&depl).unwrap()) - }; - count_check!(1, clos, |client| { - let err = mount_secret(client, "id").await.err().unwrap(); + let ctx = op_ctx_with_deployment(client, depl); + let err = mount_secret(&ctx, "id").await.err().unwrap(); assert!(err.to_string().contains("but had no spec")); }); } #[tokio::test] async fn test_mount_secret_no_pod_spec() { - let clos = async |_, _| { + let clos = async |req: Request<_>, _| panic!("unexpected API interaction: {req:?}"); + count_check!(0, clos, |client| { let mut depl = dummy_deployment(); let spec = depl.spec.as_mut().unwrap(); spec.template.spec = None; - Ok(serde_json::to_string(&depl).unwrap()) - }; - count_check!(1, clos, |client| { - let err = mount_secret(client, "id").await.err().unwrap(); + let ctx = op_ctx_with_deployment(client, depl); + let err = mount_secret(&ctx, "id").await.err().unwrap(); assert!(err.to_string().contains("but had no pod spec")); }); } #[tokio::test] async fn test_mount_secret_no_containers() { - let clos = async |_, _| { + let clos = async |req: Request<_>, _| panic!("unexpected API interaction: {req:?}"); + count_check!(0, clos, |client| { let mut depl = dummy_deployment(); let spec = depl.spec.as_mut().unwrap(); let pod_spec = spec.template.spec.as_mut().unwrap(); pod_spec.containers = vec![]; - Ok(serde_json::to_string(&depl).unwrap()) - }; - count_check!(1, clos, |client| { - let err = mount_secret(client, "id").await.err().unwrap(); + let ctx = op_ctx_with_deployment(client, depl); + let err = mount_secret(&ctx, "id").await.err().unwrap(); assert!(err.to_string().contains("but had no containers")); }); } @@ -817,31 +802,29 @@ mod tests { #[tokio::test] async fn test_unmount_secret() { let clos = async |req: Request, ctr| match (ctr, req.method()) { - (0, &Method::GET) => { - let mut depl = dummy_deployment(); - let spec = depl.spec.as_mut().unwrap(); - let pod_spec = spec.template.spec.as_mut().unwrap(); - pod_spec.volumes = Some(vec![Volume { - name: "id".to_string(), - ..Default::default() - }]); - let container = pod_spec.containers.get_mut(0).unwrap(); - container.volume_mounts = Some(vec![VolumeMount { - name: "id".to_string(), - ..Default::default() - }]); - Ok(serde_json::to_string(&depl).unwrap()) - } - (1, &Method::PUT) => { + (0, &Method::PATCH) => { let bytes = req.into_body().collect_bytes().await.unwrap().to_vec(); let body = String::from_utf8_lossy(&bytes); - assert!(!body.contains("id")); + assert!(!body.contains("\"id\"")); Ok(serde_json::to_string(&dummy_deployment()).unwrap()) } _ => panic!("unexpected API interaction: {req:?}, counter {ctr}"), }; - count_check!(2, clos, |client| { - assert!(unmount_secret(client, "id").await.is_ok()); + count_check!(1, clos, |client| { + let mut depl = dummy_deployment(); + let spec = depl.spec.as_mut().unwrap(); + let pod_spec = spec.template.spec.as_mut().unwrap(); + pod_spec.volumes = Some(vec![Volume { + name: "id".to_string(), + ..Default::default() + }]); + let container = pod_spec.containers.get_mut(0).unwrap(); + container.volume_mounts = Some(vec![VolumeMount { + name: "id".to_string(), + ..Default::default() + }]); + let ctx = op_ctx_with_deployment(client, depl); + assert!(unmount_secret(&ctx, "id").await.is_ok()); }); } From ac86d1141dca8c7e5cc892a8ad598431324e4078 Mon Sep 17 00:00:00 2001 From: Jakob Naucke Date: Thu, 9 Jul 2026 12:38:23 +0200 Subject: [PATCH 02/10] rvs: Reconcile images on jobs too so that status is updated in a timely manner Signed-off-by: Jakob Naucke --- operator/src/reference_values.rs | 3 +++ 1 file changed, 3 insertions(+) diff --git a/operator/src/reference_values.rs b/operator/src/reference_values.rs index 16c8182b..abe5db54 100644 --- a/operator/src/reference_values.rs +++ b/operator/src/reference_values.rs @@ -347,8 +347,11 @@ async fn image_remove_reconcile( pub async fn launch_rv_image_controller(ctx: Arc) { let images: Api = Api::default_namespaced(ctx.client.clone()); + let jobs: Api = Api::default_namespaced(ctx.client.clone()); + let wc = watcher::Config::default().labels(&format!("{JOB_LABEL_KEY}={PCR_COMMAND_NAME}")); tokio::spawn( Controller::new(images, Default::default()) + .owns(jobs, wc) .run(image_reconcile, controller_error_policy, ctx) .for_each(controller_info), ); From fd031384ea88050fb9c7adb493ceff7e104bc0a3 Mon Sep 17 00:00:00 2001 From: Jakob Naucke Date: Wed, 19 Aug 2026 13:12:06 +0200 Subject: [PATCH 03/10] rvs: Refactor JOB_LABEL_KEY to generic const for reuse Signed-off-by: Jakob Naucke --- operator/src/lib.rs | 1 + operator/src/reference_values.rs | 9 ++++----- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/operator/src/lib.rs b/operator/src/lib.rs index 26bea843..1b8c4280 100644 --- a/operator/src/lib.rs +++ b/operator/src/lib.rs @@ -103,6 +103,7 @@ macro_rules! create_or_info_if_exists { }; } +pub const KIND_LABEL_KEY: &str = "kind"; pub const TLS_DIR: &str = "/etc/tls"; /// As per kube-rs docs, it's possible to miss events and requeue_after = None should only be used /// when it is known another requeue is imminent. Use this requeue duration for cases where no diff --git a/operator/src/reference_values.rs b/operator/src/reference_values.rs index abe5db54..d1678c7f 100644 --- a/operator/src/reference_values.rs +++ b/operator/src/reference_values.rs @@ -29,11 +29,10 @@ use std::{collections::BTreeMap, sync::Arc, time::Duration}; use crate::COMPONENT_VERSION; use crate::trustee::{self, get_image_pcrs}; -use operator::{ControllerError, LONG_REQUEUE, OperatorContext, upsert_condition}; +use operator::{ControllerError, KIND_LABEL_KEY, LONG_REQUEUE, OperatorContext, upsert_condition}; use operator::{controller_error_policy, controller_info, create_or_info_if_exists}; use trusted_cluster_operator_lib::{conditions::*, reference_values::*, *}; -const JOB_LABEL_KEY: &str = "kind"; const APPROVED_IMAGE_ANNOTATION: &str = "approved-image"; const PCR_COMMAND_NAME: &str = "compute-pcrs"; const PCR_LABEL: &str = "org.coreos.pcrs"; @@ -145,7 +144,7 @@ async fn job_reconcile( pub async fn launch_rv_job_controller(ctx: Arc) { let jobs: Api = Api::default_namespaced(ctx.client.clone()); let watcher = watcher::Config { - label_selector: Some(format!("{JOB_LABEL_KEY}={PCR_COMMAND_NAME}")), + label_selector: Some(format!("{KIND_LABEL_KEY}={PCR_COMMAND_NAME}")), ..Default::default() }; tokio::spawn( @@ -181,7 +180,7 @@ async fn compute_fresh_pcrs(client: Client, image: &ApprovedImage) -> anyhow::Re metadata: ObjectMeta { name: Some(job_name.clone()), labels: Some(BTreeMap::from([( - JOB_LABEL_KEY.to_string(), + KIND_LABEL_KEY.to_string(), PCR_COMMAND_NAME.to_string(), )])), owner_references: Some(vec![generate_owner_reference(image)?]), @@ -348,7 +347,7 @@ async fn image_remove_reconcile( pub async fn launch_rv_image_controller(ctx: Arc) { let images: Api = Api::default_namespaced(ctx.client.clone()); let jobs: Api = Api::default_namespaced(ctx.client.clone()); - let wc = watcher::Config::default().labels(&format!("{JOB_LABEL_KEY}={PCR_COMMAND_NAME}")); + let wc = watcher::Config::default().labels(&format!("{KIND_LABEL_KEY}={PCR_COMMAND_NAME}")); tokio::spawn( Controller::new(images, Default::default()) .owns(jobs, wc) From 8b7fe38896a005b3ffbeabd1b24a8cabff2d1cc6 Mon Sep 17 00:00:00 2001 From: Jakob Naucke Date: Wed, 19 Aug 2026 13:12:51 +0200 Subject: [PATCH 04/10] operator/ak-reg: Limit secret watch to labelled Add a label to secrets and watch only by the label Signed-off-by: Jakob Naucke --- operator/src/attestation_key_register.rs | 28 ++++++++++-------------- 1 file changed, 12 insertions(+), 16 deletions(-) diff --git a/operator/src/attestation_key_register.rs b/operator/src/attestation_key_register.rs index 8d4c628a..313ec6c9 100644 --- a/operator/src/attestation_key_register.rs +++ b/operator/src/attestation_key_register.rs @@ -27,12 +27,15 @@ use trusted_cluster_operator_lib::{AttestationKey, AttestationKeyStatus, Machine use crate::conditions::attestation_key_approved_condition; use crate::trustee; -use operator::{ControllerError, LONG_REQUEUE, OperatorContext, TLS_DIR, controller_error_policy}; -use operator::{create_or_info_if_exists, read_certificate, upsert_condition}; +use operator::{ + ControllerError, KIND_LABEL_KEY, LONG_REQUEUE, OperatorContext, TLS_DIR, + controller_error_policy, create_or_info_if_exists, read_certificate, upsert_condition, +}; const INTERNAL_ATTESTATION_KEY_REGISTER_PORT: i32 = 8001; const ATTESTATION_KEY_SECRET_FINALIZER: &str = "trusted-execution-clusters.io/attestationkey-secret-finalizer"; +const ATTESTATION_KEY_LABEL_VALUE: &str = "attestationkey"; pub async fn create_attestation_key_register_deployment( client: Client, @@ -238,6 +241,10 @@ async fn approve_ak(ak: &AttestationKey, machine: &Machine, ctx: &OperatorContex let secret = Secret { metadata: ObjectMeta { name: Some(secret_name.clone()), + labels: Some(BTreeMap::from([( + KIND_LABEL_KEY.to_string(), + ATTESTATION_KEY_LABEL_VALUE.to_string(), + )])), owner_references: Some(vec![owner_reference]), finalizers: Some(vec![ATTESTATION_KEY_SECRET_FINALIZER.to_string()]), ..Default::default() @@ -258,19 +265,6 @@ async fn secret_reconcile( ctx: Arc, ) -> Result { let secret_name = secret.metadata.name.clone().unwrap_or_default(); - - // Only handle secrets owned by AttestationKey - let is_ak_secret = secret - .metadata - .owner_references - .as_ref() - .map(|owners| owners.iter().any(|owner| owner.kind == "AttestationKey")) - .unwrap_or(false); - - if !is_ak_secret { - return Ok(Action::await_change()); - } - info!("Secret reconciliation for AttestationKey secret: {secret_name}"); let secrets: Api = Api::default_namespaced(ctx.client.clone()); @@ -337,8 +331,10 @@ pub async fn launch_machine_ak_controller(ctx: Arc) { pub async fn launch_secret_ak_controller(ctx: Arc) { let secrets: Api = Api::default_namespaced(ctx.client.clone()); + let wc = watcher::Config::default() + .labels(&format!("{KIND_LABEL_KEY}={ATTESTATION_KEY_LABEL_VALUE}")); tokio::spawn( - Controller::new(secrets, watcher::Config::default()) + Controller::new(secrets, wc) .run(secret_reconcile, controller_error_policy, ctx) .for_each(|res| async move { match res { From a9c2056b562226664fbb5af40aa22ca3105c7035 Mon Sep 17 00:00:00 2001 From: Jakob Naucke Date: Wed, 19 Aug 2026 13:17:10 +0200 Subject: [PATCH 05/10] rvs: Add a missing LONG_REQUEUE Signed-off-by: Jakob Naucke --- operator/src/reference_values.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/operator/src/reference_values.rs b/operator/src/reference_values.rs index d1678c7f..cdb92dad 100644 --- a/operator/src/reference_values.rs +++ b/operator/src/reference_values.rs @@ -138,7 +138,7 @@ async fn job_reconcile( delete.map_err(Into::::into)?; let image_pcrs = cached_image_pcrs(&ctx)?; trustee::update_reference_values(&ctx.client, image_pcrs).await?; - Ok(Action::await_change()) + Ok(LONG_REQUEUE) } pub async fn launch_rv_job_controller(ctx: Arc) { @@ -542,7 +542,7 @@ mod tests { let ctx = Arc::new(op_ctx_with_cm(client, PCR_CONFIG_MAP, dummy_pcrs_map())); let job = Arc::new(dummy_job()); let result = job_reconcile(job, ctx).await.unwrap(); - assert_eq!(result, Action::await_change()); + assert_eq!(result, LONG_REQUEUE); }); } From dbf29d8c7e350201c72b1c647c5d0ac4c650f264 Mon Sep 17 00:00:00 2001 From: Jakob Naucke Date: Thu, 6 Aug 2026 18:27:58 +0200 Subject: [PATCH 06/10] Set kube-client read timeout Set read timeout of 295s (same as write) so that hanging operations can retry. Signed-off-by: Jakob Naucke --- operator/src/main.rs | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/operator/src/main.rs b/operator/src/main.rs index 0397d0a9..7a688c32 100644 --- a/operator/src/main.rs +++ b/operator/src/main.rs @@ -48,6 +48,9 @@ const COMPONENT_VERSION: &str = match option_env!("COMPONENT_VERSION") { /// Default registry const TEC_REGISTRY: &str = "quay.io/trusted-execution-clusters"; +/// Keep a read timeout to allow hanging operations to retry (same as write timeout). This breaks +/// exec/attach operations without traffic for more than 5 minutes, but we do not use those. +const KUBE_READ_TIMEOUT: Duration = Duration::from_secs(295); fn is_installed(status: Option) -> bool { let chk = |c: &Condition| c.type_ == INSTALLED_CONDITION && c.status == "True"; @@ -235,7 +238,9 @@ async fn install_attestation_key_register( async fn main() -> Result<()> { env_logger::Builder::from_env(Env::default().default_filter_or("info")).init(); - let kube_client = Client::try_default().await?; + let mut config = kube::Config::infer().await?; + config.read_timeout = Some(KUBE_READ_TIMEOUT); + let kube_client = Client::try_from(config)?; info!("trusted execution clusters operator"); const CACHE_SYNC_TIMEOUT: Duration = Duration::from_secs(60); From 85dcabf6d48e26e616c5e2c0e45565a56896bc50 Mon Sep 17 00:00:00 2001 From: Jakob Naucke Date: Wed, 19 Aug 2026 13:25:14 +0200 Subject: [PATCH 07/10] reg-srv: Retry enough for a kube timeout to hit Signed-off-by: Jakob Naucke --- register-server/src/main.rs | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/register-server/src/main.rs b/register-server/src/main.rs index 6db836b2..10b3396a 100644 --- a/register-server/src/main.rs +++ b/register-server/src/main.rs @@ -10,7 +10,7 @@ use axum::{http::StatusCode, routing::get, Router}; use axum_server::tls_openssl::OpenSSLConfig; use clap::Parser; use clevis_pin_trustee_lib::{ - AttestationKey, Config as ClevisConfig, Registration, Server as ClevisServer, + AttestationKey, Config as ClevisConfig, NumRetries, Registration, Server as ClevisServer, }; use env_logger::Env; use ignition_config::v3_6::{ @@ -28,6 +28,10 @@ use trusted_cluster_operator_lib::{ generate_owner_reference, get_trusted_execution_cluster, Machine, MachineSpec, }; +/// Allow for a operator::KUBE_READ_TIMEOUT to hit (5 minutes) plus one minute, +/// thus 360s / 5s (clevis-pin-trustee's delay) +const RETRIES: u32 = 72; + #[derive(Parser)] #[command(name = "register-server")] #[command(about = "HTTP server that generates Clevis PINs with random UUIDs")] @@ -119,7 +123,8 @@ fn generate_ignition(id: &str, endpoint_info: &EndpointInfo) -> IgnitionConfig { cert: trustee_cert, }], path: format!("default/{id}/root"), - num_retries: None, + // TODO retry forever once we don't need a debugging shell + num_retries: Some(NumRetries::Finite(RETRIES)), initdata: None, // TODO add initdata, e.g. // #[derive(Serialize)] From 413360950d93f041894ffc9e7901580746d880e0 Mon Sep 17 00:00:00 2001 From: Jakob Naucke Date: Tue, 18 Aug 2026 16:25:09 +0200 Subject: [PATCH 08/10] gha: Set test timeout multiplier > read timeout Local network on GHA has been seen to be less reliable than local development and production clusters. Set a test timeout multiplier large enough to let one read time out and retry. Signed-off-by: Jakob Naucke --- .github/workflows/integration-tests.yml | 2 ++ 1 file changed, 2 insertions(+) diff --git a/.github/workflows/integration-tests.yml b/.github/workflows/integration-tests.yml index d30dcc0b..6d1132f4 100644 --- a/.github/workflows/integration-tests.yml +++ b/.github/workflows/integration-tests.yml @@ -77,6 +77,8 @@ jobs: - name: "Run integration tests" run: | eval $(ssh-agent -s) + # Lowest timeout is 1 minute, allow for one read timeout (5 minutes) retry + export TEST_TIMEOUT_MULTIPLIER=6 make integration-tests - name: "Gather must-gather" if: always() From 6e12530b2d68775932079989f4fcb4f8906de9b4 Mon Sep 17 00:00:00 2001 From: Jakob Naucke Date: Thu, 16 Jul 2026 20:04:29 +0200 Subject: [PATCH 09/10] kv: Forward console Signed-off-by: Jakob Naucke --- Makefile | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/Makefile b/Makefile index 0a65d4dc..75efc7d5 100644 --- a/Makefile +++ b/Makefile @@ -54,7 +54,8 @@ COMPUTE_PCRS_IMAGE=$(REGISTRY)/compute-pcrs:$(TAG) REG_SERVER_IMAGE=$(REGISTRY)/registration-server:$(TAG) ATTESTATION_KEY_REGISTER_IMAGE=$(REGISTRY)/attestation-key-register:$(TAG) TRUSTEE_IMAGE ?= quay.io/trusted-execution-clusters/key-broker-service:v0.17.0 -TEST_IMAGE ?= quay.io/trusted-execution-clusters/fedora-coreos-kubevirt:42.20260622 +TEST_IMAGE ?= quay.io/trusted-execution-clusters/fedora-coreos-kubevirt:42.20251012.2.0-console-fwd + # tagged as 42.20251012.2.0 APPROVED_IMAGE ?= quay.io/trusted-execution-clusters/fedora-coreos@sha256:6997f51fd27d1be1b5fc2e6cc3ebf16c17eb94d819b5d44ea8d6cf5f826ee773 From 6c1745b5ee895e6c8aae6acb35983439dbdd0961 Mon Sep 17 00:00:00 2001 From: Jakob Naucke Date: Tue, 28 Jul 2026 13:57:56 +0200 Subject: [PATCH 10/10] gha: repeat ad inf Signed-off-by: Jakob Naucke --- .github/workflows/integration-tests.yml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/workflows/integration-tests.yml b/.github/workflows/integration-tests.yml index 6d1132f4..847bb6dd 100644 --- a/.github/workflows/integration-tests.yml +++ b/.github/workflows/integration-tests.yml @@ -32,7 +32,6 @@ jobs: github.event.pull_request.author_association == 'COLLABORATOR' || contains(github.event.pull_request.labels.*.name, 'ok-to-test') runs-on: "ubuntu-24.04" - timeout-minutes: 120 steps: - name: "Check out repository" uses: actions/checkout@v7 @@ -79,7 +78,8 @@ jobs: eval $(ssh-agent -s) # Lowest timeout is 1 minute, allow for one read timeout (5 minutes) retry export TEST_TIMEOUT_MULTIPLIER=6 - make integration-tests + while make integration-tests; do :; done + exit 1 - name: "Gather must-gather" if: always() run: must-gather/gather