libdd_telemetry/worker/
http_client.rs

1// Copyright 2021-Present Datadog, Inc. https://www.datadoghq.com/
2// SPDX-License-Identifier: Apache-2.0
3
4use http_body_util::BodyExt;
5use libdd_common::{http_common, HttpRequestBuilder};
6use std::{
7    fs::OpenOptions,
8    future::Future,
9    io::Write,
10    pin::Pin,
11    sync::{Arc, Mutex},
12};
13
14use crate::config::Config;
15use tracing::{debug, error};
16
17pub mod header {
18    #![allow(clippy::declare_interior_mutable_const)]
19    use http::header::HeaderName;
20    pub const REQUEST_TYPE: HeaderName = HeaderName::from_static("dd-telemetry-request-type");
21    pub const API_VERSION: HeaderName = HeaderName::from_static("dd-telemetry-api-version");
22    pub const LIBRARY_LANGUAGE: HeaderName = HeaderName::from_static("dd-client-library-language");
23    pub const LIBRARY_VERSION: HeaderName = HeaderName::from_static("dd-client-library-version");
24
25    pub const DEBUG_ENABLED: HeaderName = HeaderName::from_static("dd-telemetry-debug-enabled");
26
27    pub const DD_SESSION_ID: HeaderName = HeaderName::from_static("dd-session-id");
28    pub const DD_ROOT_SESSION_ID: HeaderName = HeaderName::from_static("dd-root-session-id");
29    pub const DD_PARENT_SESSION_ID: HeaderName = HeaderName::from_static("dd-parent-session-id");
30}
31
32/// `session_id`, then `parent_session_id`, then `root_session_id` (must match call sites in
33/// `build_request`).
34pub(crate) fn add_instrumentation_session_headers(
35    mut builder: HttpRequestBuilder,
36    session_id: Option<&str>,
37    parent_session_id: Option<&str>,
38    root_session_id: Option<&str>,
39) -> HttpRequestBuilder {
40    let Some(s) = session_id.filter(|id| !id.is_empty()) else {
41        return builder;
42    };
43    builder = builder.header(header::DD_SESSION_ID, s);
44    if let Some(r) = root_session_id
45        .filter(|r| !r.is_empty())
46        .filter(|r| *r != s)
47    {
48        builder = builder.header(header::DD_ROOT_SESSION_ID, r);
49    }
50    if let Some(p) = parent_session_id
51        .filter(|p| !p.is_empty())
52        .filter(|p| *p != s)
53    {
54        builder = builder.header(header::DD_PARENT_SESSION_ID, p);
55    }
56    builder
57}
58
59pub type ResponseFuture =
60    Pin<Box<dyn Future<Output = Result<http_common::HttpResponse, http_common::Error>> + Send>>;
61
62pub trait HttpClient {
63    fn request(&self, req: http_common::HttpRequest) -> ResponseFuture;
64}
65
66pub fn request_builder(c: &Config) -> anyhow::Result<HttpRequestBuilder> {
67    match &c.endpoint {
68        Some(e) => {
69            debug!(
70                endpoint.url = %e.url,
71                endpoint.timeout_ms = e.timeout_ms,
72                telemetry.version = env!("CARGO_PKG_VERSION"),
73                "Building telemetry request"
74            );
75            let mut builder =
76                e.to_request_builder(concat!("telemetry/", env!("CARGO_PKG_VERSION")));
77            if c.debug_enabled {
78                debug!(
79                    telemetry.debug_enabled = true,
80                    "Telemetry debug mode enabled"
81                );
82                builder = Ok(builder?.header(header::DEBUG_ENABLED, "true"))
83            }
84            builder
85        }
86        None => {
87            error!("No valid telemetry endpoint found, cannot build request");
88            Err(anyhow::Error::msg(
89                "no valid endpoint found, can't build the request".to_string(),
90            ))
91        }
92    }
93}
94
95pub fn from_config(c: &Config) -> Box<dyn HttpClient + Sync + Send> {
96    match &c.endpoint {
97        Some(e) if e.url.scheme_str() == Some("file") => {
98            #[allow(clippy::expect_used)]
99            let file_path = libdd_common::decode_uri_path_in_authority(&e.url)
100                .expect("file urls should always have been encoded in authority");
101            debug!(
102                file.path = ?file_path,
103                "Using file-based mock telemetry client"
104            );
105            return Box::new(MockClient {
106                #[allow(clippy::expect_used)]
107                file: Arc::new(Mutex::new(Box::new(
108                    OpenOptions::new()
109                        .create(true)
110                        .append(true)
111                        .open(file_path.as_path())
112                        .expect("Couldn't open mock client file"),
113                ))),
114            });
115        }
116        Some(e) => {
117            debug!(
118                endpoint.url = %e.url,
119                endpoint.timeout_ms = e.timeout_ms,
120                "Using HTTP telemetry client"
121            );
122        }
123        None => {
124            debug!(
125                endpoint = "default",
126                "No telemetry endpoint configured, using default HTTP client"
127            );
128        }
129    };
130    Box::new(HyperClient {
131        inner: http_common::new_client_periodic(),
132    })
133}
134
135pub struct HyperClient {
136    inner: libdd_common::HttpClient,
137}
138
139impl HttpClient for HyperClient {
140    fn request(&self, req: http_common::HttpRequest) -> ResponseFuture {
141        let resp = self.inner.request(req);
142        Box::pin(async move {
143            match resp.await {
144                Ok(response) => Ok(http_common::into_response(response)),
145                Err(e) => Err(http_common::Error::Client(e.into())),
146            }
147        })
148    }
149}
150
151#[derive(Clone)]
152pub struct MockClient {
153    file: Arc<Mutex<Box<dyn Write + Sync + Send>>>,
154}
155
156impl HttpClient for MockClient {
157    fn request(&self, req: http_common::HttpRequest) -> ResponseFuture {
158        let s = self.clone();
159        Box::pin(async move {
160            debug!("MockClient writing request to file");
161            let mut body = req.collect().await?.to_bytes().to_vec();
162            body.push(b'\n');
163
164            {
165                #[allow(clippy::expect_used)]
166                let mut writer = s.file.lock().expect("mutex poisoned");
167
168                match writer.write_all(body.as_ref()) {
169                    Ok(()) => debug!(
170                        file.bytes_written = body.len(),
171                        "Successfully wrote payload to mock file"
172                    ),
173                    Err(e) => {
174                        error!(
175                            error = %e,
176                            "Failed to write to mock file"
177                        );
178                        return Err(http_common::Error::from(e));
179                    }
180                }
181            }
182
183            debug!(http.status = 202, "MockClient returning success response");
184            http_common::empty_response(http::Response::builder().status(202))
185        })
186    }
187}
188
189#[cfg(test)]
190mod tests {
191    use libdd_common::HttpRequestBuilder;
192
193    use super::*;
194
195    #[tokio::test]
196    #[cfg_attr(miri, ignore)]
197    async fn test_mock_client() {
198        let output: Vec<u8> = Vec::new();
199        let c = MockClient {
200            file: Arc::new(Mutex::new(Box::new(output))),
201        };
202        c.request(
203            HttpRequestBuilder::new()
204                .body(http_common::Body::from("hello world\n"))
205                .unwrap(),
206        )
207        .await
208        .unwrap();
209    }
210}
libdd_telemetry/worker/http_client.rs

libdd_telemetry/worker/
http_client.rs
