brokkr-agent::kube_events

Rust

Tail Kubernetes Event objects for resources this agent manages and forward them to the broker as WsMessage::K8sEvent frames.

Filtering

Brokkr-managed resources carry an k8s.brokkr.io/stack annotation whose value is the owning stack id (see crate::k8s::objects). When a kube Event arrives the tailer fetches the referenced involvedObject and checks its annotations. Lookup results are cached for a few minutes (positive and negative) so repeated events for the same pod don’t translate to repeated GETs.

Lifecycle

• Spawned once at agent startup.
• Watches Events cluster-wide via kube::runtime::watcher.
• Restart-safe: on transient failures the watcher loops with a brief backoff. On success, the watcher itself resumes from the latest resourceVersion it has seen.

Per ADR-0008 / BROKKR-I-0019 these frames flow over the agent uplink introduced in WS-05; the broker persists them into agent_k8s_events (WS-09) under the hard 6h retention ceiling.

Functions

fn spawn

pub

fn spawn(
    client: Client,
    uplink: WsUplink,
    agent_id: Uuid,
    uid_cache_cap: usize,
    watch_namespace: Option<String>,
) -> tokio::task::JoinHandle<()>

Spawn the kube-events tailer. Returns a JoinHandle; production drops it and the task runs for the lifetime of the agent process.

Source

pub fn spawn(
    client: Client,
    uplink: WsUplink,
    agent_id: Uuid,
    uid_cache_cap: usize,
    watch_namespace: Option<String>,
) -> tokio::task::JoinHandle<()> {
    tokio::spawn(async move {
        let cache: Arc<RwLock<UidCache>> = Arc::new(RwLock::new(UidCache::new(uid_cache_cap)));
        let (tx, mut rx) = mpsc::channel::<WsMessage>(OUTBOUND_CAPACITY);

        // Forwarder task: drains the channel into the WS uplink, falling
        // back to silent drop if WS is down (REST has no equivalent for
        // streaming events — by design).
        let uplink_clone = uplink.clone();
        let _forwarder = tokio::spawn(async move {
            while let Some(msg) = rx.recv().await {
                if uplink_clone.try_send(msg).is_err() {
                    // WS not up; drop. The 6h history endpoint cannot
                    // backfill these — kube Events themselves are
                    // short-lived in the cluster.
                }
            }
        });

        // Capped exponential backoff so a persistent failure (e.g. an RBAC
        // denial) doesn't re-dial the API server every 5s forever.
        const MAX_BACKOFF: Duration = Duration::from_secs(60);
        let mut backoff = Duration::from_secs(1);
        loop {
            match watch_loop(
                client.clone(),
                agent_id,
                tx.clone(),
                cache.clone(),
                watch_namespace.as_deref(),
            )
            .await
            {
                Ok(()) => backoff = Duration::from_secs(1),
                Err(e) => {
                    warn!(error = %e, "kube events tailer fell out of watch loop; retrying in {:?}", backoff)
                }
            }
            tokio::time::sleep(backoff).await;
            backoff = (backoff * 2).min(MAX_BACKOFF);
        }
    })
}

Constants

Name	Type	Value	Description
DEFAULT_UID_CACHE_CAP	usize	10_000	Default entry cap. Sized for a cluster with up to ~10k managed pods; bump agent.kube_event_uid_cache_cap for larger fleets. The cache exists to turn the per-Event annotation lookup into an O(1) hit for already-seen UIDs; bounding it stops unmanaged-object churn from growing it without limit (the previous HashMap was unbounded).