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

libdd_common/
lib.rs
