internal(prof): configurable upload compression

profiling/Cargo.toml

@@ -19,10 +19,10 @@ cfg-if = { version = "1.0" }
 cpu-time = { version = "1.0" }
 chrono = { version = "0.4" }
 crossbeam-channel = { version = "0.5", default-features = false, features = ["std"] }
-datadog-alloc = { git = "https://github.com/DataDog/libdatadog", tag = "v18.0.0" }
-datadog-profiling = { git = "https://github.com/DataDog/libdatadog", tag = "v18.0.0" }
-ddcommon = { git = "https://github.com/DataDog/libdatadog", tag = "v18.0.0" }
-datadog-library-config-ffi = { git = "https://github.com/DataDog/libdatadog", tag = "v18.0.0" }
+datadog-alloc = { git = "https://github.com/DataDog/libdatadog", rev = "0d6cabcba390128ae6ccdb47145c5b1b6d4c3975" }
+datadog-profiling = { git = "https://github.com/DataDog/libdatadog", rev = "0d6cabcba390128ae6ccdb47145c5b1b6d4c3975" }
+ddcommon = { git = "https://github.com/DataDog/libdatadog", rev = "0d6cabcba390128ae6ccdb47145c5b1b6d4c3975" }
+datadog-library-config-ffi = { git = "https://github.com/DataDog/libdatadog", rev = "0d6cabcba390128ae6ccdb47145c5b1b6d4c3975" }
 env_logger = { version = "0.11", default-features = false }
 indexmap = { version = "2.2" }
 lazy_static = { version = "1.4" }

profiling/src/config.rs

