datadog_agent_commons/ipc/client/
mod.rs

1//! Helpers for interacting with the Datadog Agent.
2
3use std::time::Duration;
4
5use backon::{ConstantBuilder, Retryable as _};
6use datadog_protos::agent::v1::{
7    RefreshRemoteAgentRequest, RegisterRemoteAgentRequest, RegisterRemoteAgentResponse, ReportRemoteAgentEventRequest,
8    ReportRemoteAgentEventResponse,
9};
10use datadog_protos::agent::{
11    AgentClient, AgentSecureClient, AutodiscoveryStreamResponse, ConfigEvent, ConfigStreamRequest, EntityId,
12    FetchEntityRequest, HostTagReply, HostTagRequest, HostnameRequest, RemoteAgentClient as RemoteAgentServiceClient,
13    StreamTagsRequest, StreamTagsResponse, TagCardinality, WorkloadmetaEventType, WorkloadmetaFilter, WorkloadmetaKind,
14    WorkloadmetaSource, WorkloadmetaStreamRequest, WorkloadmetaStreamResponse,
15};
16use datadog_protos::remote_config::{ClientGetConfigsRequest, ClientGetConfigsResponse};
17use saluki_error::{generic_error, ErrorContext as _, GenericError};
18use saluki_io::net::client::http::HttpsCapableConnectorBuilder;
19use saluki_metadata::AppDetails;
20use tonic::{
21    service::interceptor::InterceptedService,
22    transport::{Channel, Endpoint},
23    Code, Request, Response, Status,
24};
25use tracing::warn;
26
27use crate::ipc::{config::RemoteAgentClientConfiguration, session::SessionId, tls::build_ipc_client_ipc_tls_config};
28
29mod bearer_auth;
30use self::bearer_auth::BearerAuthInterceptor;
31
32mod streaming;
33pub use self::streaming::StreamingResponse;
34
35const CONNECT_RETRY_ATTEMPTS: usize = 10;
36const CONNECT_RETRY_BACKOFF: Duration = Duration::from_secs(2);
37
38/// Formats an application name for the Agent's configuration-update stream and Remote Configuration requests.
39///
40/// Converts the full name to lowercase and replaces spaces and underscores with hyphens.
41pub fn client_name(app_details: &AppDetails) -> String {
42    app_details
43        .full_name()
44        .replace(" ", "-")
45        .replace("_", "-")
46        .to_lowercase()
47}
48
49/// A client for interacting with the Datadog Agent's internal gRPC-based API.
50#[derive(Clone)]
51pub struct RemoteAgentClient {
52    client: AgentClient<InterceptedService<Channel, BearerAuthInterceptor>>,
53    secure_client: AgentSecureClient<InterceptedService<Channel, BearerAuthInterceptor>>,
54    remote_agent_client: RemoteAgentServiceClient<InterceptedService<Channel, BearerAuthInterceptor>>,
55}
56
57impl RemoteAgentClient {
58    /// Connects to the Core Agent's gRPC IPC endpoint using the given client configuration.
59    ///
60    /// # Errors
61    ///
62    /// If the connection can't be established, the authentication token or IPC certificate can't be read, or the
63    /// initial health-check RPC fails, an error will be returned. Each of these is retried first.
64    pub async fn connect(config: &RemoteAgentClientConfiguration) -> Result<Self, GenericError> {
65        // TODO: We need to write a Tower middleware service that allows applying a backoff between failed calls,
66        // specifically so that we can throttle reconnection attempts.
67        //
68        // When the remote Agent endpoint is not available -- Agent isn't running, etc -- the gRPC client will
69        // essentially freewheel, trying to reconnect as quickly as possible, which spams the logs, wastes resources, so
70        // on and so forth. We would want to essentially apply a backoff like any other client would for the RPC calls
71        // themselves, but use it with the _connector_ instead.
72        //
73        // We could potentially just use a retry middleware, but Tonic does have its own reconnection logic, so we'd
74        // have to test it out to make sure it behaves sensibly.
75        let service_builder = || async {
76            let auth_interceptor = BearerAuthInterceptor::from_file(config.auth.auth_token_file_path()).await?;
77            let ipc_cert_file_path = config.auth.ipc_cert_file_path();
78            let client_tls_config = build_ipc_client_ipc_tls_config(ipc_cert_file_path).await?;
79            let connector_builder = HttpsCapableConnectorBuilder::default();
80            #[cfg(target_os = "linux")]
81            let connector_builder = if let Some(addr) = config.vsock_addr() {
82                connector_builder.with_vsock_addr(addr)
83            } else {
84                connector_builder
85            };
86            let https_connector = connector_builder.build(client_tls_config)?;
87            let endpoint = config.endpoint();
88            let channel = Endpoint::from(endpoint.clone())
89                .connect_timeout(Duration::from_secs(2))
90                .connect_with_connector(https_connector)
91                .await
92                .with_error_context(|| format!("Failed to connect to Datadog Agent API at '{}'.", endpoint))?;
93
94            let service = InterceptedService::new(channel, auth_interceptor);
95
96            // Health check inside the retried region. A transient partition can let the TCP connect succeed but break
97            // this first RPC stream, so retrying here keeps a boot-time blip from failing client construction.
98            let mut secure_client =
99                AgentSecureClient::new(service.clone()).max_decoding_message_size(config.grpc_max_message_size);
100            try_query_agent_api(&mut secure_client).await?;
101
102            Ok::<_, GenericError>(service)
103        };
104
105        let service = service_builder
106            .retry(
107                ConstantBuilder::default()
108                    .with_delay(CONNECT_RETRY_BACKOFF)
109                    .with_max_times(CONNECT_RETRY_ATTEMPTS),
110            )
111            .notify(|e, delay| {
112                warn!(error = %e, "Failed to create Datadog Agent API client. Retrying in {:?}...", delay);
113            })
114            .await
115            .error_context("Failed to create Datadog Agent API client.")?;
116
117        let client = AgentClient::new(service.clone()).max_decoding_message_size(config.grpc_max_message_size);
118        let secure_client =
119            AgentSecureClient::new(service.clone()).max_decoding_message_size(config.grpc_max_message_size);
120        let remote_agent_client =
121            RemoteAgentServiceClient::new(service).max_decoding_message_size(config.grpc_max_message_size);
122
123        Ok(Self {
124            client,
125            secure_client,
126            remote_agent_client,
127        })
128    }
129
130    /// Gets the detected hostname from the Agent.
131    ///
132    /// # Errors
133    ///
134    /// If there is an error querying the Agent API, an error will be returned.
135    pub async fn get_hostname(&mut self) -> Result<String, GenericError> {
136        let response = self
137            .client
138            .get_hostname(HostnameRequest {})
139            .await
140            .map(|r| r.into_inner())?;
141
142        Ok(response.hostname)
143    }
144
145    /// Gets a stream of tagger entities at the given cardinality, optionally limited to a set of entity ID prefixes.
146    ///
147    /// When `prefixes` is given, the server only sends events for entities whose ID carries one of those prefixes,
148    /// which avoids paying the serialization and bandwidth cost of entities that the caller has no use for. Passing
149    /// `None` applies no filtering, and the server sends events for every prefix it knows about.
150    ///
151    /// If there is an error with the initial request, or an error occurs while streaming, the next message in the
152    /// stream will be `Some(Err(status))`, where the status indicates the underlying error.
153    pub fn get_tagger_stream(
154        &mut self, cardinality: TagCardinality, prefixes: Option<Vec<String>>,
155    ) -> StreamingResponse<StreamTagsResponse> {
156        let mut client = self.secure_client.clone();
157        StreamingResponse::from_response_future(async move {
158            client
159                .tagger_stream_entities(StreamTagsRequest {
160                    cardinality: cardinality.into(),
161                    prefixes: prefixes.unwrap_or_default(),
162                    ..Default::default()
163                })
164                .await
165        })
166    }
167
168    /// Gets a stream of all workloadmeta entities.
169    ///
170    /// If there is an error with the initial request, or an error occurs while streaming, the next message in the
171    /// stream will be `Some(Err(status))`, where the status indicates the underlying error.
172    pub fn get_workloadmeta_stream(&mut self) -> StreamingResponse<WorkloadmetaStreamResponse> {
173        let mut client = self.secure_client.clone();
174        StreamingResponse::from_response_future(async move {
175            client
176                .workloadmeta_stream_entities(WorkloadmetaStreamRequest {
177                    filter: Some(WorkloadmetaFilter {
178                        kinds: vec![
179                            WorkloadmetaKind::Container.into(),
180                            WorkloadmetaKind::KubernetesPod.into(),
181                            WorkloadmetaKind::EcsTask.into(),
182                        ],
183                        source: WorkloadmetaSource::All.into(),
184                        event_type: WorkloadmetaEventType::EventTypeAll.into(),
185                    }),
186                })
187                .await
188        })
189    }
190
191    /// Registers a Remote Agent with the Agent.
192    ///
193    /// # Errors
194    ///
195    /// If there is an error sending the request to the Agent API, an error will be returned.
196    pub async fn register_remote_agent(
197        &mut self, pid: u32, display_name: &str, flavor: &str, api_endpoint: &str, services: Vec<String>,
198    ) -> Result<Response<RegisterRemoteAgentResponse>, GenericError> {
199        let mut client = self.remote_agent_client.clone();
200        let response = client
201            .register_remote_agent(RegisterRemoteAgentRequest {
202                pid: pid.to_string(),
203                flavor: flavor.to_string(),
204                display_name: display_name.to_string(),
205                api_endpoint_uri: api_endpoint.to_string(),
206                services,
207            })
208            .await?;
209        Ok(response)
210    }
211
212    /// Refreshes the given remote agent session with the Agent.
213    ///
214    /// # Errors
215    ///
216    /// If there is an error sending the request to the Agent API, an error will be returned.
217    pub async fn refresh_remote_agent(&mut self, session_id: &SessionId) -> Result<Response<()>, GenericError> {
218        let mut client = self.remote_agent_client.clone();
219        let response = client
220            .refresh_remote_agent(RefreshRemoteAgentRequest {
221                session_id: session_id.to_string(),
222            })
223            .await?
224            .map(|_| ());
225        Ok(response)
226    }
227
228    /// Reports one or more operational events for a remote agent session to the Agent.
229    ///
230    /// # Errors
231    ///
232    /// If there is an error sending the request to the Agent API, an error will be returned.
233    pub async fn report_remote_agent_event(
234        &mut self, request: ReportRemoteAgentEventRequest,
235    ) -> Result<Response<ReportRemoteAgentEventResponse>, GenericError> {
236        let mut client = self.remote_agent_client.clone();
237        let response = client.report_remote_agent_event(request).await?;
238        Ok(response)
239    }
240
241    /// Gets the host tags from the Agent.
242    ///
243    /// # Errors
244    ///
245    /// If there is an error querying the Agent API, an error will be returned.
246    pub async fn get_host_tags(&self) -> Result<Response<HostTagReply>, GenericError> {
247        let mut client = self.secure_client.clone();
248        let response = client.get_host_tags(HostTagRequest {}).await?;
249        Ok(response)
250    }
251
252    /// Polls the Agent for the Remote Configuration assigned to a client.
253    ///
254    /// # Errors
255    ///
256    /// Returns the gRPC status unchanged, so that callers can distinguish `Unimplemented`, which the Agent returns when
257    /// Remote Configuration is disabled, from other failures.
258    pub async fn client_get_configs(
259        &self, request: ClientGetConfigsRequest,
260    ) -> Result<ClientGetConfigsResponse, Status> {
261        let mut client = self.secure_client.clone();
262        let response = client.client_get_configs(request).await?;
263        Ok(response.into_inner())
264    }
265
266    /// Gets a stream of autodiscovery config updates.
267    ///
268    /// If there is an error with the initial request, or an error occurs while streaming, the next message in the
269    /// stream will be `Some(Err(status))`, where the status indicates the underlying error.
270    pub fn get_autodiscovery_stream(&mut self) -> StreamingResponse<AutodiscoveryStreamResponse> {
271        let mut client = self.secure_client.clone();
272        StreamingResponse::from_response_future(async move { client.autodiscovery_stream_config(()).await })
273    }
274
275    /// Gets a stream of config events.
276    ///
277    /// If there is an error with the initial request, or an error occurs while streaming, the next message in the
278    /// stream will be `Some(Err(status))`, where the status indicates the underlying error.
279    pub fn stream_config_events(&mut self, session_id: &SessionId) -> StreamingResponse<ConfigEvent> {
280        let mut client = self.secure_client.clone();
281        let mut request = Request::new(ConfigStreamRequest {
282            name: client_name(saluki_metadata::get_app_details()),
283        });
284
285        request
286            .metadata_mut()
287            .insert("session_id", session_id.to_grpc_header_value());
288
289        StreamingResponse::from_response_future(async move { client.stream_config_events(request).await })
290    }
291}
292
293async fn try_query_agent_api(
294    client: &mut AgentSecureClient<InterceptedService<Channel, BearerAuthInterceptor>>,
295) -> Result<(), GenericError> {
296    let noop_fetch_request = FetchEntityRequest {
297        id: Some(EntityId {
298            prefix: "container_id".to_string(),
299            uid: "nonexistent".to_string(),
300        }),
301        cardinality: TagCardinality::High.into(),
302    };
303    match client.tagger_fetch_entity(noop_fetch_request).await {
304        Ok(_) => Ok(()),
305        Err(e) => match e.code() {
306            Code::Unauthenticated => Err(generic_error!(
307                "Failed to authenticate to Datadog Agent API. Check that the configured authentication token is correct."
308            )),
309            _ => Err(e.into()),
310        },
311    }
312}
313
314#[cfg(test)]
315mod tests {
316    use saluki_metadata::{AppDetails, Version};
317
318    use super::client_name;
319
320    #[test]
321    fn client_name_is_the_lowercased_hyphenated_full_name() {
322        let details = AppDetails::new("Agent Data_Plane", "data-plane", "adp", Version::new("1.0.0", 1, 0, 0));
323        assert_eq!(client_name(&details), "agent-data-plane");
324    }
325}
datadog_agent_commons/ipc/client/mod.rs

datadog_agent_commons/ipc/client/
mod.rs
