Skip to content

Use a read-timeout client for non-watch operations #324

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
Jakob-Naucke wants to merge 7 commits into from
Closed

Use a read-timeout client for non-watch operations #324

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
d063668
operator/ak-reg: Use warn!, not eprintln!
Jakob-Naucke Jul 13, 2026
2301527
operator/ak-reg: Fix a log message
Jakob-Naucke Jul 29, 2026
c175d35
Add timestamps to logging
Jakob-Naucke Jul 13, 2026
935cca4
gha: run & upload must-gather
Jakob-Naucke Jul 15, 2026
f7fef12
rvs: Reconcile images on jobs too
Jakob-Naucke Jul 9, 2026
7ed1df5
Use a read-timeout client for non-watch operations
Jakob-Naucke Jul 29, 2026
1ec63fe
operator/reg: Pass error if cleanup fails
Jakob-Naucke Jul 29, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 11 additions & 0 deletions .github/workflows/integration-tests.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -76,3 +76,14 @@ jobs:
run: |
eval $(ssh-agent -s)
make integration-tests
- name: "Gather must-gather"
if: always()
run: must-gather/gather
env:
COLLECTION_PATH: must-gather-output
- name: "Upload must-gather"
if: always()
uses: actions/upload-artifact@v7
with:
name: must-gather-${GITHUB_HEAD_REF}
path: must-gather-output/
Comment thread
SpaceFace02 marked this conversation as resolved.
2 changes: 1 addition & 1 deletion Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@ chrono = "0.4.45"
clap = { version = "4.6.1", features = ["derive"] }
clevis-pin-trustee-lib = { git = "https://github.com/latchset/clevis-pin-trustee" }
compute-pcrs-lib = { git = "https://github.com/trusted-execution-clusters/compute-pcrs" }
env_logger = { version = "0.11.10", default-features = false }
env_logger = { version = "0.11.10", default-features = false, features = ["humantime"] }
http = "1.4.2"
ignition-config = "0.6.1"
k8s-openapi = { version = "0.28.0", features = ["v1_35", "schemars"] }
Expand Down
5 changes: 3 additions & 2 deletions attestation-key-register/src/main.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,8 @@ use uuid::Uuid;

use trusted_cluster_operator_lib::endpoints::ATTESTATION_KEY_REGISTER_RESOURCE;
use trusted_cluster_operator_lib::{
generate_owner_reference, get_trusted_execution_cluster, AttestationKey, AttestationKeySpec,
generate_owner_reference, get_trusted_execution_cluster, timed_client, AttestationKey,
AttestationKeySpec,
};

#[derive(Parser)]
Expand Down Expand Up @@ -131,7 +132,7 @@ async fn main() {
let args = Args::parse();
let endpoint = format!("/{ATTESTATION_KEY_REGISTER_RESOURCE}");
let err = "failed to create Kubernetes client";
let client = Client::try_default().await.expect(err);
let client = timed_client().await.expect(err);
let app = Router::new()
.route(&endpoint, put(handle_registration))
.with_state(client);
Expand Down
4 changes: 2 additions & 2 deletions compute-pcrs/src/main.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@ use anyhow::{Context, Result, anyhow};
use clap::Parser;
use compute_pcrs_lib::*;
use k8s_openapi::{api::core::v1::ConfigMap, jiff::Timestamp};
use kube::{Api, Client};
use kube::Api;
use std::{fs::File, io::Read};

use trusted_cluster_operator_lib::{conditions::INSTALLED_REASON, reference_values::*, *};
Expand Down Expand Up @@ -52,7 +52,7 @@ async fn main() -> Result<()> {
compute_pcr14(&mokvars),
];

let client = Client::try_default().await?;
let client = timed_client().await?;
let config_maps: Api<ConfigMap> = Api::default_namespaced(client.clone());

let mut image_pcrs_map = config_maps.get(PCR_CONFIG_MAP).await?;
Expand Down
25 changes: 25 additions & 0 deletions lib/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,31 @@ use anyhow::{Context, Result, anyhow};
use conditions::*;
use k8s_openapi::apimachinery::pkg::apis::meta::v1::{Condition, OwnerReference, Time};
use kube::{Api, Client, Resource};
use std::time::Duration;

/// Read timeout for non-watch Kubernetes API calls.
pub const KUBE_READ_TIMEOUT: Duration = Duration::from_secs(30);

/// Create a [`Client`] with [`KUBE_READ_TIMEOUT`] applied.
///
/// Use this for request/response API calls. For long-lived watch streams, use
/// [`Client::try_default`] instead so the connection never times out on idle.
pub async fn timed_client() -> Result<Client> {
let mut config = kube::Config::infer().await?;
config.read_timeout = Some(KUBE_READ_TIMEOUT);
Ok(Client::try_from(config)?)
}

/// A pair of Kubernetes clients
///
/// `client` has a read timeout for request/response API calls.
/// `watch` has no read timeout, keeping it safe for long-lived watch streams
/// (Controllers, Reflectors, `await_condition`).
Comment on lines +48 to +49

@alicefr alicefr Aug 6, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

can you please document when one should use one or the other, it isn't very clear from the comment

#[derive(Clone)]
pub struct KubeClients {
pub request_client: Client,
pub watch_client: Client,
}

#[macro_export]
macro_rules! update_status {
Expand Down
35 changes: 18 additions & 17 deletions operator/src/attestation_key_register.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,13 +25,13 @@ use kube::{
watcher,
},
};
use log::info;
use log::{info, warn};
use serde_json::json;
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, KubeClients, Machine};
use trusted_cluster_operator_lib::{endpoints::*, update_status};

