feat: emit collector heartbeat telemetry

Author: RoyLin

README.md

@@ -54,10 +54,13 @@ latency / TTFT, or plaintext) / **where** (peer IP / hostname).
 | `ssl`\* | OpenSSL `SSL_write` / `SSL_read` uprobes | `SslContent` — request/response plaintext (`A3S_OBSERVER_SSL=1`) |
 | `llm-api`\* | parsed from `SslContent` | `LlmApi` — **model** + token usage (`A3S_OBSERVER_SSL=1`) |
 | `security` | `setuid` / `ptrace` / `bind` syscalls | `SecurityAction` — privilege escalation (→root) / process injection / opened a listening port (rare + in-kernel-filtered) |
+| collector heartbeat | userspace timer | `CollectorHeartbeat` — collector id, node/pod, attached probes, feature flags, per-window counts, ring drops, output drops |
 
 Userspace enriches each event with **identity** (k8s cgroup→pod, `/proc` comm+ppid, or an
 in-kernel `comm` fallback for short-lived processes), a `(pid,fd)→peer` **correlation**, and
 **provider** classification (SNI → 15 LLM providers); then exports **NDJSON** (or a human log).
+`CollectorHeartbeat` is a control-plane event for platforms such as AnySentry; it is not an
+agent action and should not be fed into security policy decisions.
 
 **Example output** (`A3S_OBSERVER_JSON=1`, one event per line — wrapped here for readability):
 
@@ -77,6 +80,9 @@ in-kernel `comm` fallback for short-lived processes), a `(pid,fd)→peer` **corr
 Filter with `jq`, e.g. every LLM call and its provider:
 `… | jq -c 'select(.event.LlmCall) | {agent:.identity.agent, provider, sni:.event.LlmCall.sni}'`.
 
+Set `A3S_OBSERVER_COLLECTOR_ID` and `A3S_NODE_NAME` in DaemonSets when you want stable collector
+identity in downstream fleet-health views. If unset, the collector falls back to pod/host names.
+
 ## Intervene — egress / file / exec (opt-in)
 
 The same vantage point enforces an **external policy** — a plain file any controller writes; the

a3s-observer-collector/src/main.rs

@@ -20,7 +20,7 @@ use aya::{
     programs::{KProbe, TracePoint, UProbe},
     Ebpf,
 };
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
 use std::time::Duration;
 
@@ -205,9 +205,23 @@ async fn main() -> anyhow::Result<()> {
              livenessProbe on it will restart-loop the pod");
     }
 
+    let collector = CollectorMeta::from_env(
+        files,
+        std::env::var_os("A3S_OBSERVER_SSL").is_some(),
+        attached,
+    );
+
     let mut sigint = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::interrupt())?;
     let mut sigterm = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())?;
     let mut stats = Stats::default();
