From 99cefc742b0920c335822fe7a2295925cad05ad0 Mon Sep 17 00:00:00 2001 From: Bob Weinand Date: Fri, 26 Jun 2026 22:09:57 +0200 Subject: [PATCH 1/4] Avoid a dedicated socket for crashtracker This allows us doing an unified authentication of the socket on the sidecar side, before handing off to crashtracker. Usually the parent process would be the one connecting to the crashtracker, here it is not. This will allow custom handling, and one less endpoint to secure. The strategy chosen is upgrading to the crashtracker once an IPC socket is received. Currently we recreate a new socket, but we might also decide to reuse the existing IPC connection in future. Signed-off-by: Bob Weinand --- datadog-ipc-macros/src/lib.rs | 29 +- datadog-ipc/src/example_interface.rs | 67 ++-- datadog-ipc/src/ipc_server.rs | 32 ++ datadog-ipc/src/lib.rs | 2 + datadog-sidecar-ffi/src/lib.rs | 16 - datadog-sidecar/src/config.rs | 2 +- datadog-sidecar/src/crashtracker.rs | 360 +++++++++++++++++- datadog-sidecar/src/entry.rs | 31 +- datadog-sidecar/src/lib.rs | 1 + datadog-sidecar/src/service/blocking.rs | 9 + .../src/service/sidecar_interface.rs | 5 + datadog-sidecar/src/service/sidecar_server.rs | 109 +++--- datadog-sidecar/src/setup/mod.rs | 10 + datadog-sidecar/src/setup/thread_listener.rs | 1 - .../src/setup/thread_listener_windows.rs | 1 - datadog-sidecar/src/setup/unix.rs | 13 + datadog-sidecar/src/unix.rs | 26 +- .../src/collector/receiver_manager.rs | 28 +- libdd-crashtracker/src/lib.rs | 5 +- .../src/receiver/entry_points.rs | 6 + libdd-crashtracker/src/receiver/mod.rs | 5 +- .../src/shared/configuration/builder.rs | 10 + .../src/shared/configuration/mod.rs | 39 ++ 23 files changed, 615 insertions(+), 192 deletions(-) create mode 100644 datadog-ipc/src/ipc_server.rs diff --git a/datadog-ipc-macros/src/lib.rs b/datadog-ipc-macros/src/lib.rs index dc3a69bf52..7e82ab0269 100644 --- a/datadog-ipc-macros/src/lib.rs +++ b/datadog-ipc-macros/src/lib.rs @@ -210,7 +210,6 @@ fn gen_handler_trait( quote! { fn #name( &self, - peer: datadog_ipc::PeerCredentials, #(#params),* ) -> impl ::std::future::Future + Send + '_; } @@ -223,6 +222,9 @@ fn gen_handler_trait( /// The serve loop uses this to track received payloads. fn recv_counter(&self) -> &::std::sync::atomic::AtomicU64; + /// Storage for the connection to read from. + fn connection(&self) -> &datadog_ipc::ipc_server::OwnedServerConn; + #(#handler_methods)* } } @@ -260,29 +262,29 @@ fn gen_serve_fn( quote! { #[cfg(target_os = "linux")] if __pending_acks > 0 { - datadog_ipc::send_acks_async(&async_fd, __pending_acks).await; + datadog_ipc::send_acks_async(handler.connection().async_conn(), __pending_acks).await; __pending_acks = 0; } - let result = handler.#name(peer, #(#field_names),*).await; + let result = handler.#name(#(#field_names),*).await; let __resp_data = datadog_ipc::codec::encode(&result); - datadog_ipc::send_raw_async(&async_fd, &__resp_data).await.ok(); + datadog_ipc::send_raw_async(handler.connection().async_conn(), &__resp_data).await.ok(); } } else { // On Linux, buffer up to 20 acks and flush in a single // sendmmsg(2) syscall; on other platforms send each ack immediately. quote! { - handler.#name(peer, #(#field_names),*).await; + handler.#name(#(#field_names),*).await; #[cfg(target_os = "linux")] { __pending_acks += 1; if #force_flush || __pending_acks >= datadog_ipc::ACK_BUFFER_SIZE { - datadog_ipc::send_acks_async(&async_fd, __pending_acks).await; + datadog_ipc::send_acks_async(handler.connection().async_conn(), __pending_acks).await; __pending_acks = 0; } } #[cfg(not(target_os = "linux"))] // 1-byte ack: distinguishable from EOF (0 bytes from recvmsg on closed socket). - datadog_ipc::send_raw_async(&async_fd, &[0u8]).await.ok(); + datadog_ipc::send_raw_async(handler.connection().async_conn(), &[0u8]).await.ok(); } }; @@ -296,23 +298,14 @@ fn gen_serve_fn( quote! { pub async fn #serve_fn( - conn: datadog_ipc::SeqpacketConn, handler: ::std::sync::Arc, ) { - let peer = conn.peer_credentials().unwrap_or_default(); - let async_fd = match conn.into_async_conn() { - Ok(fd) => fd, - Err(e) => { - ::tracing::error!("IPC serve: into_async_conn failed: {e}"); - return; - } - }; // Pending 1-byte acks for fire-and-forget methods, flushed via sendmmsg(2) on Linux. #[cfg(target_os = "linux")] let mut __pending_acks: u32 = 0; loop { let (mut req, fds) = match datadog_ipc::recv_raw_async( - &async_fd, + &handler.connection().async_conn(), |buf| datadog_ipc::codec::decode::<#enum_name>(buf), ).await { Ok((Ok(req), fds)) => (req, fds), @@ -334,7 +327,7 @@ fn gen_serve_fn( break; } let recv_counter = handler.recv_counter().load(::std::sync::atomic::Ordering::Relaxed) + 1; - ::tracing::trace!(recv_counter, ?req, pid = peer.pid, "IPC recv"); + ::tracing::trace!(recv_counter, ?req, pid = handler.connection().peer().pid, "IPC recv"); match req { #(#match_arms)* diff --git a/datadog-ipc/src/example_interface.rs b/datadog-ipc/src/example_interface.rs index 8585a41b08..4e9a8706b5 100644 --- a/datadog-ipc/src/example_interface.rs +++ b/datadog-ipc/src/example_interface.rs @@ -10,6 +10,7 @@ use std::{ }; use super::platform::{FileBackedHandle, PlatformHandle, ShmHandle}; +use crate::ipc_server::OwnedServerConn; extern crate self as datadog_ipc; @@ -30,6 +31,7 @@ pub trait ExampleInterface { async fn echo_len(payload: Vec) -> u32; } +/// Shared server state. Cloned into a per-connection [`ExampleConnectionHandler`] on accept. #[derive(Default, Clone)] pub struct ExampleServer { req_cnt: Arc, @@ -38,70 +40,69 @@ pub struct ExampleServer { impl ExampleServer { pub async fn accept_connection(self, conn: crate::SeqpacketConn) { - serve_example_interface_connection(conn, Arc::new(self)).await + let connection = match OwnedServerConn::new(conn) { + Ok(c) => c, + Err(e) => { + ::tracing::error!("ExampleServer: failed to set up connection: {e}"); + return; + } + }; + serve_example_interface_connection(Arc::new(ExampleConnectionHandler { + server: self, + connection, + })) + .await } } -impl ExampleInterface for ExampleServer { +/// Per-connection handler: owns the connection and serves requests received on it. +struct ExampleConnectionHandler { + server: ExampleServer, + connection: OwnedServerConn, +} + +impl ExampleInterface for ExampleConnectionHandler { fn recv_counter(&self) -> &AtomicU64 { - &self.req_cnt + &self.server.req_cnt } - fn notify( - &self, - _peer: datadog_ipc::PeerCredentials, - ) -> impl std::future::Future + Send + '_ { + fn connection(&self) -> &OwnedServerConn { + &self.connection + } + + fn notify(&self) -> impl std::future::Future + Send + '_ { std::future::ready(()) } - fn ping( - &self, - _peer: datadog_ipc::PeerCredentials, - ) -> impl std::future::Future + Send + '_ { + fn ping(&self) -> impl std::future::Future + Send + '_ { std::future::ready(()) } - fn time_now( - &self, - _peer: datadog_ipc::PeerCredentials, - ) -> impl std::future::Future + Send + '_ { + fn time_now(&self) -> impl std::future::Future + Send + '_ { std::future::ready(Instant::now().elapsed()) } - fn req_cnt( - &self, - _peer: datadog_ipc::PeerCredentials, - ) -> impl std::future::Future + Send + '_ { - std::future::ready(self.req_cnt.load(Ordering::Relaxed) as u32) + fn req_cnt(&self) -> impl std::future::Future + Send + '_ { + std::future::ready(self.server.req_cnt.load(Ordering::Relaxed) as u32) } fn store_file( &self, - _peer: datadog_ipc::PeerCredentials, file: PlatformHandle, ) -> impl std::future::Future + Send + '_ { #[allow(clippy::unwrap_used)] - self.stored_files.lock().unwrap().push(file); + self.server.stored_files.lock().unwrap().push(file); std::future::ready(()) } - async fn shm_sum( - &self, - _peer: datadog_ipc::PeerCredentials, - handle: ShmHandle, - len: usize, - ) -> u64 { + async fn shm_sum(&self, handle: ShmHandle, len: usize) -> u64 { match handle.map() { Ok(mapped) => mapped.as_slice()[..len].iter().map(|&b| b as u64).sum(), Err(_) => u64::MAX, } } - fn echo_len( - &self, - _peer: datadog_ipc::PeerCredentials, - payload: Vec, - ) -> impl std::future::Future + Send + '_ { + fn echo_len(&self, payload: Vec) -> impl std::future::Future + Send + '_ { std::future::ready(payload.len() as u32) } } diff --git a/datadog-ipc/src/ipc_server.rs b/datadog-ipc/src/ipc_server.rs new file mode 100644 index 0000000000..9d8dbe6967 --- /dev/null +++ b/datadog-ipc/src/ipc_server.rs @@ -0,0 +1,32 @@ +// Copyright 2021-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 + +use crate::{PeerCredentials, SeqpacketConn}; +use std::io; + +pub struct OwnedServerConn { + connection: crate::AsyncConn, + peer: PeerCredentials, +} + +impl OwnedServerConn { + pub fn new(conn: SeqpacketConn) -> io::Result { + let peer = conn.peer_credentials().unwrap_or_default(); + let connection = conn.into_async_conn()?; + Ok(Self { connection, peer }) + } + + /// Construct from an already-async connection and a known peer. Useful for callers that have + /// already wrapped the fd (and for tests). + pub fn from_async(connection: crate::AsyncConn, peer: PeerCredentials) -> Self { + Self { connection, peer } + } + + pub fn async_conn(&self) -> &crate::AsyncConn { + &self.connection + } + + pub fn peer(&self) -> &PeerCredentials { + &self.peer + } +} diff --git a/datadog-ipc/src/lib.rs b/datadog-ipc/src/lib.rs index 5204784703..ddb8b88aae 100644 --- a/datadog-ipc/src/lib.rs +++ b/datadog-ipc/src/lib.rs @@ -18,6 +18,8 @@ pub mod shm_stats; mod atomic_option; pub mod client; pub mod codec; +pub mod ipc_server; + pub use atomic_option::AtomicOption; pub use client::IpcClientConn; diff --git a/datadog-sidecar-ffi/src/lib.rs b/datadog-sidecar-ffi/src/lib.rs index 13f915f9e9..0b45b4e3b4 100644 --- a/datadog-sidecar-ffi/src/lib.rs +++ b/datadog-sidecar-ffi/src/lib.rs @@ -17,7 +17,6 @@ use datadog_live_debugger::debugger_defs::DebuggerPayload; use datadog_sidecar::agent_remote_config::{new_reader, reader_from_shm, AgentRemoteConfigWriter}; use datadog_sidecar::config; use datadog_sidecar::config::LogMethod; -use datadog_sidecar::crashtracker::crashtracker_unix_socket_path; use datadog_sidecar::service::agent_info::AgentInfoReader; use datadog_sidecar::service::telemetry::InternalTelemetryAction; use datadog_sidecar::service::{ @@ -1614,21 +1613,6 @@ pub extern "C" fn ddog_sidecar_reconnect( transport.reconnect(|| unsafe { factory() }); } -/// Return the path of the crashtracker unix domain socket. -#[no_mangle] -#[allow(clippy::missing_safety_doc)] -pub unsafe extern "C" fn ddog_sidecar_get_crashtracker_unix_socket_path() -> ffi::CharSlice<'static> -{ - let socket_path = crashtracker_unix_socket_path(); - let str = socket_path.to_str().unwrap_or_default(); - - let size = str.len(); - let malloced = libc::malloc(size) as *mut u8; - let buf = slice::from_raw_parts_mut(malloced, size); - buf.copy_from_slice(str.as_bytes()); - ffi::CharSlice::from_raw_parts(malloced as *mut c_char, size) -} - /// Gets an agent info reader. #[no_mangle] #[allow(clippy::missing_safety_doc)] diff --git a/datadog-sidecar/src/config.rs b/datadog-sidecar/src/config.rs index 95ecf221eb..c2a923665f 100644 --- a/datadog-sidecar/src/config.rs +++ b/datadog-sidecar/src/config.rs @@ -174,7 +174,7 @@ impl AppSecConfig { pub struct FromEnv {} impl FromEnv { - fn ipc_mode() -> IpcMode { + pub fn ipc_mode() -> IpcMode { let mode = std::env::var(ENV_SIDECAR_IPC_MODE).unwrap_or_default(); match mode.as_str() { diff --git a/datadog-sidecar/src/crashtracker.rs b/datadog-sidecar/src/crashtracker.rs index 176ec45a29..e14088f968 100644 --- a/datadog-sidecar/src/crashtracker.rs +++ b/datadog-sidecar/src/crashtracker.rs @@ -1,18 +1,350 @@ // Copyright 2021-Present Datadog, Inc. https://www.datadoghq.com/ // SPDX-License-Identifier: Apache-2.0 -use std::path::PathBuf; - -use crate::primary_sidecar_identifier; - -pub fn crashtracker_unix_socket_path() -> PathBuf { - let base_path = format!( - concat!("libdatadog.ct.", crate::sidecar_version!(), "@{}.sock"), - primary_sidecar_identifier() - ); - #[cfg(target_os = "linux")] - let ret = base_path.into(); - #[cfg(not(target_os = "linux"))] - let ret = std::env::temp_dir().join(base_path); - ret +//! Crashtracker integration over the single sidecar IPC socket. +//! +//! Instead of a second dedicated listener, the crash-time collector connects to the sidecar IPC +//! socket and sends an +//! [`enter_crashtracker_receiver`](crate::service::SidecarInterface::enter_crashtracker_receiver) +//! request as its first message — a normal codec-encoded message (see +//! [`crashtracker_receiver_request_bytes`]), decoded by the regular serve loop with no marker or +//! peeking — then streams the crash report over the same connection. +//! +//! The receiver consumes an `AsyncBufRead` byte stream; since the socket preserves message +//! boundaries, [`SeqpacketStreamReader`] concatenates the collector's datagrams back into a +//! contiguous stream (a zero-length datagram is EOF). + +use datadog_ipc::AsyncConn; + +/// Returns the abstract socket name the Linux listener binds. +#[cfg(target_os = "linux")] +pub fn crashtracker_ipc_socket_path( + master_pid: u32, + ipc_mode: crate::config::IpcMode, +) -> std::path::PathBuf { + let liaison = if master_pid != 0 { + crate::setup::DefaultLiason::ipc_for_pid(master_pid) + } else { + crate::setup::liaison_for_ipc_mode(ipc_mode) + }; + liaison.path().to_path_buf() +} + +/// The codec-encoded `enter_crashtracker_receiver` request the collector sends as its first +/// message. Fixed bytes, safe to send from a crash handler. This function must however be called +/// once before the crash handler runs, to initialize. +pub fn crashtracker_receiver_request_bytes() -> &'static [u8] { + use crate::service::sidecar_interface::SidecarInterfaceRequest; + static BYTES: std::sync::OnceLock> = std::sync::OnceLock::new(); + BYTES.get_or_init(|| { + datadog_ipc::codec::encode(&SidecarInterfaceRequest::EnterCrashtrackerReceiver {}) + }) +} + +/// `unix_socket_connector` for the crashtracker config: connect a `SOCK_SEQPACKET` socket to the +/// sidecar IPC socket at `unix_socket_path` and send the encoded `enter_crashtracker_receiver` +/// request as the first message, returning the connected fd (`-1` on error). Runs in the crash +/// handler with no allocation, so [`crashtracker_receiver_request_bytes`] must be primed earlier. +/// Linux-only (fresh `SOCK_SEQPACKET` connect); macOS reuses the existing sidecar fd instead. +#[cfg(target_os = "linux")] +pub fn connect_to_sidecar_receiver(unix_socket_path: &str) -> std::os::fd::RawFd { + use nix::sys::socket; + use std::os::fd::{AsRawFd, IntoRawFd}; + + let fd = match socket::socket( + socket::AddressFamily::Unix, + socket::SockType::SeqPacket, + socket::SockFlag::empty(), + None, + ) { + Ok(fd) => fd, + Err(_) => return -1, + }; + // Close-on-exec (portable; SOCK_CLOEXEC isn't available everywhere). + let raw = fd.as_raw_fd(); + let flags = unsafe { libc::fcntl(raw, libc::F_GETFD) }; + if flags >= 0 { + unsafe { libc::fcntl(raw, libc::F_SETFD, flags | libc::FD_CLOEXEC) }; + } + + // A name starting with `.`/`/` is a filesystem path; anything else is an abstract name (the + // listener binds abstract names by default). + let addr = if unix_socket_path.starts_with(['.', '/']) { + socket::UnixAddr::new(unix_socket_path) + } else { + socket::UnixAddr::new_abstract(unix_socket_path.as_bytes()) + }; + let addr = match addr { + Ok(a) => a, + Err(_) => return -1, + }; + + if socket::connect(fd.as_raw_fd(), &addr).is_err() { + return -1; + } + if socket::send( + fd.as_raw_fd(), + crashtracker_receiver_request_bytes(), + socket::MsgFlags::empty(), + ) + .is_err() + { + return -1; + } + fd.into_raw_fd() +} + +mod adapter { + use super::*; + use datadog_ipc::platform::sockets::max_message_size; + use std::io; + use std::os::fd::AsRawFd; + use std::pin::Pin; + use std::task::{Context, Poll}; + use tokio::io::{AsyncRead, ReadBuf}; + + /// `AsyncRead` adapter that concatenates an ordered stream of SEQPACKET datagrams into a + /// contiguous byte stream. Borrows the handler-owned connection; reads are serial (the + /// `enter_crashtracker_receiver` handler runs it to completion), so there is no concurrent + /// reader. A zero-length datagram is EOF; a datagram larger than the caller's buffer is + /// buffered and drained on later reads, so boundaries never truncate data. + pub struct SeqpacketStreamReader<'a> { + conn: &'a AsyncConn, + /// Leftover bytes from a datagram that did not fit in the caller's buffer. + leftover: Vec, + leftover_pos: usize, + eof: bool, + } + + impl<'a> SeqpacketStreamReader<'a> { + pub fn new(conn: &'a AsyncConn) -> Self { + Self { + conn, + leftover: Vec::new(), + leftover_pos: 0, + eof: false, + } + } + } + + impl AsyncRead for SeqpacketStreamReader<'_> { + fn poll_read( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + out: &mut ReadBuf<'_>, + ) -> Poll> { + let this = self.get_mut(); + + // Drain any buffered leftover first. + if this.leftover_pos < this.leftover.len() { + let remaining = &this.leftover[this.leftover_pos..]; + let n = remaining.len().min(out.remaining()); + out.put_slice(&remaining[..n]); + this.leftover_pos += n; + if this.leftover_pos >= this.leftover.len() { + this.leftover.clear(); + this.leftover_pos = 0; + } + return Poll::Ready(Ok(())); + } + + if this.eof { + return Poll::Ready(Ok(())); + } + + loop { + let mut guard = match this.conn.poll_read_ready(cx) { + Poll::Ready(Ok(g)) => g, + Poll::Ready(Err(e)) => return Poll::Ready(Err(e)), + Poll::Pending => return Poll::Pending, + }; + + // One recv() returns one datagram; a max-sized buffer avoids truncation. + let read_result = guard.try_io(|inner| { + let mut tmp = vec![0u8; max_message_size()]; + let n = unsafe { + libc::recv( + inner.as_raw_fd(), + tmp.as_mut_ptr() as *mut libc::c_void, + tmp.len(), + 0, + ) + }; + if n < 0 { + Err(io::Error::last_os_error()) + } else { + tmp.truncate(n as usize); + Ok(tmp) + } + }); + + match read_result { + Ok(Ok(payload)) => { + if payload.is_empty() { + this.eof = true; + return Poll::Ready(Ok(())); + } + let n = payload.len().min(out.remaining()); + out.put_slice(&payload[..n]); + if n < payload.len() { + this.leftover = payload; + this.leftover_pos = n; + } + return Poll::Ready(Ok(())); + } + Ok(Err(e)) => { + // Treat peer close as EOF, not error. + if matches!( + e.kind(), + io::ErrorKind::BrokenPipe + | io::ErrorKind::UnexpectedEof + | io::ErrorKind::ConnectionReset + ) { + this.eof = true; + return Poll::Ready(Ok(())); + } + return Poll::Ready(Err(e)); + } + Err(_would_block) => continue, + } + } + } + } +} + +pub use adapter::SeqpacketStreamReader; + +/// Wrap `AsyncConn` and dispatch it to crashtracking receiver. +pub async fn run_crashtracker_receiver(conn: &AsyncConn) { + use std::os::fd::AsRawFd; + use tokio::io::BufReader; + + let reader = BufReader::new(SeqpacketStreamReader::new(conn)); + if let Err(e) = libdd_crashtracker::async_receiver_entry_point_stream(reader).await { + tracing::warn!("Got error while receiving crash report over IPC: {e}"); + } + + // The connection ends with a crash. We don't go back to receive more data. + unsafe { libc::shutdown(conn.as_raw_fd(), libc::SHUT_RD) }; +} + +#[cfg(test)] +mod tests { + use super::*; + use std::os::fd::{FromRawFd, OwnedFd}; + use tokio::io::unix::AsyncFd; + use tokio::io::{AsyncReadExt, BufReader}; + + fn dgram_pair() -> (OwnedFd, OwnedFd) { + let mut fds = [0i32; 2]; + let rc = unsafe { libc::socketpair(libc::AF_UNIX, libc::SOCK_DGRAM, 0, fds.as_mut_ptr()) }; + assert_eq!( + rc, + 0, + "socketpair failed: {}", + std::io::Error::last_os_error() + ); + unsafe { (OwnedFd::from_raw_fd(fds[0]), OwnedFd::from_raw_fd(fds[1])) } + } + + fn send_datagrams(fd: &OwnedFd, chunks: &[&[u8]]) { + use std::os::fd::AsRawFd; + for chunk in chunks { + let n = unsafe { + libc::send( + fd.as_raw_fd(), + chunk.as_ptr() as *const libc::c_void, + chunk.len(), + 0, + ) + }; + assert_eq!( + n, + chunk.len() as isize, + "send: {}", + std::io::Error::last_os_error() + ); + } + } + + /// The adapter concatenates ordered datagrams back into the original stream — including one + /// larger than the read buffer (the leftover path) — and reports EOF on peer close. + #[tokio::test] + #[cfg_attr(miri, ignore)] + async fn test_seqpacket_stream_reader_multichunk() { + let (a, b) = dgram_pair(); + + let big = "X".repeat(2000); + let chunks: Vec> = vec![ + b"DD_CRASHTRACK_BEGIN_CONFIG\n".to_vec(), + b"{\"some\":\"config\"}\n".to_vec(), + b"DD_CRASHTRACK_END_CONFIG\n".to_vec(), + format!("{big}\n").into_bytes(), + b"DD_CRASHTRACK_DONE\n".to_vec(), + ]; + + let writer = tokio::task::spawn_blocking(move || { + let refs: Vec<&[u8]> = chunks.iter().map(|c| c.as_slice()).collect(); + send_datagrams(&a, &refs); + // Drop `a` to signal EOF to the reader. + drop(a); + }); + + let async_fd = AsyncFd::new(b).expect("AsyncFd"); + // Small buffer to force the adapter's leftover path on the big chunk. + let mut reader = BufReader::with_capacity(64, SeqpacketStreamReader::new(&async_fd)); + let mut got = Vec::new(); + reader.read_to_end(&mut got).await.expect("read_to_end"); + writer.await.expect("writer task"); + + let expected = format!( + "DD_CRASHTRACK_BEGIN_CONFIG\n{{\"some\":\"config\"}}\nDD_CRASHTRACK_END_CONFIG\n{big}\nDD_CRASHTRACK_DONE\n" + ); + assert_eq!(String::from_utf8(got).unwrap(), expected); + } + + /// After the receiver consumes a full report, the collector never sends anything else on + /// this connection — it only polls for the sidecar to hang up. `run_crashtracker_receiver` + /// must therefore leave the connection in a state where the serve loop's *next* `recv` + /// fails right away (its existing "connection closed" path already breaks the loop), instead + /// of blocking forever in `readable()` for a request that will never arrive. + #[tokio::test] + #[cfg_attr(miri, ignore)] + async fn test_run_crashtracker_receiver_unblocks_next_recv() { + use datadog_ipc::recv_raw_async; + + let (collector, receiver) = dgram_pair(); + // Mirror the collector: send the report and never send/close anything else. + send_datagrams(&collector, &[b"DD_CRASHTRACK_DONE\n"]); + + let async_fd = AsyncFd::new(receiver).expect("AsyncFd"); + run_crashtracker_receiver(&async_fd).await; + + // Simulates the serve loop's next iteration: without the fix this hangs forever, so + // bound it with a timeout that would fail the test rather than hang the suite. + let result = tokio::time::timeout( + std::time::Duration::from_secs(5), + recv_raw_async::<_, ()>(&async_fd, |_| ()), + ) + .await + .expect("serve loop's next recv did not return promptly after the receiver finished"); + assert!( + result.is_err(), + "expected the next recv to observe the connection as closed" + ); + } + + /// The collector's first-message bytes must decode, via the normal IPC codec, back to the + /// `EnterCrashtrackerReceiver` request — i.e. a real protocol message, not a magic marker. + #[test] + fn receiver_request_bytes_decode_to_variant() { + use crate::service::sidecar_interface::SidecarInterfaceRequest; + let bytes = crashtracker_receiver_request_bytes(); + let decoded = datadog_ipc::codec::decode::(bytes) + .expect("request bytes must decode as a SidecarInterfaceRequest"); + assert!(matches!( + decoded, + SidecarInterfaceRequest::EnterCrashtrackerReceiver {} + )); + } } diff --git a/datadog-sidecar/src/entry.rs b/datadog-sidecar/src/entry.rs index 0919f6e649..1fd039dc1c 100644 --- a/datadog-sidecar/src/entry.rs +++ b/datadog-sidecar/src/entry.rs @@ -2,8 +2,6 @@ // SPDX-License-Identifier: Apache-2.0 use anyhow::Context; -#[cfg(unix)] -use libdd_crashtracker; #[cfg(target_os = "linux")] use spawn_worker::read_pt_interp_self; use spawn_worker::{entrypoint, Stdio}; @@ -19,8 +17,6 @@ use std::{ }; use tokio::sync::{mpsc, oneshot}; -#[cfg(unix)] -use crate::crashtracker::crashtracker_unix_socket_path; use crate::service::blocking::SidecarTransport; use crate::service::SidecarServer; @@ -36,7 +32,6 @@ use crate::{ddog_daemon_entry_point, setup_daemon_process}; /// Configuration for main_loop behavior pub struct MainLoopConfig { pub enable_ctrl_c_handler: bool, - pub enable_crashtracker: bool, pub external_shutdown_rx: Option>, /// Set to false in thread mode so the worker's UID can be obtained on the /// first connection and used to fchown the SHM. @@ -47,7 +42,6 @@ impl Default for MainLoopConfig { fn default() -> Self { Self { enable_ctrl_c_handler: true, - enable_crashtracker: true, external_shutdown_rx: None, init_shm_eagerly: true, } @@ -110,26 +104,6 @@ where }); } - #[cfg(unix)] - if loop_config.enable_crashtracker { - tokio::spawn(async move { - let socket_path = crashtracker_unix_socket_path(); - match libdd_crashtracker::get_receiver_unix_socket( - socket_path.to_str().unwrap_or_default(), - ) { - Ok(listener) => loop { - if let Err(e) = - libdd_crashtracker::async_receiver_entry_point_unix_listener(&listener) - .await - { - tracing::warn!("Got error while receiving crash report: {e}"); - } - }, - Err(e) => tracing::error!("Failed setting up the crashtracker listener: {e}"), - } - }); - } - if loop_config.init_shm_eagerly { drop(SHM_LIMITER.lock()); } @@ -311,10 +285,7 @@ pub fn start_or_connect_to_sidecar(cfg: Config) -> anyhow::Result setup::DefaultLiason::ipc_shared(), - config::IpcMode::InstancePerProcess => setup::DefaultLiason::ipc_per_process(), - }; + let liaison = setup::liaison_for_ipc_mode(cfg.ipc_mode); let err = match liaison.attempt_listen() { Ok(Some(listener)) => { diff --git a/datadog-sidecar/src/lib.rs b/datadog-sidecar/src/lib.rs index 2c22b02644..dcaea0b303 100644 --- a/datadog-sidecar/src/lib.rs +++ b/datadog-sidecar/src/lib.rs @@ -9,6 +9,7 @@ pub mod agent_remote_config; pub mod config; +#[cfg(unix)] pub mod crashtracker; mod dump; pub mod entry; diff --git a/datadog-sidecar/src/service/blocking.rs b/datadog-sidecar/src/service/blocking.rs index 9d7eb32cac..d011479f85 100644 --- a/datadog-sidecar/src/service/blocking.rs +++ b/datadog-sidecar/src/service/blocking.rs @@ -48,6 +48,15 @@ impl SidecarTransport { Ok(creds.pid) } + #[cfg(unix)] + pub fn as_raw_fd(&mut self) -> std::os::fd::RawFd { + let sender = match self.inner.get_mut() { + Ok(s) => s, + Err(poisoned) => poisoned.into_inner(), + }; + sender.channel.0.conn.as_raw_fd() + } + pub fn reconnect(&mut self, factory: F) where F: FnOnce() -> Option>, diff --git a/datadog-sidecar/src/service/sidecar_interface.rs b/datadog-sidecar/src/service/sidecar_interface.rs index a58c6fcbde..bf6c421d52 100644 --- a/datadog-sidecar/src/service/sidecar_interface.rs +++ b/datadog-sidecar/src/service/sidecar_interface.rs @@ -262,4 +262,9 @@ pub trait SidecarInterface { /// /// A string representation of the current statistics of the service. async fn stats() -> String; + + /// Repurpose this connection as a crashtracker receiver. + /// + /// The connection must right after that start emitting crashtracker messages. + async fn enter_crashtracker_receiver(); } diff --git a/datadog-sidecar/src/service/sidecar_server.rs b/datadog-sidecar/src/service/sidecar_server.rs index ea16ef1d6c..105e2ddaee 100644 --- a/datadog-sidecar/src/service/sidecar_server.rs +++ b/datadog-sidecar/src/service/sidecar_server.rs @@ -11,7 +11,7 @@ use crate::service::{ SidecarInterface, }; use datadog_ipc::platform::{FileBackedHandle, ShmHandle}; -use datadog_ipc::{PeerCredentials, SeqpacketConn}; +use datadog_ipc::SeqpacketConn; use libdd_common::{Endpoint, MutexExt}; use libdd_telemetry::metrics::MetricContext; use libdd_telemetry::worker::{LifecycleAction, TelemetryActions, TelemetryWorkerStats}; @@ -28,8 +28,6 @@ use std::sync::{Arc, Mutex, Weak}; use std::time::{Duration, SystemTime}; use tracing::{debug, error, info, trace, warn}; -use serde::{Deserialize, Serialize}; - use crate::config::get_product_endpoint; use crate::service::agent_info::AgentInfos; use crate::service::debugger_diagnostics_bookkeeper::{ @@ -45,6 +43,7 @@ use crate::service::stats_flusher::{ }; use crate::service::tracing::trace_flusher::TraceFlusherStats; use crate::tokio_util::run_or_spawn_shared; +use datadog_ipc::ipc_server::OwnedServerConn; use datadog_live_debugger::sender::{agent_info_supports_debugger_v2_endpoint, DebuggerType}; use libdd_capabilities_impl::NativeCapabilities; use libdd_common::tag::Tag; @@ -53,6 +52,7 @@ use libdd_remote_config::fetch::{ConfigInvariants, ConfigOptions, MultiTargetSta use libdd_telemetry::config::Config; use libdd_tinybytes as tinybytes; use libdd_trace_utils::tracer_header_tags::TracerHeaderTags; +use serde::{Deserialize, Serialize}; /// A Windows process handle used for remote config notification. /// @@ -130,22 +130,25 @@ struct ConnectionSidecarHandler { /// Used to auto-register metrics in newly-created telemetry clients when a metric point /// for a previously registered metric arrives for a new (service, env) combination. metric_registrations: Mutex>, + /// The connection this handler serves. + connection: OwnedServerConn, } impl ConnectionSidecarHandler { - fn new(server: SidecarServer) -> Self { + fn new(server: SidecarServer, connection: OwnedServerConn) -> Option { let submitted_payloads = Arc::new(AtomicU64::new(0)); server .connection_counters .lock_or_panic() .push(Arc::downgrade(&submitted_payloads)); - Self { + Some(Self { server, submitted_payloads, session_id: Default::default(), instances: Default::default(), metric_registrations: Default::default(), - } + connection, + }) } fn track_instance(&self, instance_id: &InstanceId) { @@ -198,9 +201,19 @@ impl SidecarServer { /// /// * `conn`: The connection to the client. pub async fn accept_connection(self, conn: SeqpacketConn) { - let handler = Arc::new(ConnectionSidecarHandler::new(self)); + let server_conn = match OwnedServerConn::new(conn) { + Ok(c) => c, + Err(e) => { + error!("IPC serve: failed to set up connection: {e}"); + return; + } + }; + let Some(handler) = ConnectionSidecarHandler::new(self, server_conn) else { + return; + }; + let handler = Arc::new(handler); let handler_for_cleanup = handler.clone(); - serve_sidecar_interface_connection(conn, handler).await; + serve_sidecar_interface_connection(handler).await; handler_for_cleanup.cleanup().await; } @@ -400,9 +413,17 @@ impl SidecarInterface for ConnectionSidecarHandler { &self.submitted_payloads } + fn connection(&self) -> &OwnedServerConn { + &self.connection + } + + async fn enter_crashtracker_receiver(&self) { + #[cfg(unix)] + crate::crashtracker::run_crashtracker_receiver(self.connection.async_conn()).await; + } + async fn enqueue_actions( &self, - _peer: PeerCredentials, instance_id: InstanceId, queue_id: QueueId, actions: Vec, @@ -640,12 +661,7 @@ impl SidecarInterface for ConnectionSidecarHandler { } } - async fn clear_queue_id( - &self, - _peer: PeerCredentials, - instance_id: InstanceId, - queue_id: QueueId, - ) { + async fn clear_queue_id(&self, instance_id: InstanceId, queue_id: QueueId) { let rt_info = self.server.get_runtime(&instance_id); let mut applications = rt_info.lock_applications(); if let Entry::Occupied(entry) = applications.entry(queue_id) { @@ -654,7 +670,7 @@ impl SidecarInterface for ConnectionSidecarHandler { } } - async fn register_telemetry_metric(&self, _peer: PeerCredentials, metric: MetricContext) { + async fn register_telemetry_metric(&self, metric: MetricContext) { self.metric_registrations .lock_or_panic() .entry(metric.name.clone()) @@ -663,7 +679,6 @@ impl SidecarInterface for ConnectionSidecarHandler { async fn set_session_config( &self, - peer: PeerCredentials, session_id: String, #[cfg(windows)] remote_config_notify_function: crate::service::remote_configs::RemoteConfigNotifyFunction, config: SessionConfig, @@ -683,7 +698,9 @@ impl SidecarInterface for ConnectionSidecarHandler { debug!("Set session config for {session_id} to {config:?}"); let session = self.server.get_session(&session_id); - session.pid.store(peer.pid as i32, Ordering::Relaxed); + session + .pid + .store(self.connection.peer().pid as i32, Ordering::Relaxed); #[cfg(windows)] #[allow(clippy::unwrap_used)] { @@ -692,7 +709,7 @@ impl SidecarInterface for ConnectionSidecarHandler { winapi::um::processthreadsapi::OpenProcess( winapi::um::winnt::PROCESS_ALL_ACCESS, 0, - peer.pid, + self.connection.peer().pid, ) }; if !handle.is_null() { @@ -820,7 +837,7 @@ impl SidecarInterface for ConnectionSidecarHandler { } } - async fn set_session_process_tags(&self, _peer: PeerCredentials, process_tags: Vec) { + async fn set_session_process_tags(&self, process_tags: Vec) { let session_id = self .session_id .get() @@ -831,7 +848,7 @@ impl SidecarInterface for ConnectionSidecarHandler { session.refresh_stats_process_tags(); } - async fn set_session_default_service_name(&self, _peer: PeerCredentials, name: Option) { + async fn set_session_default_service_name(&self, name: Option) { let session_id = self .session_id .get() @@ -842,7 +859,7 @@ impl SidecarInterface for ConnectionSidecarHandler { session.refresh_stats_process_tags(); } - async fn set_session_user_service_defined(&self, _peer: PeerCredentials, is_defined: bool) { + async fn set_session_user_service_defined(&self, is_defined: bool) { let session_id = self .session_id .get() @@ -853,12 +870,12 @@ impl SidecarInterface for ConnectionSidecarHandler { session.refresh_stats_process_tags(); } - async fn shutdown_runtime(&self, _peer: PeerCredentials, instance_id: InstanceId) { + async fn shutdown_runtime(&self, instance_id: InstanceId) { let session = self.server.get_session(&instance_id.session_id); tokio::spawn(async move { session.shutdown_runtime(&instance_id.runtime_id).await }); } - async fn shutdown_session(&self, _peer: PeerCredentials) { + async fn shutdown_session(&self) { let server = self.server.clone(); let session_id = self.session_id.get().cloned().unwrap_or_default(); tokio::spawn(async move { server.stop_session(&session_id).await }); @@ -866,7 +883,6 @@ impl SidecarInterface for ConnectionSidecarHandler { async fn send_trace_v04_shm( &self, - _peer: PeerCredentials, instance_id: InstanceId, handle: ShmHandle, _len: usize, @@ -897,7 +913,6 @@ impl SidecarInterface for ConnectionSidecarHandler { async fn send_trace_v04_bytes( &self, - _peer: PeerCredentials, instance_id: InstanceId, data: Vec, headers: SerializedTracerHeaderTags, @@ -923,7 +938,6 @@ impl SidecarInterface for ConnectionSidecarHandler { async fn send_debugger_data_shm( &self, - _peer: PeerCredentials, instance_id: InstanceId, queue_id: QueueId, handle: ShmHandle, @@ -946,7 +960,6 @@ impl SidecarInterface for ConnectionSidecarHandler { async fn send_debugger_diagnostics( &self, - _peer: PeerCredentials, instance_id: InstanceId, queue_id: QueueId, diagnostics_payload: Vec, @@ -974,7 +987,6 @@ impl SidecarInterface for ConnectionSidecarHandler { async fn acquire_exception_hash_rate_limiter( &self, - _peer: PeerCredentials, exception_hash: u64, granularity: Duration, ) { @@ -986,7 +998,6 @@ impl SidecarInterface for ConnectionSidecarHandler { #[allow(clippy::too_many_arguments)] async fn set_universal_service_tags( &self, - _peer: PeerCredentials, instance_id: InstanceId, queue_id: QueueId, service_name: String, @@ -1019,7 +1030,6 @@ impl SidecarInterface for ConnectionSidecarHandler { async fn set_request_config( &self, - _peer: PeerCredentials, instance_id: InstanceId, queue_id: QueueId, dynamic_instrumentation_state: DynamicInstrumentationConfigState, @@ -1044,7 +1054,6 @@ impl SidecarInterface for ConnectionSidecarHandler { async fn send_dogstatsd_actions( &self, - _peer: PeerCredentials, instance_id: InstanceId, actions: Vec, ) { @@ -1061,7 +1070,6 @@ impl SidecarInterface for ConnectionSidecarHandler { async fn add_span_to_concentrator( &self, - _peer: PeerCredentials, env: String, version: String, span: datadog_ipc::shm_stats::OwnedShmSpanInput, @@ -1082,7 +1090,7 @@ impl SidecarInterface for ConnectionSidecarHandler { } } - async fn flush(&self, _peer: PeerCredentials, options: SidecarFlushOptions) { + async fn flush(&self, options: SidecarFlushOptions) { if options.traces_and_stats { let flusher = self.server.trace_flusher.clone(); if let Err(e) = tokio::spawn(async move { flusher.flush().await }).await { @@ -1123,7 +1131,7 @@ impl SidecarInterface for ConnectionSidecarHandler { } } - async fn set_test_session_token(&self, _peer: PeerCredentials, token: String) { + async fn set_test_session_token(&self, token: String) { let session_id = self .session_id .get() @@ -1157,13 +1165,13 @@ impl SidecarInterface for ConnectionSidecarHandler { // }); } - async fn ping(&self, _peer: PeerCredentials) {} + async fn ping(&self) {} - async fn dump(&self, _peer: PeerCredentials) -> String { + async fn dump(&self) -> String { crate::dump::dump().await } - async fn stats(&self, _peer: PeerCredentials) -> String { + async fn stats(&self) -> String { let stats = self.server.compute_stats().await; #[allow(clippy::expect_used)] simd_json::serde::to_string(&stats).expect("unable to serialize stats to string") @@ -1177,6 +1185,16 @@ mod tests { use httpmock::{Method::POST, MockServer}; use tokio::time::{sleep, Duration as TokioDuration}; + /// Build a handler backed by a throwaway socketpair connection. These tests exercise + /// `enqueue_actions`, which uses only the shared server state and never reads the connection, + /// but the handler now requires one. + fn test_handler(server: SidecarServer) -> ConnectionSidecarHandler { + let (local, peer) = SeqpacketConn::socketpair().expect("socketpair"); + drop(peer); + let conn = OwnedServerConn::new(local).expect("OwnedServerConn"); + ConnectionSidecarHandler::new(server, conn).expect("handler") + } + fn ffe_context() -> FfeTelemetryContext { FfeTelemetryContext { service: "svc".to_owned(), @@ -1218,7 +1236,7 @@ mod tests { }) .await; - let handler = ConnectionSidecarHandler::new(SidecarServer::default()); + let handler = test_handler(SidecarServer::default()); let instance_id = InstanceId::new("session", "runtime"); let queue_id = QueueId::from(42); @@ -1241,7 +1259,6 @@ mod tests { handler .enqueue_actions( - PeerCredentials::default(), instance_id.clone(), queue_id, vec![SidecarAction::FfeExposureBatch(FfeExposureBatch { @@ -1280,7 +1297,7 @@ mod tests { }) .await; - let handler = ConnectionSidecarHandler::new(SidecarServer::default()); + let handler = test_handler(SidecarServer::default()); let instance_id = InstanceId::new("session", "runtime"); let queue_id = QueueId::from(42); @@ -1303,7 +1320,6 @@ mod tests { handler .enqueue_actions( - PeerCredentials::default(), instance_id.clone(), queue_id, vec![SidecarAction::FfeEvaluationMetrics { @@ -1346,7 +1362,7 @@ mod tests { }) .await; - let handler = ConnectionSidecarHandler::new(SidecarServer::default()); + let handler = test_handler(SidecarServer::default()); let instance_id = InstanceId::new("session", "runtime"); let queue_id = QueueId::from(42); @@ -1375,12 +1391,7 @@ mod tests { .contains_key(&queue_id)); handler - .enqueue_actions( - PeerCredentials::default(), - instance_id, - queue_id, - Vec::new(), - ) + .enqueue_actions(instance_id, queue_id, Vec::new()) .await; sleep(TokioDuration::from_millis(50)).await; diff --git a/datadog-sidecar/src/setup/mod.rs b/datadog-sidecar/src/setup/mod.rs index 6e69f4276c..3617d9589f 100644 --- a/datadog-sidecar/src/setup/mod.rs +++ b/datadog-sidecar/src/setup/mod.rs @@ -36,3 +36,13 @@ pub trait Liaison: Sized { fn ipc_shared() -> Self; fn ipc_per_process() -> Self; } + +/// The liaison a subprocess sidecar listens on / connects to for the given `ipc_mode`. Shared by +/// `start_or_connect_to_sidecar` and the crashtracker collector (`crashtracker_ipc_socket_path`) so +/// both always target the same socket. (Thread mode keys the socket by the master PID instead.) +pub fn liaison_for_ipc_mode(ipc_mode: crate::config::IpcMode) -> DefaultLiason { + match ipc_mode { + crate::config::IpcMode::Shared => DefaultLiason::ipc_shared(), + crate::config::IpcMode::InstancePerProcess => DefaultLiason::ipc_per_process(), + } +} diff --git a/datadog-sidecar/src/setup/thread_listener.rs b/datadog-sidecar/src/setup/thread_listener.rs index 26b94027b3..e1bd959ca7 100644 --- a/datadog-sidecar/src/setup/thread_listener.rs +++ b/datadog-sidecar/src/setup/thread_listener.rs @@ -195,7 +195,6 @@ fn run_listener(pid: u32, _config: Config, shutdown_rx: oneshot::Receiver<()>) - let cancel = || {}; let loop_config = MainLoopConfig { enable_ctrl_c_handler: false, - enable_crashtracker: false, external_shutdown_rx: None, // Defer SHM init to first connection so we can fchown using the worker's UID. init_shm_eagerly: false, diff --git a/datadog-sidecar/src/setup/thread_listener_windows.rs b/datadog-sidecar/src/setup/thread_listener_windows.rs index 333164c564..32b16a1c28 100644 --- a/datadog-sidecar/src/setup/thread_listener_windows.rs +++ b/datadog-sidecar/src/setup/thread_listener_windows.rs @@ -139,7 +139,6 @@ fn run_listener_windows( let loop_config = MainLoopConfig { enable_ctrl_c_handler: false, - enable_crashtracker: false, external_shutdown_rx: None, init_shm_eagerly: true, }; diff --git a/datadog-sidecar/src/setup/unix.rs b/datadog-sidecar/src/setup/unix.rs index 9859230b93..3f240757da 100644 --- a/datadog-sidecar/src/setup/unix.rs +++ b/datadog-sidecar/src/setup/unix.rs @@ -121,6 +121,11 @@ impl SharedDirLiaison { lock_path, } } + + /// The filesystem socket path this liaison binds/connects to. + pub fn socket_path(&self) -> &Path { + &self.socket_path + } } impl Default for SharedDirLiaison { @@ -187,6 +192,14 @@ mod linux { )); Self { path } } + + /// The abstract socket name this liaison binds/connects to. + /// + /// Exposed so the crashtracker collector (in another process) can target the exact same + /// IPC socket the sidecar listens on. + pub fn path(&self) -> &std::path::Path { + &self.path + } } impl Default for AbstractUnixSocketLiaison { diff --git a/datadog-sidecar/src/unix.rs b/datadog-sidecar/src/unix.rs index a7b9359be0..39451b9c78 100644 --- a/datadog-sidecar/src/unix.rs +++ b/datadog-sidecar/src/unix.rs @@ -217,8 +217,12 @@ fn shutdown_appsec() -> bool { true } +/// Allow initializing crashtracker independently for thread-mode sidecar. #[cfg(target_os = "linux")] -fn init_crashtracker(dependency_paths: Option<*const *const libc::c_char>) -> anyhow::Result<()> { +pub fn build_crashtracker_receiver_config( + dependency_paths: Option<*const *const libc::c_char>, + output: Option, +) -> anyhow::Result { let entrypoint = entrypoint!(ddog_crashtracker_entry_point); let entrypoint_path = match unsafe { get_dl_path_raw(entrypoint.ptr as *const libc::c_void) } { (Some(path), _) => path, @@ -257,12 +261,24 @@ fn init_crashtracker(dependency_paths: Option<*const *const libc::c_char>) -> an } } + CrashtrackerReceiverConfig::new( + receiver_args, + receiver_env, + format!("/proc/{}/exe", unsafe { libc::getpid() }), + output, + None, + ) +} + +#[cfg(target_os = "linux")] +fn init_crashtracker(dependency_paths: Option<*const *const libc::c_char>) -> anyhow::Result<()> { let output = match &Config::get().log_method { LogMethod::Stdout => Some(format!("/proc/{}/fd/1", unsafe { libc::getpid() })), LogMethod::Stderr => Some(format!("/proc/{}/fd/2", unsafe { libc::getpid() })), LogMethod::File(file) => file.to_str().map(|s| s.to_string()), LogMethod::Disabled => None, }; + let receiver_config = build_crashtracker_receiver_config(dependency_paths, output)?; let mut config_builder = CrashtrackerConfiguration::builder() .create_alt_stack(true) @@ -291,13 +307,7 @@ fn init_crashtracker(dependency_paths: Option<*const *const libc::c_char>) -> an libdd_crashtracker::init( config_builder.build()?, - CrashtrackerReceiverConfig::new( - receiver_args, - receiver_env, - format!("/proc/{}/exe", unsafe { libc::getpid() }), - output, - None, - )?, + receiver_config, Metadata::new( "libdatadog".to_string(), crate::sidecar_version!().to_string(), diff --git a/libdd-crashtracker/src/collector/receiver_manager.rs b/libdd-crashtracker/src/collector/receiver_manager.rs index 4701c8d72c..28a1951940 100644 --- a/libdd-crashtracker/src/collector/receiver_manager.rs +++ b/libdd-crashtracker/src/collector/receiver_manager.rs @@ -9,7 +9,6 @@ use libdd_common::unix_utils::{alt_fork, open_file_or_quiet, terminate, Prepared use nix::sys::signal::{self, SaFlags, SigAction, SigHandler, SigSet}; use nix::sys::socket; use std::os::unix::io::{IntoRawFd, RawFd}; -use std::os::unix::net::UnixStream; use std::ptr; use std::sync::atomic::AtomicPtr; use std::sync::atomic::Ordering::SeqCst; @@ -40,27 +39,22 @@ pub(crate) struct Receiver { } impl Receiver { - pub(crate) fn from_socket(unix_socket_path: &str) -> Result { + fn from_connector( + unix_socket_path: &str, + connector: fn(&str) -> RawFd, + ) -> Result { // Creates a fake "Receiver", which can be waited on like a normal receiver. // This is intended to support configurations where the collector is speaking to a // long-lived, async receiver process. if unix_socket_path.is_empty() { return Err(ReceiverError::NoReceiverPath); } - #[cfg(target_os = "linux")] - let unix_stream = if unix_socket_path.starts_with(['.', '/']) { - UnixStream::connect(unix_socket_path) - } else { - use std::os::linux::net::SocketAddrExt; - let addr = std::os::unix::net::SocketAddr::from_abstract_name(unix_socket_path) - .map_err(ReceiverError::ConnectionError)?; - UnixStream::connect_addr(&addr) - }; - #[cfg(not(target_os = "linux"))] - let unix_stream = UnixStream::connect(unix_socket_path); - let uds_fd = unix_stream - .map_err(ReceiverError::ConnectionError)? - .into_raw_fd(); + let uds_fd = connector(unix_socket_path); + if uds_fd < 0 { + return Err(ReceiverError::ConnectionError( + std::io::Error::last_os_error(), + )); + } Ok(Self { handle: ProcessHandle::new(uds_fd, None), }) @@ -115,7 +109,7 @@ impl Receiver { if unix_socket_path.is_empty() { Self::spawn_from_stored_config() } else { - Self::from_socket(unix_socket_path) + Self::from_connector(unix_socket_path, config.unix_socket_connector()) } } diff --git a/libdd-crashtracker/src/lib.rs b/libdd-crashtracker/src/lib.rs index 475664297d..6681be875b 100644 --- a/libdd-crashtracker/src/lib.rs +++ b/libdd-crashtracker/src/lib.rs @@ -88,8 +88,9 @@ pub use runtime_callback::*; #[cfg(all(unix, feature = "receiver"))] pub use receiver::{ - async_receiver_entry_point_unix_listener, async_receiver_entry_point_unix_socket, - get_receiver_unix_socket, receiver_entry_point_stdin, receiver_entry_point_unix_socket, + async_receiver_entry_point_stream, async_receiver_entry_point_unix_listener, + async_receiver_entry_point_unix_socket, get_receiver_unix_socket, receiver_entry_point_stdin, + receiver_entry_point_unix_socket, }; #[cfg(all(unix, any(feature = "collector", feature = "receiver")))] diff --git a/libdd-crashtracker/src/receiver/entry_points.rs b/libdd-crashtracker/src/receiver/entry_points.rs index 532c580cb4..3149a95a8a 100644 --- a/libdd-crashtracker/src/receiver/entry_points.rs +++ b/libdd-crashtracker/src/receiver/entry_points.rs @@ -34,6 +34,12 @@ pub async fn async_receiver_entry_point_unix_listener( receiver_entry_point(receiver_timeout(), stream).await } +pub async fn async_receiver_entry_point_stream( + stream: impl AsyncBufReadExt + std::marker::Unpin, +) -> anyhow::Result<()> { + receiver_entry_point(receiver_timeout(), stream).await +} + pub async fn async_receiver_entry_point_unix_socket( socket_path: impl AsRef, one_shot: bool, diff --git a/libdd-crashtracker/src/receiver/mod.rs b/libdd-crashtracker/src/receiver/mod.rs index 1265cfca66..7d882826e3 100644 --- a/libdd-crashtracker/src/receiver/mod.rs +++ b/libdd-crashtracker/src/receiver/mod.rs @@ -4,8 +4,9 @@ mod entry_points; pub use entry_points::{ - async_receiver_entry_point_unix_listener, async_receiver_entry_point_unix_socket, - get_receiver_unix_socket, receiver_entry_point_stdin, receiver_entry_point_unix_socket, + async_receiver_entry_point_stream, async_receiver_entry_point_unix_listener, + async_receiver_entry_point_unix_socket, get_receiver_unix_socket, receiver_entry_point_stdin, + receiver_entry_point_unix_socket, }; #[cfg(target_os = "linux")] mod ptrace_collector; diff --git a/libdd-crashtracker/src/shared/configuration/builder.rs b/libdd-crashtracker/src/shared/configuration/builder.rs index c4773d0fbd..b3239bcd44 100644 --- a/libdd-crashtracker/src/shared/configuration/builder.rs +++ b/libdd-crashtracker/src/shared/configuration/builder.rs @@ -23,6 +23,8 @@ pub struct CrashtrackerConfigurationBuilder { signals: Vec, timeout: Option, unix_socket_path: Option, + #[cfg(unix)] + unix_socket_connector: Option std::os::fd::RawFd>, use_alt_stack: bool, } @@ -104,6 +106,11 @@ impl CrashtrackerConfigurationBuilder { self } + pub fn unix_socket_connector(mut self, connector: fn(&str) -> std::os::fd::RawFd) -> Self { + self.unix_socket_connector = Some(connector); + self + } + pub fn build(self) -> anyhow::Result { // Requesting to create, but not use, the altstack is considered paradoxical. anyhow::ensure!( @@ -159,6 +166,9 @@ impl CrashtrackerConfigurationBuilder { signals, timeout, unix_socket_path: self.unix_socket_path, + unix_socket_connector: self + .unix_socket_connector + .unwrap_or(super::default_unix_socket_connector), demangle_names: self.demangle_names, }) } diff --git a/libdd-crashtracker/src/shared/configuration/mod.rs b/libdd-crashtracker/src/shared/configuration/mod.rs index 96a075f79f..fa4ccdd0b2 100644 --- a/libdd-crashtracker/src/shared/configuration/mod.rs +++ b/libdd-crashtracker/src/shared/configuration/mod.rs @@ -26,6 +26,10 @@ pub enum StacktraceCollection { EnabledWithSymbolsInReceiver, } +// The derived `PartialEq` compares `unix_socket_connector`, a fn pointer. Comparing fn pointers is +// unreliable (the lint), but harmless here: nothing compares whole `CrashtrackerConfiguration` +// values, and the connector is process-local plumbing (also `#[serde(skip)]`). +#[allow(unpredictable_function_pointer_comparisons)] #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct CrashtrackerConfiguration { // Paths to any additional files to track, if any @@ -42,6 +46,8 @@ pub struct CrashtrackerConfiguration { signals: Vec, timeout: Duration, unix_socket_path: Option, + #[serde(skip, default = "default_unix_socket_connector_value")] + unix_socket_connector: fn(&str) -> std::os::fd::RawFd, use_alt_stack: bool, } @@ -49,6 +55,31 @@ pub const fn default_max_threads() -> usize { 256 } +pub fn default_unix_socket_connector(unix_socket_path: &str) -> std::os::fd::RawFd { + use std::os::fd::IntoRawFd; + use std::os::unix::net::UnixStream; + #[cfg(target_os = "linux")] + let stream = if unix_socket_path.starts_with(['.', '/']) { + UnixStream::connect(unix_socket_path) + } else { + use std::os::linux::net::SocketAddrExt; + match std::os::unix::net::SocketAddr::from_abstract_name(unix_socket_path) { + Ok(addr) => UnixStream::connect_addr(&addr), + Err(e) => Err(e), + } + }; + #[cfg(not(target_os = "linux"))] + let stream = UnixStream::connect(unix_socket_path); + match stream { + Ok(s) => s.into_raw_fd(), + Err(_) => -1, + } +} + +fn default_unix_socket_connector_value() -> fn(&str) -> std::os::fd::RawFd { + default_unix_socket_connector +} + #[derive(Debug, Clone, PartialEq, Serialize, Deserialize, Default)] pub struct CrashtrackerReceiverConfig { pub args: Vec, @@ -128,6 +159,10 @@ impl CrashtrackerConfiguration { &self.unix_socket_path } + pub fn unix_socket_connector(&self) -> fn(&str) -> std::os::fd::RawFd { + self.unix_socket_connector + } + pub fn demangle_names(&self) -> bool { self.demangle_names } @@ -161,6 +196,10 @@ impl CrashtrackerConfiguration { pub fn set_unix_socket_path(&mut self, path: String) { self.unix_socket_path = Some(path); } + + pub fn set_unix_socket_connector(&mut self, connector: fn(&str) -> std::os::fd::RawFd) { + self.unix_socket_connector = connector; + } } #[cfg(test)] From 91dde1594e57570f834dd326a45a7df1754cf52b Mon Sep 17 00:00:00 2001 From: Bob Weinand Date: Fri, 3 Jul 2026 15:25:42 +0200 Subject: [PATCH 2/4] Use single buffer Signed-off-by: Bob Weinand --- datadog-sidecar/src/crashtracker.rs | 35 ++++++++++++++++++----------- 1 file changed, 22 insertions(+), 13 deletions(-) diff --git a/datadog-sidecar/src/crashtracker.rs b/datadog-sidecar/src/crashtracker.rs index e14088f968..30732a1fc5 100644 --- a/datadog-sidecar/src/crashtracker.rs +++ b/datadog-sidecar/src/crashtracker.rs @@ -110,6 +110,8 @@ mod adapter { /// buffered and drained on later reads, so boundaries never truncate data. pub struct SeqpacketStreamReader<'a> { conn: &'a AsyncConn, + /// Scratch buffer for one `recv()`, avoiding reallocation. + recv_buf: Vec, /// Leftover bytes from a datagram that did not fit in the caller's buffer. leftover: Vec, leftover_pos: usize, @@ -120,6 +122,13 @@ mod adapter { pub fn new(conn: &'a AsyncConn) -> Self { Self { conn, + recv_buf: { + let size = max_message_size(); + let mut vec = Vec::with_capacity(size); + // SAFETY: We only read bytes written by recv() + unsafe { vec.set_len(size); } + vec + }, leftover: Vec::new(), leftover_pos: 0, eof: false, @@ -159,36 +168,36 @@ mod adapter { Poll::Pending => return Poll::Pending, }; - // One recv() returns one datagram; a max-sized buffer avoids truncation. + let recv_buf = &mut this.recv_buf; let read_result = guard.try_io(|inner| { - let mut tmp = vec![0u8; max_message_size()]; let n = unsafe { libc::recv( inner.as_raw_fd(), - tmp.as_mut_ptr() as *mut libc::c_void, - tmp.len(), + recv_buf.as_mut_ptr() as *mut libc::c_void, + recv_buf.len(), 0, ) }; if n < 0 { Err(io::Error::last_os_error()) } else { - tmp.truncate(n as usize); - Ok(tmp) + Ok(n as usize) } }); match read_result { - Ok(Ok(payload)) => { - if payload.is_empty() { + Ok(Ok(n)) => { + if n == 0 { this.eof = true; return Poll::Ready(Ok(())); } - let n = payload.len().min(out.remaining()); - out.put_slice(&payload[..n]); - if n < payload.len() { - this.leftover = payload; - this.leftover_pos = n; + let payload = &this.recv_buf[..n]; + let copied = payload.len().min(out.remaining()); + out.put_slice(&payload[..copied]); + if copied < payload.len() { + this.leftover.clear(); + this.leftover.extend_from_slice(&payload[copied..]); + this.leftover_pos = 0; } return Poll::Ready(Ok(())); } From c8a4dcfedecfd49da2e7baa5628ba76b6579b977 Mon Sep 17 00:00:00 2001 From: Bob Weinand Date: Fri, 3 Jul 2026 18:55:36 +0200 Subject: [PATCH 3/4] Drop useless Option<> Signed-off-by: Bob Weinand --- datadog-sidecar/src/crashtracker.rs | 4 +++- datadog-sidecar/src/service/sidecar_server.rs | 13 +++++-------- 2 files changed, 8 insertions(+), 9 deletions(-) diff --git a/datadog-sidecar/src/crashtracker.rs b/datadog-sidecar/src/crashtracker.rs index 30732a1fc5..8491599613 100644 --- a/datadog-sidecar/src/crashtracker.rs +++ b/datadog-sidecar/src/crashtracker.rs @@ -126,7 +126,9 @@ mod adapter { let size = max_message_size(); let mut vec = Vec::with_capacity(size); // SAFETY: We only read bytes written by recv() - unsafe { vec.set_len(size); } + unsafe { + vec.set_len(size); + } vec }, leftover: Vec::new(), diff --git a/datadog-sidecar/src/service/sidecar_server.rs b/datadog-sidecar/src/service/sidecar_server.rs index 105e2ddaee..9bed032fab 100644 --- a/datadog-sidecar/src/service/sidecar_server.rs +++ b/datadog-sidecar/src/service/sidecar_server.rs @@ -135,20 +135,20 @@ struct ConnectionSidecarHandler { } impl ConnectionSidecarHandler { - fn new(server: SidecarServer, connection: OwnedServerConn) -> Option { + fn new(server: SidecarServer, connection: OwnedServerConn) -> Self { let submitted_payloads = Arc::new(AtomicU64::new(0)); server .connection_counters .lock_or_panic() .push(Arc::downgrade(&submitted_payloads)); - Some(Self { + Self { server, submitted_payloads, session_id: Default::default(), instances: Default::default(), metric_registrations: Default::default(), connection, - }) + } } fn track_instance(&self, instance_id: &InstanceId) { @@ -208,10 +208,7 @@ impl SidecarServer { return; } }; - let Some(handler) = ConnectionSidecarHandler::new(self, server_conn) else { - return; - }; - let handler = Arc::new(handler); + let handler = Arc::new(ConnectionSidecarHandler::new(self, server_conn)); let handler_for_cleanup = handler.clone(); serve_sidecar_interface_connection(handler).await; handler_for_cleanup.cleanup().await; @@ -1192,7 +1189,7 @@ mod tests { let (local, peer) = SeqpacketConn::socketpair().expect("socketpair"); drop(peer); let conn = OwnedServerConn::new(local).expect("OwnedServerConn"); - ConnectionSidecarHandler::new(server, conn).expect("handler") + ConnectionSidecarHandler::new(server, conn) } fn ffe_context() -> FfeTelemetryContext { From ea308fb9e5fd20b24e6ec5df7b452c9555601063 Mon Sep 17 00:00:00 2001 From: Bob Weinand Date: Fri, 3 Jul 2026 19:10:49 +0200 Subject: [PATCH 4/4] Clippy Signed-off-by: Bob Weinand --- datadog-sidecar/src/crashtracker.rs | 40 ++++++------------- .../src/shared/configuration/mod.rs | 22 +++++++--- 2 files changed, 29 insertions(+), 33 deletions(-) diff --git a/datadog-sidecar/src/crashtracker.rs b/datadog-sidecar/src/crashtracker.rs index 8491599613..405eb48e31 100644 --- a/datadog-sidecar/src/crashtracker.rs +++ b/datadog-sidecar/src/crashtracker.rs @@ -112,9 +112,7 @@ mod adapter { conn: &'a AsyncConn, /// Scratch buffer for one `recv()`, avoiding reallocation. recv_buf: Vec, - /// Leftover bytes from a datagram that did not fit in the caller's buffer. - leftover: Vec, - leftover_pos: usize, + buf_pos: usize, eof: bool, } @@ -122,17 +120,8 @@ mod adapter { pub fn new(conn: &'a AsyncConn) -> Self { Self { conn, - recv_buf: { - let size = max_message_size(); - let mut vec = Vec::with_capacity(size); - // SAFETY: We only read bytes written by recv() - unsafe { - vec.set_len(size); - } - vec - }, - leftover: Vec::new(), - leftover_pos: 0, + recv_buf: Vec::with_capacity(max_message_size()), + buf_pos: 0, eof: false, } } @@ -147,15 +136,11 @@ mod adapter { let this = self.get_mut(); // Drain any buffered leftover first. - if this.leftover_pos < this.leftover.len() { - let remaining = &this.leftover[this.leftover_pos..]; + if this.buf_pos < this.recv_buf.len() { + let remaining = &this.recv_buf[this.buf_pos..]; let n = remaining.len().min(out.remaining()); out.put_slice(&remaining[..n]); - this.leftover_pos += n; - if this.leftover_pos >= this.leftover.len() { - this.leftover.clear(); - this.leftover_pos = 0; - } + this.buf_pos += n; return Poll::Ready(Ok(())); } @@ -176,7 +161,7 @@ mod adapter { libc::recv( inner.as_raw_fd(), recv_buf.as_mut_ptr() as *mut libc::c_void, - recv_buf.len(), + recv_buf.capacity(), 0, ) }; @@ -193,14 +178,13 @@ mod adapter { this.eof = true; return Poll::Ready(Ok(())); } - let payload = &this.recv_buf[..n]; + unsafe { + this.recv_buf.set_len(n); + } + let payload = this.recv_buf.as_slice(); let copied = payload.len().min(out.remaining()); out.put_slice(&payload[..copied]); - if copied < payload.len() { - this.leftover.clear(); - this.leftover.extend_from_slice(&payload[copied..]); - this.leftover_pos = 0; - } + this.buf_pos = copied; return Poll::Ready(Ok(())); } Ok(Err(e)) => { diff --git a/libdd-crashtracker/src/shared/configuration/mod.rs b/libdd-crashtracker/src/shared/configuration/mod.rs index fa4ccdd0b2..4ec65a329e 100644 --- a/libdd-crashtracker/src/shared/configuration/mod.rs +++ b/libdd-crashtracker/src/shared/configuration/mod.rs @@ -26,11 +26,7 @@ pub enum StacktraceCollection { EnabledWithSymbolsInReceiver, } -// The derived `PartialEq` compares `unix_socket_connector`, a fn pointer. Comparing fn pointers is -// unreliable (the lint), but harmless here: nothing compares whole `CrashtrackerConfiguration` -// values, and the connector is process-local plumbing (also `#[serde(skip)]`). -#[allow(unpredictable_function_pointer_comparisons)] -#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[derive(Debug, Clone, Serialize, Deserialize)] pub struct CrashtrackerConfiguration { // Paths to any additional files to track, if any additional_files: Vec, @@ -51,6 +47,22 @@ pub struct CrashtrackerConfiguration { use_alt_stack: bool, } +impl PartialEq for CrashtrackerConfiguration { + fn eq(&self, other: &Self) -> bool { + self.additional_files == other.additional_files + && self.collect_all_threads == other.collect_all_threads + && self.create_alt_stack == other.create_alt_stack + && self.demangle_names == other.demangle_names + && self.endpoint == other.endpoint + && self.max_threads == other.max_threads + && self.resolve_frames == other.resolve_frames + && self.signals == other.signals + && self.timeout == other.timeout + && self.unix_socket_path == other.unix_socket_path + && self.use_alt_stack == other.use_alt_stack + } +} + pub const fn default_max_threads() -> usize { 256 }