From 89a636d8a8cf18eaa8a5006d91f4691ff4b77386 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gustavo=20Andr=C3=A9=20dos=20Santos=20Lopes?= Date: Tue, 14 Jul 2026 14:52:25 +0100 Subject: [PATCH] refactor(library-config): reorganize Linux process context --- libdd-library-config-ffi/Cargo.toml | 2 +- libdd-library-config/Cargo.toml | 5 +- libdd-library-config/src/lib.rs | 4 + libdd-library-config/src/otel_process_ctx.rs | 834 ++++-------------- .../src/otel_process_ctx/linux.rs | 22 + .../src/otel_process_ctx/linux/self_reader.rs | 745 ---------------- .../src/otel_process_ctx/reader.rs | 268 ++++++ .../otel_process_ctx/reader/copy_pipe_unix.rs | 240 +++++ .../src/otel_process_ctx/reader/linux.rs | 135 +++ .../src/otel_process_ctx/writer.rs | 296 +++++++ .../src/otel_process_ctx/writer/linux.rs | 264 ++++++ libdd-library-config/src/tracer_metadata.rs | 3 +- 12 files changed, 1428 insertions(+), 1390 deletions(-) create mode 100644 libdd-library-config/src/otel_process_ctx/linux.rs delete mode 100644 libdd-library-config/src/otel_process_ctx/linux/self_reader.rs create mode 100644 libdd-library-config/src/otel_process_ctx/reader.rs create mode 100644 libdd-library-config/src/otel_process_ctx/reader/copy_pipe_unix.rs create mode 100644 libdd-library-config/src/otel_process_ctx/reader/linux.rs create mode 100644 libdd-library-config/src/otel_process_ctx/writer.rs create mode 100644 libdd-library-config/src/otel_process_ctx/writer/linux.rs diff --git a/libdd-library-config-ffi/Cargo.toml b/libdd-library-config-ffi/Cargo.toml index a751035173..57124b7186 100644 --- a/libdd-library-config-ffi/Cargo.toml +++ b/libdd-library-config-ffi/Cargo.toml @@ -14,7 +14,7 @@ bench = false [dependencies] libdd-common = { path = "../libdd-common" } libdd-common-ffi = { path = "../libdd-common-ffi", default-features = false } -libdd-library-config = { path = "../libdd-library-config" } +libdd-library-config = { path = "../libdd-library-config", default-features = false, features = ["process-context-writer"] } anyhow = "1.0" constcat = "0.4.1" diff --git a/libdd-library-config/Cargo.toml b/libdd-library-config/Cargo.toml index 2f7e07e962..54b2d57083 100644 --- a/libdd-library-config/Cargo.toml +++ b/libdd-library-config/Cargo.toml @@ -15,7 +15,10 @@ crate-type = ["lib"] bench = false [features] -otel-thread-ctx = [] +default = ["process-context-reader", "process-context-writer"] +otel-thread-ctx = ["process-context-writer"] +process-context-reader = [] +process-context-writer = [] [dependencies] serde = { version = "1.0", features = ["derive"] } diff --git a/libdd-library-config/src/lib.rs b/libdd-library-config/src/lib.rs index 6d473f1d84..d801748d9f 100644 --- a/libdd-library-config/src/lib.rs +++ b/libdd-library-config/src/lib.rs @@ -2,6 +2,10 @@ // SPDX-License-Identifier: Apache-2.0 extern crate alloc; +#[cfg(all( + target_os = "linux", + any(feature = "process-context-reader", feature = "process-context-writer") +))] pub mod otel_process_ctx; pub mod tracer_metadata; diff --git a/libdd-library-config/src/otel_process_ctx.rs b/libdd-library-config/src/otel_process_ctx.rs index 1b785d986d..53229e4ce5 100644 --- a/libdd-library-config/src/otel_process_ctx.rs +++ b/libdd-library-config/src/otel_process_ctx.rs @@ -18,676 +18,226 @@ //! reads even when the clock returns the same value twice. Concurrent writers are rejected, and //! retry policy is left to the reader's caller. +#[cfg(feature = "process-context-reader")] +mod reader; +#[cfg(feature = "process-context-writer")] +mod writer; +#[cfg(all(target_os = "linux", not(target_has_atomic = "64")))] +compile_error!("OTel process context requires 64-bit atomics on Linux"); #[cfg(target_os = "linux")] -#[cfg(target_has_atomic = "64")] -pub mod linux { - use core::{ - convert::TryInto, - ffi::{c_void, CStr}, - mem::{size_of, swap, ManuallyDrop}, - ptr::{self, NonNull}, - sync::atomic::{fence, AtomicPtr, AtomicU32, AtomicU64, Ordering}, - time::Duration, - }; - use std::{ - io, - os::fd::{AsRawFd, FromRawFd, OwnedFd}, - sync::{Mutex, MutexGuard}, - }; - - use libdd_trace_protobuf::opentelemetry::proto::common::v1::ProcessContext; - use prost::Message; - - mod self_reader; - pub use self_reader::ProcessContextSelfReader; - - /// Current version of the process context format - pub const PROCESS_CTX_VERSION: u32 = 2; - /// Signature bytes for identifying process context mappings - pub const SIGNATURE: &[u8; 8] = b"OTEL_CTX"; - /// The discoverable name of the memory mapping. - pub const MAPPING_NAME: &CStr = c"OTEL_CTX"; - /// Sentinel timestamp indicating that the context is unpublished or being updated. - const UNPUBLISHED_OR_UPDATING: u64 = 0; - - /// The header structure written at the start of the mapping. This must match the C - /// layout of the specification. - /// - /// Header fields intentionally use the plain C layout types specified by OTel. There are no - /// atomics here: publication relies on naturally atomic aligned word-sized accesses on the - /// supported Linux architectures, plus explicit fences to constrain store/load ordering. - #[repr(C)] - struct MappingHeader { - signature: [u8; 8], - version: u32, - payload_size: AtomicU32, - monotonic_published_at_ns: AtomicU64, - payload_ptr: AtomicPtr, - } - - #[repr(C)] - struct MappingHeaderSnapshot { - signature: [u8; 8], - version: u32, - payload_size: u32, - monotonic_published_at_ns: u64, - payload_ptr: *const u8, - } +pub mod linux; + +#[cfg(feature = "process-context-reader")] +pub use reader::ProcessContextSelfReader; +#[cfg(feature = "process-context-writer")] +pub use writer::{publish, unpublish}; + +/// Current version of the process context format +pub const PROCESS_CTX_VERSION: u32 = 2; +/// Signature bytes for identifying process context mappings +pub const SIGNATURE: &[u8; 8] = b"OTEL_CTX"; +/// Sentinel timestamp indicating that the context is unpublished or being updated. +const UNPUBLISHED_OR_UPDATING: u64 = 0; + +#[repr(C)] +#[cfg(feature = "process-context-reader")] +struct MappingHeaderSnapshot { + signature: [u8; 8], + version: u32, + payload_size: u32, + monotonic_published_at_ns: u64, + payload_ptr: *const u8, +} - // Compile-time verification that MappingHeader matches the field offsets and total size - // mandated by the OTel process context spec: - // https://github.com/open-telemetry/opentelemetry-specification/blob/main/oteps/profiles/4719-process-ctx.md - const _: () = { - use core::mem::{offset_of, size_of}; - assert!(offset_of!(MappingHeader, signature) == 0); - assert!(offset_of!(MappingHeader, version) == 8); - assert!(offset_of!(MappingHeader, payload_size) == 12); - assert!(offset_of!(MappingHeader, monotonic_published_at_ns) == 16); - assert!(offset_of!(MappingHeader, payload_ptr) == 24); - assert!(size_of::() == 32); - assert!(core::mem::align_of::() == 8); - assert!(size_of::<*const u8>() == size_of::()); +#[cfg(all( + test, + feature = "process-context-reader", + feature = "process-context-writer" +))] +#[serial_test::serial] +mod tests { + use core::time::Duration; + + use super::ProcessContextSelfReader; + use libdd_trace_protobuf::opentelemetry::proto::common::v1::{ + any_value, AnyValue, KeyValue, ProcessContext, }; + use prost::Message; - /// The shared memory mapped area to publish the context to. The memory region is owned by a - /// [MemMapping] instance and is automatically unmapped upon drop. - /// - /// # Safety - /// - /// The following invariants MUST always hold for safety and are guaranteed by [MemMapping]: - /// - `start` is non-null, is coming from a previous call to `mmap` with a size value of - /// [mapping_size] and hasn't been unmmaped since. - /// - once `self` has been dropped, no memory access must be performed on the memory previously - /// pointed to by `start`. - struct MemMapping { - start_addr: NonNull, - } - - // SAFETY: MemMapping represents ownership over the mapped region. It never leaks or - // share the internal pointer. It's also safe to drop (`munmap`) from a different thread. - unsafe impl Send for MemMapping {} - - /// The global instance of the context for the current process. - /// - /// We need a mutex to put the handle in a static and avoid bothering the users of this API - /// with storing the handle, but we don't expect this mutex to actually be contended. Ideally a - /// single thread should handle context updates, even if it's not strictly required. - static PROCESS_CONTEXT_HANDLER: Mutex> = Mutex::new(None); - - impl MemMapping { - /// Creates a suitable memory mapping for the context protocol to be published. - /// - /// `memfd` is the preferred method, but this function fallbacks to an anonymous mapping if - /// `memfd` failed for any reason. - /// - /// Both allocation paths produce zero-filled memory: `MAP_ANONYMOUS` mappings are - /// initialized to zero, and the memfd path maps a newly-created file extended by - /// `ftruncate()`, whose extended bytes read as `\0`. This matters because a memfd-backed - /// mapping is discoverable before `set_name()` runs, so early readers may race with header - /// initialization. They must observe [`UNPUBLISHED_OR_UPDATING`] (0) and stop until the - /// final timestamp store publishes the initialized header. - fn new() -> io::Result { - let size = mapping_size(); - - try_memfd(MAPPING_NAME, libc::MFD_CLOEXEC | libc::MFD_NOEXEC_SEAL | libc::MFD_ALLOW_SEALING) - .or_else(|_| try_memfd(MAPPING_NAME, libc::MFD_CLOEXEC | libc::MFD_ALLOW_SEALING)) - .and_then(|fd| { - // SAFETY: fd is a valid open file descriptor. - check_syscall_retval( - unsafe { - libc::ftruncate(fd.as_raw_fd(), mapping_size() as libc::off_t) - }, - "ftruncate failed" - )?; - // SAFETY: we pass a null pointer to mmap which is unconditionally ok - let start_addr = check_mapping_addr( - unsafe { - libc::mmap( - ptr::null_mut(), - size, - libc::PROT_WRITE | libc::PROT_READ, - libc::MAP_PRIVATE, - fd.as_raw_fd(), - 0, - ) - }, - "mmap failed" - )?; - - // We (implicitly) close the file descriptor right away, but this ok - Ok(MemMapping { start_addr }) - }) - // If any previous step failed, we fallback to an anonymous mapping - .or_else(|_| { - // SAFETY: we pass a null pointer to mmap, no precondition to uphold - let start_addr = check_mapping_addr( - unsafe { - libc::mmap( - ptr::null_mut(), - size, - libc::PROT_WRITE | libc::PROT_READ, - libc::MAP_PRIVATE | libc::MAP_ANONYMOUS, - -1, - 0, - ) - }, - "mmap failed: couldn't create a memfd or anonymous mmapped region for process context publication" - )?; - - Ok(MemMapping { start_addr }) - }) - } - - /// Makes this mapping discoverable by giving it a name. - fn set_name(&mut self) -> io::Result<()> { - // SAFETY: self.start_addr is valid for mapping_size() bytes as per MemMapping - // invariants. name is a valid NUL-terminated string that outlives the prctl call. - check_syscall_retval( - unsafe { - // int prctl(PR_SET_VMA, long attr, unsigned long addr, unsigned long size, - // const char *_Nullable val); - libc::prctl( - libc::PR_SET_VMA, - libc::PR_SET_VMA_ANON_NAME as libc::c_ulong, - TryInto::::try_into(self.start_addr.as_ptr() as usize) - .expect("start addr overflowed"), - TryInto::::try_into(mapping_size()) - .expect("mapping size overflowed"), - MAPPING_NAME.as_ptr(), - ) - }, - "prctl PR_SET_VMA_ANON_NAME failed", - )?; - - Ok(()) - } - - /// Unmaps the underlying memory region. This has same effect as dropping `self`, but - /// propagates potential errors. - fn free(mut self) -> io::Result<()> { - // SAFETY: We put `self` in a `ManuallyDrop`, which prevents drop and future calls to - // `free()`. - unsafe { - self.unmap()?; - } - - // Prevent `Self::drop` from being called - let _ = ManuallyDrop::new(self); - - Ok(()) - } - - /// Unmaps the underlying memory region. For internal use only; prefer `free()` or `drop()`. - /// - /// # Safety - /// - /// This method must only be called once. After calling `unmap()`, no other method of - /// `MemMapping` must be ever called on `self` again, including `unmap()` and `drop()`. - /// - /// Practically, `self` must be put in a `ManuallyDrop` wrapper and forgotten, or being in - /// the process of being dropped. - unsafe fn unmap(&mut self) -> io::Result<()> { - check_syscall_retval( - // SAFETY: upheld by the caller. - unsafe { libc::munmap(self.start_addr.as_ptr(), mapping_size()) }, - "munmap failed when freeing the process context", - )?; - - Ok(()) - } - } - - impl Drop for MemMapping { - fn drop(&mut self) { - // SAFETY: `self` is being dropped - let _ = unsafe { self.unmap() }; - } - } - - /// Handle for future updates of a published process context. - struct ProcessContextHandle { - mapping: MemMapping, - /// Once published, and until the next update is complete, the backing allocation of - /// `payload` might be read by external processes and thus must not move (e.g. by resizing - /// or drop). - #[allow(unused)] - payload: Vec, - /// The process id of the last publisher. This is useful to detect forks(), and publish a - /// new context accordingly. - pid: libc::pid_t, - } - - impl ProcessContextHandle { - /// Initial publication of the process context. Creates an appropriate memory mapping. - fn publish(payload: Vec) -> io::Result { - let payload_size: u32 = payload - .len() - .try_into() - .map_err(|_| io::Error::other("payload size overflowed"))?; - - let mut mapping = MemMapping::new()?; - let size = mapping_size(); - check_syscall_retval( - // SAFETY: the invariants of MemMapping ensures `start_addr` is not null and comes - // from a previous call to `mmap` - unsafe { libc::madvise(mapping.start_addr.as_ptr(), size, libc::MADV_DONTFORK) }, - "madvise MADVISE_DONTFORK failed", - )?; - - let published_at_ns = since_boottime_ns().ok_or_else(|| { - io::Error::other("failed to get current time for process context publication") - })?; - - let header = mapping.start_addr.as_ptr() as *mut MappingHeader; - - // SAFETY: header points to a zero-filled, page-aligned mapping of at least - // mapping_size() bytes; field projections are in-bounds and aligned. - // The pointer writes do not happen while there are live &MappingHeader references - // and, to the extent the atomic stores do, this is fine because the mutated bytes - // are inside UnsafeCells. - unsafe { - ptr::addr_of_mut!((*header).signature).write(*SIGNATURE); - ptr::addr_of_mut!((*header).version).write(PROCESS_CTX_VERSION); - (*header) - .payload_size - .store(payload_size, Ordering::Relaxed); - (*header) - .payload_ptr - .store(payload.as_ptr().cast_mut(), Ordering::Relaxed); - - fence(Ordering::SeqCst); - (*header) - .monotonic_published_at_ns - .store(published_at_ns, Ordering::Relaxed); - } - - // Note that naming must be unconditionally attempted, even on kernels where we might - // know it will fail. It is ok for naming to fail - we must only make sure that at - // least we tried, as per the - // [spec](https://github.com/open-telemetry/opentelemetry-specification/pull/4719). - let _ = mapping.set_name(); - - Ok(ProcessContextHandle { - mapping, - payload, - // SAFETY: getpid() is always safe to call. - pid: unsafe { libc::getpid() }, - }) - } - - /// Updates the context after initial publication. - fn update(&mut self, payload: Vec) -> io::Result<()> { - let header = self.mapping.start_addr.as_ptr() as *mut MappingHeader; - - let monotonic_published_at_ns = since_boottime_ns() - .ok_or_else(|| io::Error::other("could not get the current timestamp"))?; - let payload_size: u32 = payload.len().try_into().map_err(|_| { - io::Error::other("couldn't update process context: new payload too large") - })?; - // A process shouldn't try to concurrently update its own context. - // - // `UNPUBLISHED_OR_UPDATING` is an out-of-band sentinel, not a value that - // `CLOCK_BOOTTIME` is expected to produce after publication. Published non-zero - // timestamp values must advance monotonically; the field may temporarily hold the - // sentinel while an update is in progress. - // - // Note: be careful of early return while `monotonic_published_at` is still zero, as - // this would effectively "lock" any future publishing. Move throwing code above this - // swap, or properly restore the previous value if the former can't be done. - // SAFETY: the mapping is live and valid for writes. - // Note: this does not use CAS because we assume the write lock is being held by the - // caller. - let previous_published_at_ns = unsafe { - (*header) - .monotonic_published_at_ns - .swap(UNPUBLISHED_OR_UPDATING, Ordering::Relaxed) - }; - // should never happen (publish() and the several update() calls are serialized by the - // lock) - if previous_published_at_ns == UNPUBLISHED_OR_UPDATING { - panic!("concurrent update of the process context is not supported"); - } - - // The timestamp also acts as the seqlock version, so it must advance even if the - // clock source returns the same value for two rapid updates. - let monotonic_published_at_ns = - monotonic_published_at_ns.max(previous_published_at_ns.saturating_add(1)); - - // Pair this with the reader's SeqCst fence before its second timestamp copy. If a - // reader starts from the previous non-zero timestamp but copies data after this update - // begins, it must not accept that copy as the previous version: its final timestamp - // check should see `UNPUBLISHED_OR_UPDATING` or the later published timestamp. - // Note: only needs - fence(Ordering::SeqCst); - self.payload = payload; - - // SAFETY: the mapping is live and valid, and the global mutex prevents concurrent - // in-process writers from mutating the plain header fields. - unsafe { - (*header) - .payload_ptr - .store(self.payload.as_ptr().cast_mut(), Ordering::Relaxed); - (*header) - .payload_size - .store(payload_size, Ordering::Relaxed); - } + #[cfg(target_os = "linux")] + mod linux { + use core::ptr; + use std::io; - fence(Ordering::SeqCst); - // SAFETY: same as above. - unsafe { - (*header) - .monotonic_published_at_ns - .store(monotonic_published_at_ns, Ordering::Relaxed); - } + use super::super::{reader, MappingHeaderSnapshot}; - Ok(()) - } - } - - /// Returns `Err` wrapping the current `errno` with `msg` as context if `addr` equals - /// `MAP_FAILED`, `Ok(addr)` otherwise. - fn check_mapping_addr(addr: *mut c_void, msg: &'static str) -> io::Result> { - if addr == libc::MAP_FAILED { - let e = io::Error::last_os_error(); - Err(io::Error::new(e.kind(), format!("{msg}: {e}"))) - } else { - // SAFETY: mmap returns a non-null pointer on success. - Ok(unsafe { NonNull::new_unchecked(addr) }) + pub(super) fn read_process_context() -> io::Result { + let mapping_addr = reader::linux::find_otel_mapping()?; + let header_ptr: *const MappingHeaderSnapshot = + ptr::with_exposed_provenance(mapping_addr); + // SAFETY: the mapping was published by this test before being read; the tests are + // serial and don't update the mapping while this header is copied. + Ok(unsafe { ptr::read(header_ptr) }) } - } - /// Returns `Err` wrapping the current `errno` with `msg` as context if `ret` is negative, - /// `Ok(ret)` otherwise. - fn check_syscall_retval(ret: libc::c_int, msg: &'static str) -> io::Result { - if ret < 0 { - let e = io::Error::last_os_error(); - Err(io::Error::new(e.kind(), format!("{msg}: {e}"))) - } else { - Ok(ret) + pub(super) fn is_published() -> bool { + reader::linux::find_otel_mapping().is_ok() } } - /// Creates a `memfd` file descriptor with the given name and flags. - fn try_memfd(name: &CStr, flags: libc::c_uint) -> io::Result { - // We use the raw syscall rather than `libc::memfd_create` because the latter requires - // glibc >= 2.27, while `syscall()` + `SYS_memfd_create` works with any glibc version. - check_syscall_retval( - // SAFETY: name is a valid NUL-terminated string; flags are constant bit flags. - unsafe { - libc::syscall(libc::SYS_memfd_create, name.as_ptr(), flags as libc::c_long) - as libc::c_int - }, - "memfd_create failed", - ) - // SAFETY: fd is a valid file descriptor just returned by memfd_create. - .map(|fd| unsafe { OwnedFd::from_raw_fd(fd) }) - } + #[cfg(target_os = "linux")] + use linux::{is_published, read_process_context}; - // The returned size is guaranteed to be larger or equal to the size of `MappingHeader`. - fn mapping_size() -> usize { - size_of::() - } - - /// Returns the value of the monotonic BOOTTIME clock in nanoseconds. - fn since_boottime_ns() -> Option { - let mut ts = libc::timespec { - tv_sec: 0, - tv_nsec: 0, + #[test] + #[cfg_attr(miri, ignore)] + fn publish_then_read_process_context() { + let context = ProcessContext { + resource: None, + extra_attributes: vec![KeyValue { + key: "service.name".to_string(), + value: Some(AnyValue { + value: Some(any_value::Value::StringValue("checkout".to_string())), + }), + key_ref: 0, + }], }; - // SAFETY: ts is a valid, writable timespec. - let ret = unsafe { libc::clock_gettime(libc::CLOCK_BOOTTIME, &mut ts) }; - if ret != 0 { - return None; - } - let secs: u64 = ts.tv_sec.try_into().ok()?; - let nanos: u32 = ts.tv_nsec.try_into().ok()?; - let duration = Duration::new(secs, nanos); - u64::try_from(duration.as_nanos()).ok() - } - - /// Locks the context handle. Returns a uniform error if the lock has been poisoned. - fn lock_context_handle() -> io::Result>> { - PROCESS_CONTEXT_HANDLER.lock().map_err(|_| { - io::Error::other("a thread panicked while operating on the process context handler") - }) - } - - /// Publishes or updates the process context for it to be visible by external readers. - /// - /// If any of the following condition holds: - /// - /// - this is the first publication - /// - [unpublish] has been called last - /// - the previous context has been published from a different process id (that is, a `fork()` - /// happened and we're the child process) - /// - /// Then we follow the Publish protocol of the OTel process context specification (allocating a - /// fresh mapping). - /// - /// Otherwise, if a context has been previously published from the same process and hasn't been - /// unpublished since, we follow the Update protocol. - /// - /// # Fork safety - /// - /// If we're a forked children of the original publisher, we are extremely restricted in the - /// set of operations that we can do (we must be async-signal-safe). On paper, heap allocation - /// is Undefined Behavior, for example. We assume that a forking runtime (such as Python or - /// Ruby) that doesn't follow with an immediate `exec` is already "taking that risk", so to - /// speak (typically, if no thread is ever spawned before the fork, things are mostly fine). - #[inline] - pub fn publish(context: &ProcessContext) -> io::Result<()> { - publish_raw_payload(context.encode_to_vec()) - } - - fn publish_raw_payload(payload: Vec) -> io::Result<()> { - let mut guard = lock_context_handle()?; - - // SAFETY: getpid() is always safe to call. - match &mut *guard { - Some(handler) if handler.pid == unsafe { libc::getpid() } => handler.update(payload), - Some(handler) => { - let mut local_handler = ProcessContextHandle::publish(payload)?; - // If we've been forked, we need to prevent the mapping from being dropped - // normally, as it would try to unmap a region that isn't mapped anymore in the - // child process, or worse, could have been remapped to something else in the - // meantime. - // - // To do so, we get the old handler back in `local_handler` and prevent `mapping` - // from being dropped specifically. - swap(&mut local_handler, handler); - let _: ManuallyDrop = ManuallyDrop::new(local_handler.mapping); - - Ok(()) - } - None => { - *guard = Some(ProcessContextHandle::publish(payload)?); - Ok(()) - } - } - } - /// Unmaps the region used to share the process context. If no context has ever been published, - /// this is no-op. - /// - /// A call to [publish] following an [unpublish] will create a new mapping. - pub fn unpublish() -> io::Result<()> { - let mut guard = lock_context_handle()?; - - if let Some(ProcessContextHandle { - mapping, payload, .. - }) = guard.take() - { - // Mark the context as unavailable before freeing the mapping/payload. The fence forces - // the writing CPU not to reorder the unavailable timestamp store and the deallocation - // stores. This gives readers more of a chance (but no guarantee) to observe an - // unavailable context before the mapping is removed. - // - // SAFETY: the mapping is still live and valid, and the global mutex prevents - // concurrent in-process writers from mutating the plain header fields. - let header = mapping.start_addr.as_ptr() as *mut MappingHeader; - unsafe { - (*header) - .monotonic_published_at_ns - .store(UNPUBLISHED_OR_UPDATING, Ordering::Relaxed); - } - fence(Ordering::SeqCst); - - mapping.free()?; // payload will still drop if it fails - // but we'll be stuck with a zero timestamp - drop(payload); - } - - Ok(()) - } - - #[cfg(test)] - #[serial_test::serial] - mod tests { - use core::{ptr, time::Duration}; - use std::io; + super::publish(&context).expect("couldn't publish the process context"); + let header = read_process_context().expect("couldn't read back the process context"); + // SAFETY: the published context must have put valid bytes of size payload_size in the + // context if the signature check succeded. + let read_payload = unsafe { + core::slice::from_raw_parts(header.payload_ptr, header.payload_size as usize) + }; + let read_context = + ProcessContext::decode(read_payload).expect("couldn't decode the process context"); + super::unpublish().expect("couldn't unpublish the context"); + + assert!(header.signature == *super::SIGNATURE, "wrong signature"); + assert!( + header.version == super::PROCESS_CTX_VERSION, + "wrong context version" + ); + assert!( + header.monotonic_published_at_ns > 0, + "monotonic_published_at_ns is zero" + ); + assert!(read_context == context, "read back a different context"); + } + + #[test] + #[cfg_attr(miri, ignore)] + fn publish_then_update_process_context() { + let payload_v1 = "example process context payload"; + let payload_v2 = "another example process context payload of different size"; + + super::writer::publish_raw_payload(payload_v1.as_bytes().to_vec()) + .expect("couldn't publish the process context"); + + let header = read_process_context().expect("couldn't read back the process context"); + // SAFETY: the published context must have put valid bytes of size payload_size in the + // context if the signature check succeded. + let read_payload = unsafe { + core::slice::from_raw_parts(header.payload_ptr, header.payload_size as usize) + }; - use super::ProcessContext; - use libdd_trace_protobuf::opentelemetry::proto::common::v1::{ - any_value, AnyValue, KeyValue, + assert!(header.signature == *super::SIGNATURE, "wrong signature"); + assert!( + header.version == super::PROCESS_CTX_VERSION, + "wrong context version" + ); + assert!( + header.payload_size == payload_v1.len() as u32, + "wrong payload size" + ); + assert!( + header.monotonic_published_at_ns > 0, + "monotonic_published_at_ns is zero" + ); + assert!(read_payload == payload_v1.as_bytes(), "payload mismatch"); + + let published_at_ns_v1 = header.monotonic_published_at_ns; + // Ensure the clock advances so the updated timestamp is strictly greater + std::thread::sleep(Duration::from_nanos(10)); + + super::writer::publish_raw_payload(payload_v2.as_bytes().to_vec()) + .expect("couldn't update the process context"); + + let header = read_process_context().expect("couldn't read back the process context"); + // SAFETY: the published context must have put valid bytes of size payload_size in the + // context if the signature check succeded. + let read_payload = unsafe { + core::slice::from_raw_parts(header.payload_ptr, header.payload_size as usize) }; - use prost::Message; - - /// Read the process context from the current process. - /// - /// This searches `/proc/self/maps` for an OTEL_CTX mapping and decodes its contents. - /// - /// **CAUTION**: Note that the reader implemented in this module, as well as the helper - /// functions it relies on, are specialized for tests (for example, it doesn't check for - /// concurrent writers after reading the header, because we know they can't be). Do not - /// extract or use as it is as a generic Rust OTel process context reader. - fn read_process_context() -> io::Result { - let mapping_addr = super::ProcessContextSelfReader::find_otel_mapping()?; - let header_ptr: *const super::MappingHeaderSnapshot = - ptr::with_exposed_provenance(mapping_addr); - // SAFETY: the mapping was published by this test before being read; the tests are - // serial and don't update the mapping while this header is copied. - Ok(unsafe { ptr::read(header_ptr) }) - } - #[test] - #[cfg_attr(miri, ignore)] - fn publish_then_read_process_context() { - let context = ProcessContext { + assert!(header.signature == *super::SIGNATURE, "wrong signature"); + assert!( + header.version == super::PROCESS_CTX_VERSION, + "wrong context version" + ); + assert!( + header.payload_size == payload_v2.len() as u32, + "wrong payload size" + ); + assert!( + header.monotonic_published_at_ns > published_at_ns_v1, + "published_at_ns should be strictly greater after update" + ); + assert!(read_payload == payload_v2.as_bytes(), "payload mismatch"); + + super::unpublish().expect("couldn't unpublish the context"); + } + + #[test] + #[cfg_attr(miri, ignore)] + fn unpublish_process_context() { + let payload = "example process context payload"; + + super::writer::publish_raw_payload(payload.as_bytes().to_vec()) + .expect("couldn't publish the process context"); + + assert!( + is_published(), + "process context should be visible after publishing" + ); + + super::unpublish().expect("couldn't unpublish the context"); + + assert!( + !is_published(), + "process context should not be visible after unpublishing" + ); + } + + /// The only end-to-end test with `ProcessContextSelfReader` + #[test] + #[cfg_attr(miri, ignore)] + fn publish_read_update_read_and_unpublish() { + fn context(service_name: &str) -> ProcessContext { + ProcessContext { resource: None, extra_attributes: vec![KeyValue { key: "service.name".to_string(), value: Some(AnyValue { - value: Some(any_value::Value::StringValue("checkout".to_string())), + value: Some(any_value::Value::StringValue(service_name.to_string())), }), key_ref: 0, }], - }; - - super::publish(&context).expect("couldn't publish the process context"); - let header = read_process_context().expect("couldn't read back the process context"); - // SAFETY: the published context must have put valid bytes of size payload_size in the - // context if the signature check succeded. - let read_payload = unsafe { - core::slice::from_raw_parts(header.payload_ptr, header.payload_size as usize) - }; - let read_context = - ProcessContext::decode(read_payload).expect("couldn't decode the process context"); - super::unpublish().expect("couldn't unpublish the context"); - - assert!(header.signature == *super::SIGNATURE, "wrong signature"); - assert!( - header.version == super::PROCESS_CTX_VERSION, - "wrong context version" - ); - assert!( - header.monotonic_published_at_ns > 0, - "monotonic_published_at_ns is zero" - ); - assert!(read_context == context, "read back a different context"); - } - - #[test] - #[cfg_attr(miri, ignore)] - fn publish_then_update_process_context() { - let payload_v1 = "example process context payload"; - let payload_v2 = "another example process context payload of different size"; - - super::publish_raw_payload(payload_v1.as_bytes().to_vec()) - .expect("couldn't publish the process context"); - - let header = read_process_context().expect("couldn't read back the process context"); - // SAFETY: the published context must have put valid bytes of size payload_size in the - // context if the signature check succeded. - let read_payload = unsafe { - core::slice::from_raw_parts(header.payload_ptr, header.payload_size as usize) - }; - - assert!(header.signature == *super::SIGNATURE, "wrong signature"); - assert!( - header.version == super::PROCESS_CTX_VERSION, - "wrong context version" - ); - assert!( - header.payload_size == payload_v1.len() as u32, - "wrong payload size" - ); - assert!( - header.monotonic_published_at_ns > 0, - "monotonic_published_at_ns is zero" - ); - assert!(read_payload == payload_v1.as_bytes(), "payload mismatch"); - - let published_at_ns_v1 = header.monotonic_published_at_ns; - // Ensure the clock advances so the updated timestamp is strictly greater - std::thread::sleep(Duration::from_nanos(10)); - - super::publish_raw_payload(payload_v2.as_bytes().to_vec()) - .expect("couldn't update the process context"); - - let header = read_process_context().expect("couldn't read back the process context"); - // SAFETY: the published context must have put valid bytes of size payload_size in the - // context if the signature check succeded. - let read_payload = unsafe { - core::slice::from_raw_parts(header.payload_ptr, header.payload_size as usize) - }; - - assert!(header.signature == *super::SIGNATURE, "wrong signature"); - assert!( - header.version == super::PROCESS_CTX_VERSION, - "wrong context version" - ); - assert!( - header.payload_size == payload_v2.len() as u32, - "wrong payload size" - ); - assert!( - header.monotonic_published_at_ns > published_at_ns_v1, - "published_at_ns should be strictly greater after update" - ); - assert!(read_payload == payload_v2.as_bytes(), "payload mismatch"); - - super::unpublish().expect("couldn't unpublish the context"); + } } - #[test] - #[cfg_attr(miri, ignore)] - fn unpublish_process_context() { - let payload = "example process context payload"; + let first = context("checkout"); + super::publish(&first).expect("publication should succeed"); - super::publish_raw_payload(payload.as_bytes().to_vec()) - .expect("couldn't publish the process context"); + let reader = ProcessContextSelfReader::new().expect("reader creation should succeed"); + assert_eq!(reader.read().expect("read should succeed"), first); - // The mapping must be discoverable right after publishing - super::ProcessContextSelfReader::find_otel_mapping() - .expect("couldn't find the otel mapping after publishing"); + let second = context("payments"); + super::publish(&second).expect("context update should succeed"); + assert_eq!(reader.read().expect("updated read should succeed"), second); - super::unpublish().expect("couldn't unpublish the context"); - - // After unpublishing the name must no longer appear in /proc/self/maps - assert!( - super::ProcessContextSelfReader::find_otel_mapping().is_err(), - "otel mapping should not be visible after unpublish" - ); - } + super::unpublish().expect("unpublish should succeed"); + assert!(reader.read().is_err()); + assert!(ProcessContextSelfReader::new().is_err()); } } diff --git a/libdd-library-config/src/otel_process_ctx/linux.rs b/libdd-library-config/src/otel_process_ctx/linux.rs new file mode 100644 index 0000000000..994bb5c4dc --- /dev/null +++ b/libdd-library-config/src/otel_process_ctx/linux.rs @@ -0,0 +1,22 @@ +// Copyright 2026-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 + +use core::ffi::CStr; + +#[cfg(feature = "process-context-reader")] +pub mod self_reader { + pub use crate::otel_process_ctx::ProcessContextSelfReader; +} + +// Re-exported for backwards compatibility. +#[cfg(feature = "process-context-writer")] +#[deprecated(note = "use libdd_library_config::otel_process_ctx directly")] +pub use super::{publish, unpublish}; +#[deprecated(note = "use libdd_library_config::otel_process_ctx directly")] +pub use super::{PROCESS_CTX_VERSION, SIGNATURE}; +#[cfg(feature = "process-context-reader")] +#[deprecated(note = "use libdd_library_config::otel_process_ctx::ProcessContextSelfReader")] +pub use self_reader::ProcessContextSelfReader; + +/// The discoverable name of the memory mapping. +pub const MAPPING_NAME: &CStr = c"OTEL_CTX"; diff --git a/libdd-library-config/src/otel_process_ctx/linux/self_reader.rs b/libdd-library-config/src/otel_process_ctx/linux/self_reader.rs deleted file mode 100644 index 4113228e87..0000000000 --- a/libdd-library-config/src/otel_process_ctx/linux/self_reader.rs +++ /dev/null @@ -1,745 +0,0 @@ -// Copyright 2026-Present Datadog, Inc. https://www.datadoghq.com/ -// SPDX-License-Identifier: Apache-2.0 - -//! Same-process reader for the OTEL process context mapping. -//! -//! By design, this reader emulates an out-of-process reader: it does not take the publisher's -//! mutex or hold any guard that prevents the mapping from being unpublished, the header from being -//! rewritten, or the payload allocation from being replaced while it copies memory. The -//! seqlock-style timestamp checks and fences only let it detect concurrent publication or update; -//! unpublish can make the memory disappear entirely. -//! -//! To handle this safely, instead of dereferencing process memory directly, we ask the kernel to -//! copy those bytes into a pipe and then read the pipe back. This gives us the same failure mode an -//! out-of-process reader has: if memory is unmapped or invalid, the copy fails instead of -//! segfaulting the process. - -use core::{ - cell::Cell, - mem::offset_of, - ptr::{self, NonNull}, - sync::atomic::{fence, Ordering}, -}; -use std::{ - fs::File, - io::{self, BufRead, BufReader}, - os::fd::{AsRawFd, FromRawFd, OwnedFd}, -}; - -use libdd_trace_protobuf::opentelemetry::proto::common::v1::{ - any_value, AnyValue, KeyValue, ProcessContext, -}; -use prost::Message; - -use crate::otel_process_ctx::linux::MappingHeaderSnapshot; - -use super::{PROCESS_CTX_VERSION, SIGNATURE, UNPUBLISHED_OR_UPDATING}; - -/// Reader for the current process's OTel process context mapping. -/// -/// Locates the OTEL_CTX mapping at construction. Call [`read`](Self::read) repeatedly to fetch -/// updated context data without re-parsing `/proc/self/maps`, as long as the process has not -/// forked. After a `fork()`, reads fail and a new reader must be constructed. -pub struct ProcessContextSelfReader { - pid: libc::pid_t, - header_ptr: NonNull, - pipe: Cell>, -} - -// SAFETY: ProcessContextSelfReader can be moved between threads because it owns its pipe file -// descriptors and only stores a process-global mapping address. `Cell` keeps it !Sync, so the -// cached pipe cannot be used concurrently through shared references. -unsafe impl Send for ProcessContextSelfReader {} -// we do not implement Sync because of the Cell - -impl ProcessContextSelfReader { - /// Locates the OTEL_CTX mapping in `/proc/self/maps`. - pub fn new() -> io::Result { - let mapping_addr = Self::find_otel_mapping()?; - // SAFETY: getpid() is always safe to call. - let pid = unsafe { libc::getpid() }; - let reader = Self { - pid, - header_ptr: Self::header_ptr_from_addr(mapping_addr)?, - pipe: Cell::new(None), - }; - Ok(reader) - } - - /// Reads and decodes the current process's OTel process context. - /// - /// Returns [`io::ErrorKind::WouldBlock`] if a writer is currently publishing or updating the - /// context, or if the context changed while it was being read. Callers may retry later. - pub fn read(&self) -> io::Result { - // SAFETY: getpid() is always safe to call. - let current_pid = unsafe { libc::getpid() }; - if current_pid != self.pid { - return Err(io::Error::new( - io::ErrorKind::InvalidInput, - "process context reader is stale after fork; construct a new reader", - )); - } - - let published_at = self.read_published_at()?; - if published_at == UNPUBLISHED_OR_UPDATING { - return Err(io::Error::new( - io::ErrorKind::WouldBlock, - "process context is currently being updated", - )); - } - - fence(Ordering::SeqCst); - - let header = self.read_header()?; - if header.signature != *SIGNATURE { - return Err(io::Error::new( - io::ErrorKind::InvalidData, - "invalid signature in process context mapping", - )); - } - if header.version != PROCESS_CTX_VERSION { - return Err(io::Error::new( - io::ErrorKind::InvalidData, - format!("unsupported process context version {}", header.version), - )); - } - - let payload_size = header.payload_size; - let payload_ptr = header.payload_ptr; - - if payload_ptr.is_null() { - return Err(io::Error::new( - io::ErrorKind::InvalidData, - "process context payload pointer is null", - )); - } - - let payload_bytes = self.read_process_memory(payload_ptr, payload_size as usize)?; - - // This is the second read-side seqlock fence. It pairs with the writer's SeqCst fences so - // that, if we copied data updated after the initial published time, the final timestamp - // copy sees `UNPUBLISHED_OR_UPDATING` or a later published timestamp. - fence(Ordering::SeqCst); - - let published_at_after = self.read_published_at()?; - if published_at != published_at_after { - return Err(io::Error::new( - io::ErrorKind::WouldBlock, - "process context changed while being read", - )); - } - - let context = ProcessContext::decode(payload_bytes.as_slice()) - .map_err(|err| io::Error::new(io::ErrorKind::InvalidData, err))?; - - Ok(context) - } - - /// Returns the thread-local attribute key map from a decoded process context. - pub fn threadlocal_attribute_key_map(context: &ProcessContext) -> Option> { - Self::find_attr(&context.extra_attributes, "threadlocal.attribute_key_map") - .and_then(Self::string_array) - } - - fn string_array(value: &AnyValue) -> Option> { - let any_value::Value::ArrayValue(array) = value.value.as_ref()? else { - return None; - }; - - array - .values - .iter() - .map(|value| match value.value.as_ref()? { - any_value::Value::StringValue(value) => Some(value.clone()), - _ => None, - }) - .collect() - } - - // The process context only carries a small resource/extra attribute set, so a linear scan - // keeps this helper allocation-free and simpler than building a temporary index. - fn find_attr<'a>(attrs: &'a [KeyValue], key: &str) -> Option<&'a AnyValue> { - attrs - .iter() - .find(|attr| attr.key == key) - .and_then(|attr| attr.value.as_ref()) - } - - fn header_ptr_from_addr(mapping_addr: usize) -> io::Result> { - NonNull::new(ptr::with_exposed_provenance::(mapping_addr).cast_mut()).ok_or_else(|| { - io::Error::new(io::ErrorKind::InvalidData, "null process context header") - }) - } - - /// Find the OTEL_CTX mapping in /proc/self/maps. - pub(super) fn find_otel_mapping() -> io::Result { - let file = File::open("/proc/self/maps")?; - let reader = BufReader::new(file); - - for line in reader.lines() { - let line = line?; - - if Self::is_named_otel_mapping(&line) { - if let Some(addr) = Self::parse_mapping_start(&line) { - return Ok(addr); - } - } - } - - Err(io::Error::new( - io::ErrorKind::NotFound, - "couldn't find the mapping of the process context", - )) - } - - /// Parses the start address from a /proc/self/maps line. - fn parse_mapping_start(line: &str) -> Option { - usize::from_str_radix(line.split('-').next()?, 16).ok() - } - - /// Checks if a mapping line refers to the OTEL_CTX mapping. - fn is_named_otel_mapping(line: &str) -> bool { - let trimmed = line.trim_end(); - - // The mapping name is the `pathname` column documented for `/proc//maps`: - // https://github.com/torvalds/linux/blob/9147566d801602c9e7fc7f85e989735735bf38ba/Documentation/filesystems/proc.rst?plain=1#L384-L386 - // For the OTEL_CTX names we care about, it is the 6th whitespace-delimited field; - // `split_whitespace()` ignores the column padding. - let Some(name) = trimmed.split_whitespace().nth(5) else { - return false; - }; - - // The OTel process context spec says to search for entries whose names start with - // these prefixes. In `/proc//maps`, however, the optional ` (deleted)` suffix is - // emitted as a separate token, so the mapping-name token itself should match exactly. - matches!( - name, - "/memfd:OTEL_CTX" | "[anon_shmem:OTEL_CTX]" | "[anon:OTEL_CTX]" - ) - } - - fn read_published_at(&self) -> io::Result { - let timestamp_ptr = self - .header_ptr - .as_ptr() - .wrapping_add(offset_of!(MappingHeaderSnapshot, monotonic_published_at_ns)) - .cast_const(); - let bytes = self.read_process_memory(timestamp_ptr, size_of::())?; - Ok(u64::from_ne_bytes( - Self::field_bytes::<{ size_of::() }>(&bytes, 0, "monotonic_published_at_ns")?, - )) - } - - fn read_header(&self) -> io::Result { - let bytes = self.read_process_memory( - self.header_ptr.as_ptr().cast_const(), - size_of::(), - )?; - Self::mapping_header_from_bytes(&bytes) - } - - /// Unmarshalls the header from the bytes - fn mapping_header_from_bytes(bytes: &[u8]) -> io::Result { - if bytes.len() != size_of::() { - return Err(io::Error::new( - io::ErrorKind::InvalidData, - format!( - "process context header copy had size {}, expected {}", - bytes.len(), - size_of::() - ), - )); - } - - let signature = Self::field_bytes::<8>( - bytes, - offset_of!(MappingHeaderSnapshot, signature), - "signature", - )?; - let version = u32::from_ne_bytes(Self::field_bytes::<{ size_of::() }>( - bytes, - offset_of!(MappingHeaderSnapshot, version), - "version", - )?); - let payload_size = u32::from_ne_bytes(Self::field_bytes::<{ size_of::() }>( - bytes, - offset_of!(MappingHeaderSnapshot, payload_size), - "payload_size", - )?); - let monotonic_published_at_ns = - u64::from_ne_bytes(Self::field_bytes::<{ size_of::() }>( - bytes, - offset_of!(MappingHeaderSnapshot, monotonic_published_at_ns), - "monotonic_published_at_ns", - )?); - let payload_addr = usize::from_ne_bytes(Self::field_bytes::<{ size_of::() }>( - bytes, - offset_of!(MappingHeaderSnapshot, payload_ptr), - "payload_ptr", - )?); - - Ok(MappingHeaderSnapshot { - signature, - version, - payload_size, - monotonic_published_at_ns, - payload_ptr: ptr::with_exposed_provenance(payload_addr), - }) - } - - fn field_bytes( - bytes: &[u8], - offset: usize, - field: &'static str, - ) -> io::Result<[u8; N]> { - let end = offset.checked_add(N).ok_or_else(|| { - io::Error::new( - io::ErrorKind::InvalidData, - format!("process context header field {field} offset overflowed"), - ) - })?; - let slice = bytes.get(offset..end).ok_or_else(|| { - io::Error::new( - io::ErrorKind::InvalidData, - format!("process context header field {field} is out of bounds"), - ) - })?; - let mut out = [0; N]; - out.copy_from_slice(slice); - Ok(out) - } - - /// Copies `len` bytes from `addr` through a pipe. - /// - /// `write(2)` copies from the source address in kernel space, so unmapped source memory is - /// reported as an error or a short write instead of raising `SIGSEGV`. - fn read_process_memory(&self, addr: *const u8, len: usize) -> io::Result> { - if len == 0 { - return Ok(Vec::new()); - } - - let pipe = match self.pipe.take() { - Some(pipe) => pipe, - None => CopyPipe::new()?, - }; - - match pipe.copy_via_pipe(addr, len) { - Ok(buf) => { - self.pipe.set(Some(pipe)); - Ok(buf) - } - Err(PipeCopyError { err, pipe_dirty }) => { - if !pipe_dirty { - // The pipe does not hold bytes from an aborted transfer - // Save it as a sort of cache - // Note that we're not Sync - self.pipe.set(Some(pipe)); - } - Err(err) - } - } - } -} - -/// A cached pipe used to probe-copy process memory via `write(2)`. -/// -/// Invariant: the pipe is empty between calls to `CopyPipe::copy_via_pipe`. -struct CopyPipe { - read_fd: OwnedFd, - write_fd: OwnedFd, - capacity: usize, -} - -impl CopyPipe { - fn new() -> io::Result { - let mut fds = [0; 2]; - // SAFETY: `fds` points to space for the two file descriptors returned by pipe2. - // `O_NONBLOCK` makes invariant bugs fail with EAGAIN instead of blocking forever. - let ret = unsafe { libc::pipe2(fds.as_mut_ptr(), libc::O_CLOEXEC | libc::O_NONBLOCK) }; - if ret != 0 { - let err = io::Error::last_os_error(); - return Err(io::Error::new( - err.kind(), - format!("failed to create process context copy pipe: {err}"), - )); - } - - // SAFETY: pipe2 initialized both file descriptors on success and ownership is - // transferred to OwnedFd exactly once. - let (read_fd, write_fd) = - unsafe { (OwnedFd::from_raw_fd(fds[0]), OwnedFd::from_raw_fd(fds[1])) }; - - // SAFETY: `write_fd` is a valid pipe file descriptor. - let capacity = unsafe { libc::fcntl(write_fd.as_raw_fd(), libc::F_GETPIPE_SZ) }; - if capacity < 0 { - let err = io::Error::last_os_error(); - return Err(io::Error::new( - err.kind(), - format!("failed to query process context copy pipe capacity: {err}"), - )); - } - if capacity == 0 { - return Err(io::Error::new( - io::ErrorKind::InvalidData, - "process context copy pipe has zero capacity", - )); - } - - Ok(Self { - read_fd, - write_fd, - capacity: capacity as usize, - }) - } - - fn copy_via_pipe(&self, addr: *const u8, len: usize) -> Result, PipeCopyError> { - // The buffer is filled sequentially from the pipe; `buf[..offset]` is always initialized, - // so there is no need to zero it up front. - let mut buf: Vec = Vec::with_capacity(len); - let mut offset = 0; - - while offset < len { - let chunk_len = (len - offset).min(self.capacity); - let chunk_addr = addr.wrapping_add(offset); - - // SAFETY: `write` copies from `chunk_addr` in kernel space without dereferencing it in - // Rust. Invalid user memory is reported by the kernel as EFAULT (nothing copied) or a - // short write (fault partway through). - let nbytes = loop { - let result = - unsafe { libc::write(self.write_fd.as_raw_fd(), chunk_addr.cast(), chunk_len) }; - if result > 0 { - break result as usize; - } - if result == 0 { - return Err(PipeCopyError { - err: io::Error::new( - io::ErrorKind::WriteZero, - "zero-length write while copying process context memory into pipe", - ), - pipe_dirty: false, - }); - } - - let err = io::Error::last_os_error(); - match err.raw_os_error() { - Some(libc::EINTR) => continue, - Some(errno) if errno == libc::EAGAIN || errno == libc::EWOULDBLOCK => { - return Err(PipeCopyError { - err: io::Error::other( - "process context copy pipe blocked despite capacity-limited write", - ), - pipe_dirty: true, - }); - } - // EFAULT is only returned when the fault occurs before any byte is copied, so - // the pipe is still empty. - Some(libc::EFAULT) => { - return Err(PipeCopyError { - err: io::Error::new( - io::ErrorKind::WouldBlock, - "process context memory was unmapped during read", - ), - pipe_dirty: false, - }); - } - _ => { - return Err(PipeCopyError { - err: io::Error::new( - err.kind(), - format!("failed to copy process context memory into pipe: {err}"), - ), - pipe_dirty: false, - }); - } - } - }; - - // A short write (`nbytes < chunk_len`) means the kernel copy stopped partway — either a - // fault or a signal after partial transfer; the two are indistinguishable here. Drain - // the bytes that made it and retry the remainder: `offset` advances by `nbytes`, so the - // next write starts exactly at the stop point. If the source really is unmapped, that - // write fails with EFAULT before copying anything, and we report it then. Since - // `nbytes >= 1`, progress is guaranteed and the loop terminates. - // - // Draining exactly `nbytes` every iteration also keeps the pipe empty before each - // write, so `chunk_len <= capacity` should leave every write ready immediately. - let mut drained = 0; - while drained < nbytes { - // SAFETY: `buf` owns `len` bytes of writable capacity and - // `offset + nbytes <= len`, so the destination range is valid. - let result = unsafe { - libc::read( - self.read_fd.as_raw_fd(), - buf.as_mut_ptr().add(offset + drained).cast(), - nbytes - drained, - ) - }; - if result > 0 { - drained += result as usize; - continue; - } - - if result == 0 { - // Unreachable in practice: this process holds the write end open, so the pipe - // cannot report EOF - return Err(PipeCopyError { - err: io::Error::new( - io::ErrorKind::UnexpectedEof, - "pipe reported EOF while draining process context payload", - ), - pipe_dirty: true, - }); - } - - let err = io::Error::last_os_error(); - match err.raw_os_error() { - Some(libc::EINTR) => continue, - Some(errno) if errno == libc::EAGAIN || errno == libc::EWOULDBLOCK => { - return Err(PipeCopyError { - err: io::Error::other( - "process context copy pipe had no bytes to drain after write", - ), - // The pipe state violated our expected write/drain sequence. - pipe_dirty: true, - }); - } - _ => { - return Err(PipeCopyError { - err: io::Error::new( - err.kind(), - format!("failed to drain process context memory from pipe: {err}"), - ), - // Undrained bytes may remain in the pipe. - pipe_dirty: true, - }); - } - } - } - - offset += nbytes; - } - - // SAFETY: the loop exits with `offset == len`, and every byte of `buf[..offset]` was - // initialized by the pipe reads above. - unsafe { buf.set_len(len) }; - Ok(buf) - } -} - -/// Error from [`copy_via_pipe`], carrying whether the pipe may still hold undrained bytes. -#[derive(Debug)] -struct PipeCopyError { - err: io::Error, - /// If true, the pipe may contain leftover bytes and must not be reused. - pipe_dirty: bool, -} - -#[cfg(test)] -mod tests { - use core::ptr; - use std::io; - - use super::{CopyPipe, ProcessContextSelfReader}; - - fn with_published_mapping(f: impl FnOnce()) { - super::super::publish_raw_payload(b"setup".to_vec()).expect("publish should succeed"); - f(); - super::super::unpublish().expect("unpublish should succeed"); - } - - fn page_size() -> usize { - let page_size = unsafe { libc::sysconf(libc::_SC_PAGESIZE) }; - assert!(page_size > 0, "page size query should succeed"); - page_size as usize - } - - fn map_anonymous(len: usize) -> *mut u8 { - let ptr = unsafe { - libc::mmap( - ptr::null_mut(), - len, - libc::PROT_READ | libc::PROT_WRITE, - libc::MAP_PRIVATE | libc::MAP_ANONYMOUS, - -1, - 0, - ) - }; - assert_ne!( - ptr, - libc::MAP_FAILED, - "anonymous page mapping should succeed" - ); - ptr.cast() - } - - fn map_anonymous_page() -> *mut u8 { - map_anonymous(page_size()) - } - - fn unmap(ptr: *mut u8, len: usize) { - let ret = unsafe { libc::munmap(ptr.cast(), len) }; - assert_eq!(ret, 0, "page unmap should succeed"); - } - - fn unmap_page(ptr: *mut u8) { - unmap(ptr, page_size()); - } - - struct MappedPage(*mut u8); - - impl MappedPage { - fn as_ptr(&self) -> *const u8 { - self.0.cast() - } - } - - impl Drop for MappedPage { - fn drop(&mut self) { - unmap_page(self.0); - } - } - - fn assert_unmapped_read_error(err: io::Error) { - assert_eq!(err.kind(), io::ErrorKind::WouldBlock); - assert!( - err.to_string().contains("unmapped"), - "unexpected error message: {err}" - ); - } - - fn assert_is_otel_mapping(line: &str) { - assert!(ProcessContextSelfReader::is_named_otel_mapping(line)); - } - - fn assert_is_not_otel_mapping(line: &str) { - assert!(!ProcessContextSelfReader::is_named_otel_mapping(line)); - } - - #[test] - #[cfg_attr(miri, ignore)] - #[serial_test::serial] - fn read_returns_would_block_while_context_is_being_updated() { - with_published_mapping(|| { - let reader = ProcessContextSelfReader::new().expect("reader creation should succeed"); - // SAFETY: the mapping was published by this test before being read, and the test is - // serial so no writer is concurrently mutating the plain header field. - let published_at_ns = unsafe { - let header = reader - .header_ptr - .as_ptr() - .cast::(); - let published_at_ns = (*header).monotonic_published_at_ns; - (*header).monotonic_published_at_ns = super::UNPUBLISHED_OR_UPDATING; - published_at_ns - }; - - let err = reader - .read() - .expect_err("read should report writer in progress"); - - assert_eq!(err.kind(), io::ErrorKind::WouldBlock); - assert!( - err.to_string().contains("currently being updated"), - "unexpected error message: {err}" - ); - - // SAFETY: same as above. - unsafe { - (*reader - .header_ptr - .as_ptr() - .cast::()) - .monotonic_published_at_ns = published_at_ns; - } - drop(reader); - }); - } - - #[test] - fn is_named_otel_mapping_matches_exact_mapping_name() { - assert_is_otel_mapping("7f000000-7f001000 rw-p 00000000 00:00 0 /memfd:OTEL_CTX"); - assert_is_otel_mapping("7f000000-7f001000 rw-p 00000000 00:00 0 /memfd:OTEL_CTX (deleted)"); - assert_is_otel_mapping("7f000000-7f001000 rw-p 00000000 00:00 0 [anon_shmem:OTEL_CTX]"); - assert_is_otel_mapping("7f000000-7f001000 rw-p 00000000 00:00 0 [anon:OTEL_CTX]"); - - assert_is_not_otel_mapping( - "7f000000-7f001000 rw-p 00000000 00:00 0 /memfd:OTEL_CTX_BACKUP", - ); - assert_is_not_otel_mapping("7f000000-7f001000 rw-p 00000000 00:00 0 [anon:OTEL_CTX_old]"); - } - - #[test] - #[cfg_attr(miri, ignore)] - #[serial_test::serial] - fn read_process_memory_copies_valid_memory() { - with_published_mapping(|| { - let payload = b"example process context payload"; - - let reader = ProcessContextSelfReader::new().expect("reader creation should succeed"); - let copy = reader - .read_process_memory(payload.as_ptr(), payload.len()) - .expect("payload copy through pipe should succeed"); - - assert_eq!(copy, payload); - }); - } - - #[test] - #[cfg_attr(miri, ignore)] - #[serial_test::serial] - fn read_process_memory_copies_more_than_pipe_capacity() { - with_published_mapping(|| { - let pipe = CopyPipe::new().expect("pipe creation should succeed"); - let len = pipe - .capacity - .checked_add(1) - .expect("pipe capacity should fit payload length"); - let payload: Vec<_> = (0..len).map(|i| i as u8).collect(); - - let copy = pipe - .copy_via_pipe(payload.as_ptr(), payload.len()) - .expect("large payload copy through pipe should succeed"); - - assert_eq!(copy, payload); - }); - } - - #[test] - #[cfg_attr(miri, ignore)] - #[serial_test::serial] - fn read_process_memory_fails_on_unmapped_address() { - with_published_mapping(|| { - let ptr = map_anonymous_page(); - unmap_page(ptr); - - let reader = ProcessContextSelfReader::new().expect("reader creation should succeed"); - let err = reader - .read_process_memory(ptr.cast(), 1) - .expect_err("read from unmapped address should fail"); - - assert_unmapped_read_error(err); - }); - } - - #[test] - #[cfg_attr(miri, ignore)] - #[serial_test::serial] - fn read_process_memory_fails_when_range_extends_past_mapped_page() { - with_published_mapping(|| { - let page_size = page_size(); - let pages = map_anonymous(page_size * 2); - let second_page = pages.wrapping_add(page_size); - unmap_page(second_page); - let first_page = MappedPage(pages); - let len = page_size + 1; - - let reader = ProcessContextSelfReader::new().expect("reader creation should succeed"); - let err = reader - .read_process_memory(first_page.as_ptr(), len) - .expect_err("read past mapped page should fail"); - - assert_unmapped_read_error(err); - }); - } -} diff --git a/libdd-library-config/src/otel_process_ctx/reader.rs b/libdd-library-config/src/otel_process_ctx/reader.rs new file mode 100644 index 0000000000..da766907ca --- /dev/null +++ b/libdd-library-config/src/otel_process_ctx/reader.rs @@ -0,0 +1,268 @@ +// Copyright 2026-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 + +use core::{ + cell::Cell, + mem::{offset_of, size_of}, + ptr::{self, NonNull}, + sync::atomic::{fence, Ordering}, +}; +use std::io; + +use libdd_trace_protobuf::opentelemetry::proto::common::v1::{ + any_value, AnyValue, KeyValue, ProcessContext, +}; +use prost::Message; + +use super::{MappingHeaderSnapshot, PROCESS_CTX_VERSION, SIGNATURE, UNPUBLISHED_OR_UPDATING}; + +#[cfg(target_os = "linux")] +mod copy_pipe_unix; +#[cfg(target_os = "linux")] +pub(super) mod linux; + +#[cfg(target_os = "linux")] +use copy_pipe_unix::CopyPipe as PlatformCopyPipe; + +#[cfg(target_os = "linux")] +type PlatformHeaderDiscovery = linux::HeaderDiscovery; + +pub(super) trait ReaderPlatform { + fn discover_header() -> io::Result>; +} + +pub(super) trait ProcessMemoryCopy: Sized { + fn new() -> io::Result; + fn copy(&self, addr: *const u8, len: usize) -> Result, PipeCopyError>; +} + +#[derive(Debug)] +pub(super) struct PipeCopyError { + pub(super) err: io::Error, + /// If true, the pipe may contain leftover bytes and must not be reused. + pub(super) pipe_dirty: bool, +} + +/// Reader for the current process's OTel process context. +/// +/// The platform implementation locates the header at construction. Reads use a kernel-mediated +/// copy so unpublish and payload replacement races return errors instead of dereferencing freed +/// memory. +pub struct ProcessContextSelfReader { + pid: u32, + pub(in crate::otel_process_ctx) header_ptr: NonNull, + pipe: Cell>, +} + +// SAFETY: the reader owns its copy pipe and only stores a process-global address. Cell keeps the +// type !Sync, so the cached pipe cannot be used concurrently through shared references. +unsafe impl Send for ProcessContextSelfReader {} + +impl ProcessContextSelfReader { + /// Locates the current process's published OTel process context header. + pub fn new() -> io::Result { + Ok(Self { + pid: std::process::id(), + header_ptr: ::discover_header()?, + pipe: Cell::new(None), + }) + } + + /// Reads and decodes the current process's OTel process context. + /// + /// Returns [`io::ErrorKind::WouldBlock`] if a writer is publishing or updating the context, + /// or if the context changed while it was being read. Callers may retry later. + pub fn read(&self) -> io::Result { + if self.pid != std::process::id() { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "process context reader is stale after fork; construct a new reader", + )); + } + + let published_at = self.read_published_at()?; + if published_at == UNPUBLISHED_OR_UPDATING { + return Err(io::Error::new( + io::ErrorKind::WouldBlock, + "process context is currently being updated", + )); + } + + fence(Ordering::SeqCst); + + let header = self.read_header()?; + if header.signature != *SIGNATURE { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "invalid signature in process context header", + )); + } + if header.version != PROCESS_CTX_VERSION { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + format!("unsupported process context version {}", header.version), + )); + } + if header.payload_ptr.is_null() { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "process context payload pointer is null", + )); + } + + let payload = self.read_process_memory(header.payload_ptr, header.payload_size as usize)?; + + fence(Ordering::SeqCst); + if self.read_published_at()? != published_at { + return Err(io::Error::new( + io::ErrorKind::WouldBlock, + "process context changed while being read", + )); + } + + ProcessContext::decode(payload.as_slice()) + .map_err(|err| io::Error::new(io::ErrorKind::InvalidData, err)) + } + + /// Returns the thread-local attribute key map from a decoded process context. + pub fn threadlocal_attribute_key_map(context: &ProcessContext) -> Option> { + Self::find_attr(&context.extra_attributes, "threadlocal.attribute_key_map") + .and_then(Self::string_array) + } + + fn string_array(value: &AnyValue) -> Option> { + let any_value::Value::ArrayValue(array) = value.value.as_ref()? else { + return None; + }; + + array + .values + .iter() + .map(|value| match value.value.as_ref()? { + any_value::Value::StringValue(value) => Some(value.clone()), + _ => None, + }) + .collect() + } + + fn find_attr<'a>(attrs: &'a [KeyValue], key: &str) -> Option<&'a AnyValue> { + attrs + .iter() + .find(|attribute| attribute.key == key) + .and_then(|attribute| attribute.value.as_ref()) + } + + fn read_published_at(&self) -> io::Result { + let timestamp_ptr = self + .header_ptr + .as_ptr() + .wrapping_add(offset_of!(MappingHeaderSnapshot, monotonic_published_at_ns)) + .cast_const(); + let bytes = self.read_process_memory(timestamp_ptr, size_of::())?; + Ok(u64::from_ne_bytes( + Self::field_bytes::<{ size_of::() }>(&bytes, 0, "monotonic_published_at_ns")?, + )) + } + + fn read_header(&self) -> io::Result { + let bytes = self.read_process_memory( + self.header_ptr.as_ptr().cast_const(), + size_of::(), + )?; + Self::mapping_header_from_bytes(&bytes) + } + + fn mapping_header_from_bytes(bytes: &[u8]) -> io::Result { + if bytes.len() != size_of::() { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + format!( + "process context header copy had size {}, expected {}", + bytes.len(), + size_of::() + ), + )); + } + + let signature = Self::field_bytes::<8>( + bytes, + offset_of!(MappingHeaderSnapshot, signature), + "signature", + )?; + let version = u32::from_ne_bytes(Self::field_bytes::<{ size_of::() }>( + bytes, + offset_of!(MappingHeaderSnapshot, version), + "version", + )?); + let payload_size = u32::from_ne_bytes(Self::field_bytes::<{ size_of::() }>( + bytes, + offset_of!(MappingHeaderSnapshot, payload_size), + "payload_size", + )?); + let monotonic_published_at_ns = + u64::from_ne_bytes(Self::field_bytes::<{ size_of::() }>( + bytes, + offset_of!(MappingHeaderSnapshot, monotonic_published_at_ns), + "monotonic_published_at_ns", + )?); + let payload_addr = usize::from_ne_bytes(Self::field_bytes::<{ size_of::() }>( + bytes, + offset_of!(MappingHeaderSnapshot, payload_ptr), + "payload_ptr", + )?); + + Ok(MappingHeaderSnapshot { + signature, + version, + payload_size, + monotonic_published_at_ns, + payload_ptr: ptr::with_exposed_provenance(payload_addr), + }) + } + + fn field_bytes( + bytes: &[u8], + offset: usize, + field: &'static str, + ) -> io::Result<[u8; N]> { + let end = offset.checked_add(N).ok_or_else(|| { + io::Error::new( + io::ErrorKind::InvalidData, + format!("process context header field {field} offset overflowed"), + ) + })?; + let slice = bytes.get(offset..end).ok_or_else(|| { + io::Error::new( + io::ErrorKind::InvalidData, + format!("process context header field {field} is out of bounds"), + ) + })?; + let mut result = [0; N]; + result.copy_from_slice(slice); + Ok(result) + } + + fn read_process_memory(&self, addr: *const u8, len: usize) -> io::Result> { + if len == 0 { + return Ok(Vec::new()); + } + + let pipe = match self.pipe.take() { + Some(pipe) => pipe, + None => PlatformCopyPipe::new()?, + }; + + match pipe.copy(addr, len) { + Ok(bytes) => { + self.pipe.set(Some(pipe)); + Ok(bytes) + } + Err(PipeCopyError { err, pipe_dirty }) => { + if !pipe_dirty { + self.pipe.set(Some(pipe)); + } + Err(err) + } + } + } +} diff --git a/libdd-library-config/src/otel_process_ctx/reader/copy_pipe_unix.rs b/libdd-library-config/src/otel_process_ctx/reader/copy_pipe_unix.rs new file mode 100644 index 0000000000..9211fd0769 --- /dev/null +++ b/libdd-library-config/src/otel_process_ctx/reader/copy_pipe_unix.rs @@ -0,0 +1,240 @@ +// Copyright 2026-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 + +use core::ffi::c_void; +use std::{ + io, + os::fd::{AsRawFd, FromRawFd, OwnedFd}, +}; + +use super::{PipeCopyError, ProcessMemoryCopy}; + +/// A cached pipe used to probe-copy process memory through the kernel. +/// +/// The pipe is empty between calls to [`ProcessMemoryCopy::copy`]. +pub(super) struct CopyPipe { + read_fd: OwnedFd, + write_fd: OwnedFd, + chunk_size: usize, +} + +impl ProcessMemoryCopy for CopyPipe { + fn new() -> io::Result { + let (read_fd, write_fd, chunk_size) = create_pipe()?; + + Ok(Self { + read_fd, + write_fd, + chunk_size, + }) + } + + fn copy(&self, addr: *const u8, len: usize) -> Result, PipeCopyError> { + let mut bytes: Vec = Vec::with_capacity(len); + let mut offset = 0; + + while offset < len { + let chunk_len = (len - offset).min(self.chunk_size); + let chunk_addr = addr.wrapping_add(offset); + + // SAFETY: write asks the kernel to copy from chunk_addr. Invalid user memory is + // reported as EFAULT or a short write without being dereferenced by Rust. + let written = loop { + let result = unsafe { + libc::write( + self.write_fd.as_raw_fd(), + chunk_addr.cast::(), + chunk_len, + ) + }; + if result > 0 { + break result as usize; + } + if result == 0 { + return Err(PipeCopyError { + err: io::Error::new( + io::ErrorKind::WriteZero, + "zero-length write while copying process context memory", + ), + pipe_dirty: false, + }); + } + + let err = io::Error::last_os_error(); + match err.raw_os_error() { + Some(libc::EINTR) => continue, + Some(errno) if errno == libc::EAGAIN || errno == libc::EWOULDBLOCK => { + return Err(PipeCopyError { + err: io::Error::other( + "process context copy pipe blocked despite an empty pipe", + ), + pipe_dirty: true, + }); + } + Some(libc::EFAULT) => { + return Err(PipeCopyError { + err: io::Error::new( + io::ErrorKind::WouldBlock, + "process context memory was unmapped during read", + ), + pipe_dirty: false, + }); + } + _ => { + return Err(PipeCopyError { + err: io::Error::new( + err.kind(), + format!("failed to copy process context memory: {err}"), + ), + pipe_dirty: false, + }); + } + } + }; + + let mut drained = 0; + while drained < written { + // SAFETY: bytes owns len bytes of spare capacity and + // offset + written <= len, so the destination is writable. + let result = unsafe { + libc::read( + self.read_fd.as_raw_fd(), + bytes.as_mut_ptr().add(offset + drained).cast::(), + written - drained, + ) + }; + if result > 0 { + drained += result as usize; + continue; + } + if result == 0 { + return Err(PipeCopyError { + err: io::Error::new( + io::ErrorKind::UnexpectedEof, + "process context copy pipe reported EOF", + ), + pipe_dirty: true, + }); + } + + let err = io::Error::last_os_error(); + match err.raw_os_error() { + Some(libc::EINTR) => continue, + _ => { + return Err(PipeCopyError { + err: io::Error::new( + err.kind(), + format!("failed to drain process context copy pipe: {err}"), + ), + pipe_dirty: true, + }); + } + } + } + + offset += written; + } + + // SAFETY: every byte was initialized by the pipe reads above. + unsafe { bytes.set_len(len) }; + Ok(bytes) + } +} + +#[cfg(target_os = "linux")] +fn create_pipe() -> io::Result<(OwnedFd, OwnedFd, usize)> { + let mut fds = [0; 2]; + // SAFETY: fds points to space for the two descriptors returned by pipe2. + if unsafe { libc::pipe2(fds.as_mut_ptr(), libc::O_CLOEXEC | libc::O_NONBLOCK) } != 0 { + return Err(last_error("failed to create process context copy pipe")); + } + + // SAFETY: pipe2 initialized both descriptors and ownership is transferred exactly once. + let (read_fd, write_fd) = + unsafe { (OwnedFd::from_raw_fd(fds[0]), OwnedFd::from_raw_fd(fds[1])) }; + + // SAFETY: write_fd is a valid pipe descriptor. + let capacity = unsafe { libc::fcntl(write_fd.as_raw_fd(), libc::F_GETPIPE_SZ) }; + if capacity < 0 { + return Err(last_error( + "failed to query process context copy pipe capacity", + )); + } + if capacity == 0 { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "process context copy pipe has zero capacity", + )); + } + + Ok((read_fd, write_fd, capacity as usize)) +} + +fn last_error(context: &'static str) -> io::Error { + let err = io::Error::last_os_error(); + io::Error::new(err.kind(), format!("{context}: {err}")) +} + +#[cfg(test)] +mod tests { + use core::ptr; + + use super::{io, CopyPipe, ProcessMemoryCopy}; + + #[test] + #[cfg_attr(miri, ignore)] + fn copies_valid_memory_across_multiple_chunks() { + let pipe = CopyPipe::new().expect("pipe creation should succeed"); + let len = pipe.chunk_size + 1; + let source: Vec<_> = (0..len).map(|index| index as u8).collect(); + + let copied = pipe + .copy(source.as_ptr(), source.len()) + .expect("memory copy should succeed"); + + assert_eq!(copied, source); + } + + #[test] + #[cfg_attr(miri, ignore)] + fn rejects_a_range_crossing_into_inaccessible_memory() { + // SAFETY: the arguments reserve two writable anonymous pages. + let page_size = unsafe { libc::sysconf(libc::_SC_PAGESIZE) }; + assert!(page_size > 0, "page size query should succeed"); + let page_size = page_size as usize; + let len = page_size * 2; + let address = unsafe { + libc::mmap( + ptr::null_mut(), + len, + libc::PROT_READ | libc::PROT_WRITE, + libc::MAP_PRIVATE | libc::MAP_ANON, + -1, + 0, + ) + }; + assert_ne!(address, libc::MAP_FAILED); + // SAFETY: the second page is part of the mapping above. + assert_eq!( + unsafe { + libc::mprotect( + address.cast::().add(page_size).cast(), + page_size, + libc::PROT_NONE, + ) + }, + 0 + ); + + let err = CopyPipe::new() + .expect("pipe creation should succeed") + // SAFETY: the last byte of the first page is inside the live mapping. + .copy(unsafe { address.cast::().add(page_size - 1) }, 2) + .expect_err("a copy crossing into inaccessible memory should fail"); + + assert_eq!(err.err.kind(), io::ErrorKind::WouldBlock); + assert!(!err.pipe_dirty); + // SAFETY: address and len came from mmap above. + assert_eq!(unsafe { libc::munmap(address, len) }, 0); + } +} diff --git a/libdd-library-config/src/otel_process_ctx/reader/linux.rs b/libdd-library-config/src/otel_process_ctx/reader/linux.rs new file mode 100644 index 0000000000..cf5332dca9 --- /dev/null +++ b/libdd-library-config/src/otel_process_ctx/reader/linux.rs @@ -0,0 +1,135 @@ +// Copyright 2026-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 + +use core::{ptr, ptr::NonNull}; +use std::{ + fs::File, + io::{self, BufRead, BufReader}, +}; + +use super::ReaderPlatform; + +pub(super) struct HeaderDiscovery; + +impl ReaderPlatform for HeaderDiscovery { + fn discover_header() -> io::Result> { + let address = find_otel_mapping()?; + NonNull::new(ptr::with_exposed_provenance::(address).cast_mut()).ok_or_else(|| { + io::Error::new(io::ErrorKind::InvalidData, "null process context header") + }) + } +} + +/// Finds the OTEL_CTX mapping in `/proc/self/maps`. +pub(in crate::otel_process_ctx) fn find_otel_mapping() -> io::Result { + let file = File::open("/proc/self/maps")?; + let reader = BufReader::new(file); + + for line in reader.lines() { + let line = line?; + if is_named_otel_mapping(&line) { + if let Some(address) = parse_mapping_start(&line) { + return Ok(address); + } + } + } + + Err(io::Error::new( + io::ErrorKind::NotFound, + "couldn't find the mapping of the process context", + )) +} + +fn parse_mapping_start(line: &str) -> Option { + usize::from_str_radix(line.split('-').next()?, 16).ok() +} + +fn is_named_otel_mapping(line: &str) -> bool { + let Some(name) = line.split_whitespace().nth(5) else { + return false; + }; + + matches!( + name, + "/memfd:OTEL_CTX" | "[anon_shmem:OTEL_CTX]" | "[anon:OTEL_CTX]" + ) +} + +#[cfg(test)] +mod tests { + use super::is_named_otel_mapping; + + #[test] + fn recognizes_exact_mapping_names() { + assert!(is_named_otel_mapping( + "7f000000-7f001000 rw-p 00000000 00:00 0 /memfd:OTEL_CTX" + )); + assert!(is_named_otel_mapping( + "7f000000-7f001000 rw-p 00000000 00:00 0 /memfd:OTEL_CTX (deleted)" + )); + assert!(is_named_otel_mapping( + "7f000000-7f001000 rw-p 00000000 00:00 0 [anon_shmem:OTEL_CTX]" + )); + assert!(is_named_otel_mapping( + "7f000000-7f001000 rw-p 00000000 00:00 0 [anon:OTEL_CTX]" + )); + assert!(!is_named_otel_mapping( + "7f000000-7f001000 rw-p 00000000 00:00 0 /memfd:OTEL_CTX_BACKUP" + )); + } + + #[cfg(feature = "process-context-writer")] + mod with_writer { + use core::sync::atomic::Ordering; + + use super::super::io; + use crate::otel_process_ctx::ProcessContextSelfReader; + use crate::otel_process_ctx::{writer::MappingHeader, UNPUBLISHED_OR_UPDATING}; + + #[test] + #[cfg_attr(miri, ignore)] + #[serial_test::serial] + fn read_returns_would_block_while_context_is_being_updated() { + crate::otel_process_ctx::writer::publish_raw_payload(b"published payload".to_vec()) + .expect("publish should succeed"); + let reader = ProcessContextSelfReader::new().expect("reader creation should succeed"); + // SAFETY: the mapping is live and this serial test excludes concurrent publishers. + let header = reader.header_ptr.as_ptr().cast::(); + let published_at = unsafe { + (*header) + .monotonic_published_at_ns + .swap(UNPUBLISHED_OR_UPDATING, Ordering::Relaxed) + }; + + let error = reader + .read() + .expect_err("read should report writer in progress"); + assert_eq!(error.kind(), io::ErrorKind::WouldBlock); + + // SAFETY: the mapping remains live and this restores the field changed above. + unsafe { + (*header) + .monotonic_published_at_ns + .store(published_at, Ordering::Relaxed); + } + crate::otel_process_ctx::unpublish().expect("unpublish should succeed"); + } + + #[test] + #[cfg_attr(miri, ignore)] + #[serial_test::serial] + fn discovers_and_reads_published_context() { + crate::otel_process_ctx::writer::publish_raw_payload(b"published payload".to_vec()) + .expect("publish should succeed"); + + let reader = ProcessContextSelfReader::new().expect("reader creation should succeed"); + let error = reader + .read() + .expect_err("raw test payload is not a protobuf context"); + assert_eq!(error.kind(), io::ErrorKind::InvalidData); + + crate::otel_process_ctx::unpublish().expect("unpublish should succeed"); + assert!(ProcessContextSelfReader::new().is_err()); + } + } +} diff --git a/libdd-library-config/src/otel_process_ctx/writer.rs b/libdd-library-config/src/otel_process_ctx/writer.rs new file mode 100644 index 0000000000..e4367b3f6d --- /dev/null +++ b/libdd-library-config/src/otel_process_ctx/writer.rs @@ -0,0 +1,296 @@ +// Copyright 2026-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 + +use core::{ + convert::TryInto, + marker::PhantomData, + mem::swap, + ptr, + sync::atomic::{fence, AtomicPtr, AtomicU32, AtomicU64, Ordering}, +}; +use std::{ + io, + sync::{Mutex, MutexGuard}, +}; + +use libdd_trace_protobuf::opentelemetry::proto::common::v1::ProcessContext; +use prost::Message; + +use super::{PROCESS_CTX_VERSION, SIGNATURE, UNPUBLISHED_OR_UPDATING}; + +#[cfg(target_os = "linux")] +pub(super) mod linux; + +/// The header structure written at the start of the mapping. This must match the C +/// layout of the specification. +/// +/// The atomic fields have the same size and alignment as their corresponding C fields. They +/// provide the aligned word-sized accesses required by the publication protocol, while explicit +/// fences constrain store/load ordering. +#[repr(C)] +pub(super) struct MappingHeader { + pub(super) signature: [u8; 8], + pub(super) version: u32, + pub(super) payload_size: AtomicU32, + pub(super) monotonic_published_at_ns: AtomicU64, + pub(super) payload_ptr: AtomicPtr, +} + +// Compile-time verification that MappingHeader matches the field offsets and total size +// mandated by the OTel process context spec: +// https://github.com/open-telemetry/opentelemetry-specification/blob/main/oteps/profiles/4719-process-ctx.md +const _: () = { + use core::mem::{offset_of, size_of}; + assert!(offset_of!(MappingHeader, signature) == 0); + assert!(offset_of!(MappingHeader, version) == 8); + assert!(offset_of!(MappingHeader, payload_size) == 12); + assert!(offset_of!(MappingHeader, monotonic_published_at_ns) == 16); + assert!(offset_of!(MappingHeader, payload_ptr) == 24); + assert!(size_of::() == 32); + assert!(core::mem::align_of::() == 8); + assert!(size_of::<*const u8>() == size_of::()); +}; + +#[cfg(target_os = "linux")] +type HeaderMemory = linux::MemMapping; + +#[cfg(target_os = "linux")] +type PlatformMonotonicClock = linux::MonotonicClock; + +type ProcessContextHandle = ProcessContextHandleGen; + +/// The global instance of the context for the current process. +/// +/// We need a mutex to put the handle in a static and avoid bothering the users of this API +/// with storing the handle, but we don't expect this mutex to actually be contended. Ideally a +/// single thread should handle context updates, even if it's not strictly required. +static PROCESS_CONTEXT_HANDLER: Mutex> = Mutex::new(None); + +pub(super) trait HeaderMemoryHolder: Sized { + fn new() -> io::Result; + fn as_ptr(&self) -> *mut MappingHeader; + fn make_discoverable(&mut self); + fn unpublish_and_release(self) -> io::Result<()>; + fn after_fork(self); +} + +pub(super) trait MonotonicTime { + fn monotonic_time_ns() -> io::Result; +} + +/// Handle for future updates of a published process context. +struct ProcessContextHandleGen { + mapping: M, + /// Once published, and until the next update is complete, the backing allocation of + /// `payload` might be read and thus must not move (e.g. by resizing or drop). + payload: Vec, + /// The process id of the last publisher. This is useful to detect forks(), and publish a + /// new context accordingly. + pid: u32, + monotonic_clock: PhantomData, +} + +impl ProcessContextHandleGen { + /// Initial publication of the process context. Creates an appropriate header allocation. + fn publish(payload: Vec) -> io::Result { + let payload_size: u32 = payload + .len() + .try_into() + .map_err(|_| io::Error::other("payload size overflowed"))?; + + let mut mapping = M::new()?; + let published_at_ns = T::monotonic_time_ns()?; + + let header = mapping.as_ptr(); + + // SAFETY: header points to a zero-filled, writable allocation of at least + // mapping_size() bytes with MappingHeader alignment; field projections are in-bounds. + // The pointer writes do not happen while there are live &MappingHeader references + // and, to the extent the atomic stores do, this is fine because the mutated bytes + // are inside UnsafeCells. + unsafe { + ptr::addr_of_mut!((*header).signature).write(*SIGNATURE); + ptr::addr_of_mut!((*header).version).write(PROCESS_CTX_VERSION); + (*header) + .payload_ptr + .store(payload.as_ptr().cast_mut(), Ordering::Relaxed); + (*header) + .payload_size + .store(payload_size, Ordering::Relaxed); + + fence(Ordering::SeqCst); + (*header) + .monotonic_published_at_ns + .store(published_at_ns, Ordering::Relaxed); + } + + mapping.make_discoverable(); + + Ok(ProcessContextHandleGen { + mapping, + payload, + pid: std::process::id(), + monotonic_clock: PhantomData, + }) + } + + /// Updates the context after initial publication. + fn update(&mut self, payload: Vec) -> io::Result<()> { + let header = self.mapping.as_ptr(); + + let monotonic_published_at_ns = T::monotonic_time_ns()?; + let payload_size: u32 = payload.len().try_into().map_err(|_| { + io::Error::other("couldn't update process context: new payload too large") + })?; + // A process shouldn't try to concurrently update its own context. + // + // `UNPUBLISHED_OR_UPDATING` is an out-of-band sentinel, not a value that + // the monotonic clock is expected to produce after publication. Published non-zero + // timestamp values must advance monotonically; the field may temporarily hold the sentinel + // while an update is in progress. + // + // Note: be careful of early return while `monotonic_published_at` is still zero, as + // subsequent updates would get stuck. + let last_monotonic_published_at_ns = unsafe { + (*header) + .monotonic_published_at_ns + .swap(UNPUBLISHED_OR_UPDATING, Ordering::Relaxed) + }; + if last_monotonic_published_at_ns == UNPUBLISHED_OR_UPDATING { + panic!("concurrent update of the process context is not supported"); + } + + let monotonic_published_at_ns = + monotonic_published_at_ns.max(last_monotonic_published_at_ns.saturating_add(1)); + + // Prevent the payload metadata and payload replacement below from moving above the + // unavailable marker. In particular, if a reader starts from the previous non-zero + // timestamp but copies data after this update begins, it must not accept that copy as the + // previous version: its final timestamp check should see `UNPUBLISHED_OR_UPDATING` or the + // later published timestamp. + fence(Ordering::SeqCst); + self.payload = payload; + + unsafe { + (*header) + .payload_ptr + .store(self.payload.as_ptr().cast_mut(), Ordering::Relaxed); + (*header) + .payload_size + .store(payload_size, Ordering::Relaxed); + } + + // Prevent the final timestamp publication from moving above either the payload metadata + // writes or the payload bytes written before this method was called. Readers fence after + // observing this non-zero timestamp before copying both. + fence(Ordering::SeqCst); + + unsafe { + (*header) + .monotonic_published_at_ns + .store(monotonic_published_at_ns, Ordering::Relaxed); + } + + Ok(()) + } +} + +// The returned size is guaranteed to be larger or equal to the size of `MappingHeader`. +#[cfg(target_os = "linux")] +pub(super) const fn mapping_size() -> usize { + core::mem::size_of::() +} + +/// Locks the context handle. Returns a uniform error if the lock has been poisoned. +fn lock_context_handle() -> io::Result>> { + PROCESS_CONTEXT_HANDLER.lock().map_err(|_| { + io::Error::other("a thread panicked while operating on the process context handler") + }) +} + +/// Publishes or updates the process context for it to be visible by external readers. +/// +/// If any of the following condition holds: +/// +/// - this is the first publication +/// - [unpublish] has been called last +/// - the previous context has been published from a different process id (that is, a `fork()` +/// happened and we're the child process) +/// +/// Then we follow the Publish protocol of the OTel process context specification (allocating a +/// fresh header). +/// +/// Otherwise, if a context has been previously published from the same process and hasn't been +/// unpublished since, we follow the Update protocol. +/// +/// # Fork safety +/// +/// If we're a forked children of the original publisher, we are extremely restricted in the +/// set of operations that we can do (we must be async-signal-safe). On paper, heap allocation +/// is Undefined Behavior, for example. We assume that a forking runtime (such as Python or +/// Ruby) that doesn't follow with an immediate `exec` is already "taking that risk", so to +/// speak (typically, if no thread is ever spawned before the fork, things are mostly fine). +#[inline] +pub fn publish(context: &ProcessContext) -> io::Result<()> { + publish_raw_payload(context.encode_to_vec()) +} + +pub(super) fn publish_raw_payload(payload: Vec) -> io::Result<()> { + let mut guard = lock_context_handle()?; + + match &mut *guard { + Some(handler) if handler.pid == std::process::id() => handler.update(payload), + Some(handler) => { + let mut local_handler = ProcessContextHandleGen::publish(payload)?; + // If we've been forked, we need to prevent the mapping from being dropped + // normally, as it would try to unmap a region that isn't mapped anymore in the + // child process, or worse, could have been remapped to something else in the + // meantime. + // + // To do so, we get the old handler back in `local_handler` and prevent `mapping` + // from being dropped specifically. + swap(&mut local_handler, handler); + local_handler.mapping.after_fork(); + + Ok(()) + } + None => { + *guard = Some(ProcessContextHandleGen::publish(payload)?); + Ok(()) + } + } +} + +/// Removes the process context publication and releases its header allocation. If no context has +/// ever been published, this is a no-op. +/// +/// A call to [publish] following an [unpublish] will create a new mapping. +pub fn unpublish() -> io::Result<()> { + let mut guard = lock_context_handle()?; + + if let Some(ProcessContextHandleGen { + mapping, payload, .. + }) = guard.take() + { + // Mark the context as unavailable before freeing the mapping/payload. The fence forces + // the writing CPU not to reorder the unavailable timestamp store and the deallocation + // stores. This gives readers more of a chance (but no guarantee) to observe an + // unavailable context before the publication is removed. + // + // SAFETY: the mapping is still live and valid, and the global mutex prevents + // concurrent in-process writers from mutating the plain header fields. + let header = mapping.as_ptr(); + unsafe { + (*header) + .monotonic_published_at_ns + .store(UNPUBLISHED_OR_UPDATING, Ordering::Relaxed); + } + fence(Ordering::SeqCst); + + // The payload will still drop if this fails, leaving a zero timestamp behind. + mapping.unpublish_and_release()?; + drop(payload); + } + + Ok(()) +} diff --git a/libdd-library-config/src/otel_process_ctx/writer/linux.rs b/libdd-library-config/src/otel_process_ctx/writer/linux.rs new file mode 100644 index 0000000000..7c5ef2f145 --- /dev/null +++ b/libdd-library-config/src/otel_process_ctx/writer/linux.rs @@ -0,0 +1,264 @@ +// Copyright 2026-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 + +use core::{ + convert::TryInto, + ffi::{c_void, CStr}, + mem::{forget, ManuallyDrop}, + ptr::{self, NonNull}, + time::Duration, +}; +use std::{ + io, + os::fd::{AsRawFd, FromRawFd, OwnedFd}, +}; + +/// The shared memory mapped area to publish the context to. The memory region is owned by a +/// [MemMapping] instance and is automatically unmapped upon drop. +/// +/// # Safety +/// +/// The following invariants MUST always hold for safety and are guaranteed by [MemMapping]: +/// - `start` is non-null, is coming from a previous call to `mmap` with a size value of +/// `mapping_size()` and hasn't been unmmaped since. +/// - once `self` has been dropped, no memory access must be performed on the memory previously +/// pointed to by `start`. +pub(super) struct MemMapping { + start_addr: NonNull, +} + +// SAFETY: MemMapping represents ownership over the mapped region. It never leaks or +// share the internal pointer. It's also safe to drop (`munmap`) from a different thread. +unsafe impl Send for MemMapping {} + +impl super::HeaderMemoryHolder for MemMapping { + fn new() -> io::Result { + let mapping = Self::new()?; + check_syscall_retval( + // SAFETY: MemMapping owns a live mapping of mapping_size() bytes. + unsafe { + libc::madvise( + mapping.start_addr.as_ptr(), + super::mapping_size(), + libc::MADV_DONTFORK, + ) + }, + "madvise MADVISE_DONTFORK failed", + )?; + Ok(mapping) + } + + fn as_ptr(&self) -> *mut super::MappingHeader { + self.start_addr.as_ptr() as *mut super::MappingHeader + } + + fn make_discoverable(&mut self) { + // Naming must be attempted even when it is expected to fail. A naming failure does not + // invalidate the publication protocol. + let _ = self.set_name(); + } + + fn unpublish_and_release(self) -> io::Result<()> { + self.free() + } + + fn after_fork(self) { + forget(self); + } +} + +pub(super) struct MonotonicClock; + +impl super::MonotonicTime for MonotonicClock { + fn monotonic_time_ns() -> io::Result { + since_boottime_ns().ok_or_else(|| { + io::Error::other("failed to get current time for process context publication") + }) + } +} + +impl MemMapping { + /// Creates a suitable memory mapping for the context protocol to be published. + /// + /// `memfd` is the preferred method, but this function fallbacks to an anonymous mapping if + /// `memfd` failed for any reason. + /// + /// Both allocation paths produce zero-filled memory: `MAP_ANONYMOUS` mappings are + /// initialized to zero, and the memfd path maps a newly-created file extended by + /// `ftruncate()`, whose extended bytes read as `\0`. This matters because a memfd-backed + /// mapping is discoverable before `set_name()` runs, so early readers may race with header + /// initialization. They must observe the unpublished/updating sentinel (0) and stop until the + /// final timestamp store publishes the initialized header. + fn new() -> io::Result { + let size = super::mapping_size(); + + try_memfd(crate::otel_process_ctx::linux::MAPPING_NAME, libc::MFD_CLOEXEC | libc::MFD_NOEXEC_SEAL | libc::MFD_ALLOW_SEALING) + .or_else(|_| try_memfd(crate::otel_process_ctx::linux::MAPPING_NAME, libc::MFD_CLOEXEC | libc::MFD_ALLOW_SEALING)) + .and_then(|fd| { + // SAFETY: fd is a valid open file descriptor. + check_syscall_retval( + unsafe { + libc::ftruncate(fd.as_raw_fd(), super::mapping_size() as libc::off_t) + }, + "ftruncate failed" + )?; + // SAFETY: we pass a null pointer to mmap which is unconditionally ok + let start_addr = check_mapping_addr( + unsafe { + libc::mmap( + ptr::null_mut(), + size, + libc::PROT_WRITE | libc::PROT_READ, + libc::MAP_PRIVATE, + fd.as_raw_fd(), + 0, + ) + }, + "mmap failed" + )?; + + // We (implicitly) close the file descriptor right away, but this ok + Ok(MemMapping { start_addr }) + }) + // If any previous step failed, we fallback to an anonymous mapping + .or_else(|_| { + // SAFETY: we pass a null pointer to mmap, no precondition to uphold + let start_addr = check_mapping_addr( + unsafe { + libc::mmap( + ptr::null_mut(), + size, + libc::PROT_WRITE | libc::PROT_READ, + libc::MAP_PRIVATE | libc::MAP_ANONYMOUS, + -1, + 0, + ) + }, + "mmap failed: couldn't create a memfd or anonymous mmapped region for process context publication" + )?; + + Ok(MemMapping { start_addr }) + }) + } + + /// Makes this mapping discoverable by giving it a name. + fn set_name(&mut self) -> io::Result<()> { + // SAFETY: self.start_addr is valid for mapping_size() bytes as per MemMapping + // invariants. name is a valid NUL-terminated string that outlives the prctl call. + check_syscall_retval( + unsafe { + // int prctl(PR_SET_VMA, long attr, unsigned long addr, unsigned long size, + // const char *_Nullable val); + libc::prctl( + libc::PR_SET_VMA, + libc::PR_SET_VMA_ANON_NAME as libc::c_ulong, + TryInto::::try_into(self.start_addr.as_ptr() as usize) + .expect("start addr overflowed"), + TryInto::::try_into(super::mapping_size()) + .expect("mapping size overflowed"), + crate::otel_process_ctx::linux::MAPPING_NAME.as_ptr(), + ) + }, + "prctl PR_SET_VMA_ANON_NAME failed", + )?; + + Ok(()) + } + + /// Unmaps the underlying memory region. This has same effect as dropping `self`, but + /// propagates potential errors. + fn free(mut self) -> io::Result<()> { + // SAFETY: We put `self` in a `ManuallyDrop`, which prevents drop and future calls to + // `free()`. + unsafe { + self.unmap()?; + } + + // Prevent `Self::drop` from being called + let _ = ManuallyDrop::new(self); + + Ok(()) + } + + /// Unmaps the underlying memory region. For internal use only; prefer `free()` or `drop()`. + /// + /// # Safety + /// + /// This method must only be called once. After calling `unmap()`, no other method of + /// `MemMapping` must be ever called on `self` again, including `unmap()` and `drop()`. + /// + /// Practically, `self` must be put in a `ManuallyDrop` wrapper and forgotten, or being in + /// the process of being dropped. + unsafe fn unmap(&mut self) -> io::Result<()> { + check_syscall_retval( + // SAFETY: upheld by the caller. + unsafe { libc::munmap(self.start_addr.as_ptr(), super::mapping_size()) }, + "munmap failed when freeing the process context", + )?; + + Ok(()) + } +} + +impl Drop for MemMapping { + fn drop(&mut self) { + // SAFETY: `self` is being dropped + let _ = unsafe { self.unmap() }; + } +} + +/// Returns `Err` wrapping the current `errno` with `msg` as context if `ret` is negative, +/// `Ok(ret)` otherwise. +fn check_syscall_retval(ret: libc::c_int, msg: &'static str) -> io::Result { + if ret < 0 { + let e = io::Error::last_os_error(); + Err(io::Error::new(e.kind(), format!("{msg}: {e}"))) + } else { + Ok(ret) + } +} + +/// Returns `Err` wrapping the current `errno` with `msg` as context if `addr` equals +/// `MAP_FAILED`, `Ok(addr)` otherwise. +fn check_mapping_addr(addr: *mut c_void, msg: &'static str) -> io::Result> { + if addr == libc::MAP_FAILED { + let e = io::Error::last_os_error(); + Err(io::Error::new(e.kind(), format!("{msg}: {e}"))) + } else { + // SAFETY: mmap returns a non-null pointer on success. + Ok(unsafe { NonNull::new_unchecked(addr) }) + } +} + +/// Creates a `memfd` file descriptor with the given name and flags. +fn try_memfd(name: &CStr, flags: libc::c_uint) -> io::Result { + // We use the raw syscall rather than `libc::memfd_create` because the latter requires + // glibc >= 2.27, while `syscall()` + `SYS_memfd_create` works with any glibc version. + check_syscall_retval( + // SAFETY: name is a valid NUL-terminated string; flags are constant bit flags. + unsafe { + libc::syscall(libc::SYS_memfd_create, name.as_ptr(), flags as libc::c_long) + as libc::c_int + }, + "memfd_create failed", + ) + // SAFETY: fd is a valid file descriptor just returned by memfd_create. + .map(|fd| unsafe { OwnedFd::from_raw_fd(fd) }) +} + +/// Returns the value of the monotonic BOOTTIME clock in nanoseconds. +fn since_boottime_ns() -> Option { + let mut ts = libc::timespec { + tv_sec: 0, + tv_nsec: 0, + }; + // SAFETY: ts is a valid, writable timespec. + let ret = unsafe { libc::clock_gettime(libc::CLOCK_BOOTTIME, &mut ts) }; + if ret != 0 { + return None; + } + let secs: u64 = ts.tv_sec.try_into().ok()?; + let nanos: u32 = ts.tv_nsec.try_into().ok()?; + let duration = Duration::new(secs, nanos); + u64::try_from(duration.as_nanos()).ok() +} diff --git a/libdd-library-config/src/tracer_metadata.rs b/libdd-library-config/src/tracer_metadata.rs index 55cd56e56b..aa9c3b9146 100644 --- a/libdd-library-config/src/tracer_metadata.rs +++ b/libdd-library-config/src/tracer_metadata.rs @@ -220,7 +220,8 @@ mod linux { pub fn store_tracer_metadata( data: &super::TracerMetadata, ) -> anyhow::Result { - let _ = crate::otel_process_ctx::linux::publish(&data.to_otel_process_ctx()); + #[cfg(feature = "process-context-writer")] + let _ = crate::otel_process_ctx::publish(&data.to_otel_process_ctx()); let uid: String = rand::thread_rng() .sample_iter(&Alphanumeric)