use crate::conditions::attestation_key_approved_condition;
use crate::trustee;
Expand All @@ -44,26 +44,29 @@ use operator::{
/// Stores give local cache access to avoid repeated API-server reads.
pub struct AkContextData {
pub client: Client,
watch_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 {
pub fn new(clients: &KubeClients) -> 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");
let watch = &clients.watch_client;
crate::spawn_reflector::<Machine>(machine_writer, watch.clone(), "Machine");
crate::spawn_reflector::<AttestationKey>(ak_writer, watch.clone(), "AttestationKey");
crate::spawn_reflector::<Secret>(secret_writer, watch.clone(), "Secret");
crate::spawn_reflector::<Deployment>(deployment_writer, watch.clone(), "Deployment");

Self {
client,
client: clients.request_client.clone(),
watch_client: watch.clone(),
machine_store,
ak_store,
secret_store,
Expand Down Expand Up @@ -213,7 +216,7 @@ async fn machine_reconcile(
// Check if the machine is being deleted
if machine.metadata.deletion_timestamp.is_some() {
info!(
"Machine {} is being deleted, updating attestation key volumes",
"Machine {} is being deleted, skipping update of attestation key volumes",
Comment thread
SpaceFace02 marked this conversation as resolved.
machine.metadata.name.clone().unwrap_or_default()
);
return Ok(Action::await_change());
Expand Down Expand Up @@ -335,7 +338,7 @@ async fn secret_reconcile(
.await
.map(|_| Action::await_change())
.map_err(|e| {
eprintln!("Error updating attestation key volumes on secret apply: {e}");
warn!("Error updating attestation key volumes on secret apply: {e}");
finalizer::Error::<ControllerError>::ApplyFailed(e.into())
})
}
Expand All @@ -349,9 +352,7 @@ async fn secret_reconcile(
.await
.map(|_| Action::await_change())
.map_err(|e| {
eprintln!(
"Error updating attestation key volumes during secret deletion: {e}"
);
warn!("Error updating attestation key volumes during secret deletion: {e}");
finalizer::Error::<ControllerError>::CleanupFailed(e.into())
})
}
Expand All @@ -362,7 +363,7 @@ async fn secret_reconcile(
}

pub async fn launch_ak_controller(ctx: Arc<AkContextData>) {
let aks: Api<AttestationKey> = Api::default_namespaced(ctx.client.clone());
let aks: Api<AttestationKey> = Api::default_namespaced(ctx.watch_client.clone());
tokio::spawn(
Controller::new(aks, watcher::Config::default())
.run(ak_reconcile, controller_error_policy, ctx)
Expand All @@ -376,7 +377,7 @@ pub async fn launch_ak_controller(ctx: Arc<AkContextData>) {
}

pub async fn launch_machine_ak_controller(ctx: Arc<AkContextData>) {
let machines: Api<Machine> = Api::default_namespaced(ctx.client.clone());
let machines: Api<Machine> = Api::default_namespaced(ctx.watch_client.clone());
tokio::spawn(
Controller::new(machines, watcher::Config::default())
.run(machine_reconcile, controller_error_policy, ctx)
Expand All @@ -390,7 +391,7 @@ pub async fn launch_machine_ak_controller(ctx: Arc<AkContextData>) {
}

pub async fn launch_secret_ak_controller(ctx: Arc<AkContextData>) {
let secrets: Api<Secret> = Api::default_namespaced(ctx.client.clone());
let secrets: Api<Secret> = Api::default_namespaced(ctx.watch_client.clone());
tokio::spawn(
Controller::new(secrets, watcher::Config::default())
.run(secret_reconcile, controller_error_policy, ctx)
Expand Down
32 changes: 19 additions & 13 deletions operator/src/main.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,8 +18,10 @@ 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 trusted_cluster_operator_lib::{
KubeClients, TrustedExecutionCluster, TrustedExecutionClusterStatus, conditions::*, images::*,
timed_client, update_status,
};

mod attestation_key_register;
mod conditions;
Expand All @@ -45,15 +47,16 @@ struct ClusterContext {
}

impl ClusterContext {
fn new(client: Client) -> Self {
fn new(clients: &KubeClients) -> Self {
let (tec_store, tec_writer) = reflector::store::<TrustedExecutionCluster>();

spawn_reflector::<TrustedExecutionCluster>(
tec_writer,
client.clone(),
clients.watch_client.clone(),
"TrustedExecutionCluster",
);

let client = clients.request_client.clone();
Self { client, tec_store }
}

Expand Down Expand Up @@ -248,19 +251,22 @@ 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 kube_client = timed_client().await?;
let watch_client = Client::try_default().await?;
let clients = KubeClients {
request_client: kube_client.clone(),
watch_client: watch_client.clone(),
};
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;
register_server::launch_keygen_controller(clients.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()));
let ak_ctx = Arc::new(attestation_key_register::AkContextData::new(&clients));
let ctx = Arc::new(ClusterContext::new(&clients));

// Best-effort wait for caches; controllers will work with
// partially-filled stores if the sync times out.
Expand All @@ -273,14 +279,14 @@ async fn main() -> Result<()> {

info!("Starting controllers");

let cl: Api<TrustedExecutionCluster> = Api::default_namespaced(kube_client.clone());
let cl: Api<TrustedExecutionCluster> = Api::default_namespaced(watch_client);

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;
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(clients.clone()).await;
reference_values::launch_rv_job_controller(clients).await;

Controller::new(cl, watcher::Config::default())
.run(reconcile, controller_error_policy, ctx)
Expand Down
17 changes: 11 additions & 6 deletions operator/src/reference_values.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -133,15 +133,16 @@ async fn job_reconcile(job: Arc<Job>, client: Arc<Client>) -> Result<Action, Con
Ok(Action::await_change())
}

pub async fn launch_rv_job_controller(client: Client) {
let jobs: Api<Job> = Api::default_namespaced(client.clone());
pub async fn launch_rv_job_controller(clients: KubeClients) {
let jobs: Api<Job> = Api::default_namespaced(clients.watch_client);
let watcher = watcher::Config {
label_selector: Some(format!("{JOB_LABEL_KEY}={PCR_COMMAND_NAME}")),
..Default::default()
};
let client = Arc::new(clients.request_client);
tokio::spawn(
Controller::new(jobs, watcher)
.run(job_reconcile, controller_error_policy, Arc::new(client))
.run(job_reconcile, controller_error_policy, client)
.for_each(controller_info),
);
}
Expand Down Expand Up @@ -339,11 +340,15 @@ async fn image_remove_reconcile(
Ok(Action::await_change())
}

pub async fn launch_rv_image_controller(client: Client) {
let images: Api<ApprovedImage> = Api::default_namespaced(client.clone());
pub async fn launch_rv_image_controller(clients: KubeClients) {
let images: Api<ApprovedImage> = Api::default_namespaced(clients.watch_client.clone());
let jobs: Api<Job> = Api::default_namespaced(clients.watch_client);
let wc = watcher::Config::default().labels(&format!("{JOB_LABEL_KEY}={PCR_COMMAND_NAME}"));
let client = Arc::new(clients.request_client);
tokio::spawn(
Controller::new(images, Default::default())
.run(image_reconcile, controller_error_policy, Arc::new(client))
.owns(jobs, wc)
.run(image_reconcile, controller_error_policy, client)
.for_each(controller_info),
);
}
Expand Down
15 changes: 11 additions & 4 deletions operator/src/register_server.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,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::{KubeClients, Machine, TrustedExecutionCluster, endpoints::*};

/// Finalizer name to discard decryption keys when a machine is deleted
const MACHINE_FINALIZER: &str = "finalizer.machine.trusted-execution-clusters.io";
Expand Down Expand Up @@ -179,6 +179,12 @@ async fn keygen_reconcile(
);
return Ok(Action::await_change());
}
Err(e) => {
let err: anyhow::Error = e.into();
let fin_err =
finalizer::Error::<ControllerError>::CleanupFailed(err.into());
return Err(fin_err);
}
_ => {
// TEC exists and is not being deleted, proceed with unmount_secret
}
Expand All @@ -196,11 +202,12 @@ 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<Machine> = Api::default_namespaced(client.clone());
pub async fn launch_keygen_controller(clients: KubeClients) {
let machines: Api<Machine> = Api::default_namespaced(clients.watch_client);
let client = Arc::new(clients.request_client);
tokio::spawn(
Controller::new(machines, Default::default())
.run(keygen_reconcile, controller_error_policy, Arc::new(client))
.run(keygen_reconcile, controller_error_policy, client)
.for_each(controller_info),
);
}
Expand Down
4 changes: 2 additions & 2 deletions register-server/src/main.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,7 @@ use uuid::Uuid;

use trusted_cluster_operator_lib::endpoints::*;
use trusted_cluster_operator_lib::{
generate_owner_reference, get_trusted_execution_cluster, Machine, MachineSpec,
generate_owner_reference, get_trusted_execution_cluster, timed_client, Machine, MachineSpec,
};

#[derive(Parser)]
Expand Down Expand Up @@ -237,7 +237,7 @@ async fn main() {
let err = "failed to create Kubernetes client";
let app = Router::new()
.route(&endpoint, get(register_handler))
.with_state(Client::try_default().await.expect(err));
.with_state(timed_client().await.expect(err));
let addr = SocketAddr::from(([0, 0, 0, 0], args.port));
let service = app.into_make_service();

Expand Down
Loading