brokkr-agent::pod_logs

Rust

Stream container logs for Brokkr-managed pods to the broker as WsMessage::PodLogLine frames.

Opt-in

Off by default. A pod is eligible only when both annotations are present on the Pod itself:

• k8s.brokkr.io/stack → the owning stack id (set by the agent’s reconciler on every managed object).
• brokkr.io/stream-logs: "true" → explicit opt-in. Today this is expected to live on the Pod (e.g. via a PodTemplateSpec annotation on the Deployment that owns it); a future agent-side improvement can propagate stack-level opt-in down to pods automatically.

Rate limiting

Each container stream gets a token bucket (default 100 lines/sec). Over-rate lines are dropped and a WsMessage::LogGap{reason: RateLimit} is emitted so the UI renders a visible gap rather than silently swallowing data. Per ADR-0008 / project_log_retention_stance, Brokkr is not a log warehouse — when an opted-in stack saturates the bucket the right answer is “ship to Datadog”, not “raise the limit”.

Functions

fn spawn

pub

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

Source

pub fn spawn(
    client: Client,
    uplink: WsUplink,
    agent_id: Uuid,
    watch_namespace: Option<String>,
) -> JoinHandle<()> {
    tokio::spawn(async move {
        let active: ActiveTails = Arc::new(RwLock::new(HashMap::new()));
        // 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_pods(
                client.clone(),
                uplink.clone(),
                agent_id,
                active.clone(),
                watch_namespace.as_deref(),
            )
            .await
            {
                Ok(()) => backoff = Duration::from_secs(1),
                Err(e) => warn!(error = %e, "pod-logs watcher fell out; retrying in {:?}", backoff),
            }
            tokio::time::sleep(backoff).await;
            backoff = (backoff * 2).min(MAX_BACKOFF);
        }
    })
}

Constants

Name	Type	Value	Description
STREAM_LOGS_ANNOTATION	&str	"brokkr.io/stream-logs"	Annotation that opts a workload into log streaming.