format and copyright headers and fix windows

Author: danielsn

libdd-profiling/src/exporter/file_exporter.rs

@@ -1,3 +1,6 @@
+// Copyright 2021-Present Datadog, Inc. https://www.datadoghq.com/
+// SPDX-License-Identifier: Apache-2.0
+
 #[cfg(any(unix, windows))]
 use std::path::PathBuf;
 
@@ -36,7 +39,7 @@ pub(crate) fn spawn_dump_server(output_path: PathBuf) -> anyhow::Result<PathBuf>
 
     let socket_path_clone = socket_path.clone();
     let (tx, rx) = std::sync::mpsc::channel();
-    
+
     std::thread::spawn(move || {
         // Top-level error handler - all errors logged here
         let result = (|| -> anyhow::Result<()> {
@@ -47,7 +50,7 @@ pub(crate) fn spawn_dump_server(output_path: PathBuf) -> anyhow::Result<PathBuf>
                 run_dump_server_unix(output_path, listener).await
             })
         })();
-        
+
         if let Err(e) = result {
             eprintln!("[dump-server] Error: {}", e);
             let _ = tx.send(Err(e));
@@ -66,7 +69,7 @@ async fn run_dump_server_windows(output_path: PathBuf, pipe_name: String) -> any
 
     loop {
         // Create server instance
-        let mut server = ServerOptions::new()
+        let server = ServerOptions::new()
             .first_pipe_instance(true)
             .create(&pipe_name)?;
 
@@ -92,7 +95,7 @@ pub(crate) fn spawn_dump_server(output_path: PathBuf) -> anyhow::Result<PathBuf>
     let pipe_path = PathBuf::from(&pipe_name);
 
     let (tx, rx) = std::sync::mpsc::channel();
-    
+
     std::thread::spawn(move || {
         // Top-level error handler - all errors logged here
         let result = (|| -> anyhow::Result<()> {
@@ -102,7 +105,7 @@ pub(crate) fn spawn_dump_server(output_path: PathBuf) -> anyhow::Result<PathBuf>
                 run_dump_server_windows(output_path, pipe_name).await
             })
         })();
-        
+
         if let Err(e) = result {
             eprintln!("[dump-server] Error: {}", e);
             let _ = tx.send(Err(e));
@@ -155,9 +158,7 @@ fn is_request_complete(
 
 /// Read complete HTTP request from an async stream
 #[cfg(any(unix, windows))]
-async fn read_http_request_async<R: tokio::io::AsyncReadExt + Unpin>(
-    stream: &mut R,
-) -> Vec<u8> {
+async fn read_http_request_async<R: tokio::io::AsyncReadExt + Unpin>(stream: &mut R) -> Vec<u8> {
     let mut request_data = Vec::new();
     let mut buffer = [0u8; 8192];
     let mut content_length: Option<usize> = None;
@@ -197,7 +198,10 @@ async fn read_http_request_async<R: tokio::io::AsyncReadExt + Unpin>(
 async fn write_request_to_file_async(output_path: &PathBuf, request_data: &[u8]) {
     if !request_data.is_empty() {
         if let Err(e) = tokio::fs::write(output_path, request_data).await {
-            eprintln!("[dump-server] Failed to write request dump to {:?}: {}", output_path, e);
+            eprintln!(
+                "[dump-server] Failed to write request dump to {:?}: {}",
+                output_path, e
+            );
         }
     }
 }
@@ -210,7 +214,7 @@ where
 {
     let request_data = read_http_request_async(&mut stream).await;
     write_request_to_file_async(&output_path, &request_data).await;
-    
+
     if let Err(e) = stream.write_all(HTTP_200_RESPONSE).await {
         eprintln!("[dump-server] Failed to send HTTP response: {}", e);
     }

libdd-profiling/src/exporter/profile_exporter.rs

@@ -86,7 +86,7 @@ impl ProfileExporter {
                 let output_path = libdd_common::decode_uri_path_in_authority(&endpoint.url)
                     .context("Failed to decode file path from URI")?;
                 let pipe_path = spawn_dump_server(output_path)?;
-                builder = builder.windows_named_pipe(pipe_path);
+                builder = builder.windows_named_pipe(pipe_path.to_string_lossy().to_string());
                 "http://localhost/v1/input".to_string()
             }
 
@@ -108,7 +108,7 @@ impl ProfileExporter {
             Some("windows") => {
                 use libdd_common::connector::named_pipe::named_pipe_path_from_uri;
                 let pipe_path = named_pipe_path_from_uri(&endpoint.url)?;
-                builder = builder.windows_named_pipe(pipe_path);
+                builder = builder.windows_named_pipe(pipe_path.to_string_lossy().to_string());
                 format!("http://localhost{}", endpoint.url.path())
             }
             #[cfg(unix)]

libdd-profiling/src/exporter/utils.rs

@@ -58,9 +58,7 @@ pub fn parse_http_request(data: &[u8]) -> anyhow::Result<HttpRequest> {
     let mut lines = header_str.lines();
 
     // Parse request line
-    let request_line = lines
-        .next()
-        .context("No request line found")?;
+    let request_line = lines.next().context("No request line found")?;
     let parts: Vec<&str> = request_line.split_whitespace().collect();
     if parts.len() < 2 {
         anyhow::bail!("Invalid request line");
@@ -174,12 +172,12 @@ pub fn parse_multipart(body: &[u8], boundary: &str) -> anyhow::Result<Vec<Multip
 
         // Move past the delimiter
         pos = part_end + delimiter_bytes.len();
-        
+
         // Check if this is the end delimiter (delimiter followed by --)
         if pos + 2 <= body.len() && &body[pos..pos + 2] == b"--" {
             break;
         }
-        
+
         // Skip CRLF after boundary
         if pos + 2 <= body.len() && &body[pos..pos + 2] == b"\r\n" {
             pos += 2;

libdd-profiling/tests/exporter_e2e.rs

@@ -147,7 +147,6 @@ async fn read_and_capture_request<S>(
 ) where
     S: tokio::io::AsyncReadExt + tokio::io::AsyncWriteExt + Unpin,
 {
-
     let mut buffer = Vec::new();
     let mut temp_buf = [0u8; 8192];
     let mut headers_complete = false;
@@ -177,9 +176,7 @@ async fn read_and_capture_request<S>(
                     }
                 }
 
-                if let (Some(headers_end), Some(expected_len)) =
-                    (headers_end_pos, content_length)
-                {
+                if let (Some(headers_end), Some(expected_len)) = (headers_end_pos, content_length) {
                     if buffer.len() - headers_end >= expected_len {
                         break;
                     }
@@ -337,10 +334,7 @@ fn validate_full_export(req: &ReceivedRequest, expected_path: &str) -> anyhow::R
     }
 
     // Verify process_tags
-    assert_eq!(
-        event_json["process_tags"],
-        "entrypoint.name:main,pid:12345"
-    );
+    assert_eq!(event_json["process_tags"], "entrypoint.name:main,pid:12345");
 
     // Verify attachments
     let attachments = event_json["attachments"]
@@ -375,7 +369,11 @@ fn validate_full_export(req: &ReceivedRequest, expected_path: &str) -> anyhow::R
             .iter()
             .find(|p| p.name == *part_name)
             .ok_or_else(|| anyhow::anyhow!("Missing part: {}", part_name))?;
-        assert!(!part.content.is_empty(), "{} should not be empty", part_name);
+        assert!(
+            !part.content.is_empty(),
+            "{} should not be empty",
+            part_name
+        );
     }
 
     Ok(())
@@ -405,7 +403,8 @@ async fn test_agent_with_transport(transport: Transport) -> anyhow::Result<()> {
     };
 
     // Run the full export test
-    let req = export_full_profile(endpoint, RequestSource::Captured(server.received_requests)).await?;
+    let req =
+        export_full_profile(endpoint, RequestSource::Captured(server.received_requests)).await?;
 
     // Validate
     validate_full_export(&req, "/profiling/v1/input")?;
@@ -446,7 +445,8 @@ async fn test_agentless_with_transport(transport: Transport) -> anyhow::Result<(
         {
             let pipe_path = server.pipe_path.as_ref().unwrap();
             // For named pipes, we need to create endpoint manually
-            let endpoint_url = libdd_common::connector::named_pipe::named_pipe_path_to_uri(pipe_path)?;
+            let endpoint_url =
+                libdd_common::connector::named_pipe::named_pipe_path_to_uri(pipe_path)?;
             let mut parts = endpoint_url.into_parts();
             parts.path_and_query = Some("/api/v2/profile".parse()?);
             let url = http::Uri::from_parts(parts)?;
@@ -459,7 +459,8 @@ async fn test_agentless_with_transport(transport: Transport) -> anyhow::Result<(
     };
 
     // Run the full export test
-    let req = export_full_profile(endpoint, RequestSource::Captured(server.received_requests)).await?;
+    let req =
+        export_full_profile(endpoint, RequestSource::Captured(server.received_requests)).await?;
 
     // Validate - agentless uses /api/v2/profile path
     validate_full_export(&req, "/api/v2/profile")?;