Add kubernetes events

api/v1alpha1/crds.go

@@ -33,6 +33,7 @@ var (
 // +kubebuilder:rbac:groups=trusted-execution-clusters.io,resources=trustedexecutionclusters;machines;approvedimages;attestationkeys,verbs=create;delete;get;list;patch;update;watch
 // +kubebuilder:rbac:groups=trusted-execution-clusters.io,resources=trustedexecutionclusters/finalizers;machines/finalizers;attestationkeys/finalizers;approvedimages/finalizers,verbs=update
 // +kubebuilder:rbac:groups=trusted-execution-clusters.io,resources=trustedexecutionclusters/status;machines/status;approvedimages/status;attestationkeys/status,verbs=get;patch;update
+// +kubebuilder:rbac:groups=events.k8s.io,resources=events,verbs=create;patch
 
 // TrustedExecutionClusterSpec defines the desired state of TrustedExecutionCluster
 // +kubebuilder:validation:XValidation:rule="!has(oldSelf.publicAttestationKeyRegisterAddr) || has(self.publicAttestationKeyRegisterAddr)", message="Value is required once set"

attestation-key-register/src/main.rs

@@ -8,16 +8,19 @@ use axum::{http::StatusCode, routing::put, Router};
 use axum_server::tls_openssl::OpenSSLConfig;
 use clap::Parser;
 use env_logger::Env;
+use k8s_openapi::api::core::v1::ObjectReference;
 use k8s_openapi::apimachinery::pkg::apis::meta::v1::ObjectMeta;
-use kube::{Api, Client};
+use kube::runtime::events::{EventType, Recorder, Reporter};
+use kube::{Api, Client, Resource};
 use log::{error, info};
 use serde::{Deserialize, Serialize};
 use std::net::SocketAddr;
 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, record_event, AttestationKey,
+    AttestationKeySpec,
 };
 
 #[derive(Parser)]
@@ -32,6 +35,12 @@ struct Args {
     key_path: Option<String>,
 }
 
