Fix ftruncate() causing SIGBUS on /dev/shm full

Add dogstatsd stats
Add hostname
Reduce flush mutex contention

Signed-off-by: Bob Weinand <bob.weinand@datadoghq.com>

Author: bwoebi

Cargo.lock

@@ -7,6 +7,7 @@ use crate::platform::{mmap_handle, munmap_handle, OwnedFileHandle, PlatformHandl
 use libdd_tinybytes::UnderlyingBytes;
 use serde::{Deserialize, Serialize};
 use std::{ffi::CString, io, ptr::NonNull};
+use std::os::fd::AsRawFd;
 
 #[derive(Clone, Serialize, Deserialize, Debug)]
 pub struct ShmHandle {
@@ -87,10 +88,14 @@ where
         unsafe {
             self.set_mapping_size(size)?;
         }
-        nix::unistd::ftruncate(
-            self.get_shm().handle.as_owned_fd()?,
-            self.get_shm().size as libc::off_t,
-        )?;
+        let new_size = self.get_shm().size as libc::off_t;
+        let fd = self.get_shm().handle.as_owned_fd()?;
+        // Use fallocate on Linux to eagerly commit the new pages: ENOSPC at resize time is
+        // recoverable; a later SIGBUS mid-execution is not.
+        #[cfg(target_os = "linux")]
+        nix::fcntl::fallocate(fd.as_raw_fd(), nix::fcntl::FallocateFlags::empty(), 0, new_size)?;
+        #[cfg(not(target_os = "linux"))]
+        nix::unistd::ftruncate(&fd, new_size)?;
         Ok(())
     }
     /// # Safety

datadog-ipc/src/platform/mem_handle.rs

@@ -8,6 +8,8 @@ use io_lifetimes::OwnedFd;
 use libc::{chmod, off_t};
 use nix::errno::Errno;
 use nix::fcntl::{open, OFlag};
+#[cfg(target_os = "linux")]
+use nix::fcntl::{fallocate, FallocateFlags};
 use nix::sys::mman::{self, mmap, munmap, MapFlags, ProtFlags};
 use nix::sys::stat::Mode;
 use nix::unistd::{fchown, ftruncate, mkdir, unlink, Uid};
@@ -163,6 +165,11 @@ impl NamedShmHandle {
 
     pub fn create_mode(path: CString, size: usize, mode: Mode) -> io::Result<NamedShmHandle> {
         let fd = shm_open(path.as_bytes(), OFlag::O_CREAT | OFlag::O_RDWR, mode)?;
+        // Use fallocate on Linux to eagerly commit pages: if /dev/shm is full we get ENOSPC
+        // here (recoverable) rather than SIGBUS mid-execution when a worker writes a slot.
+        #[cfg(target_os = "linux")]
+        fallocate(fd.as_raw_fd(), FallocateFlags::empty(), 0, size as off_t)?;
+        #[cfg(not(target_os = "linux"))]
         ftruncate(&fd, size as off_t)?;
         if let Some(uid) = shm_owner_uid() {
             let _ = fchown(fd.as_raw_fd(), Some(Uid::from_raw(uid)), None);

datadog-ipc/src/platform/unix/mem_handle.rs

@@ -619,13 +619,11 @@ impl ShmSpanConcentrator {
     pub fn flush(
         &self,
         force: bool,
-        hostname: &str,
-        env: &str,
-        version: &str,
-        service: &str,
-        tracer_version: &str,
-        runtime_id: &str,
-        container_id: &str,
+        hostname: String,
+        env: String,
+        version: String,
+        service: String,
+        runtime_id: String,
     ) -> Option<pb::ClientStatsPayload> {
         let stat_buckets = self.drain_buckets(force);
         if stat_buckets.is_empty() {
@@ -634,15 +632,12 @@ impl ShmSpanConcentrator {
 
         let seq = self.header().flush_seq.fetch_add(1, Relaxed);
         Some(pb::ClientStatsPayload {
-            hostname: hostname.to_owned(),
-            env: env.to_owned(),
-            version: version.to_owned(),
+            hostname,
+            env,
+            version,
             stats: stat_buckets,
-            lang: "php".to_owned(),
-            tracer_version: tracer_version.to_owned(),
-            runtime_id: runtime_id.to_owned(),
-            service: service.to_owned(),
-            container_id: container_id.to_owned(),
+            runtime_id,
+            service,
             sequence: seq,
             ..Default::default()
         })
@@ -796,7 +791,7 @@ impl ShmSpanConcentrator {
                 .grpc_status_code
                 .map(|c| c.to_string())
                 .unwrap_or_default(),
-            service_source: String::new(),
+            service_source: read_str!(f.service_source),
             span_derived_primary_tags: vec![],
         }
     }
@@ -858,7 +853,7 @@ mod tests {
         .unwrap();
         c.add_span(&span("svc", "res", 1_000_000));
         c.add_span(&span("svc", "res", 2_000_000));
-        let bytes = c.flush(true, "h", "e", "v", "s", "t", "r", "c");
+        let bytes = c.flush(true, "h", "e", "v", "s", "r");
         assert!(bytes.is_some());
     }
 
@@ -875,7 +870,7 @@ mod tests {
         .unwrap();
         let worker = ShmSpanConcentrator::open(path.as_c_str()).unwrap();
         worker.add_span(&span("svc2", "res2", 5_000_000));
-        let bytes = creator.flush(true, "h", "", "", "", "t", "", "");
+        let bytes = creator.flush(true, "h", "", "", "", "r");
         assert!(bytes.is_some());
     }
 
@@ -915,7 +910,7 @@ mod tests {
             DEFAULT_STRING_POOL_BYTES,
         )
         .unwrap();
-        assert!(c.flush(false, "h", "e", "v", "s", "t", "r", "c").is_none());
+        assert!(c.flush(false, "h", "e", "v", "s", "r").is_none());
     }
 
     #[test]

datadog-ipc/src/shm_stats.rs

@@ -626,6 +626,8 @@ pub unsafe extern "C" fn ddog_sidecar_session_set_config(
     remote_config_enabled: bool,
     is_fork: bool,
     process_tags: &libdd_common_ffi::Vec<Tag>,
+    hostname: ffi::CharSlice,
+    root_service: ffi::CharSlice,
 ) -> MaybeError {
     let session_id_str: String = session_id.to_utf8_lossy().into();
     let session_config = SessionConfig {
@@ -665,6 +667,8 @@ pub unsafe extern "C" fn ddog_sidecar_session_set_config(
         process_tags: process_tags.to_vec(),
         peer_tag_keys: vec![],
         span_kinds_stats_computed: vec![],
+        hostname: hostname.to_utf8_lossy().into(),
+        root_service: root_service.to_utf8_lossy().into(),
     };
     #[cfg(unix)]
     try_c!(blocking::set_session_config(

datadog-sidecar-ffi/src/lib.rs

@@ -23,6 +23,7 @@ datadog-sidecar-macros = { path = "../datadog-sidecar-macros" }
 libdd-telemetry = { path = "../libdd-telemetry", features = ["tracing"] }
 libdd-data-pipeline = { path = "../libdd-data-pipeline" }
 libdd-trace-utils = { path = "../libdd-trace-utils" }
+libdd-trace-stats = { path = "../libdd-trace-stats" }
 datadog-remote-config = { path = "../datadog-remote-config" , features = ["live-debugger"]}
 datadog-live-debugger = { path = "../datadog-live-debugger" }
 libdd-crashtracker = { path = "../libdd-crashtracker" }

datadog-sidecar/Cargo.toml

@@ -68,6 +68,10 @@ pub struct SessionConfig {
     pub process_tags: Vec<Tag>,
     pub peer_tag_keys: Vec<String>,
     pub span_kinds_stats_computed: Vec<String>,
+    /// Tracer-configured hostname (from `DD_HOSTNAME`).  Empty means "not configured".
+    pub hostname: String,
+    /// Process-level service name (from `DD_SERVICE`), used as the stats concentrator key.
+    pub root_service: String,
 }
 
 #[derive(Debug, Deserialize, Serialize)]

datadog-sidecar/src/service/mod.rs

@@ -166,6 +166,12 @@ impl SessionInfo {
         self.dogstatsd.lock_or_panic()
     }
 
+    /// Clone the Arc wrapping the DogStatsD client so it can be shared with long-lived tasks
+    /// (e.g. the stats flush loop) without creating a new UDP socket.
+    pub(crate) fn clone_dogstatsd(&self) -> Arc<Mutex<Option<libdd_dogstatsd_client::Client>>> {
+        self.dogstatsd.clone()
+    }
+
     pub(crate) fn configure_dogstatsd<F>(&self, f: F)
     where
         F: FnOnce(&mut Option<libdd_dogstatsd_client::Client>),

datadog-sidecar/src/service/session_info.rs

@@ -35,6 +35,7 @@ use crate::service::debugger_diagnostics_bookkeeper::{
 };
 use crate::service::exception_hash_rate_limiter::EXCEPTION_HASH_LIMITER;
 use crate::service::remote_configs::{RemoteConfigNotifyTarget, RemoteConfigs};
+use crate::service::stats_flusher::{ensure_stats_concentrator, flush_all_stats_now, get_hostname, stats_endpoint, ConcentratorKey, SpanConcentratorState, StatsConfig};
 use crate::service::tracing::trace_flusher::TraceFlusherStats;
 use crate::tokio_util::run_or_spawn_shared;
 use datadog_live_debugger::sender::{agent_info_supports_debugger_v2_endpoint, DebuggerType};
@@ -103,14 +104,7 @@ pub struct SidecarServer {
     /// Diagnostics bookkeeper
     debugger_diagnostics_bookkeeper: Arc<DebuggerDiagnosticsBookkeeper>,
     /// Per-env&version SHM span concentrators (global across all sessions).
-    pub(crate) span_concentrators: Arc<
-        Mutex<
-            HashMap<
-                crate::service::stats_flusher::ConcentratorKey,
-                crate::service::stats_flusher::SpanConcentratorState,
-            >,
-        >,
-    >,
+    pub(crate) span_concentrators: Arc<Mutex<HashMap<ConcentratorKey, Arc<SpanConcentratorState>>>>,
 }
 
 /// Per-connection handler wrapper that tracks sessions/instances for cleanup on disconnect.
@@ -635,10 +629,14 @@ impl SidecarInterface for ConnectionSidecarHandler {
             });
             *session.agent_infos.lock_or_panic() = Some(agent_info);
         }
-        *session.stats_config.lock_or_panic() = Some(crate::service::stats_flusher::StatsConfig {
-            endpoint: config.endpoint.clone(),
-            tracer_version: config.tracer_version.clone(),
+        *session.stats_config.lock_or_panic() = Some(StatsConfig {
+            endpoint: stats_endpoint(&config.endpoint).unwrap_or_else(|| config.endpoint.clone()),
             flush_interval: config.flush_interval,
+            hostname: if config.hostname.is_empty() { get_hostname() } else { config.hostname.clone() },
+            process_tags: config.process_tags.iter().map(|t| t.to_string()).collect::<Vec<_>>().join(","),
+            root_service: config.root_service.clone(),
+            language: config.language.clone(),
+            tracer_version: config.tracer_version.clone(),
         });
 
         session.set_remote_config_invariants(ConfigOptions {
@@ -848,10 +846,18 @@ impl SidecarInterface for ConnectionSidecarHandler {
         debug!("Registered remote config metadata: instance {instance_id:?}, queue_id: {queue_id:?}, service: {service_name}, env: {env_name}, version: {app_version}");
 
         let session = self.server.get_session(&instance_id.session_id);
-        let concentrator_guard = crate::service::stats_flusher::ensure_stats_concentrator(
+        let concentrator_service = session
+            .stats_config
+            .lock()
+            .unwrap_or_else(|e| e.into_inner())
+            .as_ref()
+            .map(|c| c.root_service.clone())
+            .unwrap_or_default();
+        let concentrator_guard = ensure_stats_concentrator(
             &self.server.span_concentrators,
             &env_name,
             &app_version,
+            &concentrator_service,
             &instance_id.session_id,
             &session,
         );
@@ -920,7 +926,16 @@ impl SidecarInterface for ConnectionSidecarHandler {
         version: String,
         span: datadog_ipc::shm_stats::OwnedShmSpanInput,
     ) {
-        let map_key = crate::service::stats_flusher::ConcentratorKey { env, version };
+        let session_id = self.session_id.get().map(|s| s.as_str()).unwrap_or("");
+        let session = self.server.get_session(session_id);
+        let service = session
+            .stats_config
+            .lock()
+            .unwrap_or_else(|e| e.into_inner())
+            .as_ref()
+            .map(|c| c.root_service.clone())
+            .unwrap_or_default();
+        let map_key = ConcentratorKey { env, version, root_service: service };
         let guard = self
             .server
             .span_concentrators
@@ -938,7 +953,7 @@ impl SidecarInterface for ConnectionSidecarHandler {
         if let Err(e) = tokio::spawn(async move { flusher.flush().await }).await {
             error!("Failed flushing traces: {e:?}");
         }
-        crate::service::stats_flusher::flush_all_stats_now(&self.server.span_concentrators).await;
+        flush_all_stats_now(&self.server.span_concentrators).await;
     }
 
     async fn set_test_session_token(&self, _peer: PeerCredentials, token: String) {