From f9ff7b4a4de6d306af4959d6371cd96db260a876 Mon Sep 17 00:00:00 2001 From: Levi Morrison Date: Wed, 22 Jul 2026 09:29:19 -0600 Subject: [PATCH 1/2] Add caller-owned OTel context APIs --- libdd-library-config/Cargo.toml | 2 +- libdd-library-config/src/lib.rs | 4 - libdd-library-config/src/otel_process_ctx.rs | 22 +- .../src/otel_process_ctx/mapping.rs | 380 ++++++++++++++++++ libdd-library-config/src/tracer_metadata.rs | 56 ++- libdd-otel-thread-ctx-ffi/Cargo.toml | 9 +- libdd-otel-thread-ctx-ffi/build.rs | 2 +- libdd-otel-thread-ctx-ffi/cbindgen.toml | 2 + libdd-otel-thread-ctx-ffi/src/lib.rs | 141 ++++++- libdd-otel-thread-ctx/Cargo.toml | 6 +- libdd-otel-thread-ctx/src/lib.rs | 203 +--------- libdd-otel-thread-ctx/src/record.rs | 144 +++++++ 12 files changed, 758 insertions(+), 213 deletions(-) create mode 100644 libdd-library-config/src/otel_process_ctx/mapping.rs create mode 100644 libdd-otel-thread-ctx/src/record.rs diff --git a/libdd-library-config/Cargo.toml b/libdd-library-config/Cargo.toml index 0258caff00..599c7140f4 100644 --- a/libdd-library-config/Cargo.toml +++ b/libdd-library-config/Cargo.toml @@ -35,7 +35,7 @@ libdd-trace-protobuf = { version = "4.0.0", path = "../libdd-trace-protobuf" } tempfile = { version = "3.3" } serial_test = "3.2" -[target.'cfg(unix)'.dependencies] +[target.'cfg(target_os = "linux")'.dependencies] memfd = { version = "0.6" } libc = "0.2" diff --git a/libdd-library-config/src/lib.rs b/libdd-library-config/src/lib.rs index d801748d9f..6d473f1d84 100644 --- a/libdd-library-config/src/lib.rs +++ b/libdd-library-config/src/lib.rs @@ -2,10 +2,6 @@ // 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 48a548726c..9f179ee3bd 100644 --- a/libdd-library-config/src/otel_process_ctx.rs +++ b/libdd-library-config/src/otel_process_ctx.rs @@ -1,7 +1,7 @@ // Copyright 2026-Present Datadog, Inc. https://www.datadoghq.com/ // SPDX-License-Identifier: Apache-2.0 -//! Implementation of the Linux parts of the [OTEL process +//! Platform-neutral mapping operations and Linux publication/discovery for the [OTEL process //! context specification](https://github.com/open-telemetry/opentelemetry-specification/blob/main/oteps/profiles/4719-process-ctx.md). //! //! The update/read protocol is seqlock-style: the publisher marks the mapping as unavailable, @@ -18,20 +18,28 @@ //! 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(target_os = "linux")] use std::io; -#[cfg(feature = "process-context-reader")] +mod mapping; + +pub use mapping::{ + decode, initialize, invalidate, read, threadlocal_attribute_key_map, update, validate, + ProcessContextMapping, +}; + +#[cfg(all(target_os = "linux", feature = "process-context-reader"))] mod reader; -#[cfg(feature = "process-context-writer")] +#[cfg(all(target_os = "linux", 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")] pub mod linux; -#[cfg(feature = "process-context-reader")] +#[cfg(all(target_os = "linux", feature = "process-context-reader"))] pub use reader::ProcessContextSelfReader; -#[cfg(feature = "process-context-writer")] +#[cfg(all(target_os = "linux", feature = "process-context-writer"))] pub use writer::{publish, unpublish}; /// Current version of the process context format @@ -42,7 +50,7 @@ pub const SIGNATURE: &[u8; 8] = b"OTEL_CTX"; const UNPUBLISHED_OR_UPDATING: u64 = 0; #[repr(C)] -#[cfg(feature = "process-context-reader")] +#[cfg(all(target_os = "linux", feature = "process-context-reader"))] struct MappingHeaderSnapshot { signature: [u8; 8], version: u32, @@ -52,6 +60,7 @@ struct MappingHeaderSnapshot { } /// Runs an operation until it succeeds or fails for a reason other than `EINTR`. +#[cfg(target_os = "linux")] fn retry_on_eintr(mut operation: impl FnMut() -> io::Result) -> io::Result { loop { match operation() { @@ -63,6 +72,7 @@ fn retry_on_eintr(mut operation: impl FnMut() -> io::Result) -> io::Result #[cfg(all( test, + target_os = "linux", feature = "process-context-reader", feature = "process-context-writer" ))] diff --git a/libdd-library-config/src/otel_process_ctx/mapping.rs b/libdd-library-config/src/otel_process_ctx/mapping.rs new file mode 100644 index 0000000000..b0e5014b68 --- /dev/null +++ b/libdd-library-config/src/otel_process_ctx/mapping.rs @@ -0,0 +1,380 @@ +// Copyright 2026-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 + +//! Platform-neutral process-context operations over caller-owned storage. + +use core::{ + mem::{align_of, offset_of, size_of}, + ptr::{self, NonNull}, + sync::atomic::{fence, AtomicPtr, AtomicU32, AtomicU64, AtomicU8, Ordering}, +}; +use std::{io, sync::OnceLock, time::Instant}; + +use libdd_trace_protobuf::opentelemetry::proto::common::v1::{ + any_value, AnyValue, KeyValue, ProcessContext, +}; +use prost::Message; + +use super::{PROCESS_CTX_VERSION, SIGNATURE, UNPUBLISHED_OR_UPDATING}; + +#[repr(C)] +struct MappingHeader { + signature: [u8; 8], + version: u32, + payload_size: AtomicU32, + monotonic_published_at_ns: AtomicU64, + payload_ptr: AtomicPtr, +} + +const _: () = { + 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!(align_of::() == 8); +}; + +/// A caller-owned memory region containing a process-context header and inline payload. +#[derive(Clone, Copy)] +pub struct ProcessContextMapping { + base: NonNull, + len: usize, +} + +// SAFETY: this type does not own or dereference the region implicitly. Synchronization is supplied +// by the process-context protocol, and callers retain responsibility for the region lifetime. +unsafe impl Send for ProcessContextMapping {} +unsafe impl Sync for ProcessContextMapping {} + +impl ProcessContextMapping { + /// Constructs a view over caller-owned storage. + /// + /// # Safety + /// + /// `base..base + len` must remain mapped and readable for every operation on this value, and + /// writable for initialization, update, and invalidation. The base must be 8-byte aligned. + pub unsafe fn from_raw_parts(base: *mut u8, len: usize) -> io::Result { + let base = NonNull::new(base).ok_or_else(|| { + io::Error::new(io::ErrorKind::InvalidInput, "null process context mapping") + })?; + if len < size_of::() { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "process context mapping is smaller than its header", + )); + } + if base.as_ptr().align_offset(align_of::()) != 0 { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "process context mapping is not 8-byte aligned", + )); + } + Ok(Self { base, len }) + } + + fn header(self) -> *mut MappingHeader { + self.base.as_ptr().cast() + } + + fn payload(self) -> *mut u8 { + self.base.as_ptr().wrapping_add(size_of::()) + } + + fn capacity(self) -> usize { + self.len - size_of::() + } +} + +pub fn initialize(mapping: ProcessContextMapping, context: &ProcessContext) -> io::Result<()> { + let payload = context.encode_to_vec(); + let header = mapping.header(); + + // Construct the typed header before using its atomic fields. The unpublished value makes the + // mapping fail closed if validation or payload sizing fails below. + unsafe { + header.write(MappingHeader { + signature: *SIGNATURE, + version: PROCESS_CTX_VERSION, + payload_size: AtomicU32::new(0), + monotonic_published_at_ns: AtomicU64::new(UNPUBLISHED_OR_UPDATING), + payload_ptr: AtomicPtr::new(mapping.payload()), + }); + } + let payload_size = payload_size(mapping, payload.len())?; + + // SAFETY: construction validates size and alignment; the caller guarantees writable lifetime. + unsafe { + copy_payload_to_mapping(mapping, &payload); + (*header) + .payload_size + .store(payload_size, Ordering::Relaxed); + fence(Ordering::SeqCst); + (*header) + .monotonic_published_at_ns + .store(next_published_at(0), Ordering::Relaxed); + } + Ok(()) +} + +pub fn update(mapping: ProcessContextMapping, context: &ProcessContext) -> io::Result<()> { + let payload = context.encode_to_vec(); + let payload_size = match payload_size(mapping, payload.len()) { + Ok(size) => size, + Err(error) => { + invalidate(mapping); + return Err(error); + } + }; + let header = mapping.header(); + + // SAFETY: construction validates size and alignment; the caller guarantees writable lifetime. + let previous = unsafe { + (*header) + .monotonic_published_at_ns + .swap(UNPUBLISHED_OR_UPDATING, Ordering::Relaxed) + }; + if previous == UNPUBLISHED_OR_UPDATING { + return Err(io::Error::new( + io::ErrorKind::WouldBlock, + "process context is already being updated or invalidated", + )); + } + + unsafe { + fence(Ordering::SeqCst); + copy_payload_to_mapping(mapping, &payload); + (*header) + .payload_ptr + .store(mapping.payload(), Ordering::Relaxed); + (*header) + .payload_size + .store(payload_size, Ordering::Relaxed); + fence(Ordering::SeqCst); + (*header) + .monotonic_published_at_ns + .store(next_published_at(previous), Ordering::Relaxed); + } + Ok(()) +} + +pub fn invalidate(mapping: ProcessContextMapping) { + // SAFETY: the caller guarantees writable lifetime. + unsafe { + (*mapping.header()) + .monotonic_published_at_ns + .store(UNPUBLISHED_OR_UPDATING, Ordering::Relaxed); + } + fence(Ordering::SeqCst); +} + +pub fn validate(mapping: ProcessContextMapping) -> io::Result<()> { + let header = mapping.header(); + // SAFETY: the caller guarantees readable lifetime. + let published_at = unsafe { (*header).monotonic_published_at_ns.load(Ordering::Relaxed) }; + if published_at == UNPUBLISHED_OR_UPDATING { + return Err(io::Error::new( + io::ErrorKind::WouldBlock, + "process context is unavailable or being updated", + )); + } + fence(Ordering::SeqCst); + + // SAFETY: the validated region contains a complete header. + let (signature, version, size, payload) = unsafe { + ( + ptr::addr_of!((*header).signature).read(), + ptr::addr_of!((*header).version).read(), + (*header).payload_size.load(Ordering::Relaxed) as usize, + (*header).payload_ptr.load(Ordering::Relaxed), + ) + }; + if signature != *SIGNATURE || version != PROCESS_CTX_VERSION { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "invalid process context header", + )); + } + if payload != mapping.payload() || size > mapping.capacity() { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "process context payload is not inline or exceeds the mapping", + )); + } + Ok(()) +} + +pub fn decode(mapping: ProcessContextMapping) -> io::Result { + let (context, _) = read(mapping)?; + Ok(context) +} + +pub fn read(mapping: ProcessContextMapping) -> io::Result<(ProcessContext, u64)> { + validate(mapping)?; + let header = mapping.header(); + // SAFETY: validate checked that the inline range is in bounds. + let (published_at, size) = unsafe { + ( + (*header).monotonic_published_at_ns.load(Ordering::Relaxed), + (*header).payload_size.load(Ordering::Relaxed) as usize, + ) + }; + let mut payload = vec![0; size]; + copy_payload_from_mapping(mapping, &mut payload); + fence(Ordering::SeqCst); + let published_after = unsafe { (*header).monotonic_published_at_ns.load(Ordering::Relaxed) }; + if published_at == 0 || published_at != published_after { + return Err(io::Error::new( + io::ErrorKind::WouldBlock, + "process context changed while being read", + )); + } + let context = ProcessContext::decode(payload.as_slice()) + .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))?; + Ok((context, published_at)) +} + +pub fn threadlocal_attribute_key_map(context: &ProcessContext) -> Option> { + find_attr(&context.extra_attributes, "threadlocal.attribute_key_map") + .or_else(|| { + context.resource.as_ref().and_then(|resource| { + find_attr(&resource.attributes, "threadlocal.attribute_key_map") + }) + }) + .and_then(string_array) +} + +fn payload_size(mapping: ProcessContextMapping, len: usize) -> io::Result { + if len > mapping.capacity() { + return Err(io::Error::new( + io::ErrorKind::StorageFull, + format!( + "encoded process context needs {len} bytes but mapping capacity is {}", + mapping.capacity() + ), + )); + } + u32::try_from(len).map_err(|_| io::Error::other("process context payload size overflowed")) +} + +fn next_published_at(previous: u64) -> u64 { + static ORIGIN: OnceLock = OnceLock::new(); + let elapsed = ORIGIN.get_or_init(Instant::now).elapsed().as_nanos(); + let elapsed = u64::try_from(elapsed).unwrap_or(u64::MAX); + elapsed.max(previous.saturating_add(1)).max(1) +} + +fn copy_payload_to_mapping(mapping: ProcessContextMapping, payload: &[u8]) { + for (offset, byte) in payload.iter().copied().enumerate() { + // SAFETY: payload_size checked that offset is in the caller-provided mapping. + unsafe { AtomicU8::from_ptr(mapping.payload().add(offset)).store(byte, Ordering::Relaxed) }; + } +} + +fn copy_payload_from_mapping(mapping: ProcessContextMapping, payload: &mut [u8]) { + for (offset, byte) in payload.iter_mut().enumerate() { + // SAFETY: validate checked that offset is in the caller-provided mapping. + *byte = + unsafe { AtomicU8::from_ptr(mapping.payload().add(offset)).load(Ordering::Relaxed) }; + } +} + +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 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() +} + +#[cfg(test)] +mod tests { + use super::*; + use libdd_trace_protobuf::opentelemetry::proto::common::v1::ProcessContext; + + fn storage() -> Box<[u64; 2048]> { + Box::new([0; 2048]) + } + + fn mapping(storage: &mut [u64; 2048]) -> ProcessContextMapping { + unsafe { + ProcessContextMapping::from_raw_parts(storage.as_mut_ptr().cast(), size_of_val(storage)) + .expect("valid storage") + } + } + + #[test] + fn caller_storage_lifecycle() { + let mut storage = storage(); + let mapping = mapping(&mut storage); + let first = ProcessContext::default(); + initialize(mapping, &first).expect("initialize"); + assert_eq!(decode(mapping).expect("decode"), first); + let first_published_at = read(mapping).expect("read").1; + update(mapping, &first).expect("update"); + assert!(read(mapping).expect("read").1 > first_published_at); + invalidate(mapping); + assert_eq!( + validate(mapping).expect_err("invalidated").kind(), + io::ErrorKind::WouldBlock + ); + } + + #[test] + fn oversized_payload_invalidates_mapping() { + let mut words = [0u64; 8]; + let mapping = unsafe { + ProcessContextMapping::from_raw_parts(words.as_mut_ptr().cast(), size_of_val(&words)) + .expect("valid storage") + }; + initialize(mapping, &ProcessContext::default()).expect("initialize"); + let oversized = ProcessContext { + extra_attributes: vec![KeyValue { + key: "large".into(), + value: Some(AnyValue { + value: Some(any_value::Value::StringValue("x".repeat(256))), + }), + key_ref: 0, + }], + ..Default::default() + }; + assert_eq!( + update(mapping, &oversized).expect_err("too large").kind(), + io::ErrorKind::StorageFull + ); + assert_eq!( + validate(mapping).expect_err("invalidated").kind(), + io::ErrorKind::WouldBlock + ); + } + + #[test] + fn update_detects_an_update_in_progress() { + let mut storage = storage(); + let mapping = mapping(&mut storage); + let context = ProcessContext::default(); + initialize(mapping, &context).expect("initialize"); + invalidate(mapping); + + assert_eq!( + update(mapping, &context) + .expect_err("concurrent update must be rejected") + .kind(), + io::ErrorKind::WouldBlock + ); + } +} diff --git a/libdd-library-config/src/tracer_metadata.rs b/libdd-library-config/src/tracer_metadata.rs index aa9c3b9146..4aa524a125 100644 --- a/libdd-library-config/src/tracer_metadata.rs +++ b/libdd-library-config/src/tracer_metadata.rs @@ -133,7 +133,7 @@ impl TracerMetadata { threadlocal_metadata, } = self; - let resource_attrs = vec![ + let mut resource_attrs = vec![ key_value_opt("service.name", service_name), key_value_opt("service.instance.id", runtime_id), key_value_opt("service.version", service_version), @@ -141,9 +141,13 @@ impl TracerMetadata { key_value("telemetry.sdk.language", tracer_language.clone()), key_value("telemetry.sdk.version", tracer_version.clone()), key_value("telemetry.sdk.name", Self::OTEL_SDK_NAME.to_owned()), - key_value("host.name", hostname.clone()), - key_value_opt("container.id", container_id), ]; + if !hostname.is_empty() { + resource_attrs.push(key_value("host.name", hostname.clone())); + } + if let Some(container_id) = container_id.as_ref().filter(|id| !id.is_empty()) { + resource_attrs.push(key_value("container.id", container_id.clone())); + } // `mut` is only needed when `otel-thread-ctx` is enabled. #[allow(unused_mut)] @@ -270,7 +274,6 @@ pub use other::*; #[cfg(test)] mod tests { use super::*; - #[cfg(feature = "otel-thread-ctx")] use libdd_trace_protobuf::opentelemetry::proto::common::v1::any_value; use libdd_trace_protobuf::opentelemetry::proto::common::v1::{AnyValue, ProcessContext}; @@ -282,6 +285,51 @@ mod tests { .as_ref() } + fn find_resource_attr<'a>(ctx: &'a ProcessContext, key: &str) -> Option<&'a AnyValue> { + ctx.resource + .as_ref()? + .attributes + .iter() + .find(|kv| kv.key == key)? + .value + .as_ref() + } + + #[test] + fn process_context_sdk_and_optional_host_attributes() { + let absent = TracerMetadata { + tracer_language: "php".to_owned(), + tracer_version: "1.2.3".to_owned(), + ..Default::default() + } + .to_otel_process_ctx(); + assert!(find_resource_attr(&absent, "host.name").is_none()); + assert!(find_resource_attr(&absent, "container.id").is_none()); + assert!(matches!( + find_resource_attr(&absent, "telemetry.sdk.language").and_then(|v| v.value.as_ref()), + Some(any_value::Value::StringValue(value)) if value == "php" + )); + assert!(matches!( + find_resource_attr(&absent, "telemetry.sdk.version").and_then(|v| v.value.as_ref()), + Some(any_value::Value::StringValue(value)) if value == "1.2.3" + )); + + let present = TracerMetadata { + hostname: "host-a".to_owned(), + container_id: Some("container-a".to_owned()), + ..Default::default() + } + .to_otel_process_ctx(); + assert!(matches!( + find_resource_attr(&present, "host.name").and_then(|v| v.value.as_ref()), + Some(any_value::Value::StringValue(value)) if value == "host-a" + )); + assert!(matches!( + find_resource_attr(&present, "container.id").and_then(|v| v.value.as_ref()), + Some(any_value::Value::StringValue(value)) if value == "container-a" + )); + } + #[test] fn tracer_metadata_equality() { let a = TracerMetadata { diff --git a/libdd-otel-thread-ctx-ffi/Cargo.toml b/libdd-otel-thread-ctx-ffi/Cargo.toml index 4a2ef4f84e..9ddeda4d26 100644 --- a/libdd-otel-thread-ctx-ffi/Cargo.toml +++ b/libdd-otel-thread-ctx-ffi/Cargo.toml @@ -15,13 +15,14 @@ crate-type = ["staticlib", "cdylib", "lib"] bench = false [dependencies] -libdd-common-ffi = { path = "../libdd-common-ffi", default-features = false, optional = true } -libdd-otel-thread-ctx = { path = "../libdd-otel-thread-ctx" } +libdd-common-ffi = { path = "../libdd-common-ffi", default-features = false } +libdd-otel-thread-ctx = { path = "../libdd-otel-thread-ctx", default-features = false } [features] -default = ["cbindgen"] +default = ["cbindgen", "tls-storage"] cbindgen = ["build_common/cbindgen", "libdd-common-ffi/cbindgen"] -sanity-check = ["dep:libdd-common-ffi", "libdd-otel-thread-ctx/sanity-check"] +tls-storage = ["libdd-otel-thread-ctx/tls-storage"] +sanity-check = ["libdd-otel-thread-ctx/sanity-check"] [dev-dependencies] libdd-otel-thread-ctx = { path = "../libdd-otel-thread-ctx", features = [ diff --git a/libdd-otel-thread-ctx-ffi/build.rs b/libdd-otel-thread-ctx-ffi/build.rs index 930272f09a..13bc51daa7 100644 --- a/libdd-otel-thread-ctx-ffi/build.rs +++ b/libdd-otel-thread-ctx-ffi/build.rs @@ -12,7 +12,7 @@ fn main() { println!("cargo:rustc-env=LIBDD_OTEL_THREAD_CTX_FFI_CROSS_COMPILING={cross_compiling}"); let target_os = env::var("CARGO_CFG_TARGET_OS").unwrap(); - if target_os != "linux" { + if target_os != "linux" || env::var_os("CARGO_FEATURE_TLS_STORAGE").is_none() { return; } diff --git a/libdd-otel-thread-ctx-ffi/cbindgen.toml b/libdd-otel-thread-ctx-ffi/cbindgen.toml index b5cb437108..d5bb97fc9f 100644 --- a/libdd-otel-thread-ctx-ffi/cbindgen.toml +++ b/libdd-otel-thread-ctx-ffi/cbindgen.toml @@ -11,6 +11,7 @@ include_guard = "DDOG_OTEL_THREAD_CTX_H" style = "both" pragma_once = true no_includes = true +includes = ["common.h"] sys_includes = ["stdbool.h", "stddef.h", "stdint.h"] [parse] @@ -27,6 +28,7 @@ renaming_overrides_prefixing = true [export.rename] "VoidResult" = "ddog_VoidResult" "Error" = "ddog_Error" +"AtomicU8" = "uint8_t" [export.mangle] rename_types = "PascalCase" diff --git a/libdd-otel-thread-ctx-ffi/src/lib.rs b/libdd-otel-thread-ctx-ffi/src/lib.rs index ec77d043d7..bab2884b68 100644 --- a/libdd-otel-thread-ctx-ffi/src/lib.rs +++ b/libdd-otel-thread-ctx-ffi/src/lib.rs @@ -3,9 +3,144 @@ //! FFI bindings for the OTel thread-level context publisher. //! -//! All symbols are only available on Linux, since spec is currently Linux-specific. +//! Record layout and explicit record operations are platform-neutral. Linux additionally exposes +//! the libdatadog-owned TLS convenience API. -#[cfg(target_os = "linux")] +use libdd_otel_thread_ctx::ThreadContextRecord; + +#[repr(C)] +pub struct OtelThreadContextAttribute<'a> { + pub key_index: u8, + pub value: libdd_common_ffi::slice::CharSlice<'a>, +} + +unsafe fn attrs_from_raw<'a>( + attrs: *const OtelThreadContextAttribute<'a>, + attrs_len: usize, +) -> &'a [OtelThreadContextAttribute<'a>] { + if attrs.is_null() || attrs_len == 0 { + &[] + } else { + unsafe { core::slice::from_raw_parts(attrs, attrs_len) } + } +} + +fn attrs_iter<'a>( + attrs: &'a [OtelThreadContextAttribute<'a>], +) -> impl Iterator { + use libdd_common_ffi::slice::AsBytes; + attrs.iter().filter_map(|attr| { + attr.value + .try_to_utf8() + .ok() + .map(|value| (attr.key_index, value)) + }) +} + +/// Initialize a caller-owned thread-context record. +/// +/// # Safety +/// +/// `ctx` must point to writable storage for a `ThreadContextRecord`. The identifier pointers and +/// `attrs..attrs + attrs_len` must be valid for the duration of the call. +#[no_mangle] +pub unsafe extern "C" fn ddog_otel_thread_ctx_record_init( + ctx: *mut ThreadContextRecord, + trace_id: &[u8; 16], + span_id: &[u8; 8], + local_root_span_id: &[u8; 8], + attrs: *const OtelThreadContextAttribute<'_>, + attrs_len: usize, +) -> bool { + if ctx.is_null() { + return false; + } + let attrs = unsafe { attrs_from_raw(attrs, attrs_len) }; + unsafe { ctx.write(ThreadContextRecord::default()) }; + unsafe { &mut *ctx }.initialize(*trace_id, *span_id, *local_root_span_id, attrs_iter(attrs)) +} + +/// Replace all fields in a caller-owned thread-context record. +/// +/// # Safety +/// +/// `ctx` must point to a live, writable `ThreadContextRecord`. The identifier pointers and +/// `attrs..attrs + attrs_len` must be valid for the duration of the call. +#[no_mangle] +pub unsafe extern "C" fn ddog_otel_thread_ctx_record_update( + ctx: *mut ThreadContextRecord, + trace_id: &[u8; 16], + span_id: &[u8; 8], + local_root_span_id: &[u8; 8], + attrs: *const OtelThreadContextAttribute<'_>, + attrs_len: usize, +) -> bool { + let Some(ctx) = (unsafe { ctx.as_mut() }) else { + return false; + }; + let attrs = unsafe { attrs_from_raw(attrs, attrs_len) }; + ctx.update(*trace_id, *span_id, *local_root_span_id, attrs_iter(attrs)) +} + +/// Replace the span identifier in a caller-owned thread-context record. +/// +/// # Safety +/// +/// `ctx` must point to a live, writable `ThreadContextRecord`, and `span_id` must be valid for the +/// duration of the call. +#[no_mangle] +pub unsafe extern "C" fn ddog_otel_thread_ctx_record_update_span_id( + ctx: *mut ThreadContextRecord, + span_id: &[u8; 8], +) -> bool { + let Some(ctx) = (unsafe { ctx.as_mut() }) else { + return false; + }; + ctx.update_span_id(*span_id); + true +} + +#[cfg(test)] +mod record_tests { + use super::*; + use core::mem::{size_of, MaybeUninit}; + + #[test] + fn ffi_initializes_caller_owned_storage() { + let mut record = MaybeUninit::::uninit(); + let trace_id = [1; 16]; + let span_id = [2; 8]; + let local_root_span_id = [3; 8]; + let attrs = [OtelThreadContextAttribute { + key_index: 1, + value: "service".into(), + }]; + + assert!(unsafe { + ddog_otel_thread_ctx_record_init( + record.as_mut_ptr(), + &trace_id, + &span_id, + &local_root_span_id, + attrs.as_ptr(), + attrs.len(), + ) + }); + let record = unsafe { record.assume_init() }; + let bytes = unsafe { + core::slice::from_raw_parts( + (&raw const record).cast::(), + size_of::(), + ) + }; + assert_eq!(&bytes[..16], &trace_id); + assert_eq!(&bytes[16..24], &span_id); + assert_eq!(bytes[24], 1); + assert_eq!(&bytes[48..55], b"service"); + } +} + +#[cfg(all(target_os = "linux", feature = "tls-storage"))] pub use linux::*; /// Verify that this binary was linked with the correct options such that the thread contexts are @@ -22,7 +157,7 @@ pub extern "C" fn ddog_otel_thread_ctx_sanity_check() -> libdd_common_ffi::VoidR } } -#[cfg(target_os = "linux")] +#[cfg(all(target_os = "linux", feature = "tls-storage"))] mod linux { use libdd_otel_thread_ctx::linux::{ThreadContext, ThreadContextHandle}; use std::ptr::NonNull; diff --git a/libdd-otel-thread-ctx/Cargo.toml b/libdd-otel-thread-ctx/Cargo.toml index 98af49c4d4..be5ff00484 100644 --- a/libdd-otel-thread-ctx/Cargo.toml +++ b/libdd-otel-thread-ctx/Cargo.toml @@ -22,7 +22,11 @@ elf = { version = "0.7", optional = true } object = { version = "0.36", default-features = false, features = ["archive", "read_core"], optional = true } [features] -sanity-check = ["dep:elf", "dep:anyhow"] +default = ["tls-storage"] +# The standalone publisher owns the discoverable Linux TLS slot. Embedders that provide the +# slot themselves can disable default features and use only the common record operations. +tls-storage = [] +sanity-check = ["tls-storage", "dep:elf", "dep:anyhow"] # Test helpers used by this crate's own tests and `libdd-otel-thread-ctx-ffi`'s # integration tests. test-utils = ["dep:elf", "dep:object"] diff --git a/libdd-otel-thread-ctx/src/lib.rs b/libdd-otel-thread-ctx/src/lib.rs index cb1c1c5bab..0cc83f87ac 100644 --- a/libdd-otel-thread-ctx/src/lib.rs +++ b/libdd-otel-thread-ctx/src/lib.rs @@ -76,10 +76,15 @@ //! `atomic_signal_fence`) to keep field writes boxed between the `valid = 0` and `valid = 1` //! stores during in-place updates. +mod record; + +pub use record::{ThreadContextRecord, MAX_ATTRS_DATA_SIZE}; + // The `linux` module below resolves the TLS slot with TLSDESC inline assembly that is only written // for x86_64 and aarch64. Reject any other architecture on Linux at compile time. On non-Linux // targets the `linux` module is not compiled, so there's no such constraint. #[cfg(all( + feature = "tls-storage", target_os = "linux", not(any(target_arch = "x86_64", target_arch = "aarch64")) ))] @@ -88,21 +93,24 @@ compile_error!( supported." ); -#[cfg(all(target_os = "linux", feature = "sanity-check"))] +#[cfg(all(feature = "tls-storage", target_os = "linux", feature = "sanity-check"))] pub mod sanity_check; #[cfg(feature = "test-utils")] pub mod test_utils; #[cfg(all( + feature = "tls-storage", target_os = "linux", any(target_arch = "x86_64", target_arch = "aarch64") ))] pub mod linux { + use super::ThreadContextRecord; + pub use super::MAX_ATTRS_DATA_SIZE; use std::{ mem, ptr::{self, NonNull}, - sync::atomic::{compiler_fence, AtomicPtr, AtomicU8, Ordering}, + sync::atomic::{compiler_fence, AtomicPtr, Ordering}, }; // Define the thread-local pointer that external readers (e.g. the eBPF profiler) discover via @@ -206,178 +214,6 @@ pub mod linux { f(slot) } - // We maintain the convention in libdatadog that the `local_root_span_id` attribute key is - // always the very first in the string table, so its key index is guaranteed to be zero. - const ROOT_SPAN_KEY_INDEX: u8 = 0; - - /// Maximum size in bytes of the `attrs_data` field. - /// - /// Chosen so that the total record size (`28 + MAX_ATTRS_DATA_SIZE`) stays within the 640-byte - /// limit recommended by the spec (the eBPF profiler read limit). - pub const MAX_ATTRS_DATA_SIZE: usize = 612; - - /// In-memory layout of a thread-level context. - /// - /// **CAUTION**: The structure MUST match exactly the OTel thread-level context specification. - /// It is read by external, out-of-process code. Do not re-order fields or modify in any way, - /// unless you know exactly what you're doing. - /// - /// # Synchronization - /// - /// Readers are async-signal handlers. The writer is always stopped while a reader runs. - /// Sharing memory with a signal handler still requires some form of synchronization, which is - /// achieved through atomics and compiler fence, using `valid` and/or the TLS slot as - /// synchronization points. - /// - /// - The writer stores `valid = 0` *before* modifying fields in-place, guarded by a fence. - /// - The writer stores `valid = 1` *after* all fields are populated, guarded by a fence. - /// - `valid` starts at `1` on construction and is never set to `0` except during an in-place - /// update. - // Note: we don't need to make this struct packed, because it's already designed to avoid - // padding. Moreover, doing so would make it 1-aligned, potentially making access to - // `attrs_data_size` unaligned and thus slower, and prevent us from using `AtomicU8` for - // `valid`. We use const assertions below to verify size and offsets at compile time. - #[repr(C)] - struct ThreadContextRecord { - /// Trace identifier; all-zeroes means "no trace". - trace_id: [u8; 16], - /// Span identifier. - span_id: [u8; 8], - /// Whether the record is ready/consistent. Always set to `1` except during in-place update - /// of the current record. - valid: AtomicU8, - _reserved: u8, - /// Number of populated bytes in `attrs_data`. - attrs_data_size: u16, - /// Packed variable-length key-value records. - /// - /// It's a contiguous list of blocks with layout: - /// - /// 1. 1-byte `key_index` - /// 2. 1-byte `val_len` - /// 3. `val_len` bytes of a string value. - /// - /// # Size - /// - /// Currently, we always allocate the max recommended size. This potentially wastes a few - /// hundred bytes per thread, but it guarantees that we can modify the context in-place - /// without (re)allocation in the hot path. Having a hybrid scheme (starting smaller and - /// resizing up a few times) is not out of the question. - attrs_data: [u8; MAX_ATTRS_DATA_SIZE], - } - - const _: () = { - assert!(size_of::() == 640); - assert!(mem::offset_of!(ThreadContextRecord, trace_id) == 0); - assert!(mem::offset_of!(ThreadContextRecord, span_id) == 16); - assert!(mem::offset_of!(ThreadContextRecord, valid) == 24); - assert!(mem::offset_of!(ThreadContextRecord, _reserved) == 25); - assert!(mem::offset_of!(ThreadContextRecord, attrs_data_size) == 26); - assert!(mem::offset_of!(ThreadContextRecord, attrs_data) == 28); - }; - - impl ThreadContextRecord { - /// Build a record with the given trace id, span id and attributes. The - /// `local_root_span_id` is a distinguished attribute with special handling for - /// convenience, but it ends up as other attributes in `attrs_data`. - fn new( - trace_id: [u8; 16], - span_id: [u8; 8], - local_root_span_id: [u8; 8], - attrs: &[(u8, &str)], - ) -> Self { - let mut record = Self { - trace_id, - span_id, - ..Default::default() - }; - record.set_attrs(local_root_span_id, attrs); - record - } - - /// Encode `attributes` into `self.attrs_data` as packed key-value records. Existing data - /// are overridden (and if there were more entries than `attributes.len()`, they aren't - /// zeroed, but they will be ignored by readers). - /// - /// # Return - /// - /// Returns `true` if all attributes were properly encoded, or `false` if some of the data - /// needed to be dropped. See Size limits below. - /// - /// # Arguments - /// - /// Each input entry is a pair of a 1-byte `key_index` and a string value. - /// - /// # Size limits - /// - /// Any value over 255 bytes will be capped at this size. If the total size of the encoded - /// attributes is over [MAX_ATTRS_DATA_SIZE], extra attributes are ignored. We do this - /// instead of raising an error because we encode the attributes on-the-fly. Proper error - /// recovery would require us to be able to rollback to the previous attributes which would - /// hurt the happy path, or leave the record in an inconsistent state. Another possibility - /// would be to error out and reset the record in that situation. - fn set_attrs(&mut self, local_root_span_id: [u8; 8], attributes: &[(u8, &str)]) -> bool { - let mut fully_encoded = true; - - const { assert!(MAX_ATTRS_DATA_SIZE >= 18) } - // The local root span id is provided as raw bytes (can be seen as a big-endian u64), - // but readers will expect a string hex representation. We convert it to a fixed - // 16-characters string in the usual lowercase hex format. - // - // There's currently no easy way to use Rust format capabilities to write directly in a - // fixed-size array. Since the conversion is simple, we do it manually. - const HEX_DIGITS: &[u8; 16] = b"0123456789abcdef"; - self.attrs_data[0] = ROOT_SPAN_KEY_INDEX; - self.attrs_data[1] = 16; - for (i, &byte) in local_root_span_id.iter().enumerate() { - self.attrs_data[2 + i * 2] = HEX_DIGITS[(byte >> 4) as usize]; - self.attrs_data[2 + i * 2 + 1] = HEX_DIGITS[(byte & 0xF) as usize]; - } - - let mut offset = 18; - - for &(key_index, val) in attributes { - let val_bytes = val.as_bytes(); - let val_len = u8::try_from(val_bytes.len()).unwrap_or_else(|_| { - fully_encoded = false; - u8::MAX - }); - let entry_size = 2 + val_len as usize; - - if offset + entry_size > MAX_ATTRS_DATA_SIZE { - fully_encoded = false; - break; - } - - self.attrs_data[offset] = key_index; - self.attrs_data[offset + 1] = val_len; - self.attrs_data[offset + 2..offset + 2 + val_len as usize] - .copy_from_slice(&val_bytes[..val_len as usize]); - offset += entry_size; - } - - // `offset < MAX_ATTRS_DATA_SIZE`, which guarantees it fits in a `u16`. This also - // effectively hides the remaining of the previous `attrs` bytes, so we don't have to - // zero them. - self.attrs_data_size = offset as u16; - fully_encoded - } - } - - impl Default for ThreadContextRecord { - fn default() -> Self { - Self { - trace_id: [0u8; 16], - span_id: [0u8; 8], - // We only ever set `valid` to `0` during in-place update of an attached context. - valid: AtomicU8::new(1), - _reserved: 0, - attrs_data_size: 0, - attrs_data: [0u8; MAX_ATTRS_DATA_SIZE], - } - } - } - /// An owned (and non-moving) thread context record allocation. /// /// We don't use `Box` under the hood because it precludes aliasing, while we share the context @@ -403,12 +239,9 @@ pub mod linux { local_root_span_id: [u8; 8], attrs: &[(u8, &str)], ) -> Self { - Self::from(ThreadContextRecord::new( - trace_id, - span_id, - local_root_span_id, - attrs, - )) + let mut record = ThreadContextRecord::default(); + record.initialize(trace_id, span_id, local_root_span_id, attrs.iter().copied()); + Self::from(record) } /// Turn this thread context into a pointer to the underlying [ThreadContextRecord]. @@ -518,15 +351,7 @@ pub mod linux { // and only this thread ever writes to the slot, so the pointer is valid and not // accessed for the duration of this closure. if let Some(current) = unsafe { slot.load(Ordering::Relaxed).as_mut() } { - current.valid.store(0, Ordering::Relaxed); - compiler_fence(Ordering::SeqCst); - - current.trace_id = trace_id; - current.span_id = span_id; - current.set_attrs(local_root_span_id, attrs); - - compiler_fence(Ordering::SeqCst); - current.valid.store(1, Ordering::Relaxed); + current.update(trace_id, span_id, local_root_span_id, attrs.iter().copied()); } else { let ctxt = ThreadContext::new(trace_id, span_id, local_root_span_id, attrs) .into_ptr() diff --git a/libdd-otel-thread-ctx/src/record.rs b/libdd-otel-thread-ctx/src/record.rs new file mode 100644 index 0000000000..e2c673eba8 --- /dev/null +++ b/libdd-otel-thread-ctx/src/record.rs @@ -0,0 +1,144 @@ +// Copyright 2026-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 + +use core::{ + mem, + sync::atomic::{compiler_fence, AtomicU8, Ordering}, +}; + +pub const MAX_ATTRS_DATA_SIZE: usize = 612; +const ROOT_SPAN_KEY_INDEX: u8 = 0; + +/// Platform-neutral in-memory layout of an OTel thread context record. +#[repr(C)] +pub struct ThreadContextRecord { + pub(crate) trace_id: [u8; 16], + pub(crate) span_id: [u8; 8], + pub(crate) valid: AtomicU8, + pub(crate) reserved: u8, + pub(crate) attrs_data_size: u16, + pub(crate) attrs_data: [u8; MAX_ATTRS_DATA_SIZE], +} + +const _: () = { + assert!(mem::size_of::() == 640); + assert!(mem::align_of::() == 2); + assert!(mem::offset_of!(ThreadContextRecord, trace_id) == 0); + assert!(mem::offset_of!(ThreadContextRecord, span_id) == 16); + assert!(mem::offset_of!(ThreadContextRecord, valid) == 24); + assert!(mem::offset_of!(ThreadContextRecord, reserved) == 25); + assert!(mem::offset_of!(ThreadContextRecord, attrs_data_size) == 26); + assert!(mem::offset_of!(ThreadContextRecord, attrs_data) == 28); +}; + +impl ThreadContextRecord { + pub fn initialize( + &mut self, + trace_id: [u8; 16], + span_id: [u8; 8], + local_root_span_id: [u8; 8], + attributes: I, + ) -> bool + where + I: IntoIterator, + S: AsRef, + { + self.update(trace_id, span_id, local_root_span_id, attributes) + } + + pub fn update( + &mut self, + trace_id: [u8; 16], + span_id: [u8; 8], + local_root_span_id: [u8; 8], + attributes: I, + ) -> bool + where + I: IntoIterator, + S: AsRef, + { + self.valid.store(0, Ordering::Relaxed); + compiler_fence(Ordering::SeqCst); + self.trace_id = trace_id; + self.span_id = span_id; + let complete = self.set_attrs(local_root_span_id, attributes); + compiler_fence(Ordering::SeqCst); + self.valid.store(1, Ordering::Relaxed); + complete + } + + pub fn update_span_id(&mut self, span_id: [u8; 8]) { + self.valid.store(0, Ordering::Relaxed); + compiler_fence(Ordering::SeqCst); + self.span_id = span_id; + compiler_fence(Ordering::SeqCst); + self.valid.store(1, Ordering::Relaxed); + } + + fn set_attrs(&mut self, local_root_span_id: [u8; 8], attributes: I) -> bool + where + I: IntoIterator, + S: AsRef, + { + const HEX: &[u8; 16] = b"0123456789abcdef"; + self.attrs_data[0] = ROOT_SPAN_KEY_INDEX; + self.attrs_data[1] = 16; + for (index, byte) in local_root_span_id.iter().copied().enumerate() { + self.attrs_data[2 + index * 2] = HEX[(byte >> 4) as usize]; + self.attrs_data[3 + index * 2] = HEX[(byte & 0xf) as usize]; + } + + let mut offset = 18; + let mut complete = true; + for (key_index, value) in attributes { + let bytes = value.as_ref().as_bytes(); + let len = match u8::try_from(bytes.len()) { + Ok(len) => len, + Err(_) => { + complete = false; + u8::MAX + } + }; + let end = offset + 2 + len as usize; + if end > MAX_ATTRS_DATA_SIZE { + complete = false; + break; + } + self.attrs_data[offset] = key_index; + self.attrs_data[offset + 1] = len; + self.attrs_data[offset + 2..end].copy_from_slice(&bytes[..len as usize]); + offset = end; + } + self.attrs_data_size = offset as u16; + complete + } +} + +impl Default for ThreadContextRecord { + fn default() -> Self { + Self { + trace_id: [0; 16], + span_id: [0; 8], + valid: AtomicU8::new(1), + reserved: 0, + attrs_data_size: 0, + attrs_data: [0; MAX_ATTRS_DATA_SIZE], + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn explicit_record_lifecycle() { + let mut record = ThreadContextRecord::default(); + assert!(record.initialize([1; 16], [2; 8], [3; 8], [(1, "service")])); + assert_eq!(record.valid.load(Ordering::Relaxed), 1); + assert_eq!(record.span_id, [2; 8]); + assert_eq!(&record.attrs_data[20..27], b"service"); + record.update_span_id([4; 8]); + assert_eq!(record.span_id, [4; 8]); + } +} From 6234054034902ec0d3377f6f81823d086febfcc2 Mon Sep 17 00:00:00 2001 From: Levi Morrison Date: Thu, 23 Jul 2026 19:51:17 -0600 Subject: [PATCH 2/2] feat(otel-thread-context): publish W3C trace flags --- libdd-otel-thread-ctx-ffi/src/lib.rs | 27 +++++++-- libdd-otel-thread-ctx/src/lib.rs | 90 +++++++++++++++++++--------- libdd-otel-thread-ctx/src/record.rs | 20 +++++-- 3 files changed, 99 insertions(+), 38 deletions(-) diff --git a/libdd-otel-thread-ctx-ffi/src/lib.rs b/libdd-otel-thread-ctx-ffi/src/lib.rs index bab2884b68..7e8bcd1991 100644 --- a/libdd-otel-thread-ctx-ffi/src/lib.rs +++ b/libdd-otel-thread-ctx-ffi/src/lib.rs @@ -48,6 +48,7 @@ pub unsafe extern "C" fn ddog_otel_thread_ctx_record_init( ctx: *mut ThreadContextRecord, trace_id: &[u8; 16], span_id: &[u8; 8], + trace_flags: u8, local_root_span_id: &[u8; 8], attrs: *const OtelThreadContextAttribute<'_>, attrs_len: usize, @@ -57,7 +58,13 @@ pub unsafe extern "C" fn ddog_otel_thread_ctx_record_init( } let attrs = unsafe { attrs_from_raw(attrs, attrs_len) }; unsafe { ctx.write(ThreadContextRecord::default()) }; - unsafe { &mut *ctx }.initialize(*trace_id, *span_id, *local_root_span_id, attrs_iter(attrs)) + unsafe { &mut *ctx }.initialize( + *trace_id, + *span_id, + trace_flags, + *local_root_span_id, + attrs_iter(attrs), + ) } /// Replace all fields in a caller-owned thread-context record. @@ -71,6 +78,7 @@ pub unsafe extern "C" fn ddog_otel_thread_ctx_record_update( ctx: *mut ThreadContextRecord, trace_id: &[u8; 16], span_id: &[u8; 8], + trace_flags: u8, local_root_span_id: &[u8; 8], attrs: *const OtelThreadContextAttribute<'_>, attrs_len: usize, @@ -79,7 +87,13 @@ pub unsafe extern "C" fn ddog_otel_thread_ctx_record_update( return false; }; let attrs = unsafe { attrs_from_raw(attrs, attrs_len) }; - ctx.update(*trace_id, *span_id, *local_root_span_id, attrs_iter(attrs)) + ctx.update( + *trace_id, + *span_id, + trace_flags, + *local_root_span_id, + attrs_iter(attrs), + ) } /// Replace the span identifier in a caller-owned thread-context record. @@ -121,6 +135,7 @@ mod record_tests { record.as_mut_ptr(), &trace_id, &span_id, + 0x03, &local_root_span_id, attrs.as_ptr(), attrs.len(), @@ -136,6 +151,7 @@ mod record_tests { assert_eq!(&bytes[..16], &trace_id); assert_eq!(&bytes[16..24], &span_id); assert_eq!(bytes[24], 1); + assert_eq!(bytes[25], 0x03); assert_eq!(&bytes[48..55], b"service"); } } @@ -183,9 +199,11 @@ mod linux { pub extern "C" fn ddog_otel_thread_ctx_new( trace_id: &[u8; 16], span_id: &[u8; 8], + trace_flags: u8, local_root_span_id: &[u8; 8], ) -> NonNull { - ThreadContext::new(*trace_id, *span_id, *local_root_span_id, &[]).into_opaque_ptr() + ThreadContext::new(*trace_id, *span_id, trace_flags, *local_root_span_id, &[]) + .into_opaque_ptr() } /// Free an owned thread context. @@ -236,8 +254,9 @@ mod linux { pub extern "C" fn ddog_otel_thread_ctx_update( trace_id: &[u8; 16], span_id: &[u8; 8], + trace_flags: u8, local_root_span_id: &[u8; 8], ) { - ThreadContext::update(*trace_id, *span_id, *local_root_span_id, &[]); + ThreadContext::update(*trace_id, *span_id, trace_flags, *local_root_span_id, &[]); } } diff --git a/libdd-otel-thread-ctx/src/lib.rs b/libdd-otel-thread-ctx/src/lib.rs index 0cc83f87ac..63632971fc 100644 --- a/libdd-otel-thread-ctx/src/lib.rs +++ b/libdd-otel-thread-ctx/src/lib.rs @@ -27,8 +27,8 @@ //! let local_root_span_id = [2u8; 8]; //! //! // First call allocates a record and attaches it. -//! ThreadContext::new(trace_id, span_id, local_root_span_id, &[(0, "first")]).attach(); -//! ThreadContext::update(trace_id, span_id, local_root_span_id, &[(0, "second")]); +//! ThreadContext::new(trace_id, span_id, 0, local_root_span_id, &[(0, "first")]).attach(); +//! ThreadContext::update(trace_id, span_id, 0, local_root_span_id, &[(0, "second")]); //! ThreadContext::detach(); //! # } //! # #[cfg(not(all(target_os = "linux", any(target_arch = "x86_64", target_arch = "aarch64"))))] @@ -52,7 +52,7 @@ //! let attrs: &[(u8, &str)] = &[(0, "GET"), (1, "/api/v1")]; //! //! // Publish a new context and save the previously attached one (if any). -//! let ctx = ThreadContext::new(trace_id, span_id, local_root_span_id, attrs); +//! let ctx = ThreadContext::new(trace_id, span_id, 0, local_root_span_id, attrs); //! let previous = ctx.attach(); //! //! // ... do work inside the span ... @@ -236,11 +236,18 @@ pub mod linux { pub fn new( trace_id: [u8; 16], span_id: [u8; 8], + trace_flags: u8, local_root_span_id: [u8; 8], attrs: &[(u8, &str)], ) -> Self { let mut record = ThreadContextRecord::default(); - record.initialize(trace_id, span_id, local_root_span_id, attrs.iter().copied()); + record.initialize( + trace_id, + span_id, + trace_flags, + local_root_span_id, + attrs.iter().copied(), + ); Self::from(record) } @@ -343,6 +350,7 @@ pub mod linux { pub fn update( trace_id: [u8; 16], span_id: [u8; 8], + trace_flags: u8, local_root_span_id: [u8; 8], attrs: &[(u8, &str)], ) { @@ -351,11 +359,23 @@ pub mod linux { // and only this thread ever writes to the slot, so the pointer is valid and not // accessed for the duration of this closure. if let Some(current) = unsafe { slot.load(Ordering::Relaxed).as_mut() } { - current.update(trace_id, span_id, local_root_span_id, attrs.iter().copied()); + current.update( + trace_id, + span_id, + trace_flags, + local_root_span_id, + attrs.iter().copied(), + ); } else { - let ctxt = ThreadContext::new(trace_id, span_id, local_root_span_id, attrs) - .into_ptr() - .as_ptr(); + let ctxt = ThreadContext::new( + trace_id, + span_id, + trace_flags, + local_root_span_id, + attrs, + ) + .into_ptr() + .as_ptr(); // No need for `AcqRel`, see [^tls-slot-ordering]. compiler_fence(Ordering::Release); // `ThreadContext::new` already initialises `valid = 1`. @@ -405,7 +425,7 @@ pub mod linux { read_tls_context_ptr().is_null(), "TLS must be null initially" ); - ThreadContext::new(trace_id, span_id, root_span_id, &[]).attach(); + ThreadContext::new(trace_id, span_id, 0, root_span_id, &[]).attach(); assert!( !read_tls_context_ptr().is_null(), "TLS must not be null after attach" @@ -437,27 +457,30 @@ pub mod linux { let span_id = [2u8; 8]; let root_span_id = [3u8; 8]; - ThreadContext::new(trace_id, span_id, root_span_id, &[]).attach(); + for trace_flags in [0x00, 0x01, 0x02, 0x03] { + ThreadContext::new(trace_id, span_id, trace_flags, root_span_id, &[]).attach(); - let ptr = read_tls_context_ptr(); - assert!(!ptr.is_null(), "TLS must be non-null after attach"); + let ptr = read_tls_context_ptr(); + assert!(!ptr.is_null(), "TLS must be non-null after attach"); - // Safety: context is still live. - let record = unsafe { &*ptr }; - assert_eq!(record.trace_id, trace_id); - assert_eq!(record.span_id, span_id); - assert_eq!(record.valid.load(Ordering::Relaxed), 1); - // 1 (key) + 1 (len) + 16 (root_span_id hex chars) = 18 - assert_eq!(record.attrs_data_size, 18); + // Safety: context is still live. + let record = unsafe { &*ptr }; + assert_eq!(record.trace_id, trace_id); + assert_eq!(record.span_id, span_id); + assert_eq!(record.valid.load(Ordering::Relaxed), 1); + assert_eq!(record.trace_flags, trace_flags); + // 1 (key) + 1 (len) + 16 (root_span_id hex chars) = 18 + assert_eq!(record.attrs_data_size, 18); - let _ = ThreadContext::detach(); + let _ = ThreadContext::detach(); + } } #[test] #[cfg_attr(miri, ignore)] fn attribute_encoding_basic() { let attrs: &[(u8, &str)] = &[(1, "GET"), (2, "/api/v1")]; - ThreadContext::new([0u8; 16], [0u8; 8], [0u8; 8], attrs).attach(); + ThreadContext::new([0u8; 16], [0u8; 8], 0, [0u8; 8], attrs).attach(); let ptr = read_tls_context_ptr(); assert!(!ptr.is_null()); @@ -497,7 +520,7 @@ pub mod linux { (3, val_c.as_str()), ]; - ThreadContext::new([0u8; 16], [0u8; 8], [0u8; 8], attrs).attach(); + ThreadContext::new([0u8; 16], [0u8; 8], 0, [0u8; 8], attrs).attach(); let ptr = read_tls_context_ptr(); assert!(!ptr.is_null()); @@ -523,7 +546,7 @@ pub mod linux { let root_span_id2 = [0x79, 0x7A, 0x7B, 0x7C, 0x7D, 0x7E, 0x7F, 0x80]; // Updating before any context is attached should be equivalent to `attach()` - ThreadContext::update(trace_id1, span_id1, root_span_id1, &[(0, "v1")]); + ThreadContext::update(trace_id1, span_id1, 0x01, root_span_id1, &[(0, "v1")]); let ptr_before = read_tls_context_ptr(); assert!(!ptr_before.is_null()); @@ -531,6 +554,7 @@ pub mod linux { assert_eq!(record.trace_id, trace_id1); assert_eq!(record.span_id, span_id1); assert_eq!(record.valid.load(Ordering::Relaxed), 1); + assert_eq!(record.trace_flags, 0x01); assert_eq!(record.attrs_data[0], 0); assert_eq!(record.attrs_data[1], 16); assert_eq!(&record.attrs_data[2..18], b"78797a7b7c7d7e7f"); @@ -538,7 +562,7 @@ pub mod linux { assert_eq!(record.attrs_data[19], 2); assert_eq!(&record.attrs_data[20..22], b"v1"); - ThreadContext::update(trace_id2, span_id2, root_span_id2, &[(0, "v2")]); + ThreadContext::update(trace_id2, span_id2, 0x03, root_span_id2, &[(0, "v2")]); let ptr_after = read_tls_context_ptr(); assert_eq!( @@ -550,6 +574,7 @@ pub mod linux { assert_eq!(record.trace_id, trace_id2); assert_eq!(record.span_id, span_id2); assert_eq!(record.valid.load(Ordering::Relaxed), 1); + assert_eq!(record.trace_flags, 0x03); assert_eq!(record.attrs_data[0], 0); assert_eq!(record.attrs_data[1], 16); assert_eq!(&record.attrs_data[2..18], b"797a7b7c7d7e7f80"); @@ -564,7 +589,7 @@ pub mod linux { #[test] #[cfg_attr(miri, ignore)] fn explicit_detach_nulls_tls() { - ThreadContext::new([0u8; 16], [0u8; 8], [0u8; 8], &[]).attach(); + ThreadContext::new([0u8; 16], [0u8; 8], 0, [0u8; 8], &[]).attach(); assert!(!read_tls_context_ptr().is_null()); let _ = ThreadContext::detach(); @@ -579,7 +604,8 @@ pub mod linux { #[cfg_attr(miri, ignore)] fn long_value_capped_at_255_bytes() { let long_val = "a".repeat(300); - ThreadContext::new([0u8; 16], [0u8; 8], [0u8; 8], &[(0, long_val.as_str())]).attach(); + ThreadContext::new([0u8; 16], [0u8; 8], 0, [0u8; 8], &[(0, long_val.as_str())]) + .attach(); let ptr = read_tls_context_ptr(); assert!(!ptr.is_null()); @@ -610,8 +636,14 @@ pub mod linux { let main_root_span_id = [0x33, 0x44, 0x55, 0x66, 0x77, 0x88, 0x99, 0xAA]; let handle = std::thread::spawn(move || { - ThreadContext::new(spawned_trace_id, spawned_span_id, spawned_root_span_id, &[]) - .attach(); + ThreadContext::new( + spawned_trace_id, + spawned_span_id, + 0, + spawned_root_span_id, + &[], + ) + .attach(); // Let the main thread attach its own record and verify its slot. b.wait(); @@ -638,7 +670,7 @@ pub mod linux { "main thread should see a null pointer and not another thread's context" ); - ThreadContext::new(main_trace_id, main_span_id, main_root_span_id, &[]).attach(); + ThreadContext::new(main_trace_id, main_span_id, 0, main_root_span_id, &[]).attach(); let ptr = read_tls_context_ptr(); assert!(!ptr.is_null(), "main thread TLS must be set"); diff --git a/libdd-otel-thread-ctx/src/record.rs b/libdd-otel-thread-ctx/src/record.rs index e2c673eba8..0822d3629a 100644 --- a/libdd-otel-thread-ctx/src/record.rs +++ b/libdd-otel-thread-ctx/src/record.rs @@ -15,7 +15,7 @@ pub struct ThreadContextRecord { pub(crate) trace_id: [u8; 16], pub(crate) span_id: [u8; 8], pub(crate) valid: AtomicU8, - pub(crate) reserved: u8, + pub(crate) trace_flags: u8, pub(crate) attrs_data_size: u16, pub(crate) attrs_data: [u8; MAX_ATTRS_DATA_SIZE], } @@ -26,7 +26,7 @@ const _: () = { assert!(mem::offset_of!(ThreadContextRecord, trace_id) == 0); assert!(mem::offset_of!(ThreadContextRecord, span_id) == 16); assert!(mem::offset_of!(ThreadContextRecord, valid) == 24); - assert!(mem::offset_of!(ThreadContextRecord, reserved) == 25); + assert!(mem::offset_of!(ThreadContextRecord, trace_flags) == 25); assert!(mem::offset_of!(ThreadContextRecord, attrs_data_size) == 26); assert!(mem::offset_of!(ThreadContextRecord, attrs_data) == 28); }; @@ -36,6 +36,7 @@ impl ThreadContextRecord { &mut self, trace_id: [u8; 16], span_id: [u8; 8], + trace_flags: u8, local_root_span_id: [u8; 8], attributes: I, ) -> bool @@ -43,13 +44,20 @@ impl ThreadContextRecord { I: IntoIterator, S: AsRef, { - self.update(trace_id, span_id, local_root_span_id, attributes) + self.update( + trace_id, + span_id, + trace_flags, + local_root_span_id, + attributes, + ) } pub fn update( &mut self, trace_id: [u8; 16], span_id: [u8; 8], + trace_flags: u8, local_root_span_id: [u8; 8], attributes: I, ) -> bool @@ -61,6 +69,7 @@ impl ThreadContextRecord { compiler_fence(Ordering::SeqCst); self.trace_id = trace_id; self.span_id = span_id; + self.trace_flags = trace_flags; let complete = self.set_attrs(local_root_span_id, attributes); compiler_fence(Ordering::SeqCst); self.valid.store(1, Ordering::Relaxed); @@ -120,7 +129,7 @@ impl Default for ThreadContextRecord { trace_id: [0; 16], span_id: [0; 8], valid: AtomicU8::new(1), - reserved: 0, + trace_flags: 0, attrs_data_size: 0, attrs_data: [0; MAX_ATTRS_DATA_SIZE], } @@ -134,9 +143,10 @@ mod tests { #[test] fn explicit_record_lifecycle() { let mut record = ThreadContextRecord::default(); - assert!(record.initialize([1; 16], [2; 8], [3; 8], [(1, "service")])); + assert!(record.initialize([1; 16], [2; 8], 0x03, [3; 8], [(1, "service")])); assert_eq!(record.valid.load(Ordering::Relaxed), 1); assert_eq!(record.span_id, [2; 8]); + assert_eq!(record.trace_flags, 0x03); assert_eq!(&record.attrs_data[20..27], b"service"); record.update_span_id([4; 8]); assert_eq!(record.span_id, [4; 8]);