+#[derive(Clone)]
+struct AppState {
+    client: Client,
+    recorder: Recorder,
+}
+
 #[derive(Debug, Deserialize, Serialize)]
 struct AttestationKeyRegistration {
     /// Public attestation key
@@ -44,10 +53,12 @@ struct AttestationKeyRegistration {
 }
 
 async fn handle_registration(
-    State(client): State<Client>,
+    State(state): State<AppState>,
     Json(registration): Json<AttestationKeyRegistration>,
 ) -> impl IntoResponse {
     info!("Received registration request: {registration:?}");
+    let client = state.client;
+    let recorder = state.recorder;
 
     let internal_error = |e: anyhow::Error| {
         let code = StatusCode::INTERNAL_SERVER_ERROR;
@@ -76,10 +87,23 @@ async fn handle_registration(
         Ok(existing_keys) => {
             for key in existing_keys.items {
                 if key.spec.public_key == registration.public_key {
+                    let key_ref: ObjectReference = key.object_ref(&());
                     let existing_name = key.metadata.name.unwrap_or_default();
                     error!(
                         "Duplicate public key detected: already exists in AttestationKey '{existing_name}'"
                     );
+                    record_event(
+                        &recorder,
+                        &key_ref,
+                        EventType::Warning,
+                        "DuplicateKeyRejected",
+                        format!(
+                            "Duplicate registration attempt for AttestationKey '{existing_name}'"
+                        ),
+                        "Registering",
+                        None,
+                    )
+                    .await;
                     return (
                         StatusCode::CONFLICT,
                         Json(serde_json::json!({
@@ -93,7 +117,7 @@ async fn handle_registration(
         Err(e) => {
             return internal_error(
                 anyhow::Error::from(e).context("Failed to check for existing keys"),
-            )
+            );
         }
     }
 
@@ -113,8 +137,19 @@ async fn handle_registration(
 
     match api.create(&Default::default(), &attestation_key).await {
         Ok(created) => {
+            let created_ref: ObjectReference = created.object_ref(&());
             let name = created.metadata.name.unwrap_or_default();
             info!("Successfully created AttestationKey: {name}",);
+            record_event(
+                &recorder,
+                &created_ref,
+                EventType::Normal,
+                "AttestationKeyRegistered",
+                format!("AttestationKey '{name}' registered"),
+                "Registering",
+                None,
+            )
+            .await;
             let json = Json(serde_json::json!({
                 "status": "success",
             }));
@@ -132,9 +167,17 @@ async fn main() {
     let endpoint = format!("/{ATTESTATION_KEY_REGISTER_RESOURCE}");
     let err = "failed to create Kubernetes client";
     let client = Client::try_default().await.expect(err);
+    let reporter = Reporter {
+        controller: "attestation-key-register".into(),
+        instance: std::env::var("CONTROLLER_POD_NAME").ok(),
+    };
+    let state = AppState {
+        recorder: Recorder::new(client.clone(), reporter),
+        client,
+    };
     let app = Router::new()
         .route(&endpoint, put(handle_registration))
-        .with_state(client);
+        .with_state(state);
     let addr = SocketAddr::from(([0, 0, 0, 0], args.port));
     let service = app.into_make_service();
 

lib/Cargo.toml

@@ -15,6 +15,7 @@ anyhow.workspace = true
 compute-pcrs-lib.workspace = true
 k8s-openapi.workspace = true
 kube.workspace = true
+log.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 

lib/src/lib.rs

@@ -25,7 +25,9 @@ pub use vendor_kopium::virtualmachines;
 
 use anyhow::{Context, Result, anyhow};
 use conditions::*;
+use k8s_openapi::api::core::v1::ObjectReference;
 use k8s_openapi::apimachinery::pkg::apis::meta::v1::{Condition, OwnerReference, Time};
+use kube::runtime::events::{Event as K8sEvent, EventType, Recorder};
 use kube::{Api, Client, Resource};
 
 #[macro_export]
@@ -106,6 +108,27 @@ pub fn committed_condition(
     }
 }
 
+pub async fn record_event(
+    recorder: &Recorder,
+    reference: &ObjectReference,
+    type_: EventType,
+    reason: &str,
+    note: String,
+    action: &str,
+    secondary: Option<ObjectReference>,
+) {
+    let ev = K8sEvent {
+        type_,
+        reason: reason.into(),
+        note: Some(note),
+        action: action.into(),
+        secondary,
+    };
+    if let Err(e) = recorder.publish(&ev, reference).await {
+        log::warn!("Failed to publish event: {e}");
+    }
+}
+
 /// Generate an OwnerReference for any Kubernetes resource
 pub fn generate_owner_reference<T: Resource<DynamicType = ()>>(
     object: &T,

operator/src/attestation_key_register.rs

@@ -6,6 +6,7 @@ use anyhow::{Result, anyhow};
 use futures_util::StreamExt;
 use k8s_openapi::ByteString;
 use k8s_openapi::api::apps::v1::{Deployment, DeploymentSpec};
+use k8s_openapi::api::core::v1::ObjectReference;
 use k8s_openapi::api::core::v1::{
     Container, ContainerPort, PodSpec, PodTemplateSpec, Secret, Service, ServicePort, ServiceSpec,
 };
@@ -19,6 +20,7 @@ use kube::{
     runtime::{
         Controller,
         controller::Action,
+        events::{EventType, Recorder},
         finalizer,
         finalizer::Event,
         reflector::{self, ObjectRef, Store},
@@ -37,11 +39,13 @@ 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};
+use trusted_cluster_operator_lib::record_event;
 
 /// 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 recorder: Recorder,
     pub machine_store: Store<Machine>,
     pub ak_store: Store<AttestationKey>,
     pub secret_store: Store<Secret>,
@@ -60,8 +64,10 @@ impl AkContextData {
         crate::spawn_reflector::<Secret>(secret_writer, client.clone(), "Secret");
         crate::spawn_reflector::<Deployment>(deployment_writer, client.clone(), "Deployment");
 
+        let recorder = operator::new_recorder(client.clone(), "ak-controller");
         Self {
             client,
+            recorder,
             machine_store,
             ak_store,
             secret_store,
@@ -238,14 +244,36 @@ async fn approve_ak(ak: &AttestationKey, machine: &Machine, ctx: &AkContextData)
     let condition = attestation_key_approved_condition(approve_reason, generation, &ak.status);
     let mut conditions = ak.status.as_ref().and_then(|s| s.conditions.clone());
     let changed = upsert_condition(&mut conditions, condition);
+    let machine_name = machine.metadata.name.clone().unwrap_or_default();
 
     if changed {
         let status = AttestationKeyStatus { conditions };
         update_status!(aks, &name, status)?;
         info!("Approved attestation key {name}");
-    }
 
-    let machine_name = machine.metadata.name.clone().unwrap_or_default();
+        let ak_ref: ObjectReference = ak.object_ref(&());
+        let machine_ref: ObjectReference = machine.object_ref(&());
+        record_event(
+            &ctx.recorder,
+            &ak_ref,
+            EventType::Normal,
+            "AttestationKeyApproved",
+            format!("Attestation key {name} approved for machine {machine_name}"),
+            "Approving",
+            Some(machine_ref.clone()),
+        )
+        .await;
+        record_event(
+            &ctx.recorder,
+            &machine_ref,
+            EventType::Normal,
+            "AttestationKeyApproved",
+            format!("Machine {machine_name} matched attestation key {name}"),
+            "Approving",
+            Some(ak_ref),
+        )
+        .await;
+    }
     let has_machine_owner = ak
         .metadata
         .owner_references

operator/src/lib.rs

@@ -14,6 +14,7 @@ use k8s_openapi::api::core::v1::{Secret, SecretVolumeSource, Volume, VolumeMount
 use k8s_openapi::apimachinery::pkg::apis::meta::v1::{Condition, Time};
 use k8s_openapi::jiff::Timestamp;
 use kube::Resource;
+use kube::runtime::events::{Recorder, Reporter};
 use kube::runtime::reflector::{self, Store};
 use kube::runtime::watcher::watcher;
 use kube::{Api, Client, runtime::controller::Action};
@@ -43,6 +44,19 @@ pub async fn controller_info<T: Debug, E: Debug>(res: Result<T, E>) {
     }
 }
 
+pub fn new_recorder(client: Client, controller_name: &str) -> Recorder {
+    let reporter = Reporter {
+        controller: controller_name.into(),
+        instance: std::env::var("CONTROLLER_POD_NAME").ok(),
+    };
+    Recorder::new(client, reporter)
+}
+
+pub struct ControllerContext {
+    pub client: Client,
+    pub recorder: Recorder,
+}
+
 #[macro_export]
 macro_rules! create_or_info_if_exists {
     ($client:expr, $type:ident, $resource:ident) => {