feat(telemetry)!: Add Installation signature and AppProduct changes payloads (#2213)

And some smaller stuff like more metric namespaces.

Adding endpoint batching as well.

This bridges the gap to the current capabilities exposed by dd-trace-py.

Also contains a small debugger change to abort on timeout.

Co-authored-by: bob.weinand <bob.weinand@datadoghq.com>

Author: bwoebi

Cargo.lock

@@ -305,12 +305,22 @@ pub fn generate_tags(
     percent_encode(tags.as_bytes(), CONTROLS).to_string()
 }
 
+/// Owns the spawned request task and aborts it when dropped.
+struct AbortOnDrop(JoinHandle<anyhow::Result<http::Response<Bytes>>>);
+
+impl Drop for AbortOnDrop {
+    fn drop(&mut self) {
+        // A no-op once the task has completed, so the success path is unaffected.
+        self.0.abort();
+    }
+}
+
 #[derive(Default)]
 enum SenderFuture {
     #[default]
     Error,
     Outstanding(ResponseFuture),
-    Submitted(JoinHandle<anyhow::Result<http::Response<Bytes>>>),
+    Submitted(AbortOnDrop),
 }
 
 pub struct PayloadSender {
@@ -322,7 +332,7 @@ pub struct PayloadSender {
 }
 
 const BOUNDARY: &str = "------------------------44617461646f67";
-const BOUNDARY_LINE: &str = concat!("--", BOUNDARY, "\r\n");
+const BOUNDARY_LINE: &str = concat!("--", BOUNDARY);
 
 impl PayloadSender {
     pub fn new(
@@ -426,18 +436,19 @@ impl PayloadSender {
             SenderFuture::Outstanding(future) => {
                 if self.needs_boundary {
                     let header = concat!(
-                        BOUNDARY_LINE,
-                        "Content-Disposition: form-data; name=\"event\"; filename=\"event.json\"\r\n",
-                        "Content-Type: application/json\r\n",
-                        "\r\n",
+                    BOUNDARY_LINE,
+                    "\r\n",
+                    "Content-Disposition: form-data; name=\"event\"; filename=\"event.json\"\r\n",
+                    "Content-Type: application/json\r\n",
+                    "\r\n",
                     );
                     self.sender.send_chunk(header.into()).await?;
                 }
 
-                self.future = SenderFuture::Submitted(tokio::spawn(async move {
+                self.future = SenderFuture::Submitted(AbortOnDrop(tokio::spawn(async move {
                     let resp = future.await?;
                     Ok(resp)
-                }));
+                })));
                 true
             }
             future => {
@@ -463,7 +474,7 @@ impl PayloadSender {
             // insert a trailing ]
             if self.needs_boundary {
                 self.sender
-                    .send_chunk(concat!("]\r\n", BOUNDARY_LINE).into())
+                    .send_chunk(concat!("]\r\n", BOUNDARY_LINE, "--\r\n").into())
                     .await?;
             } else {
                 self.sender.send_chunk(Bytes::from_static(b"]")).await?;
@@ -472,17 +483,14 @@ impl PayloadSender {
             drop(self.sender);
             // Once the body is fully sent, bound the wait for the response headers and (if
             // needed) the response body under a single timeout - a slow/stalled server must
-            // not be able to hang this indefinitely. Abort the spawned task on timeout so the
-            // underlying request is actually cancelled instead of left running detached.
+            // not be able to hang this indefinitely. Returning here drops `future`, whose
+            // `AbortOnDrop` cancels the underlying request rather than detaching it.
             let response =
-                match tokio::time::timeout(Duration::from_millis(self.timeout_ms), &mut future)
+                match tokio::time::timeout(Duration::from_millis(self.timeout_ms), &mut future.0)
                     .await
                 {
                     Ok(joined) => joined??,
-                    Err(_) => {
-                        future.abort();
-                        return Err(anyhow::anyhow!("debugger payload request timed out"));
-                    }
+                    Err(_) => return Err(anyhow::anyhow!("debugger payload request timed out")),
                 };
 
             let status = response.status().as_u16();
@@ -673,7 +681,10 @@ pub fn generate_new_id() -> Uuid {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use libdd_capabilities::{ChunkFuture, HttpError, MaybeSend, StreamingBodySender};
     use std::borrow::Cow;
+    use std::sync::atomic::{AtomicBool, Ordering};
+    use std::sync::Arc;
 
     fn agent_endpoint() -> Endpoint {
         Endpoint::from_slice("http://localhost:8126")
@@ -786,6 +797,107 @@ mod tests {
         }
     }
 
+    struct SetOnDrop(Arc<AtomicBool>);
+
+    impl Drop for SetOnDrop {
+        fn drop(&mut self) {
+            self.0.store(true, Ordering::SeqCst);
+        }
+    }
+
+    /// A response that never arrives, holding `guard` for as long as it lives.
+    ///
+    /// The guard is captured by move rather than constructed in the body, because an
+    /// async block does not run its body until first polled: a task aborted before
+    /// its first poll would otherwise never arm the observer.
+    async fn stalled_response(guard: SetOnDrop) -> Result<http::Response<Bytes>, HttpError> {
+        let _guard = guard;
+        future::pending::<()>().await;
+        unreachable!()
+    }
+
+    /// Accepts and discards body chunks, so the test's stalled response future is the
+    /// only thing the spawned task ever waits on.
+    struct DiscardingBodySender;
+
+    impl StreamingBodySender for DiscardingBodySender {
+        fn send_chunk(&mut self, _data: Bytes) -> ChunkFuture<'_> {
+            Box::pin(async { Ok(()) })
+        }
+    }
+
+    /// A client whose requests never complete, so a test can observe what happens to
+    /// the in-flight task once the sender goes away. The flag is set when the
+    /// response future is dropped, which happens only if the spawned task was
+    /// aborted rather than detached.
+    ///
+    /// `request_streamed` is overridden rather than relying on the default
+    /// implementation: that one parks on the body channel until the `BodySender` is
+    /// dropped, so a task abandoned before `finish()` would be cancelled before it
+    /// ever reached `request()` and the flag would never be armed.
+    #[derive(Clone, Debug)]
+    struct StalledClient(Arc<AtomicBool>);
+
+    impl HttpClientCapability for StalledClient {
+        fn new_client() -> Self {
+            Self(Arc::new(AtomicBool::new(false)))
+        }
+
+        fn new_without_connection_pooling() -> Self {
+            Self::new_client()
+        }
+
+        fn request(
+            &self,
+            _req: http::Request<Bytes>,
+        ) -> impl std::future::Future<Output = Result<http::Response<Bytes>, HttpError>> + MaybeSend
+        {
+            stalled_response(SetOnDrop(self.0.clone()))
+        }
+
+        fn request_streamed(&self, _req: http::Request<()>) -> (BodySender, ResponseFuture) {
+            (
+                Box::new(DiscardingBodySender),
+                Box::pin(stalled_response(SetOnDrop(self.0.clone()))),
+            )
+        }
+    }
+
+    #[tokio::test]
+    async fn test_dropping_payload_sender_aborts_in_flight_request() {
+        let cancelled = Arc::new(AtomicBool::new(false));
+
+        let mut config = Config::default();
+        config.set_endpoint(agent_endpoint()).unwrap();
+
+        let mut sender = PayloadSender::new_to_endpoint_with_client(
+            config.snapshots_endpoint.as_ref().unwrap(),
+            DebuggerType::Snapshots,
+            "",
+            StalledClient(cancelled.clone()),
+        )
+        .unwrap();
+
+        // Spawns the request task, which then never completes.
+        sender.append(b"[{}]").await.unwrap();
+
+        // Abandoning the sender (an enclosing timeout, a `select!`, a cancelled task)
+        // must cancel that request instead of leaving it running with its connection.
+        drop(sender);
+
+        for _ in 0..100 {
+            if cancelled.load(Ordering::SeqCst) {
+                break;
+            }
+            tokio::task::yield_now().await;
+        }
+
+        assert!(
+            cancelled.load(Ordering::SeqCst),
+            "in-flight request was detached instead of aborted"
+        );
+    }
+
     #[test]
     fn test_payload_rejected_display_is_downcastable() {
         let err: anyhow::Error = anyhow::Error::from(PayloadRejected {

datadog-live-debugger/src/sender.rs

@@ -456,7 +456,7 @@ pub unsafe extern "C" fn ddog_sidecar_telemetry_enqueueConfig(
     let seq_id = seq_id.to_std();
     let config_entry = TelemetryActions::AddConfig(data::Configuration {
         name: config_key.to_utf8_lossy().into_owned(),
-        value: config_value.to_utf8_lossy().into_owned(),
+        value: Some(config_value.to_utf8_lossy().into_owned()),
         origin,
         config_id,
         seq_id,
@@ -487,6 +487,9 @@ pub unsafe extern "C" fn ddog_sidecar_telemetry_addEndpoint(
         path: Some(path.to_utf8_lossy().into_owned()),
         operation_name: operation_name.to_utf8_lossy().into_owned(),
         resource_name: resource_name.to_utf8_lossy().into_owned(),
+        request_body_type: None,
+        response_body_type: None,
+        response_code: None,
     });
 
     try_c!(blocking::enqueue_actions(
@@ -515,6 +518,8 @@ pub unsafe extern "C" fn ddog_sidecar_telemetry_addDependency(
     let dependency = TelemetryActions::AddDependency(Dependency {
         name: dependency_name.to_utf8_lossy().into_owned(),
         version,
+        hash: None,
+        metadata: None,
     });
 
     try_c!(blocking::enqueue_actions(
@@ -547,6 +552,7 @@ pub unsafe extern "C" fn ddog_sidecar_telemetry_addIntegration(
         version,
         compatible: None,
         auto_enabled: None,
+        error: None,
     });
 
     try_c!(blocking::enqueue_actions(

datadog-sidecar-ffi/src/lib.rs

@@ -59,23 +59,51 @@ mod native {
         }
     }
 
-    /// Write `body` as a newline-terminated record to the file referenced by `uri` (which must
-    /// have a `file://` scheme), then return a synthetic 202 response.
+    /// Record `body` to the on-disk location referenced by `uri` (which must have a `file://`
+    /// scheme), then return a synthetic 202 response.
+    ///
+    /// If the location is a directory (the path ends with a separator, as the offline
+    /// telemetry writer passes `file:///dir/`), each request is written to its own
+    /// `telemetry-<seq>-<pid>-<ts>.json` file inside it — ordinal-first so a lexicographic sort
+    /// reproduces emission order. Otherwise the body is appended as a newline-terminated record to
+    /// that single file.
     fn write_to_file_endpoint(
         uri: &http::Uri,
         body: bytes::Bytes,
     ) -> Result<http::Response<bytes::Bytes>, HttpError> {
         let path = libdd_common::decode_uri_path_in_authority(uri)
             .map_err(|e| HttpError::Other(anyhow::anyhow!("invalid file:// URI: {e}")))?;
-        let mut file = OpenOptions::new()
-            .create(true)
-            .append(true)
-            .open(&path)
-            .map_err(|e| HttpError::Other(anyhow::anyhow!("opening {path:?}: {e}")))?;
-        let mut record = body.to_vec();
-        record.push(b'\n');
-        file.write_all(&record)
-            .map_err(|e| HttpError::Other(anyhow::anyhow!("writing {path:?}: {e}")))?;
+
+        let is_dir = path.to_string_lossy().ends_with(std::path::MAIN_SEPARATOR) || path.is_dir();
+        if is_dir {
+            std::fs::create_dir_all(&path)
+                .map_err(|e| HttpError::Other(anyhow::anyhow!("creating {path:?}: {e}")))?;
+            // Process-wide sequence so successive requests get distinct, ordered filenames.
+            static SEQ: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
+            let seq = SEQ.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+            let ts = std::time::SystemTime::now()
+                .duration_since(std::time::UNIX_EPOCH)
+                .map(|d| d.as_nanos())
+                .unwrap_or(0);
+            let name = format!("telemetry-{seq:020}-{}-{ts}.json", std::process::id());
+            // Write to a temp file then rename, so a concurrent reader never sees a partial file.
+            let dest = path.join(name);
+            let tmp = dest.with_extension("json.tmp");
+            std::fs::write(&tmp, &body)
+                .map_err(|e| HttpError::Other(anyhow::anyhow!("writing {tmp:?}: {e}")))?;
+            std::fs::rename(&tmp, &dest)
+                .map_err(|e| HttpError::Other(anyhow::anyhow!("renaming to {dest:?}: {e}")))?;
+        } else {
+            let mut file = OpenOptions::new()
+                .create(true)
+                .append(true)
+                .open(&path)
+                .map_err(|e| HttpError::Other(anyhow::anyhow!("opening {path:?}: {e}")))?;
+            let mut record = body.to_vec();
+            record.push(b'\n');
+            file.write_all(&record)
+                .map_err(|e| HttpError::Other(anyhow::anyhow!("writing {path:?}: {e}")))?;
+        }
 
         http::Response::builder()
             .status(http::StatusCode::ACCEPTED)

libdd-capabilities-impl/src/http.rs

@@ -9,8 +9,13 @@
 //! This is primarily used for testing to validate the exact bytes sent over the wire.
 
 use anyhow::Context;
+use core::sync::atomic::{AtomicU64, Ordering};
 use std::path::PathBuf;
-use tokio::io::AsyncReadExt;
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+
+/// Monotonic counter for directory-mode dump filenames, so each captured request lands in
+/// its own lexicographically-ordered file instead of overwriting a single one.
+static DUMP_SEQ: AtomicU64 = AtomicU64::new(0);
 
 /// Helper to find subsequence in bytes
 fn find_subsequence(haystack: &[u8], needle: &[u8]) -> Option<usize> {
@@ -385,13 +390,55 @@ async fn write_request_to_file(
         return Ok(());
     }
 
+    // A path ending in a separator (e.g. `file:///dir/`) selects "directory mode": write each
+    // request's body as its own lexicographically-ordered file so requests accumulate instead of
+    // overwriting a single dump file.
+    let path_str = output_path.as_os_str().to_string_lossy();
+    if path_str.ends_with('/') || path_str.ends_with(std::path::MAIN_SEPARATOR) {
+        let body = if parsed_request.is_chunked {
+            decode_chunked_body(&parsed_request.raw_data[parsed_request.headers_len..])?
+        } else {
+            parsed_request.raw_data[parsed_request.headers_len..].to_vec()
+        };
+        if body.is_empty() {
+            return Ok(());
+        }
+        tokio::fs::create_dir_all(output_path)
+            .await
+            .context("Failed to create dump directory")?;
+        // Ordinal-first name so files sort in emission order; the PID avoids cross-process
+        // collisions when several processes share the directory.
+        let name = format!(
+            "{:020}-{}.json",
+            DUMP_SEQ.fetch_add(1, Ordering::Relaxed),
+            std::process::id()
+        );
+        let dest = output_path.join(&name);
+        let tmp = output_path.join(format!("{name}.tmp"));
+        {
+            let mut file = tokio::fs::File::create(&tmp)
+                .await
+                .context("Failed to create dump file")?;
+            file.write_all(&body)
+                .await
+                .context("Failed to write request dump")?;
+            file.sync_all()
+                .await
+                .context("Failed to sync request dump to disk")?;
+        }
+        // Atomic publish via rename so readers never observe a partial file.
+        tokio::fs::rename(&tmp, &dest)
+            .await
+            .context("Failed to publish request dump")?;
+        return Ok(());
+    }
+
     let data_to_write = if parsed_request.is_chunked && parsed_request.headers_len > 0 {
         reconstruct_with_content_length(&parsed_request.raw_data, parsed_request.headers_len)?
     } else {
         parsed_request.raw_data.clone()
     };
 
-    use tokio::io::AsyncWriteExt;
     let mut file = tokio::fs::File::create(output_path)
         .await
         .context("Failed to create dump file")?;