libdd_common/
lib.rs

1// Copyright 2021-Present Datadog, Inc. https://www.datadoghq.com/
2// SPDX-License-Identifier: Apache-2.0
3#![cfg_attr(not(test), deny(clippy::panic))]
4#![cfg_attr(not(test), deny(clippy::unwrap_used))]
5#![cfg_attr(not(test), deny(clippy::expect_used))]
6#![cfg_attr(not(test), deny(clippy::todo))]
7#![cfg_attr(not(test), deny(clippy::unimplemented))]
8
9extern crate alloc;
10
11use alloc::borrow::Cow;
12use anyhow::Context;
13use core::{ops::Deref, str::FromStr};
14use http::uri::{self, PathAndQuery, Uri};
15use serde::de::Error;
16use serde::{Deserialize, Deserializer, Serialize, Serializer};
17use std::path::PathBuf;
18use std::sync::{Mutex, MutexGuard, RwLock, RwLockReadGuard, RwLockWriteGuard};
19
20pub mod azure_app_services;
21#[cfg(not(target_arch = "wasm32"))]
22pub mod cc_utils;
23#[cfg(not(target_arch = "wasm32"))]
24pub mod connector;
25#[cfg(feature = "reqwest")]
26pub mod dump_server;
27pub mod entity_id;
28pub mod machine_id;
29pub mod regex_engine;
30#[macro_use]
31pub mod cstr;
32#[cfg(feature = "bench-utils")]
33pub mod bench_utils;
34pub mod config;
35pub mod error;
36pub mod http_common;
37pub mod multipart;
38#[cfg(not(target_arch = "wasm32"))]
39pub mod rate_limiter;
40pub mod tag;
41#[cfg(any(test, feature = "test-utils"))]
42pub mod test_utils;
43#[cfg(not(target_arch = "wasm32"))]
44pub mod threading;
45#[cfg(not(target_arch = "wasm32"))]
46pub mod timeout;
47pub mod unix_utils;
48
49/// Extension trait for `Mutex` to provide a method that acquires a lock, panicking if the lock is
50/// poisoned.
51///
52/// This helper function is intended to be used to avoid having to add many
53/// `#[allow(clippy::unwrap_used)]` annotations if there are a lot of usages of `Mutex`.
54///
55/// # Arguments
56///
57/// * `self` - A reference to the `Mutex` to lock.
58///
59/// # Returns
60///
61/// A `MutexGuard` that provides access to the locked data.
62///
63/// # Panics
64///
65/// This function will panic if the `Mutex` is poisoned.
66///
67/// # Examples
68///
69/// ```
70/// use libdd_common::MutexExt;
71/// use std::sync::{Arc, Mutex};
72///
73/// let data = Arc::new(Mutex::new(5));
74/// let data_clone = Arc::clone(&data);
75///
76/// std::thread::spawn(move || {
77///     let mut num = data_clone.lock_or_panic();
78///     *num += 1;
79/// })
80/// .join()
81/// .expect("Thread panicked");
82///
83/// assert_eq!(*data.lock_or_panic(), 6);
84/// ```
85pub trait MutexExt<T> {
86    fn lock_or_panic(&self) -> MutexGuard<'_, T>;
87}
88
89impl<T> MutexExt<T> for Mutex<T> {
90    #[inline(always)]
91    #[track_caller]
92    fn lock_or_panic(&self) -> MutexGuard<'_, T> {
93        #[allow(clippy::unwrap_used)]
94        self.lock().unwrap()
95    }
96}
97
98/// Extension trait for `RwLock` to provide methods that acquire read/write locks, panicking if
99/// the lock is poisoned.
100///
101/// Mirrors [`MutexExt`] for `RwLock` so callers avoid `#[allow(clippy::unwrap_used)]` at each
102/// lock site.
103///
104/// # Examples
105///
106/// ```
107/// use libdd_common::RwLockExt;
108/// use std::sync::{Arc, RwLock};
109///
110/// let data = Arc::new(RwLock::new(5));
111/// let data_clone = Arc::clone(&data);
112///
113/// std::thread::spawn(move || {
114///     let mut num = data_clone.write_or_panic();
115///     *num += 1;
116/// })
117/// .join()
118/// .expect("Thread panicked");
119///
120/// assert_eq!(*data.read_or_panic(), 6);
121/// ```
122pub trait RwLockExt<T> {
123    fn read_or_panic(&self) -> RwLockReadGuard<'_, T>;
124    fn write_or_panic(&self) -> RwLockWriteGuard<'_, T>;
125}
126
127impl<T> RwLockExt<T> for RwLock<T> {
128    #[inline(always)]
129    #[track_caller]
130    fn read_or_panic(&self) -> RwLockReadGuard<'_, T> {
131        #[allow(clippy::unwrap_used)]
132        self.read().unwrap()
133    }
134
135    #[inline(always)]
136    #[track_caller]
137    fn write_or_panic(&self) -> RwLockWriteGuard<'_, T> {
138        #[allow(clippy::unwrap_used)]
139        self.write().unwrap()
140    }
141}
142
143/// Extension trait that extracts the value from a `Result` whose error type is uninhabited.
144///
145/// The signature constrains callers at compile time: the method is only available when the
146/// error type is [`core::convert::Infallible`]. No panics — the compiler proves the `Err`
147/// arm unreachable from the type.
148///
149/// # Examples
150///
151/// ```
152/// use libdd_common::ResultInfallibleExt;
153/// use std::convert::Infallible;
154///
155/// let result: Result<i32, Infallible> = Ok(42);
156/// assert_eq!(result.unwrap_infallible(), 42);
157/// ```
158pub trait ResultInfallibleExt<T>: sealed::Sealed {
159    fn unwrap_infallible(self) -> T;
160}
161
162impl<T> ResultInfallibleExt<T> for Result<T, core::convert::Infallible> {
163    #[inline(always)]
164    fn unwrap_infallible(self) -> T {
165        match self {
166            Ok(value) => value,
167            Err(never) => match never {},
168        }
169    }
170}
171
172mod sealed {
173    pub trait Sealed {}
174    impl<T> Sealed for Result<T, core::convert::Infallible> {}
175}
176
177pub mod header {
178    #![allow(clippy::declare_interior_mutable_const)]
179    use http::{header::HeaderName, HeaderValue};
180
181    pub const APPLICATION_MSGPACK_STR: &str = "application/msgpack";
182    pub const APPLICATION_PROTOBUF_STR: &str = "application/x-protobuf";
183
184    pub const DATADOG_CONTAINER_ID: HeaderName = HeaderName::from_static("datadog-container-id");
185    pub const DATADOG_ENTITY_ID: HeaderName = HeaderName::from_static("datadog-entity-id");
186    pub const DATADOG_EXTERNAL_ENV: HeaderName = HeaderName::from_static("datadog-external-env");
187    pub const DATADOG_TRACE_COUNT: HeaderName = HeaderName::from_static("x-datadog-trace-count");
188    /// Signal to the agent to send 429 responses when a payload is dropped
189    /// If this is not set then the agent will always return a 200 regardless if the payload is
190    /// dropped.
191    pub const DATADOG_SEND_REAL_HTTP_STATUS: HeaderName =
192        HeaderName::from_static("datadog-send-real-http-status");
193    pub const DATADOG_API_KEY: HeaderName = HeaderName::from_static("dd-api-key");
194    pub const APPLICATION_JSON: HeaderValue = HeaderValue::from_static("application/json");
195    pub const APPLICATION_MSGPACK: HeaderValue = HeaderValue::from_static(APPLICATION_MSGPACK_STR);
196    pub const APPLICATION_PROTOBUF: HeaderValue =
197        HeaderValue::from_static(APPLICATION_PROTOBUF_STR);
198    pub const X_DATADOG_TEST_SESSION_TOKEN: HeaderName =
199        HeaderName::from_static("x-datadog-test-session-token");
200}
201
202#[cfg(not(target_arch = "wasm32"))]
203pub type HttpClient = http_common::GenericHttpClient<connector::Connector>;
204#[cfg(not(target_arch = "wasm32"))]
205pub type HttpResponse = http_common::HttpResponse;
206pub type HttpRequestBuilder = http::request::Builder;
207#[cfg(not(target_arch = "wasm32"))]
208pub trait Connect:
209    hyper_util::client::legacy::connect::Connect + Clone + Send + Sync + 'static
210{
211}
212#[cfg(not(target_arch = "wasm32"))]
213impl<C: hyper_util::client::legacy::connect::Connect + Clone + Send + Sync + 'static> Connect
214    for C
215{
216}
217
218// Used by tag! macro
219pub use const_format;
220
221#[derive(Clone, PartialEq, Eq, Hash, Debug, Serialize, Deserialize)]
222pub struct Endpoint {
223    #[serde(serialize_with = "serialize_uri", deserialize_with = "deserialize_uri")]
224    pub url: http::Uri,
225    pub api_key: Option<Cow<'static, str>>,
226    pub timeout_ms: u64,
227    /// Sets X-Datadog-Test-Session-Token header on any request
228    pub test_token: Option<Cow<'static, str>>,
229    /// Use the system DNS resolver when building the HTTP client. If false, the default
230    /// in-process resolver is used.
231    #[serde(default)]
232    pub use_system_resolver: bool,
233}
234
235impl Default for Endpoint {
236    fn default() -> Self {
237        Endpoint {
238            url: http::Uri::default(),
239            api_key: None,
240            timeout_ms: Self::DEFAULT_TIMEOUT,
241            test_token: None,
242            use_system_resolver: false,
243        }
244    }
245}
246
247#[derive(serde::Deserialize, serde::Serialize)]
248struct SerializedUri<'a> {
249    scheme: Option<Cow<'a, str>>,
250    authority: Option<Cow<'a, str>>,
251    path_and_query: Option<Cow<'a, str>>,
252}
253
254fn serialize_uri<S>(uri: &http::Uri, serializer: S) -> Result<S::Ok, S::Error>
255where
256    S: Serializer,
257{
258    let parts = uri.clone().into_parts();
259    let uri = SerializedUri {
260        scheme: parts.scheme.as_ref().map(|s| Cow::Borrowed(s.as_str())),
261        authority: parts.authority.as_ref().map(|s| Cow::Borrowed(s.as_str())),
262        path_and_query: parts
263            .path_and_query
264            .as_ref()
265            .map(|s| Cow::Borrowed(s.as_str())),
266    };
267    uri.serialize(serializer)
268}
269
270fn deserialize_uri<'de, D>(deserializer: D) -> Result<http::Uri, D::Error>
271where
272    D: Deserializer<'de>,
273{
274    let uri = SerializedUri::deserialize(deserializer)?;
275    let mut builder = http::Uri::builder();
276    if let Some(v) = uri.authority {
277        builder = builder.authority(v.deref());
278    }
279    if let Some(v) = uri.scheme {
280        builder = builder.scheme(v.deref());
281    }
282    if let Some(v) = uri.path_and_query {
283        builder = builder.path_and_query(v.deref());
284    }
285
286    builder.build().map_err(Error::custom)
287}
288
289/// Converts a human-facing URL string into the internal [`http::Uri`]
290/// representation.
291///
292/// NOTE: the name is misleading. For `http`/`https` this is an ordinary parse,
293/// but for the `file`/`unix`/`windows` schemes it *encodes* the path into the
294/// URI authority (see `encode_uri_path_in_authority`), so it is a
295/// URL-string-to-`Uri` *constructor*, not a pure parser.
296///
297/// WARNING: this is NOT idempotent for those three schemes. The `Uri` it
298/// returns stringifies back to the encoded form (`file://<hex>/`), and feeding
299/// that string in again re-encodes it, double-encoding the path. Only ever call
300/// this on an original URL string — never on the `.to_string()` of a `Uri` that
301/// already came out of here.
302///
303/// TODO: we should properly handle malformed urls
304/// * For windows and unix schemes:
305///     * For compatibility reasons with existing implementation this parser stores the encoded path
306///       in authority section as there is no existing standard [see](https://github.com/whatwg/url/issues/577)
307///       that covers this. We need to pick one hack or another
308///     * For windows, interprets everything after windows: as path
309///     * For unix, interprets everything after unix:// as path
310/// * For file scheme implementation will simply backfill missing authority section
311pub fn parse_uri(uri: &str) -> anyhow::Result<http::Uri> {
312    if let Some(path) = uri.strip_prefix("unix://") {
313        encode_uri_path_in_authority("unix", path)
314    } else if let Some(path) = uri.strip_prefix("windows:") {
315        encode_uri_path_in_authority("windows", path)
316    } else if let Some(path) = uri.strip_prefix("file://") {
317        encode_uri_path_in_authority("file", path)
318    } else {
319        Ok(http::Uri::from_str(uri)?)
320    }
321}
322
323fn encode_uri_path_in_authority(scheme: &str, path: &str) -> anyhow::Result<http::Uri> {
324    let mut parts = uri::Parts::default();
325    parts.scheme = uri::Scheme::from_str(scheme).ok();
326
327    let path = hex::encode(path);
328
329    parts.authority = uri::Authority::from_str(path.as_str()).ok();
330    parts.path_and_query = Some(uri::PathAndQuery::from_static("/"));
331    Ok(http::Uri::from_parts(parts)?)
332}
333
334pub fn decode_uri_path_in_authority(uri: &http::Uri) -> anyhow::Result<PathBuf> {
335    let path = hex::decode(uri.authority().context("missing uri authority")?.as_str())?;
336    #[cfg(unix)]
337    {
338        use std::os::unix::ffi::OsStringExt;
339        Ok(PathBuf::from(std::ffi::OsString::from_vec(path)))
340    }
341    #[cfg(not(unix))]
342    {
343        match String::from_utf8(path) {
344            Ok(s) => Ok(PathBuf::from(s.as_str())),
345            _ => Err(anyhow::anyhow!("file uri should be utf-8")),
346        }
347    }
348}
349
350impl Endpoint {
351    /// Default value for the timeout field in milliseconds.
352    pub const DEFAULT_TIMEOUT: u64 = 3_000;
353
354    pub fn agentless(site: &str, api_key: String) -> anyhow::Result<Self> {
355        Ok(Self {
356            url: Uri::builder()
357                .scheme("https")
358                .authority(
359                    uri::Authority::try_from(site)
360                        .with_context(|| format!("dd_site is an invalid url: {site}"))?,
361                )
362                .path_and_query(PathAndQuery::from_static(""))
363                .build()
364                .with_context(|| format!("rc url is invalid for site: {site}"))?,
365            api_key: Some(api_key.into()),
366            timeout_ms: Self::DEFAULT_TIMEOUT,
367            test_token: None,
368            use_system_resolver: true,
369        })
370    }
371
372    /// Returns an iterator of optional endpoint-specific headers (api-key, test-token)
373    /// as (header_name, header_value) string tuples for any that are available.
374    pub fn get_optional_headers(&self) -> impl Iterator<Item = (&'static str, &str)> {
375        [
376            self.api_key.as_ref().map(|v| ("dd-api-key", v.as_ref())),
377            self.test_token
378                .as_ref()
379                .map(|v| ("x-datadog-test-session-token", v.as_ref())),
380        ]
381        .into_iter()
382        .flatten()
383    }
384
385    /// Apply standard headers (user-agent, api-key, test-token, entity headers) to an
386    /// [`http::request::Builder`].
387    pub fn set_standard_headers(
388        &self,
389        mut builder: http::request::Builder,
390        user_agent: &str,
391    ) -> http::request::Builder {
392        builder = builder.header("user-agent", user_agent);
393        for (name, value) in self.get_optional_headers() {
394            builder = builder.header(name, value);
395        }
396        for (name, value) in entity_id::get_entity_headers() {
397            builder = builder.header(name, value);
398        }
399        builder
400    }
401
402    /// Return a request builder with the following headers:
403    /// - User agent
404    /// - Api key
405    /// - Container Id/Entity Id
406    pub fn to_request_builder(&self, user_agent: &str) -> anyhow::Result<HttpRequestBuilder> {
407        let mut builder = http::Request::builder()
408            .uri(self.url.clone())
409            .header(http::header::USER_AGENT, user_agent);
410
411        // Add optional endpoint headers (api-key, test-token)
412        for (name, value) in self.get_optional_headers() {
413            builder = builder.header(name, value);
414        }
415
416        // Add entity-related headers (container-id, entity-id, external-env)
417        for (name, value) in entity_id::get_entity_headers() {
418            builder = builder.header(name, value);
419        }
420
421        Ok(builder)
422    }
423
424    #[inline]
425    pub fn from_slice(url: &str) -> Endpoint {
426        Endpoint {
427            #[allow(clippy::unwrap_used)]
428            url: parse_uri(url).unwrap(),
429            ..Default::default()
430        }
431    }
432
433    #[inline]
434    pub fn from_url(url: http::Uri) -> Endpoint {
435        Endpoint {
436            url,
437            ..Default::default()
438        }
439    }
440
441    pub fn is_file_endpoint(&self) -> bool {
442        self.url.scheme_str() == Some("file")
443    }
444
445    /// Set a custom timeout for this endpoint.
446    /// If not called, uses the default timeout of 3000ms.
447    ///
448    /// # Arguments
449    /// * `timeout_ms` - Timeout in milliseconds. Pass 0 to use the default timeout (3000ms).
450    ///
451    /// # Returns
452    /// Self with the timeout set, allowing for method chaining
453    pub fn with_timeout(mut self, timeout_ms: u64) -> Self {
454        self.timeout_ms = if timeout_ms == 0 {
455            Self::DEFAULT_TIMEOUT
456        } else {
457            timeout_ms
458        };
459        self
460    }
461
462    /// Use the system DNS resolver when building the reqwest client. Only has effect for
463    /// HTTP(S) endpoints.
464    pub fn with_system_resolver(mut self, use_system_resolver: bool) -> Self {
465        self.use_system_resolver = use_system_resolver;
466        self
467    }
468
469    /// Creates a reqwest ClientBuilder configured for this endpoint.
470    ///
471    /// This method handles various endpoint schemes:
472    /// - `http`/`https`: Standard HTTP(S) endpoints
473    /// - `unix`: Unix domain sockets (Unix only)
474    /// - `windows`: Windows named pipes (Windows only)
475    /// - `file`: File dump endpoints for debugging (spawns a local server to capture requests)
476    ///
477    /// The default in-process resolver is used for DNS (fork-safe). To use the system DNS resolver
478    /// instead (less fork-safe), set [`Endpoint::use_system_resolver`] to true via
479    /// [`Endpoint::with_system_resolver`].
480    ///
481    /// # Returns
482    /// A tuple of (ClientBuilder, request_url) where:
483    /// - ClientBuilder is configured with the appropriate transport and timeout
484    /// - request_url is the URL string to use for HTTP requests
485    ///
486    /// # Errors
487    /// Returns an error if:
488    /// - The endpoint scheme is unsupported
489    /// - Path decoding fails
490    /// - The dump server fails to start (for file:// scheme)
491    #[cfg(feature = "reqwest")]
492    pub fn to_reqwest_client_builder(&self) -> anyhow::Result<(reqwest::ClientBuilder, String)> {
493        use anyhow::Context;
494
495        // Don't use proxies, as this calls `getenv` which is unsafe and not
496        // just in theory. It can cause crashes with PHP where php-fpm's env
497        // configuration will mutate the system environment (it doesn't pass
498        // it as part of the SAPI env, it changes the actual system env).
499        let mut builder = reqwest::Client::builder()
500            .timeout(core::time::Duration::from_millis(self.timeout_ms))
501            .hickory_dns(!self.use_system_resolver)
502            .no_proxy();
503
504        let request_url = match self.url.scheme_str() {
505            // HTTP/HTTPS endpoints
506            Some("http") | Some("https") => self.url.to_string(),
507
508            // File dump endpoint (debugging) - uses platform-specific local transport
509            Some("file") => {
510                let output_path = decode_uri_path_in_authority(&self.url)
511                    .context("Failed to decode file path from URI")?;
512                let socket_or_pipe_path = dump_server::spawn_dump_server(output_path)?;
513
514                // Configure the client to use the local socket/pipe
515                #[cfg(unix)]
516                {
517                    builder = builder.unix_socket(socket_or_pipe_path);
518                }
519                #[cfg(windows)]
520                {
521                    builder = builder
522                        .windows_named_pipe(socket_or_pipe_path.to_string_lossy().to_string());
523                }
524
525                "http://localhost/".to_string()
526            }
527
528            // Unix domain sockets
529            #[cfg(unix)]
530            Some("unix") => {
531                use connector::uds::socket_path_from_uri;
532                let socket_path = socket_path_from_uri(&self.url)?;
533                builder = builder.unix_socket(socket_path);
534                format!("http://localhost{}", self.url.path())
535            }
536
537            // Windows named pipes
538            #[cfg(windows)]
539            Some("windows") => {
540                use connector::named_pipe::named_pipe_path_from_uri;
541                let pipe_path = named_pipe_path_from_uri(&self.url)?;
542                builder = builder.windows_named_pipe(pipe_path.to_string_lossy().to_string());
543                format!("http://localhost{}", self.url.path())
544            }
545
546            // Unsupported schemes
547            scheme => anyhow::bail!("Unsupported endpoint scheme: {:?}", scheme),
548        };
549
550        Ok((builder, request_url))
551    }
552}
553
554#[cfg(test)]
555mod tests {
556    use super::parse_uri;
557
558    /// A scheme prefix with an empty path produces an empty (and therefore
559    /// dropped) authority. parsing must reject these as malformed rather
560    /// than accept them.
561    #[test]
562    fn empty_authority_uris_are_rejected() {
563        for input in ["unix://", "windows:", "file://"] {
564            let result = parse_uri(input);
565            assert!(
566                result.is_err(),
567                "expected {input:?} to be rejected, got {result:?}"
568            );
569        }
570    }
571}
libdd_common/lib.rs

libdd_common/
lib.rs
