datadog_agent_commons/ipc/client/
mod.rs

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

datadog_agent_commons/ipc/client/
mod.rs
