saluki_env/workload/collectors/
containerd.rs

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

saluki_env/workload/collectors/
containerd.rs
