fix: Handle containers which were running before Pulsar

crates/bpf-common/src/parsing/containers.rs

@@ -43,6 +43,15 @@ pub enum ContainerId {
     Libpod(String),
 }
 
+impl fmt::Display for ContainerId {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            ContainerId::Docker(id) => write!(f, "{id}"),
+            ContainerId::Libpod(id) => write!(f, "{id}"),
+        }
+    }
+}
+
 #[derive(Debug, Deserialize)]
 struct DockerConfig {
     #[serde(rename = "Config")]

crates/bpf-common/src/parsing/procfs.rs

@@ -178,6 +178,10 @@ fn get_container_id_from_cgroup(cgroup_info: &str) -> Option<ContainerId> {
 }
 
 pub fn get_process_container_id(pid: Pid) -> Result<Option<ContainerId>, ProcfsError> {
+    if pid.as_raw() == 0 {
+        return Ok(None);
+    }
+
     let path = format!("/proc/{pid}/cgroup");
     let file = File::open(&path).map_err(|source| ProcfsError::ReadFile { source, path })?;
 

crates/bpf-filtering/src/process_tree.rs

@@ -66,12 +66,14 @@ fn get_process_namespace(pid: Pid, ns_type: &str) -> Result<u32, Error> {
 fn get_process_namespace_or_log(pid: Pid, namespace_type: &str) -> u32 {
     get_process_namespace(pid, namespace_type).map_or_else(
         |e| {
-            log::warn!(
-                "Failed to determine {} namespace for process {:?}: {}",
-                namespace_type,
-                pid,
-                e
-            );
+            if pid.as_raw() != 0 {
+                log::warn!(
+                    "Failed to determine {} namespace for process {:?}: {}",
+                    namespace_type,
+                    pid,
+                    e
+                );
+            }
             u32::default()
         },
         |v| v,

crates/pulsar-core/src/pdk/process_tracker.rs

@@ -140,8 +140,6 @@ struct ProcessTracker {
     rx: mpsc::UnboundedReceiver<TrackerRequest>,
     /// current processes
     processes: HashMap<Pid, ProcessData>,
-    /// current containers
-    containers: HashMap<Namespaces, ContainerInfo>,
     /// scheduled removal of exited processes
     next_cleanup: Timestamp,
     /// pending info requests arrived before the process was created
@@ -162,6 +160,7 @@ struct ProcessData {
     >,
     argv: Vec<String>,
     namespaces: Namespaces,
+    container: Option<ContainerInfo>,
 }
 
 /// Cleanup timeout in nanoseconds. This is how long an exited process
@@ -186,12 +185,12 @@ impl ProcessTracker {
                 exec_changes: BTreeMap::new(),
                 argv: Vec::new(),
                 namespaces: Namespaces::default(),
+                container: None,
             },
         );
         Self {
             rx,
             processes,
-            containers: HashMap::new(),
             next_cleanup: Timestamp::now() + CLEANUP_TIMEOUT,
             pending_requests: Vec::new(),
             pending_updates: HashMap::new(),
@@ -263,15 +262,24 @@ impl ProcessTracker {
         }
     }
 
-    fn handle_container(&mut self, pid: Pid, namespaces: Namespaces) -> Result<(), ContainerError> {
+    fn get_container_info(
+        &mut self,
+        pid: Pid,
+        is_new_container: bool,
+    ) -> Result<Option<ContainerInfo>, ContainerError> {
         let container_id = procfs::get_process_container_id(pid)?;
-        if let Some(id) = container_id {
-            let container_info = ContainerInfo::from_container_id(id)?;
-            self.containers.insert(namespaces, container_info);
-        } else {
-            log::warn!("could not determine a continer ID of a new containerized process {pid}");
+        match container_id {
+            Some(id) => {
+                let container_info = ContainerInfo::from_container_id(id.clone())?;
+                if is_new_container {
+                    log::debug!("Detected a new container {id}");
+                } else {
+                    log::debug!("Detected an already existing container {id}");
+                }
+                Ok(Some(container_info))
+            }
+            None => Ok(None),
         }
-        Ok(())
     }
 
     fn handle_update(&mut self, mut update: TrackerUpdate) {
@@ -283,10 +291,13 @@ impl ProcessTracker {
                 namespaces,
                 is_new_container,
             } => {
-                if is_new_container {
-                    self.handle_container(pid, namespaces)
-                        .unwrap_or_else(|err| log::error!("{err}"));
-                }
+                let container = match self.get_container_info(pid, is_new_container) {
+                    Ok(container) => container,
+                    Err(err) => {
+                        log::error!("{err}");
+                        None
+                    }
+                };
                 self.processes.insert(
                     pid,
                     ProcessData {
@@ -301,6 +312,7 @@ impl ProcessTracker {
                             .map(|parent| parent.argv.clone())
                             .unwrap_or_default(),
                         namespaces,
+                        container,
                     },
                 );
                 if let Some(pending_updates) = self.pending_updates.remove(&pid) {
@@ -317,13 +329,18 @@ impl ProcessTracker {
                 namespaces,
                 is_new_container,
             } => {
-                if is_new_container {
-                    self.handle_container(pid, namespaces)
-                        .unwrap_or_else(|err| log::error!("{err}"));
-                }
+                let container = match self.get_container_info(pid, is_new_container) {
+                    Ok(container) => container,
+                    Err(err) => {
+                        log::error!("{err}");
+                        None
+                    }
+                };
                 if let Some(p) = self.processes.get_mut(&pid) {
                     p.exec_changes.insert(timestamp, std::mem::take(image));
-                    p.argv = std::mem::take(argv)
+                    p.argv = std::mem::take(argv);
+                    p.namespaces = namespaces;
+                    p.container = container;
                 } else {
                     // if exec arrived before the fork, we save the event as pending
                     log::debug!("(exec) Process {pid} not found in process tree, saving for later");
@@ -354,7 +371,6 @@ impl ProcessTracker {
             .processes
             .get(&pid)
             .ok_or(TrackerError::ProcessNotFound)?;
-        let container: Option<ContainerInfo> = self.containers.get(&process.namespaces).cloned();
         if ts < process.fork_time {
             log::warn!(
                 "{} not forked yet {} < {} ({}ms)",
@@ -377,7 +393,7 @@ impl ProcessTracker {
             fork_time: process.fork_time,
             argv: process.argv.clone(),
             namespaces: process.namespaces,
-            container,
+            container: process.container.clone(),
         })
     }