profiling: test using reqwest instead of hyper for exporting

Author: danielsn

Cargo.lock

@@ -50,9 +50,11 @@ thiserror = "2"
 tokio = {version = "1.23", features = ["rt", "macros"]}
 tokio-util = "0.7.1"
 zstd = { version = "0.13", default-features = false }
+reqwest = { version = "0.12.23", features = ["multipart", "rustls-tls"], default-features = false }
 
 [dev-dependencies]
 bolero = "0.13"
 criterion = "0.5.1"
 lz4_flex = { version = "0.9", default-features = false, features = ["std", "frame"] }
 proptest = "1"
+wiremock = "0.5"

libdd-profiling/Cargo.toml

@@ -21,6 +21,7 @@ use libdd_common::{
 
 pub mod config;
 mod errors;
+pub mod reqwest_exporter;
 
 #[cfg(unix)]
 pub use connector::uds::{socket_path_from_uri, socket_path_to_uri};

libdd-profiling/src/exporter/mod.rs

@@ -0,0 +1,266 @@
+// Copyright 2021-Present Datadog, Inc. https://www.datadoghq.com/
+// SPDX-License-Identifier: Apache-2.0
+
+//! Reqwest-based profiling exporter
+//! 
+//! This is a simplified async implementation using reqwest.
+
+use anyhow::Context;
+use libdd_common::{azure_app_services, tag, Endpoint};
+use libdd_common::tag::Tag;
+use serde_json::json;
+use std::{future, io::Write};
+use tokio_util::sync::CancellationToken;
+
+use crate::internal::{EncodedProfile, Profile};
+use crate::profiles::{Compressor, DefaultProfileCodec};
+
+pub struct ProfileExporter {
+    client: reqwest::Client,
+    endpoint: Endpoint,
+    family: String,
+    profiling_library_name: String,
+    profiling_library_version: String,
+    tags: Vec<Tag>,
+}
+
+pub struct File<'a> {
+    pub name: &'a str,
+    pub bytes: &'a [u8],
+}
+
+impl ProfileExporter {
+    /// Creates a new exporter to be used to report profiling data.
+    /// 
+    /// Note: Reqwest v0.12.23+ includes automatic retry support for transient failures.
+    /// The default configuration automatically retries safe errors and low-level protocol NACKs.
+    /// For custom retry policies, users can configure the reqwest client before creating the exporter.
+    pub fn new(
+        profiling_library_name: impl Into<String>,
+        profiling_library_version: impl Into<String>,
+        family: impl Into<String>,
+        tags: Vec<Tag>,
+        endpoint: Endpoint,
+    ) -> anyhow::Result<Self> {
+        let mut builder = reqwest::Client::builder().use_rustls_tls();
+        
+        // Check if this is a Unix Domain Socket
+        #[cfg(unix)]
+        if endpoint.url.scheme_str() == Some("unix") {
+            use libdd_common::connector::uds::socket_path_from_uri;
+            let socket_path = socket_path_from_uri(&endpoint.url)?;
+            builder = builder.unix_socket(socket_path);
+        }
+        
+        Ok(Self {
+            client: builder.build()?,
+            endpoint,
+            family: family.into(),
+            profiling_library_name: profiling_library_name.into(),
+            profiling_library_version: profiling_library_version.into(),
+            tags,
+        })
+    }
+
+    /// Build and send a profile. Returns the HTTP status code.
+    pub async fn send<'a>(
+        &self,
+        profile: EncodedProfile,
+        files_to_compress_and_export: &'a [File<'a>],
+        files_to_export_unmodified: &'a [File<'a>],
+        additional_tags: &[Tag],
+        internal_metadata: Option<serde_json::Value>,
+        info: Option<serde_json::Value>,
+        cancel: Option<&CancellationToken>,
+    ) -> anyhow::Result<reqwest::StatusCode> {
+        let tags_profiler = self.build_tags_string(additional_tags);
+        let event = self.build_event_json(&profile, files_to_compress_and_export, 
+            files_to_export_unmodified, &tags_profiler, internal_metadata, info);
+        
+        let form = self.build_multipart_form(event, profile, files_to_compress_and_export, 
+            files_to_export_unmodified)?;
+
+        // For Unix Domain Sockets, convert the URL to use http:// scheme
+        // The socket path was already configured in the client builder
+        let url = if self.endpoint.url.scheme_str() == Some("unix") {
+            // Use localhost as a placeholder - the actual connection goes through the UDS
+            format!("http://localhost{}", self.endpoint.url.path())
+        } else {
+            self.endpoint.url.to_string()
+        };
+
+        // Build request
+        let request = self.client
+            .post(url)
+            .header("Connection", "close")
+            .header("DD-EVP-ORIGIN", &self.profiling_library_name)
+            .header("DD-EVP-ORIGIN-VERSION", &self.profiling_library_version)
+            .header("User-Agent", format!("DDProf/{}", env!("CARGO_PKG_VERSION")))
+            .headers(self.build_optional_headers())
+            .timeout(std::time::Duration::from_millis(self.endpoint.timeout_ms))
+            .multipart(form)
+            .build()?;
+
+        // Send request with cancellation support
+        tokio::select! {
+            _ = async {
+                match cancel {
+                    Some(token) => token.cancelled().await,
+                    None => future::pending().await,
+                }
+            } => Err(anyhow::anyhow!("Operation cancelled by user")),
+            result = self.client.execute(request) => {
+                Ok(result?.status())
+            }
+        }
+    }
+
+    // Helper methods
+    
+    fn build_tags_string(&self, additional_tags: &[Tag]) -> String {
+        let mut tags = String::new();
+        
+        // Add configured tags
+        for tag in &self.tags {
+            tags.push_str(tag.as_ref());
+            tags.push(',');
+        }
+        
+        // Add additional tags
+        for tag in additional_tags {
+            tags.push_str(tag.as_ref());
+            tags.push(',');
+        }
+        
+        // Add Azure App Services tags if available
+        if let Some(aas) = &*azure_app_services::AAS_METADATA {
+            for (name, value) in [
+                ("aas.resource.id", aas.get_resource_id()),
+                ("aas.environment.extension_version", aas.get_extension_version()),
+                ("aas.environment.instance_id", aas.get_instance_id()),
+                ("aas.environment.instance_name", aas.get_instance_name()),
+                ("aas.environment.os", aas.get_operating_system()),
+                ("aas.resource.group", aas.get_resource_group()),
+                ("aas.site.name", aas.get_site_name()),
+                ("aas.site.kind", aas.get_site_kind()),
+                ("aas.site.type", aas.get_site_type()),
+                ("aas.subscription.id", aas.get_subscription_id()),
+            ] {
+                if let Ok(tag) = Tag::new(name, value) {
+                    tags.push_str(tag.as_ref());
+                    tags.push(',');
+                }
+            }
+        }
+        
+        // Add runtime platform tag (last, no trailing comma)
+        tags.push_str(tag!("runtime_platform", target_triple::TARGET).as_ref());
+        tags
+    }
+
+    fn build_event_json(
+        &self,
+        profile: &EncodedProfile,
+        files_to_compress: &[File],
+        files_unmodified: &[File],
+        tags_profiler: &str,
+        internal_metadata: Option<serde_json::Value>,
+        info: Option<serde_json::Value>,
+    ) -> String {
+        let attachments: Vec<_> = files_to_compress.iter()
+            .chain(files_unmodified.iter())
+            .map(|f| f.name)
+            .chain(std::iter::once("profile.pprof"))
+            .collect();
+
+        let mut internal = internal_metadata.unwrap_or_else(|| json!({}));
+        internal["libdatadog_version"] = json!(env!("CARGO_PKG_VERSION"));
+
+        json!({
+            "attachments": attachments,
+            "tags_profiler": tags_profiler,
+            "start": chrono::DateTime::<chrono::Utc>::from(profile.start)
+                .format("%Y-%m-%dT%H:%M:%S%.9fZ").to_string(),
+            "end": chrono::DateTime::<chrono::Utc>::from(profile.end)
+                .format("%Y-%m-%dT%H:%M:%S%.9fZ").to_string(),
+            "family": self.family,
+            "version": "4",
+            "endpoint_counts": if profile.endpoints_stats.is_empty() { 
+                None 
+            } else { 
+                Some(&profile.endpoints_stats) 
+            },
+            "internal": internal,
+            "info": info.unwrap_or_else(|| json!({})),
+        })
+        .to_string()
+    }
+
+    fn build_multipart_form(
+        &self,
+        event: String,
+        profile: EncodedProfile,
+        files_to_compress: &[File],
+        files_unmodified: &[File],
+    ) -> anyhow::Result<reqwest::multipart::Form> {
+        let mut form = reqwest::multipart::Form::new()
+            .part(
+                "event",
+                reqwest::multipart::Part::text(event)
+                    .file_name("event.json")
+                    .mime_str("application/json")?,
+            );
+
+        // Add compressed files
+        for file in files_to_compress {
+            let mut encoder = Compressor::<DefaultProfileCodec>::try_new(
+                (file.bytes.len() >> 3).next_power_of_two(),
+                10 * 1024 * 1024,
+                Profile::COMPRESSION_LEVEL,
+            )
+            .context("failed to create compressor")?;
+            encoder.write_all(file.bytes)?;
+            
+            form = form.part(
+                file.name.to_string(),
+                reqwest::multipart::Part::bytes(encoder.finish()?)
+                    .file_name(file.name.to_string()),
+            );
+        }
+
+        // Add uncompressed files
+        for file in files_unmodified {
+            form = form.part(
+                file.name.to_string(),
+                reqwest::multipart::Part::bytes(file.bytes.to_vec())
+                    .file_name(file.name.to_string()),
+            );
+        }
+
+        // Add profile
+        Ok(form.part(
+            "profile.pprof",
+            reqwest::multipart::Part::bytes(profile.buffer)
+                .file_name("profile.pprof"),
+        ))
+    }
+
+    fn build_optional_headers(&self) -> reqwest::header::HeaderMap {
+        let mut headers = reqwest::header::HeaderMap::new();
+        
+        if let Some(api_key) = &self.endpoint.api_key {
+            if let Ok(value) = reqwest::header::HeaderValue::from_str(api_key) {
+                headers.insert("DD-API-KEY", value);
+            }
+        }
+        
+        if let Some(test_token) = &self.endpoint.test_token {
+            if let Ok(value) = reqwest::header::HeaderValue::from_str(test_token) {
+                headers.insert("X-Datadog-Test-Session-Token", value);
+            }
+        }
+        
+        headers
+    }
+}
+