+    emit_collector_heartbeat(
+        exporter.as_ref(),
+        &collector,
+        0,
+        &stats,
+        0,
+        exporter.output_drops(),
+    );
     let mut report = tokio::time::interval(Duration::from_secs(60));
     report.tick().await; // consume the immediate first tick
     loop {
@@ -221,6 +235,7 @@ async fn main() -> anyhow::Result<()> {
                     .map(|v| v.iter().copied().sum())
                     .unwrap_or(0);
                 let output_dropped = exporter.output_drops();
+                emit_collector_heartbeat(exporter.as_ref(), &collector, 60, &stats, dropped, output_dropped);
                 tracing::info!(
                     exec = stats.exec,
                     exit = stats.exit,
@@ -543,6 +558,110 @@ struct Stats {
     llm: u64,
     ssl: u64,
     sec: u64,
+    agents: HashSet<String>,
+}
+
+struct CollectorMeta {
+    collector_id: String,
+    node_name: Option<String>,
+    namespace: Option<String>,
+    pod_name: Option<String>,
+    version: String,
+    mode: String,
+    attached_probes: u32,
+    enabled_features: Vec<String>,
+}
+
+impl CollectorMeta {
+    fn from_env(files: bool, ssl: bool, attached: usize) -> Self {
+        let node_name = env_any(&["A3S_NODE_NAME", "NODE_NAME", "K8S_NODE_NAME"]).or_else(hostname);
+        let namespace = env_any(&["A3S_NAMESPACE", "POD_NAMESPACE", "K8S_NAMESPACE"]);
+        let pod_name = env_any(&["A3S_POD_NAME", "POD_NAME", "HOSTNAME"]);
+        let collector_id = env_any(&["A3S_OBSERVER_COLLECTOR_ID", "COLLECTOR_ID"])
+            .or_else(|| pod_name.clone())
+            .or_else(|| node_name.clone())
+            .unwrap_or_else(|| "a3s-observer".to_string());
+        let mut enabled_features = vec![
+            "exec".to_string(),
+            "network".to_string(),
+            "dns".to_string(),
+            "security".to_string(),
+        ];
+        if files {
+            enabled_features.push("files".to_string());
+        }
+        if ssl {
+            enabled_features.push("ssl".to_string());
+        }
+        let mode = if files || ssl {
+            "observe+extensions"
+        } else {
+            "observe"
+        }
+        .to_string();
+        Self {
+            collector_id,
+            node_name,
+            namespace,
+            pod_name,
+            version: env!("CARGO_PKG_VERSION").to_string(),
+            mode,
+            attached_probes: attached as u32,
+            enabled_features,
+        }
+    }
+}
+
+fn env_any(names: &[&str]) -> Option<String> {
+    names.iter().find_map(|name| {
+        std::env::var(name)
+            .ok()
+            .map(|v| v.trim().to_string())
+            .filter(|v| !v.is_empty())
+    })
+}
+
+fn hostname() -> Option<String> {
+    std::fs::read_to_string("/etc/hostname")
+        .ok()
+        .map(|s| s.trim().to_string())
+        .filter(|s| !s.is_empty())
+}
+
+fn emit_collector_heartbeat(
+    exporter: &dyn Exporter,
+    meta: &CollectorMeta,
+    interval_secs: u64,
+    stats: &Stats,
+    dropped: u64,
+    output_dropped: u64,
+) {
+    exporter.export(&EnrichedEvent {
+        identity: Identity::default(),
+        provider: None,
+        event: AgentEvent::CollectorHeartbeat {
+            collector_id: meta.collector_id.clone(),
+            node_name: meta.node_name.clone(),
+            namespace: meta.namespace.clone(),
+            pod_name: meta.pod_name.clone(),
+            version: meta.version.clone(),
+            mode: meta.mode.clone(),
+            attached_probes: meta.attached_probes,
+            enabled_features: meta.enabled_features.clone(),
+            interval_secs,
+            observed_agents: stats.agents.len() as u64,
+            exec: stats.exec,
+            exit: stats.exit,
+            egress: stats.egress,
+            dns: stats.dns,
+            file: stats.file,
+            llm: stats.llm,
+            ssl: stats.ssl,
+            sec: stats.sec,
+            dropped,
+            output_dropped,
+        },
+    });
 }
 
 /// Export an event and count it by kind for the throughput report.
@@ -558,6 +677,12 @@ fn emit(exporter: &dyn Exporter, stats: &mut Stats, ev: EnrichedEvent) {
         AgentEvent::SslContent { .. } => stats.ssl += 1,
         AgentEvent::LlmApi { .. } => stats.llm += 1,
         AgentEvent::SecurityAction { .. } => stats.sec += 1,
+        AgentEvent::CollectorHeartbeat { .. } => {}
+    }
+    if !matches!(ev.event, AgentEvent::CollectorHeartbeat { .. }) {
+        if let Some(agent) = &ev.identity.agent {
+            stats.agents.insert(agent.clone());
+        }
     }
     exporter.export(&ev);
 }

src/model.rs

@@ -91,6 +91,31 @@ pub enum AgentEvent {
         /// ptrace: target pid · bind: port · setuid-root: 0.
         detail: u64,
     },
+    /// Collector liveness and throughput telemetry. This is an observer-side control-plane event,
+    /// not an agent action. It lets downstream platforms detect node/DaemonSet coverage gaps,
+    /// slow consumers, ring drops, and feature enablement without requiring any agent SDK.
+    CollectorHeartbeat {
+        collector_id: String,
+        node_name: Option<String>,
+        namespace: Option<String>,
+        pod_name: Option<String>,
+        version: String,
+        mode: String,
+        attached_probes: u32,
+        enabled_features: Vec<String>,
+        interval_secs: u64,
+        observed_agents: u64,
+        exec: u64,
+        exit: u64,
+        egress: u64,
+        dns: u64,
+        file: u64,
+        llm: u64,
+        ssl: u64,
+        sec: u64,
+        dropped: u64,
+        output_dropped: u64,
+    },
 }
 
 /// An [`AgentEvent`] tagged with the resolved [`Identity`] and, for LLM calls, the