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..405eb48e31 100644 --- a/datadog-sidecar/src/crashtracker.rs +++ b/datadog-sidecar/src/crashtracker.rs @@ -1,18 +1,345 @@ // 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, + /// Scratch buffer for one `recv()`, avoiding reallocation. + recv_buf: Vec, + buf_pos: usize, + eof: bool, + } + + impl<'a> SeqpacketStreamReader<'a> { + pub fn new(conn: &'a AsyncConn) -> Self { + Self { + conn, + recv_buf: Vec::with_capacity(max_message_size()), + buf_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.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.buf_pos += n; + 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, + }; + + let recv_buf = &mut this.recv_buf; + let read_result = guard.try_io(|inner| { + let n = unsafe { + libc::recv( + inner.as_raw_fd(), + recv_buf.as_mut_ptr() as *mut libc::c_void, + recv_buf.capacity(), + 0, + ) + }; + if n < 0 { + Err(io::Error::last_os_error()) + } else { + Ok(n as usize) + } + }); + + match read_result { + Ok(Ok(n)) => { + if n == 0 { + this.eof = true; + return Poll::Ready(Ok(())); + } + 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]); + this.buf_pos = copied; + 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..9bed032fab 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,10 +130,12 @@ 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) -> Self { let submitted_payloads = Arc::new(AtomicU64::new(0)); server .connection_counters @@ -145,6 +147,7 @@ impl ConnectionSidecarHandler { session_id: Default::default(), instances: Default::default(), metric_registrations: Default::default(), + connection, } } @@ -198,9 +201,16 @@ 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 handler = Arc::new(ConnectionSidecarHandler::new(self, server_conn)); 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 +410,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 +658,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 +667,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 +676,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 +695,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 +706,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 +834,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 +845,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 +856,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 +867,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 +880,6 @@ impl SidecarInterface for ConnectionSidecarHandler { async fn send_trace_v04_shm( &self, - _peer: PeerCredentials, instance_id: InstanceId, handle: ShmHandle, _len: usize, @@ -897,7 +910,6 @@ impl SidecarInterface for ConnectionSidecarHandler { async fn send_trace_v04_bytes( &self, - _peer: PeerCredentials, instance_id: InstanceId, data: Vec, headers: SerializedTracerHeaderTags, @@ -923,7 +935,6 @@ impl SidecarInterface for ConnectionSidecarHandler { async fn send_debugger_data_shm( &self, - _peer: PeerCredentials, instance_id: InstanceId, queue_id: QueueId, handle: ShmHandle, @@ -946,7 +957,6 @@ impl SidecarInterface for ConnectionSidecarHandler { async fn send_debugger_diagnostics( &self, - _peer: PeerCredentials, instance_id: InstanceId, queue_id: QueueId, diagnostics_payload: Vec, @@ -974,7 +984,6 @@ impl SidecarInterface for ConnectionSidecarHandler { async fn acquire_exception_hash_rate_limiter( &self, - _peer: PeerCredentials, exception_hash: u64, granularity: Duration, ) { @@ -986,7 +995,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 +1027,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 +1051,6 @@ impl SidecarInterface for ConnectionSidecarHandler { async fn send_dogstatsd_actions( &self, - _peer: PeerCredentials, instance_id: InstanceId, actions: Vec, ) { @@ -1061,7 +1067,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 +1087,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 +1128,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 +1162,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 +1182,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) + } + fn ffe_context() -> FfeTelemetryContext { FfeTelemetryContext { service: "svc".to_owned(), @@ -1218,7 +1233,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 +1256,6 @@ mod tests { handler .enqueue_actions( - PeerCredentials::default(), instance_id.clone(), queue_id, vec![SidecarAction::FfeExposureBatch(FfeExposureBatch { @@ -1280,7 +1294,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 +1317,6 @@ mod tests { handler .enqueue_actions( - PeerCredentials::default(), instance_id.clone(), queue_id, vec![SidecarAction::FfeEvaluationMetrics { @@ -1346,7 +1359,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 +1388,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..4ec65a329e 100644 --- a/libdd-crashtracker/src/shared/configuration/mod.rs +++ b/libdd-crashtracker/src/shared/configuration/mod.rs @@ -26,7 +26,7 @@ pub enum StacktraceCollection { EnabledWithSymbolsInReceiver, } -#[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, @@ -42,13 +42,56 @@ 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, } +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 } +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 +171,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 +208,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)]