saluki_env/workload/collectors/
containerd.rs

1use std::{path::PathBuf, time::Duration};
2
3use async_stream::stream;
4use async_trait::async_trait;
5use containerd_protos::services::namespaces::v1::Namespace;
6use futures::{stream::select_all, Stream, StreamExt as _};
7use saluki_core::accounting::{MemoryBounds, MemoryBoundsBuilder};
8use saluki_core::health::Health;
9use saluki_error::GenericError;
10use saluki_metrics::{static_metrics, Counter};
11use stringtheory::interning::{GenericMapInterner, Interner as _};
12use tokio::{select, sync::mpsc, time::sleep};
13use tracing::{debug, error, warn};
14
15use super::MetadataCollector;
16use crate::workload::{
17    entity::EntityId,
18    helpers::containerd::{
19        events::{ContainerdEvent, ContainerdTopic},
20        ContainerdClient, ContainerdConfiguration,
21    },
22    metadata::MetadataOperation,
23};
24
25static CONTAINERD_WATCH_EVENTS: &[ContainerdTopic] = &[ContainerdTopic::TaskStarted, ContainerdTopic::TaskDeleted];
26
27#[static_metrics(prefix = containerd_metadata_collector, labels(namespace))]
28#[derive(Clone)]
29struct Telemetry {
30    rpc_errors_total: Counter,
31    intern_failed_total: Counter,
32    events_task_started_total: Counter,
33    events_task_deleted_total: Counter,
34}
35
36/// A metadata collector that watches for updates from containerd.
37pub struct ContainerdMetadataCollector {
38    client: ContainerdClient,
39    watched_namespaces: Vec<Namespace>,
40    tag_interner: GenericMapInterner,
41    health: Health,
42}
43
44impl ContainerdMetadataCollector {
45    /// Creates a new `ContainerdMetadataCollector` from the given containerd configuration.
46    ///
47    /// If `socket_path` is given, that path is used. Otherwise, well-known paths are probed.
48    ///
49    /// # Errors
50    ///
51    /// If the containerd gRPC client can't be created, or listing the namespaces in the containerd runtime fails, an
52    /// error will be returned.
53    pub async fn new(
54        socket_path: Option<PathBuf>, containerd_config: &ContainerdConfiguration, health: Health,
55        tag_interner: GenericMapInterner,
56    ) -> Result<Self, GenericError> {
57        let client = ContainerdClient::new(socket_path, containerd_config).await?;
58        let watched_namespaces = client.list_namespaces().await?;
59
60        Ok(Self {
61            client,
62            watched_namespaces,
63            tag_interner,
64            health,
65        })
66    }
67}
68
69#[async_trait]
70impl MetadataCollector for ContainerdMetadataCollector {
71    fn name(&self) -> &'static str {
72        "containerd"
73    }
74
75    async fn watch(&mut self, operations_tx: &mut mpsc::Sender<MetadataOperation>) -> Result<(), GenericError> {
76        self.health.mark_ready();
77
78        // Create a watcher for each namespace, and then join all of their watch streams, which then we'll just funnel
79        // back to the operations channel.
80        let watchers = self
81            .watched_namespaces
82            .iter()
83            .map(|ns| NamespaceWatcher::new(self.client.clone(), ns.clone(), self.tag_interner.clone()).watch());
84
85        let mut operations_stream = select_all(watchers);
86
87        loop {
88            select! {
89                _ = self.health.live() => {},
90                maybe_operation = operations_stream.next() => match maybe_operation {
91                    Some(operation) => {
92                        operations_tx.send(operation).await?;
93                    },
94                    None => break,
95                },
96            }
97        }
98
99        self.health.mark_not_ready();
100
101        Ok(())
102    }
103}
104
105impl MemoryBounds for ContainerdMetadataCollector {
106    fn specify_bounds(&self, builder: &mut MemoryBoundsBuilder) {
107        // TODO: Kind of a throwaway calculation because nothing about the gRPC client can really be bounded at the
108        // moment, and we also don't have any way to know the number of namespaces we'll be monitoring a priori.
109        builder
110            .firm()
111            .with_fixed_amount("self struct", std::mem::size_of::<Self>());
112    }
113}
114
115struct NamespaceWatcher {
116    namespace: Namespace,
117    client: ContainerdClient,
118    tag_interner: GenericMapInterner,
119    telemetry: Telemetry,
120}
121
122impl NamespaceWatcher {
123    fn new(client: ContainerdClient, namespace: Namespace, tag_interner: GenericMapInterner) -> Self {
124        let telemetry = Telemetry::new(&namespace.name);
125        Self {
126            client,
127            namespace,
128            tag_interner,
129            telemetry,
130        }
131    }
132
133    async fn process_event(&self, event: ContainerdEvent) -> Option<MetadataOperation> {
134        match event {
135            ContainerdEvent::TaskStarted { id, pid } => {
136                self.telemetry.events_task_started_total().increment(1);
137                let pid_entity_id = EntityId::ContainerPid(pid);
138                let container_entity_id = EntityId::Container(id);
139                Some(MetadataOperation::add_alias(pid_entity_id, container_entity_id))
140            }
141            ContainerdEvent::TaskDeleted { pid, .. } => {
142                self.telemetry.events_task_deleted_total().increment(1);
143                Some(MetadataOperation::delete(EntityId::ContainerPid(pid)))
144            }
145        }
146    }
147
148    async fn build_initial_metadata_operations(&self) -> Option<Vec<MetadataOperation>> {
149        let mut operations = Vec::new();
150
151        // Get a list of all containers in the namespace.
152        let containers = match self.client.list_containers(&self.namespace).await {
153            Ok(containers) => containers,
154            Err(e) => {
155                self.telemetry.rpc_errors_total().increment(1);
156                error!(namespace = self.namespace.name, error = %e, "Error listing containers.");
157                return None;
158            }
159        };
160
161        for container in containers {
162            let pids = match self
163                .client
164                .list_pids_for_container(&self.namespace, container.id.clone())
165                .await
166            {
167                Ok(pids) => pids,
168                Err(e) => {
169                    if let Some(status) = e.as_response_error() {
170                        if status.code() == tonic::Code::NotFound {
171                            // The container may have been deleted before we could get the PIDs for it, so we'll just
172                            // skip it without making a fuss.
173                            continue;
174                        }
175                    }
176
177                    self.telemetry.rpc_errors_total().increment(1);
178                    error!(namespace = self.namespace.name, container_id = container.id, error = %e, "Error getting PIDs for container.");
179                    continue;
180                }
181            };
182
183            for pid in pids {
184                let pid_entity_id = EntityId::ContainerPid(pid);
185
186                match self.tag_interner.try_intern(container.id.as_str()) {
187                    Some(container_id) => {
188                        let container_entity_id = EntityId::Container(container_id.into());
189                        operations.push(MetadataOperation::add_alias(pid_entity_id, container_entity_id));
190                    }
191                    None => {
192                        self.telemetry.intern_failed_total().increment(1);
193                        warn!(
194                            namespace = self.namespace.name,
195                            container_id = container.id,
196                            container_task_pid = pid,
197                            "Failed to intern container ID. Container ID/task PID link will not be created."
198                        );
199                    }
200                }
201            }
202        }
203
204        Some(operations)
205    }
206
207    fn watch(self) -> impl Stream<Item = MetadataOperation> + Unpin {
208        debug!(
209            namespace = self.namespace.name,
210            "Starting containerd namespace watcher."
211        );
212
213        // We watch the given namespace for all of the relevant events, and convert those into metadata operations that
214        // we pass back to be collected by the parent watcher task, which then forwards them to the metadata aggregator.
215        Box::pin(stream! {
216            // Do an initial scan of the namespace to get all of the existing containers, their tasks and images, and
217            // so on, and generate metadata operations from that as a way to prime the store.
218            if let Some(initial_operations) = self.build_initial_metadata_operations().await {
219                for operation in initial_operations {
220                    yield operation;
221                }
222            }
223
224            // Now watch for events.
225            loop {
226                // TODO: We should be creating this stream -- and polling it! -- before we build our initial metadata
227                // operations in order to ensure that we have an overlap between new events and the initial scan.
228                let mut event_stream = match self.client.watch_events(CONTAINERD_WATCH_EVENTS, &self.namespace).await {
229                    Ok(stream) => stream,
230                    Err(e) => {
231                        self.telemetry.rpc_errors_total().increment(1);
232                        error!(namespace = self.namespace.name, error = %e, "Error watching container events.");
233
234                        sleep(Duration::from_secs(1)).await;
235                        continue;
236                    },
237                };
238
239                while let Some(event_result) = event_stream.next().await {
240                    let event = match event_result {
241                        Ok(event) => event,
242                        Err(e) => {
243                            error!(namespace = self.namespace.name, error = %e, "Error watching container events.");
244                            continue;
245                        },
246                    };
247
248                    if let Some(operation) = self.process_event(event).await {
249                        yield operation;
250                    }
251                }
252            }
253        })
254    }
255}
saluki_env/workload/collectors/containerd.rs

saluki_env/workload/collectors/
containerd.rs