@@ -33,6 +33,7 @@ pub struct SystemSettings {
     pub profiling_exception_message_enabled: bool,
     pub profiling_wall_time_enabled: bool,
     pub profiling_io_enabled: bool,
+    pub profiling_debug_upload_compression: Cow<'static, str>,
 
     // todo: can't this be Option<String>? I don't think the string can ever be static.
     pub output_pprof: Option<Cow<'static, str>>,
@@ -76,6 +77,7 @@ impl SystemSettings {
             profiling_exception_message_enabled: profiling_exception_message_enabled(),
             profiling_wall_time_enabled: profiling_wall_time_enabled(),
             profiling_io_enabled: profiling_io_enabled(),
+            profiling_debug_upload_compression: profiling_upload_compression(),
             output_pprof: profiling_output_pprof(),
             profiling_exception_sampling_distance: profiling_exception_sampling_distance(),
             profiling_log_level: profiling_log_level(),
@@ -150,6 +152,7 @@ impl SystemSettings {
             profiling_exception_message_enabled: false,
             profiling_wall_time_enabled: false,
             profiling_io_enabled: false,
+            profiling_debug_upload_compression: Cow::from("on"),
             output_pprof: None,
             profiling_exception_sampling_distance: 0,
             profiling_log_level: LevelFilter::Off,
@@ -356,19 +359,20 @@ unsafe fn get_system_value(id: ConfigId) -> &'static mut zval {
 #[repr(u16)]
 #[derive(Clone, Copy)]
 pub(crate) enum ConfigId {
-    ProfilingEnabled = 0,
-    ProfilingExperimentalFeaturesEnabled,
-    ProfilingEndpointCollectionEnabled,
-    ProfilingExperimentalCpuTimeEnabled,
     ProfilingAllocationEnabled,
     ProfilingAllocationSamplingDistance,
-    ProfilingTimelineEnabled,
+    ProfilingDebugUploadCompression,
+    ProfilingEnabled,
+    ProfilingEndpointCollectionEnabled,
     ProfilingExceptionEnabled,
     ProfilingExceptionMessageEnabled,
     ProfilingExceptionSamplingDistance,
+    ProfilingExperimentalCpuTimeEnabled,
+    ProfilingExperimentalFeaturesEnabled,
     ProfilingExperimentalIOEnabled,
     ProfilingLogLevel,
     ProfilingOutputPprof,
+    ProfilingTimelineEnabled,
     ProfilingWallTimeEnabled,
 
     // todo: do these need to be kept in sync with the tracer?
@@ -402,6 +406,7 @@ impl ConfigId {
             // Note: this group is meant only for debugging and testing. Please
             // don't advertise this group of settings in the docs.
             ProfilingOutputPprof => b"DD_PROFILING_OUTPUT_PPROF\0",
+            ProfilingDebugUploadCompression => b"DD_PROFILING_DEBUG_UPLOAD_COMPRESSION\0",
             ProfilingWallTimeEnabled => b"DD_PROFILING_WALLTIME_ENABLED\0",
 
             AgentHost => b"DD_AGENT_HOST\0",
@@ -442,6 +447,7 @@ lazy_static::lazy_static! {
         profiling_exception_message_enabled: false,
         profiling_wall_time_enabled: false,
         profiling_io_enabled: false,
+        profiling_debug_upload_compression: Cow::from("on"),
         output_pprof: None,
         profiling_exception_sampling_distance: u32::MAX,
         profiling_log_level: LevelFilter::Off,
@@ -461,6 +467,7 @@ lazy_static::lazy_static! {
         profiling_exception_message_enabled: false,
         profiling_wall_time_enabled: true,
         profiling_io_enabled: false,
+        profiling_debug_upload_compression: Cow::from("on"),
         output_pprof: None,
         profiling_exception_sampling_distance: 100,
         profiling_log_level: LevelFilter::Off,
@@ -591,6 +598,13 @@ unsafe fn profiling_output_pprof() -> Option<Cow<'static, str>> {
     get_system_str(ProfilingOutputPprof)
 }
 
+unsafe fn profiling_upload_compression() -> Cow<'static, str> {
+    match get_system_str(ProfilingDebugUploadCompression) {
+        Some(str) => str,
+        None => Cow::Borrowed("on"),
+    }
+}
+
 /// # Safety
 /// This function must only be called after config has been initialized in
 /// first rinit, and before it is uninitialized in mshutdown.
@@ -782,7 +796,7 @@ unsafe extern "C" fn parse_level_filter(
 }
 
 /// This function is used to parse the profiling enabled config value.
-/// It behaves similarlry to the "zai_config_decode_bool" but also accepts "auto" as true.
+/// It behaves similarly to the "zai_config_decode_bool" but also accepts "auto" as true.
 unsafe extern "C" fn parse_profiling_enabled(
     value: ZaiStr,
     decoded_value: *mut zval,
@@ -811,6 +825,36 @@ unsafe extern "C" fn parse_profiling_enabled(
     }
 }
 
+unsafe extern "C" fn parse_profiling_upload_compression(
+    value: ZaiStr,
+    decoded_value: *mut zval,
+    persistent: bool,
+) -> bool {
+    if decoded_value.is_null() {
+        return false;
+    }
+
+    let decoded_value = &mut *decoded_value;
+    match value.into_utf8() {
+        Ok(utf8) => {
+            let view = if utf8.eq_ignore_ascii_case("off") {
+                ZaiStr::literal(b"off\0")
+            } else if utf8.eq_ignore_ascii_case("on") {
+                ZaiStr::literal(b"on\0")
+            } else if utf8.eq_ignore_ascii_case("lz4") {
+                ZaiStr::literal(b"lz4\0")
+            } else if utf8.eq_ignore_ascii_case("zstd") {
+                ZaiStr::literal(b"zstd\0")
+            } else {
+                return false;
+            };
+            datadog_php_profiling_copy_string_view_into_zval(decoded_value, view, persistent);
+            true
+        }
+        _ => false,
+    }
+}
+
 /// Display the profiling enabled config value
 unsafe extern "C" fn display_profiling_enabled(ini_entry: *mut zend_ini_entry, type_: c_int) {
     let tmp_value: *mut zend_string =
@@ -1061,6 +1105,18 @@ pub(crate) fn minit(module_number: libc::c_int) {
                     displayer: None,
                     env_config_fallback: None,
                 },
+                zai_config_entry {
+                    id: transmute::<ConfigId, u16>(ProfilingDebugUploadCompression),
+                    name: ProfilingOutputPprof.env_var_name(),
+                    type_: ZAI_CONFIG_TYPE_STRING,
+                    default_encoded_value: ZaiStr::literal(b"on\0"),
+                    aliases: ptr::null_mut(),
+                    aliases_count: 0,
+                    ini_change: Some(zai_config_system_ini_change),
+                    parser: Some(parse_profiling_upload_compression),
+                    displayer: None,
+                    env_config_fallback: None,
+                },
                 // At the moment, wall-time cannot be fully disabled. This only
                 // controls automatic collection (manual collection is still
                 // possible).
@@ -1232,6 +1288,10 @@ mod tests {
                 b"DD_PROFILING_ALLOCATION_ENABLED\0",
                 "datadog.profiling.allocation_enabled",
             ),
+            (
+                b"DD_PROFILING_DEBUG_UPLOAD_COMPRESSION\0",
+                "datadog.profiling.debug_upload_compression",
+            ),
             #[cfg(feature = "timeline")]
             (
                 b"DD_PROFILING_EXPERIMENTAL_TIMELINE_ENABLED\0",

profiling/src/profiling/mod.rs

@@ -53,15 +53,15 @@ use crate::io::{
     SOCKET_WRITE_SIZE_PROFILING_INTERVAL, SOCKET_WRITE_TIME_PROFILING_INTERVAL,
 };
 
+#[cfg(feature = "exception_profiling")]
+use crate::exception::EXCEPTION_PROFILING_INTERVAL;
 #[cfg(any(
     feature = "allocation_profiling",
     feature = "exception_profiling",
     feature = "io_profiling"
 ))]
 use datadog_profiling::api::UpscalingInfo;
-
-#[cfg(feature = "exception_profiling")]
-use crate::exception::EXCEPTION_PROFILING_INTERVAL;
+use datadog_profiling::serializer::UploadCompression;
 
 const UPLOAD_PERIOD: Duration = Duration::from_secs(67);
 
@@ -700,10 +700,23 @@ impl Profiler {
             upload_period: UPLOAD_PERIOD,
         };
 
+        let compression_type = system_settings.profiling_debug_upload_compression.as_ref();
+        let upload_compression = match compression_type {
+            "off" => UploadCompression::Off,
+            "on" => UploadCompression::On,
+            "lz4" => UploadCompression::Lz4,
+            "zstd" => UploadCompression::Zstd,
+            _ => {
+                warn!("unknown profiling upload compression type \"{compression_type}\", defaulting to on");
+                UploadCompression::On
+            }
+        };
+
         let uploader = Uploader::new(
             fork_barrier.clone(),
             upload_receiver,
             system_settings.output_pprof.clone(),
+            upload_compression,
             system_settings.uri.clone(),
             Utc::now(),
         );
@@ -1583,6 +1596,7 @@ mod tests {
             profiling_exception_message_enabled: false,
             profiling_wall_time_enabled: true,
             profiling_io_enabled: false,
+            profiling_debug_upload_compression: Cow::from("on"),
             output_pprof: None,
             profiling_exception_sampling_distance: 100,
             profiling_log_level: LevelFilter::Off,

profiling/src/profiling/uploader.rs

@@ -7,6 +7,7 @@ use crate::profiling::{UploadMessage, UploadRequest};
 use crate::{PROFILER_NAME_STR, PROFILER_VERSION_STR};
 use chrono::{DateTime, Utc};
 use crossbeam_channel::{select, Receiver};
+use datadog_profiling::serializer::UploadCompression;
 use ddcommon::Endpoint;
 use log::{debug, info, warn};
 use serde_json::json;
@@ -25,6 +26,7 @@ pub struct Uploader {
     fork_barrier: Arc<Barrier>,
     receiver: Receiver<UploadMessage>,
     output_pprof: Option<Cow<'static, str>>,
+    upload_compression: UploadCompression,
     endpoint: AgentEndpoint,
     start_time: String,
 }
@@ -34,13 +36,15 @@ impl Uploader {
         fork_barrier: Arc<Barrier>,
         receiver: Receiver<UploadMessage>,
         output_pprof: Option<Cow<'static, str>>,
+        upload_compression: UploadCompression,
         endpoint: AgentEndpoint,
         start_time: DateTime<Utc>,
     ) -> Self {
         Self {
             fork_barrier,
             receiver,
             output_pprof,
+            upload_compression,
             endpoint,
             start_time: start_time.to_rfc3339_opts(chrono::SecondsFormat::Secs, true),
         }
@@ -98,8 +102,11 @@ impl Uploader {
             endpoint,
         )?;
 
-        let serialized =
-            profile.serialize_into_compressed_pprof(Some(message.end_time), message.duration)?;
+        let serialized = profile.serialize_into_compressed_pprof(
+            Some(message.end_time),
+            message.duration,
+            self.upload_compression,
+        )?;
         exporter.set_timeout(10000); // 10 seconds in milliseconds
         let request = exporter.build(
             serialized,
@@ -121,6 +128,8 @@ impl Uploader {
         let pprof_filename = &self.output_pprof;
         let mut i = 0;
 
+        let upload_compression = self.upload_compression;
+
         loop {
             /* Since profiling uploads are going over the Internet and not just
              * the local network, it would be ideal if they were the lowest
@@ -140,7 +149,7 @@ impl Uploader {
                         match pprof_filename {
                             Some(filename) => {
                                 let filename_prefix = filename.as_ref();
-                                let r = request.profile.serialize_into_compressed_pprof(None, None).unwrap();
+                                let r = request.profile.serialize_into_compressed_pprof(None, None, upload_compression).unwrap();
                                 i += 1;
                                 let name = format!("{filename_prefix}.{i}.lz4");
                                 std::fs::write(&name, r.buffer).expect("write to succeed");