Skip to content

Emit process context with observer events #6

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

Merged
ZhiXiao-Lin merged 1 commit into from
Jul 7, 2026
Merged

Emit process context with observer events #6

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
39 changes: 38 additions & 1 deletion a3s-observer-collector/src/main.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,7 @@

use a3s_observer::{
read_ppid, AgentEvent, EnrichedEvent, Exporter, Identity, IdentityResolver, JsonExporter,
KubeResolver, LogExporter, Provider, ServiceClassifier, SniClassifier,
KubeResolver, LogExporter, ProcessContext, Provider, ServiceClassifier, SniClassifier,
};
use a3s_observer_common::{
ConnectEvent, DnsEvent, ExecEvent, ExitEvent, FileEvent, LlmEvent, SecEvent, SslEvent,
Expand Down Expand Up @@ -260,6 +260,7 @@ async fn main() -> anyhow::Result<()> {
if let Some(ev) = read_pod::<ExecEvent>(&item) {
emit(exporter.as_ref(), &mut stats, EnrichedEvent {
identity: identity_for(&resolver, ev.pid, &ev.comm),
process: Some(process_context(ev.pid, &ev.comm)),
provider: None,
event: AgentEvent::ToolExec {
pid: ev.pid,
Expand All @@ -275,6 +276,7 @@ async fn main() -> anyhow::Result<()> {
if let Some(ev) = read_pod::<ExitEvent>(&item) {
emit(exporter.as_ref(), &mut stats, EnrichedEvent {
identity: identity_for(&resolver, ev.pid, &ev.comm),
process: Some(process_context(ev.pid, &ev.comm)),
provider: None,
event: AgentEvent::ProcessExit {
pid: ev.pid,
Expand All @@ -294,6 +296,7 @@ async fn main() -> anyhow::Result<()> {
};
emit(exporter.as_ref(), &mut stats, EnrichedEvent {
identity: identity_for(&resolver, ev.pid, &ev.comm),
process: Some(process_context(ev.pid, &ev.comm)),
provider: None,
event: AgentEvent::SecurityAction {
pid: ev.pid,
Expand All @@ -313,6 +316,7 @@ async fn main() -> anyhow::Result<()> {
peers.insert(sock_key(ev.pid, ev.fd), (peer, ev.port));
emit(exporter.as_ref(), &mut stats, EnrichedEvent {
identity: identity_for(&resolver, ev.pid, &ev.comm),
process: Some(process_context(ev.pid, &ev.comm)),
provider: None,
event: AgentEvent::Egress {
pid: ev.pid,
Expand Down Expand Up @@ -343,6 +347,7 @@ async fn main() -> anyhow::Result<()> {
llm_meta.insert(sock_key(ev.pid, ev.fd), (sni.clone(), provider.clone(), peer));
emit(exporter.as_ref(), &mut stats, EnrichedEvent {
identity: identity_for(&resolver, ev.pid, &ev.comm),
process: Some(process_context(ev.pid, &ev.comm)),
provider,
event: AgentEvent::Egress {
pid: ev.pid,
Expand All @@ -360,6 +365,7 @@ async fn main() -> anyhow::Result<()> {
if let Some(query) = parse_dns_qname(&ev.data[..len]) {
emit(exporter.as_ref(), &mut stats, EnrichedEvent {
identity: identity_for(&resolver, ev.pid, &ev.comm),
process: Some(process_context(ev.pid, &ev.comm)),
provider: None,
event: AgentEvent::Dns { pid: ev.pid, query },
});
Expand All @@ -382,6 +388,7 @@ async fn main() -> anyhow::Result<()> {
};
emit(exporter.as_ref(), &mut stats, EnrichedEvent {
identity: identity_for(&resolver, ev.pid, &ev.comm),
process: Some(process_context(ev.pid, &ev.comm)),
provider: None,
event,
});
Expand All @@ -398,6 +405,7 @@ async fn main() -> anyhow::Result<()> {
if provider.is_some() {
emit(exporter.as_ref(), &mut stats, EnrichedEvent {
identity: identity_for(&resolver, ev.pid, &ev.comm),
process: Some(process_context(ev.pid, &ev.comm)),
provider,
event: AgentEvent::LlmCall {
pid: ev.pid,
Expand Down Expand Up @@ -426,6 +434,7 @@ async fn main() -> anyhow::Result<()> {
{
emit(exporter.as_ref(), &mut stats, EnrichedEvent {
identity: identity.clone(),
process: Some(process_context(ev.pid, &ev.comm)),
provider: None,
event: AgentEvent::LlmApi {
pid: ev.pid,
Expand All @@ -438,6 +447,7 @@ async fn main() -> anyhow::Result<()> {
}
emit(exporter.as_ref(), &mut stats, EnrichedEvent {
identity,
process: Some(process_context(ev.pid, &ev.comm)),
provider: None,
event: AgentEvent::SslContent {
pid: ev.pid,
Expand Down Expand Up @@ -528,6 +538,32 @@ fn read_cwd(pid: u32) -> String {
.unwrap_or_default()
}

fn read_exe(pid: u32) -> Option<String> {
std::fs::read_link(format!("/proc/{pid}/exe"))
.ok()
.map(|p| p.to_string_lossy().into_owned())
.filter(|s| !s.is_empty())
}

fn read_cgroup(pid: u32) -> Option<String> {
std::fs::read_to_string(format!("/proc/{pid}/cgroup"))
.ok()
.map(|s| s.trim().to_string())
.filter(|s| !s.is_empty())
}

fn process_context(pid: u32, comm: &[u8; 16]) -> ProcessContext {
let cwd = read_cwd(pid);
ProcessContext {
pid,
ppid: read_ppid(pid),
comm: cstr(comm),
exe: read_exe(pid),
cwd: (!cwd.is_empty()).then_some(cwd),
cgroup: read_cgroup(pid),
}
}

/// A NUL-terminated byte buffer (from a kernel copy) as a lossy String.
fn cstr(buf: &[u8]) -> String {
let end = buf.iter().position(|&b| b == 0).unwrap_or(buf.len());
Expand Down Expand Up @@ -638,6 +674,7 @@ fn emit_collector_heartbeat(
) {
exporter.export(&EnrichedEvent {
identity: Identity::default(),
process: None,
provider: None,
event: AgentEvent::CollectorHeartbeat {
collector_id: meta.collector_id.clone(),
Expand Down
2 changes: 1 addition & 1 deletion src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@ pub mod model;
pub mod policy;
pub mod traits;

pub use model::{AgentEvent, EnrichedEvent};
pub use model::{AgentEvent, EnrichedEvent, ProcessContext};
pub use policy::{parse_egress_policy, AllowAll, Policy, ProviderPolicy, Verdict};
pub use traits::{
read_ppid, Exporter, Identity, IdentityResolver, JsonExporter, KubeResolver, LogExporter,
Expand Down
16 changes: 16 additions & 0 deletions src/model.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,20 @@ use serde::Serialize;
use std::net::IpAddr;
use std::time::Duration;

/// Kernel-observed process context used by downstream attribution engines.
#[derive(Debug, Clone, Default, Serialize)]
pub struct ProcessContext {
pub pid: u32,
pub ppid: u32,
pub comm: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub exe: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub cwd: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub cgroup: Option<String>,
}

/// A raw event captured by an eBPF probe, before identity enrichment.
#[derive(Debug, Clone, Serialize)]
pub enum AgentEvent {
Expand Down Expand Up @@ -123,6 +137,8 @@ pub enum AgentEvent {
#[derive(Debug, Clone, Serialize)]
pub struct EnrichedEvent {
pub identity: Identity,
#[serde(skip_serializing_if = "Option::is_none")]
pub process: Option<ProcessContext>,
pub provider: Option<Provider>,
pub event: AgentEvent,
}
1 change: 1 addition & 0 deletions src/traits.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -346,6 +346,7 @@ mod tests {
task: Some("1".into()),
session: None,
},
process: None,
provider: None,
event: AgentEvent::ProcessExit {
pid: 1,
Expand Down
Loading