Add integration test to send_data to cover UDS transport with agent (#…

…843)

data-pipeline/tests/test_fetch_info.rs

@@ -12,7 +12,7 @@ mod tracing_integration_tests {
     #[cfg_attr(miri, ignore)]
     #[tokio::test]
     async fn test_fetch_info_from_test_agent() {
-        let test_agent = DatadogTestAgent::new(None).await;
+        let test_agent = DatadogTestAgent::new(None, None).await;
         let endpoint = Endpoint::from_url(test_agent.get_uri_for_endpoint("info", None).await);
         let info = fetch_info(&endpoint)
             .await
@@ -28,7 +28,7 @@ mod tracing_integration_tests {
     #[cfg_attr(miri, ignore)]
     #[tokio::test]
     async fn test_agent_info_fetcher_with_test_agent() {
-        let test_agent = DatadogTestAgent::new(None).await;
+        let test_agent = DatadogTestAgent::new(None, None).await;
         let endpoint = Endpoint::from_url(test_agent.get_uri_for_endpoint("info", None).await);
         let fetcher = AgentInfoFetcher::new(endpoint, Duration::from_secs(1));
         let info_arc = fetcher.get_info();

trace-utils/Cargo.toml

@@ -50,6 +50,7 @@ httpmock = { version = "0.7.0"}
 serde_json = "1.0"
 tokio = { version = "1", features = ["macros", "rt-multi-thread"] }
 datadog-trace-utils = { path = ".", features = ["test-utils"] }
+tempfile = "3.3.0"
 
 [features]
 default = ["proxy"]

trace-utils/src/test_utils/datadog_test_agent.rs

@@ -1,6 +1,7 @@
 // Copyright 2021-Present Datadog, Inc. https://www.datadoghq.com/
 // SPDX-License-Identifier: Apache-2.0
 
+use std::collections::HashMap;
 use std::path::Path;
 use std::str::FromStr;
 use std::time::Duration;
@@ -25,6 +26,7 @@ const TEST_AGENT_PORT: u16 = 8126;
 #[derive(Debug)]
 struct DatadogTestAgentContainer {
     mounts: Vec<Mount>,
+    env_vars: HashMap<String, String>,
 }
 
 impl Image for DatadogTestAgentContainer {
@@ -54,30 +56,49 @@ impl Image for DatadogTestAgentContainer {
     fn mounts(&self) -> Box<dyn Iterator<Item = &Mount> + '_> {
         Box::new(self.mounts.iter())
     }
+
     fn expose_ports(&self) -> Vec<u16> {
         vec![TEST_AGENT_PORT]
     }
+
+    fn env_vars(&self) -> Box<dyn Iterator<Item = (&String, &String)> + '_> {
+        Box::new(self.env_vars.iter())
+    }
 }
 
 impl DatadogTestAgentContainer {
-    fn new(relative_snapshot_path: Option<&str>) -> Self {
+    fn new(relative_snapshot_path: Option<&str>, absolute_socket_path: Option<&str>) -> Self {
+        let mut env_vars = HashMap::new();
+        let mut mounts = Vec::new();
+
+        if let Some(absolute_socket_path) = absolute_socket_path {
+            env_vars.insert(
+                "DD_APM_RECEIVER_SOCKET".to_string(),
+                "/tmp/ddsockets/apm.socket".to_owned(),
+            );
+            mounts.push(
+                Mount::bind_mount(absolute_socket_path, "/tmp/ddsockets")
+                    .with_access_mode(AccessMode::ReadWrite),
+            );
+        }
+
         if let Some(relative_snapshot_path) = relative_snapshot_path {
-            let mount = Mount::bind_mount(
-                DatadogTestAgentContainer::calculate_snapshot_absolute_path(relative_snapshot_path),
-                "/snapshots",
-            )
-            .with_access_mode(AccessMode::ReadWrite);
-
-            DatadogTestAgentContainer {
-                mounts: vec![mount],
-            }
-        } else {
-            DatadogTestAgentContainer { mounts: vec![] }
+            mounts.push(
+                Mount::bind_mount(
+                    DatadogTestAgentContainer::calculate_volume_absolute_path(
+                        relative_snapshot_path,
+                    ),
+                    "/snapshots",
+                )
+                .with_access_mode(AccessMode::ReadWrite),
+            );
         }
+
+        DatadogTestAgentContainer { mounts, env_vars }
     }
     // The docker image requires an absolute path when mounting a volume. This function gets the
     // absolute path of the workspace and appends the provided relative path.
-    fn calculate_snapshot_absolute_path(relative_snapshot_path: &str) -> String {
+    fn calculate_volume_absolute_path(relative_snapshot_path: &str) -> String {
         let metadata = MetadataCommand::new()
             .exec()
             .expect("Failed to fetch metadata");
@@ -115,7 +136,7 @@ impl DatadogTestAgentContainer {
 /// #[tokio::main]
 /// async fn main() {
 ///     // Create a new DatadogTestAgent instance
-///     let test_agent = DatadogTestAgent::new(Some("relative/path/to/snapshot")).await;
+///     let test_agent = DatadogTestAgent::new(Some("relative/path/to/snapshot"), None).await;
 ///
 ///     // Get the URI for a specific endpoint
 ///     let uri = test_agent
@@ -154,12 +175,19 @@ impl DatadogTestAgent {
     ///   test-agent. The relative path should include the crate name. If no relative path is
     ///   provided, no snapshot directory will be mounted.
     ///
+    /// * `absolute_socket_path` - An optional string slice that holds the absolute path to the
+    ///   socket directory. This directory will get mounted in the docker container running the
+    ///   test-agent. It is recommended to use a temporary directory for this purpose. If no socket
+    ///   path is provided the test agent will not be configured for UDS transport.
     /// # Returns
     ///
     /// A new `DatadogTestAgent`.
-    pub async fn new(relative_snapshot_path: Option<&str>) -> Self {
+    pub async fn new(
+        relative_snapshot_path: Option<&str>,
+        absolute_socket_path: Option<&str>,
+    ) -> Self {
         DatadogTestAgent {
-            container: DatadogTestAgentContainer::new(relative_snapshot_path)
+            container: DatadogTestAgentContainer::new(relative_snapshot_path, absolute_socket_path)
                 .start()
                 .await
                 .expect("Unable to start DatadogTestAgent, is the Docker Daemon running?"),

trace-utils/tests/test_send_data.rs

@@ -8,14 +8,22 @@ mod tracing_integration_tests {
     use datadog_trace_utils::test_utils::datadog_test_agent::DatadogTestAgent;
     use datadog_trace_utils::trace_utils::TracerHeaderTags;
     use datadog_trace_utils::tracer_payload::TracerPayloadCollection;
+    #[cfg(target_os = "linux")]
+    use ddcommon::connector::uds::socket_path_to_uri;
     use ddcommon::Endpoint;
+    #[cfg(target_os = "linux")]
+    use hyper::Uri;
+    #[cfg(target_os = "linux")]
+    use std::fs::Permissions;
+    #[cfg(target_os = "linux")]
+    use std::os::unix::fs::PermissionsExt;
     use tinybytes::BytesString;
 
     #[cfg_attr(miri, ignore)]
     #[tokio::test]
     async fn compare_v04_trace_snapshot_test() {
         let relative_snapshot_path = "trace-utils/tests/snapshots/";
-        let test_agent = DatadogTestAgent::new(Some(relative_snapshot_path)).await;
+        let test_agent = DatadogTestAgent::new(Some(relative_snapshot_path), None).await;
 
         let header_tags = TracerHeaderTags {
             lang: "test-lang",
@@ -32,6 +40,99 @@ mod tracing_integration_tests {
                 .get_uri_for_endpoint("v0.4/traces", Some("compare_v04_trace_snapshot_test"))
                 .await,
         );
+        let mut span_1 = create_test_no_alloc_span(1234, 12342, 12341, 1, false);
+        span_1.metrics.insert(
+            BytesString::from_slice("_dd_metric1".as_ref()).unwrap(),
+            1.0,
+        );
+        span_1.metrics.insert(
+            BytesString::from_slice("_dd_metric2".as_ref()).unwrap(),
+            2.0,
+        );
+
+        let span_2 = create_test_no_alloc_span(1234, 12343, 12341, 1, false);
+
+        let mut root_span = create_test_no_alloc_span(1234, 12341, 0, 0, true);
+        root_span.r#type = BytesString::from_slice("web".as_ref()).unwrap();
+
+        let trace = vec![span_1, span_2, root_span];
+
+        let data = SendData::new(
+            300,
+            TracerPayloadCollection::V04(vec![trace.clone()]),
+            header_tags,
+            &endpoint,
+        );
+
+        let _result = data.send().await;
+
+        test_agent
+            .assert_snapshot("compare_v04_trace_snapshot_test")
+            .await;
+    }
+
+    #[cfg_attr(miri, ignore)]
+    #[tokio::test]
+    #[cfg(target_os = "linux")]
+    // Validate that we can correctly send traces to the agent via UDS
+    async fn uds_snapshot_test() {
+        let relative_snapshot_path = "trace-utils/tests/snapshots/";
+
+        // Create a temporary directory for the socket to be mounted in the test agent container
+        let socket_dir = tempfile::Builder::new()
+            .prefix("dd-trace-test-")
+            .tempdir()
+            .expect("Failed to create temporary directory");
+
+        std::fs::set_permissions(socket_dir.path(), Permissions::from_mode(0o755))
+            .expect("Failed to set directory permissions");
+
+        let absolute_socket_dir_path = socket_dir
+            .path()
+            .to_str()
+            .expect("Failed to convert path to string")
+            .to_owned();
+
+        let absolute_socket_path = socket_dir.path().join("apm.socket");
+        let socket_path = socket_path_to_uri(absolute_socket_path.as_path());
+        let socket_uri = socket_path.unwrap();
+
+        let mut parts = socket_uri.into_parts();
+        let p_q = match parts.path_and_query {
+            None => None,
+            Some(pq) => {
+                let path = pq.path();
+                let path = path.strip_suffix('/').unwrap_or(path);
+                Some(
+                    format!(
+                        "{path}/v0.4/traces?test_session_token=compare_v04_trace_snapshot_test"
+                    )
+                    .parse()
+                    .unwrap(),
+                )
+            }
+        };
+        parts.path_and_query = p_q;
+
+        let url = Uri::from_parts(parts).unwrap();
+
+        let test_agent = DatadogTestAgent::new(
+            Some(relative_snapshot_path),
+            Some(&absolute_socket_dir_path),
+        )
+        .await;
+
+        let endpoint = Endpoint::from_url(url);
+
+        let header_tags = TracerHeaderTags {
+            lang: "test-lang",
+            lang_version: "2.0",
+            lang_interpreter: "interpreter",
+            lang_vendor: "vendor",
+            tracer_version: "1.0",
+            container_id: "id",
+            ..Default::default()
+        };
 
         let mut span_1 = create_test_no_alloc_span(1234, 12342, 12341, 1, false);
         span_1.metrics.insert(