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

libdd_common/
lib.rs
