saluki_env/workload/helpers/containerd/
mod.rs

1use std::{
2    path::{Path, PathBuf},
3    time::Duration,
4};
5
6use containerd_protos::services::{
7    containers::v1::{containers_client::ContainersClient, Container, ListContainersRequest},
8    events::v1::{events_client::EventsClient, SubscribeRequest},
9    namespaces::v1::{namespaces_client::NamespacesClient, ListNamespacesRequest, Namespace},
10    tasks::v1::{tasks_client::TasksClient, ListPidsRequest},
11};
12use futures::{Stream, StreamExt as _, TryStreamExt as _};
13use hyper_util::rt::TokioIo;
14use saluki_error::{generic_error, GenericError};
15use snafu::{ResultExt as _, Snafu};
16use tokio::net::UnixStream;
17use tonic::{
18    transport::{Channel, Endpoint},
19    IntoRequest, Request,
20};
21use tower::service_fn;
22
23use crate::features::ContainerdDetector;
24
25pub mod events;
26use self::events::{decode_envelope_to_event, ContainerdEvent, ContainerdTopic};
27
28const MAX_LIST_CONTAINERS_RESPONSE_SIZE: usize = 16 * 1024 * 1024;
29
30/// Containerd gRPC client configuration.
31pub struct ContainerdConfiguration {
32    /// Timeout for establishing the gRPC connection to the containerd socket.
33    pub connection_timeout: Duration,
34
35    /// Per-RPC timeout for containerd API calls.
36    pub query_timeout: Duration,
37}
38
39/// A [`ContainerdClient`] error.
40#[derive(Debug, Snafu)]
41#[snafu(context(suffix(false)))]
42pub enum ClientError {
43    /// Failed to send a gRPC request.
44    #[snafu(display("failed to make gRPC request: {}", source))]
45    Response { source: tonic::Status },
46
47    /// Received an invalid event response from containerd.
48    #[snafu(display("invalid containerd event response: {}", reason))]
49    InvalidEvent { reason: String },
50}
51
52impl ClientError {
53    /// Gets the status of the response, if this is a response error.
54    pub fn as_response_error(&self) -> Option<tonic::Status> {
55        match self {
56            ClientError::Response { source } => Some(source.clone()),
57            _ => None,
58        }
59    }
60}
61
62/// Containerd gRPC client.
63#[derive(Clone)]
64pub struct ContainerdClient {
65    channel: Channel,
66    query_timeout: Duration,
67}
68
69impl ContainerdClient {
70    /// Creates a new `ContainerdClient` from the given socket path and client configuration.
71    ///
72    /// If `socket_path` is given, that path is used. Otherwise, well-known paths are probed.
73    ///
74    /// ## Errors
75    ///
76    /// If the containerd socket path wasn't given and couldn't be detected, or if the gRPC transport to containerd
77    /// couldn't be created, an error will be returned.
78    pub async fn new(
79        socket_path: Option<PathBuf>, containerd_config: &ContainerdConfiguration,
80    ) -> Result<Self, GenericError> {
81        let detected_socket_path = ContainerdDetector::detect_grpc_socket_path(socket_path)
82            .ok_or(generic_error!(
83                "failed to detect containerd socket path; not available at default path and not specified in configuration (`cri_socket_path`)"
84            ))?;
85
86        if !path_exists(&detected_socket_path).await {
87            return Err(generic_error!(
88                "Detected containerd socket path ({}) but path does not exist, or process lacks permissions.",
89                detected_socket_path.to_string_lossy()
90            ));
91        }
92
93        let channel = Endpoint::try_from("https://[::]")
94            .unwrap()
95            .connect_timeout(containerd_config.connection_timeout)
96            .connect_with_connector(service_fn(move |_| {
97                let socket_path = detected_socket_path.clone();
98                async move { UnixStream::connect(socket_path).await.map(TokioIo::new) }
99            }))
100            .await?;
101
102        Ok(Self {
103            channel,
104            query_timeout: containerd_config.query_timeout,
105        })
106    }
107
108    /// Lists all namespaces.
109    ///
110    /// ## Errors
111    ///
112    /// If an error occurs while sending the request or receiving the response, an error will be returned.
113    pub async fn list_namespaces(&self) -> Result<Vec<Namespace>, ClientError> {
114        let request = create_timed_request(ListNamespacesRequest::default(), self.query_timeout);
115
116        let mut client = NamespacesClient::new(self.channel.clone());
117        let namespaces = client.list(request).await.context(Response)?.into_inner();
118
119        Ok(namespaces.namespaces)
120    }
121
122    /// Lists all containers in the given namespace.
123    ///
124    /// ## Errors
125    ///
126    /// If an error occurs while sending the request or receiving the response, an error will be returned.
127    pub async fn list_containers(&self, namespace: &Namespace) -> Result<Vec<Container>, ClientError> {
128        let request = ListContainersRequest::default();
129        let request = create_timed_namespaced_request(request, namespace, self.query_timeout);
130
131        let client = ContainersClient::new(self.channel.clone());
132        let response = client
133            .max_decoding_message_size(MAX_LIST_CONTAINERS_RESPONSE_SIZE)
134            .list(request)
135            .await
136            .context(Response)?
137            .into_inner();
138
139        Ok(response.containers)
140    }
141
142    /// Watches for specific containerd events in the given namespace.
143    ///
144    /// Multiple topics (topics map directly to event types) can be watched on the same stream.
145    ///
146    /// ## Errors
147    ///
148    /// If an error occurs while sending the request or receiving the response, an error will be returned.
149    pub async fn watch_events(
150        &self, topics: &[ContainerdTopic], namespace: &Namespace,
151    ) -> Result<impl Stream<Item = Result<ContainerdEvent, ClientError>> + Unpin, ClientError> {
152        // Create our subscribe request, which requires a filter for each topic we're interested in, that is also
153        // combined with the namespace that should be watched.
154        let mut filters = Vec::new();
155        for topic in topics {
156            filters.push(format!(
157                "topic==\"{}\",namespace=={}",
158                topic.as_topic_str(),
159                namespace.name
160            ));
161        }
162
163        // Match datadog-agent behavior: `cri_query_timeout` is used for unary CRI/containerd queries, but event
164        // subscriptions are long-lived streams that should stay open until the collector is canceled.
165        let request = SubscribeRequest { filters };
166
167        let mut client = EventsClient::new(self.channel.clone());
168        let response = client.subscribe(request).await.context(Response)?.into_inner();
169
170        Ok(response
171            .map_err(|source| ClientError::Response { source })
172            .filter_map(|result| async move {
173                // Filter out all of the non-container events from the stream, converting container events to
174                // `ContainerEvent` in the process.
175                //
176                // If the message is an error, we pass it through unfiltered.
177                result
178                    .and_then(|envelope| {
179                        decode_envelope_to_event(envelope).map_err(|_| ClientError::InvalidEvent {
180                            reason: "failed to decode envelope payload".to_string(),
181                        })
182                    })
183                    .transpose()
184            })
185            .boxed())
186    }
187
188    /// Lists all process IDs for the given container in the given namespace.
189    ///
190    /// ## Errors
191    ///
192    /// If an error occurs while sending the request or receiving the response, an error will be returned.
193    pub async fn list_pids_for_container(
194        &self, namespace: &Namespace, container_id: String,
195    ) -> Result<Vec<u32>, ClientError> {
196        let request = ListPidsRequest { container_id };
197        let request = create_timed_namespaced_request(request, namespace, self.query_timeout);
198
199        let mut client = TasksClient::new(self.channel.clone());
200        let response = client.list_pids(request).await.context(Response)?.into_inner();
201
202        Ok(response.processes.into_iter().map(|p| p.pid).collect())
203    }
204}
205
206fn create_timed_request<R>(req: R, timeout: Duration) -> Request<R>
207where
208    R: IntoRequest<R>,
209{
210    let mut req = req.into_request();
211    req.set_timeout(timeout);
212    req
213}
214
215fn create_timed_namespaced_request<R>(req: R, ns: &Namespace, timeout: Duration) -> Request<R>
216where
217    R: IntoRequest<R>,
218{
219    let mut req = create_timed_request(req, timeout);
220    let md = req.metadata_mut();
221    md.insert("containerd-namespace", ns.name.parse().unwrap());
222    req
223}
224
225async fn path_exists(path: &Path) -> bool {
226    tokio::fs::metadata(path).await.is_ok()
227}
228
229#[cfg(test)]
230mod tests {
231    use std::time::Duration;
232
233    use super::*;
234
235    #[test]
236    fn create_timed_request_sets_grpc_timeout() {
237        let request = create_timed_request(ListNamespacesRequest::default(), Duration::from_secs(3));
238
239        assert_eq!(
240            Some("3000000u"),
241            request.metadata().get("grpc-timeout").map(|v| v.to_str().unwrap())
242        );
243    }
244
245    #[test]
246    fn create_timed_namespaced_request_sets_namespace_and_grpc_timeout() {
247        let namespace = Namespace {
248            name: "k8s.io".to_string(),
249            labels: Default::default(),
250        };
251        let request =
252            create_timed_namespaced_request(ListContainersRequest::default(), &namespace, Duration::from_secs(4));
253
254        assert_eq!(
255            Some("4000000u"),
256            request.metadata().get("grpc-timeout").map(|v| v.to_str().unwrap())
257        );
258        assert_eq!(
259            Some("k8s.io"),
260            request
261                .metadata()
262                .get("containerd-namespace")
263                .map(|v| v.to_str().unwrap())
264        );
265    }
266}
saluki_env/workload/helpers/containerd/mod.rs

saluki_env/workload/helpers/containerd/
mod.rs
