agentsight_capture/runners/
process.rs

1// SPDX-License-Identifier: MIT
2// Copyright (c) 2026 eunomia-bpf org.
3
4use super::common::{AnalyzerProcessor, BinaryExecutor, current_boot_time_ns, parse_json_event};
5use super::{EventStream, Runner, RunnerError};
6use crate::analyzers::Analyzer;
7use crate::event::Event;
8use crate::sources::proc::PidSeed;
9use async_trait::async_trait;
10use futures::stream::StreamExt;
11use std::path::Path;
12use std::sync::{Arc, atomic::AtomicU64};
13
14pub struct ProcessRunner {
15    analyzers: Vec<Box<dyn Analyzer>>,
16    executor: BinaryExecutor,
17    args: Vec<String>,
18}
19
20impl ProcessRunner {
21    pub fn from_binary_extractor(binary_path: impl AsRef<Path>) -> Self {
22        Self {
23            analyzers: Vec::new(),
24            executor: BinaryExecutor::new(binary_path.as_ref().to_string_lossy().into_owned())
25                .with_runner_name("Process".to_string()),
26            args: Vec::new(),
27        }
28    }
29
30    pub fn with_args<I, S>(mut self, args: I) -> Self
31    where
32        I: IntoIterator<Item = S>,
33        S: AsRef<str>,
34    {
35        self.args = args.into_iter().map(|s| s.as_ref().to_string()).collect();
36        self.executor.set_args(&self.args);
37        self
38    }
39
40    pub fn with_seed_pids(mut self, seeds: &[PidSeed]) -> Self {
41        for seed in seeds {
42            self.args.push("--seed-pid".to_string());
43            self.args.push(seed.arg_value());
44        }
45        self.executor.set_args(&self.args);
46        self
47    }
48
49    fn parse_process_event(json_value: serde_json::Value, errors: &AtomicU64) -> Event {
50        if json_value.get("event").and_then(|v| v.as_str()) == Some("CLOCK_SYNC") {
51            return Event::new_with_timestamp(
52                current_boot_time_ns(),
53                "diagnostic".to_string(),
54                0,
55                "process".to_string(),
56                json_value,
57            );
58        }
59        parse_json_event("process", "timestamp", json_value, errors)
60    }
61}
62
63#[async_trait]
64impl Runner for ProcessRunner {
65    async fn run(&mut self) -> Result<EventStream, RunnerError> {
66        let json_stream = self.executor.get_json_stream().await?;
67        let errors = Arc::new(AtomicU64::new(0));
68        let stream = json_stream.map(move |v| Self::parse_process_event(v, &errors));
69        AnalyzerProcessor::process_through_analyzers(Box::pin(stream), &mut self.analyzers).await
70    }
71
72    fn add_analyzer(mut self, analyzer: Box<dyn Analyzer>) -> Self {
73        self.analyzers.push(analyzer);
74        self
75    }
76}
77
78#[cfg(test)]
79mod tests {
80    use super::*;
81
82    #[tokio::test]
83    #[ignore = "requires real binary and sudo"]
84    async fn test_process_runner_with_real_binary() {
85        use tokio::time::timeout;
86        let binary_path = "../src/process";
87        if !Path::new(binary_path).exists() {
88            return;
89        }
90        let mut runner = ProcessRunner::from_binary_extractor(binary_path);
91        if let Ok(mut stream) = runner.run().await {
92            let _ = timeout(std::time::Duration::from_secs(30), async {
93                while futures::StreamExt::next(&mut stream).await.is_some() {}
94            })
95            .await;
96        }
97    }
98}
agentsight_capture/runners/process.rs

agentsight_capture/runners/
process.rs
