Use reflector for all repeated read operations, set read timeout & a test timeout greater than that, CI stress edition

.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()

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<ApprovedImage> = Api::default_namespaced(client);
     let image = approved_images.get(&args.resource_name).await?;

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<TrustedExecutionCluster> {
     let namespace = client.default_namespace().to_string();
     let cluster = get_opt_trusted_execution_cluster(client).await;

lib/src/reference_values.rs

@@ -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

@@ -13,74 +13,29 @@ 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::*;
 use trusted_cluster_operator_lib::{AttestationKey, AttestationKeyStatus, Machine, update_status};
 
 use crate::conditions::attestation_key_approved_condition;
 use crate::trustee;
-use operator::{ControllerError, LONG_REQUEUE, 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<Machine>,
-    pub ak_store: Store<AttestationKey>,
-    pub secret_store: Store<Secret>,
-    pub deployment_store: Store<Deployment>,
-}
-
-impl AkContextData {
-    pub fn new(client: Client) -> Self {
-        let (machine_store, machine_writer) = reflector::store::<Machine>();
-        let (ak_store, ak_writer) = reflector::store::<AttestationKey>();
-        let (secret_store, secret_writer) = reflector::store::<Secret>();
-        let (deployment_store, deployment_writer) = reflector::store::<Deployment>();
-
-        crate::spawn_reflector::<Machine>(machine_writer, client.clone(), "Machine");
-        crate::spawn_reflector::<AttestationKey>(ak_writer, client.clone(), "AttestationKey");
-        crate::spawn_reflector::<Secret>(secret_writer, client.clone(), "Secret");
-        crate::spawn_reflector::<Deployment>(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(())
-    }
-}
+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,
@@ -185,7 +140,7 @@ pub async fn create_attestation_key_register_service(
 
 async fn ak_reconcile(
     ak: Arc<AttestationKey>,
-    ctx: Arc<AkContextData>,
+    ctx: Arc<OperatorContext>,
 ) -> Result<Action, ControllerError> {
     let ak_name = ak.metadata.name.clone().unwrap_or_default();
     info!("Attestation Key reconciliation for: {ak_name}");
@@ -201,7 +156,7 @@ async fn ak_reconcile(
 
 async fn machine_reconcile(
     machine: Arc<Machine>,
-    ctx: Arc<AkContextData>,
+    ctx: Arc<OperatorContext>,
 ) -> Result<Action, ControllerError> {
     info!(
         "Machine reconciliation for: {}",
@@ -228,7 +183,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<AttestationKey> = Api::default_namespaced(client.clone());
@@ -274,10 +229,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());
@@ -288,6 +241,10 @@ async fn approve_ak(ak: &AttestationKey, machine: &Machine, ctx: &AkContextData)
         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()
@@ -305,22 +262,9 @@ async fn approve_ak(ak: &AttestationKey, machine: &Machine, ctx: &AkContextData)
 
 async fn secret_reconcile(
     secret: Arc<Secret>,
-    ctx: Arc<AkContextData>,
+    ctx: Arc<OperatorContext>,
 ) -> Result<Action, ControllerError> {
     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<Secret> = Api::default_namespaced(ctx.client.clone());
@@ -357,7 +301,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<AkContextData>) {
+pub async fn launch_ak_controller(ctx: Arc<OperatorContext>) {
     let aks: Api<AttestationKey> = Api::default_namespaced(ctx.client.clone());
     tokio::spawn(
         Controller::new(aks, watcher::Config::default())
@@ -371,7 +315,7 @@ pub async fn launch_ak_controller(ctx: Arc<AkContextData>) {
     );
 }
 
-pub async fn launch_machine_ak_controller(ctx: Arc<AkContextData>) {
+pub async fn launch_machine_ak_controller(ctx: Arc<OperatorContext>) {
     let machines: Api<Machine> = Api::default_namespaced(ctx.client.clone());
     tokio::spawn(
         Controller::new(machines, watcher::Config::default())
@@ -385,10 +329,12 @@ pub async fn launch_machine_ak_controller(ctx: Arc<AkContextData>) {
     );
 }
 
-pub async fn launch_secret_ak_controller(ctx: Arc<AkContextData>) {
+pub async fn launch_secret_ak_controller(ctx: Arc<OperatorContext>) {
     let secrets: Api<Secret> = 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 {

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<TrustedExecutionCluster>,
+    pub cm_store: Store<ConfigMap>,
+    pub deployment_store: Store<Deployment>,
+    pub machine_store: Store<Machine>,
+    pub ak_store: Store<AttestationKey>,
+    pub secret_store: Store<Secret>,
+    pub image_store: Store<ApprovedImage>,
+}
+
+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<Option<TrustedExecutionCluster>> {
+        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 {
@@ -58,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