libdd-profiling/src/exporter/reqwest_exporter.rs

@@ -536,6 +536,46 @@ impl Profile {
         Ok(self)
     }
 
+    /// Convenience method to serialize and export the profile to a reqwest exporter.
+    /// 
+    /// # Arguments
+    /// * `exporter` - The reqwest ProfileExporter to send the profile to
+    /// * `files_to_compress_and_export` - Additional files to compress and attach to the profile
+    /// * `files_to_export_unmodified` - Additional files to attach without compression
+    /// * `additional_tags` - Tags to add to this specific profile (in addition to the exporter's configured tags)
+    /// * `internal_metadata` - Internal metadata to include in the event JSON
+    /// * `info` - System info to include in the event JSON
+    /// * `end_time` - Optional end time (defaults to now)
+    /// * `duration` - Optional duration (defaults to end_time - start_time)
+    /// * `cancel` - Optional cancellation token to abort the upload
+    /// 
+    /// # Returns
+    /// The HTTP status code from the upload
+    #[allow(clippy::too_many_arguments)]
+    pub async fn export_to_endpoint<'a>(
+        self,
+        exporter: &crate::exporter::reqwest_exporter::ProfileExporter,
+        files_to_compress_and_export: &'a [crate::exporter::reqwest_exporter::File<'a>],
+        files_to_export_unmodified: &'a [crate::exporter::reqwest_exporter::File<'a>],
+        additional_tags: &[libdd_common::tag::Tag],
+        internal_metadata: Option<serde_json::Value>,
+        info: Option<serde_json::Value>,
+        end_time: Option<SystemTime>,
+        duration: Option<Duration>,
+        cancel: Option<&tokio_util::sync::CancellationToken>,
+    ) -> anyhow::Result<reqwest::StatusCode> {
+        let encoded = self.serialize_into_compressed_pprof(end_time, duration)?;
+        exporter.send(
+            encoded,
+            files_to_compress_and_export,
+            files_to_export_unmodified,
+            additional_tags,
+            internal_metadata,
+            info,
+            cancel,
+        ).await
+    }
+
     /// In incident 35390 (JIRA PROF-11456) we observed invalid location_ids being present in
     /// emitted profiles. We're doing extra checks here so that if we see incorrect ids again,
     /// we are 100% sure they were not introduced prior to this stage.