Skip to content

feat: scan containers on host during daemon initialisation #623

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鈥檒l occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
VaibhavUpreti merged 2 commits into from
Dec 19, 2025
Merged

feat: scan containers on host during daemon initialisation #623

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
183ad1b
feat: scan containers on host during daemon initialisation
VaibhavUpreti Dec 18, 2025
e77cf21
chore: refactor code
VaibhavUpreti Dec 19, 2025
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
179 changes: 113 additions & 66 deletions src/tracer/src/extracts/containers/docker_watcher/watcher.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,7 @@ use anyhow::Result;
use bollard::models::EventMessage;
use bollard::query_parameters::{EventsOptionsBuilder, InspectContainerOptions};
use bollard::Docker;
use chrono::{TimeZone, Utc};
use chrono::{DateTime, TimeZone, Utc};
use futures_util::StreamExt;
use std::collections::HashMap;
use std::sync::Arc;
Expand Down Expand Up @@ -59,6 +59,11 @@ impl DockerWatcher {
};

if let Some(ref docker) = docker {
// Scan existing containers first
if let Err(e) = self.scan_existing_containers(docker).await {
tracing::warn!("Failed to scan existing containers: {:?}", e);
}

let filters = HashMap::from_iter([("type", vec!["container".to_string()])]);
let events_options = EventsOptionsBuilder::default().filters(&filters).build();
let mut events_stream = docker.events(Some(events_options));
Expand Down Expand Up @@ -109,90 +114,132 @@ impl DockerWatcher {
Ok(())
}

async fn process_event(docker: &Docker, event: EventMessage) -> Option<ContainerEvent> {
let id = event.actor.as_ref().map(|action| action.id.clone())??;
let action = event.action.as_deref()?;
let time = Utc
.timestamp_opt(event.time.unwrap_or_default(), 0)
.single()?;
async fn scan_existing_containers(&self, docker: &Docker) -> Result<()> {
let containers = docker
.list_containers(None::<bollard::query_parameters::ListContainersOptions>)
.await?;

let inspect = docker
.inspect_container(&id, None::<InspectContainerOptions>)
.await
.ok()?;
for container in containers {
let Some(ref id) = container.id else { continue };

// environment variables of the container to get the TRACER_TRACE_ID and the AWS_BATCH_JOB_ID
let container_environment_variables = inspect
.config
.as_ref()
.and_then(|container_config| container_config.env.clone())
.unwrap_or_default();
let Ok(inspect) = docker
.inspect_container(id, None::<InspectContainerOptions>)
.await
else {
continue;
};

// each element of the vector is a string "key=value"
let trace_id = container_environment_variables
.iter()
.find(|environment_variable| environment_variable.starts_with("TRACER_TRACE_ID="))
.map(|environment_variable| {
environment_variable.split("=").nth(1).unwrap().to_string()
}); // getting the value of the environment variable
if !matches!(inspect.state.as_ref().and_then(|s| s.running), Some(true)) {
continue;
}

let timestamp = inspect
.state
.as_ref()
.and_then(|s| s.started_at.as_ref()?.parse::<DateTime<Utc>>().ok())
.unwrap_or_else(Utc::now);

let Some(event) = Self::inspect_to_event(inspect, timestamp, ContainerState::Started)
else {
continue;
};

self.container_state
.write()
.await
.insert(ContainerId(event.id.clone()), event.clone());

if let Err(e) = self
.recorder
.log_with_metadata(
ProcessStatus::ContainerExecution,
"[existing container]".to_string(),
Some(EventAttributes::ContainerEvents(event.into())),
Some(timestamp),
)
.await
{
tracing::error!("Failed to log existing container: {:?}", e);
}
}

tracing::info!("Scanned existing containers");
Ok(())
}

let job_id = container_environment_variables
fn get_container_environment_variable(env_vars: &[String], name: &str) -> Option<String> {
env_vars
.iter()
.find(|environment_variable| environment_variable.starts_with("AWS_BATCH_JOB_ID="))
.map(|environment_variable| {
environment_variable.split("=").nth(1).unwrap().to_string()
}); // getting the value of the environment variable

let name = inspect
.name
.unwrap_or_default()
.trim_start_matches('/')
.to_string();
let image = inspect
.config
.as_ref()
.and_then(|cfg| cfg.image.clone())
.unwrap_or_default();
let labels = inspect
.config
.and_then(|cfg| cfg.labels)
.unwrap_or_default();
let ip = inspect.network_settings.and_then(|net| {
net.ip_address.or_else(|| {
net.networks
.and_then(|m| m.values().next().and_then(|n| n.ip_address.clone()))
})
});
.find(|v| v.starts_with(&format!("{}=", name)))
.and_then(|v| v.split('=').nth(1).map(String::from))
}

fn inspect_to_event(
inspect: bollard::models::ContainerInspectResponse,
timestamp: DateTime<Utc>,
state: ContainerState,
) -> Option<ContainerEvent> {
let env_vars = inspect.config.as_ref()?.env.clone().unwrap_or_default();

Some(ContainerEvent {
id: inspect.id?,
name: inspect
.name
.unwrap_or_default()
.trim_start_matches('/')
.to_string(),
image: inspect
.config
.as_ref()
.and_then(|cfg| cfg.image.clone())
.unwrap_or_default(),
ip: inspect.network_settings.and_then(|net| {

@MicheleVerriello MicheleVerriello Dec 18, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

we don't really use these, so we may completely remove them from the ContainerEvent

net.ip_address
.or_else(|| net.networks?.values().next()?.ip_address.clone())
}),
labels: inspect
.config
.and_then(|cfg| cfg.labels)
.unwrap_or_default(),
timestamp,
state,
trace_id: Self::get_container_environment_variable(&env_vars, "TRACER_TRACE_ID"),
job_id: Self::get_container_environment_variable(&env_vars, "AWS_BATCH_JOB_ID"),
environment_variables: env_vars,
})
}

async fn process_event(docker: &Docker, event: EventMessage) -> Option<ContainerEvent> {
let id = event.actor.as_ref()?.id.as_ref()?;
let action = event.action.as_deref()?;
let time = Utc
.timestamp_opt(event.time.unwrap_or_default(), 0)
.single()?;

let state = match action {
"start" => ContainerState::Started,
"die" => {
let exit_code = event
.actor
.and_then(|a| a.attributes)
.and_then(|attrs| attrs.get("exitCode")?.parse().ok())
.as_ref()?
.attributes
.as_ref()?
.get("exitCode")?
.parse()
.ok()
.unwrap_or(-1);
// TODO: if the attributes contain the invoked command, add the command name to
// the error messages when relevant
let reason = exit_code_explanation(exit_code);
ContainerState::Exited { exit_code, reason }
}
"destroy" => ContainerState::Died,
_ => return None,
};

Some(ContainerEvent {
id,
name,
image,
ip,
labels,
timestamp: time,
state,
environment_variables: container_environment_variables,
trace_id,
job_id,
})
let inspect = docker
.inspect_container(id, None::<InspectContainerOptions>)
.await
.ok()?;
Self::inspect_to_event(inspect, time, state)
}

pub async fn get_container_event(&self, id: &str) -> Option<ContainerEvent> {
Expand Down
Loading