feat(sqs): extract SNS-to-SQS trace context

Read `_datadog` from both string and binary SQS message attributes so
SNS-propagated context can be extracted.

Author: ygree

instrumentation/Cargo.lock

@@ -2,7 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use std::collections::HashMap;
-use std::sync::{Arc, Mutex};
+use std::sync::{Arc, Mutex, OnceLock};
 
 use aws_credential_types::provider::SharedCredentialsProvider;
 use aws_credential_types::Credentials;
@@ -17,6 +17,8 @@ use opentelemetry_sdk::trace::{
 use wiremock::matchers::any;
 use wiremock::{Mock, MockServer as WireMockServer, Request, Respond, ResponseTemplate};
 
+const HTTP_URL: &str = "http.url";
+
 #[derive(Clone)]
 struct CaptureBodyResponder {
     status: u16,
@@ -56,27 +58,28 @@ impl MockAwsServer {
 /// 2. Build the AWS client with `harness.sdk_config()`.
 /// 3. Execute SDK calls, then assert on `harness.finished_spans()` and `harness.server.bodies()`.
 ///
-/// Tests should remain serialized (`#[serial]`) because harness creation sets global
-/// OpenTelemetry tracer/propagator state via `init_test_tracer`.
+/// The global OpenTelemetry tracer and propagator are initialized once per test
+/// binary. Each harness filters exported spans to requests sent to its mock
+/// server, so tests can run concurrently.
 pub struct TestHarness {
-    pub exporter: InMemorySpanExporter,
+    exporter: InMemorySpanExporter,
     pub server: MockAwsServer,
 }
 
 impl TestHarness {
-    /// Initializes tracing and starts a mock AWS endpoint returning `200 OK`
+    /// Ensures tracing is initialized and starts a mock AWS endpoint returning `200 OK`
     /// for every request.
     pub async fn ok() -> Self {
         Self::from_status(200).await
     }
 
-    /// Initializes tracing and starts a mock AWS endpoint returning `200 OK`
+    /// Ensures tracing is initialized and starts a mock AWS endpoint returning `200 OK`
     /// with the supplied response body.
     pub async fn ok_with_body(response_body: impl Into<String>) -> Self {
         Self::from_status_and_body(200, response_body).await
     }
 
-    /// Initializes tracing and starts a mock AWS endpoint returning
+    /// Ensures tracing is initialized and starts a mock AWS endpoint returning
     /// `400 Bad Request` for every request.
     pub async fn bad_request() -> Self {
         Self::from_status(400).await
@@ -89,7 +92,12 @@ impl TestHarness {
 
     /// Returns all finished spans currently exported by this harness.
     pub fn finished_spans(&self) -> Vec<SpanData> {
-        self.exporter.get_finished_spans().unwrap()
+        self.exporter
+            .get_finished_spans()
+            .unwrap()
+            .into_iter()
+            .filter(|span| span_targets_server(span, &self.server.url))
+            .collect()
     }
 
     async fn from_status(status: u16) -> Self {
@@ -135,15 +143,35 @@ pub async fn mock_aws_with_body(status: u16, response_body: impl Into<String>) -
 }
 
 /// Registers an `InMemorySpanExporter` as the global tracer provider and
-/// a `TraceContextPropagator` as the global propagator.
+/// a `TraceContextPropagator` as the global propagator once per test binary.
 pub fn init_test_tracer() -> InMemorySpanExporter {
-    let exporter = InMemorySpanExporter::default();
-    let provider = SdkTracerProvider::builder()
-        .with_span_processor(SimpleSpanProcessor::new(exporter.clone()))
-        .build();
-    global::set_tracer_provider(provider);
-    global::set_text_map_propagator(TraceContextPropagator::new());
-    exporter
+    static EXPORTER: OnceLock<InMemorySpanExporter> = OnceLock::new();
+
+    EXPORTER
+        .get_or_init(|| {
+            let exporter = InMemorySpanExporter::default();
+            let provider = SdkTracerProvider::builder()
+                .with_span_processor(SimpleSpanProcessor::new(exporter.clone()))
+                .build();
+            global::set_tracer_provider(provider);
+            global::set_text_map_propagator(TraceContextPropagator::new());
+            exporter
+        })
+        .clone()
+}
+
+fn span_targets_server(span: &SpanData, server_url: &str) -> bool {
+    span.attributes.iter().any(|kv| {
+        kv.key.as_str() == HTTP_URL && url_targets_server(&kv.value.to_string(), server_url)
+    })
+}
+
+fn url_targets_server(url: &str, server_url: &str) -> bool {
+    let Some(rest) = url.strip_prefix(server_url) else {
+        return false;
+    };
+
+    rest.is_empty() || rest.starts_with('/') || rest.starts_with('?')
 }
 
 /// Builds an `SdkConfig` pointing at `endpoint` with test credentials.
@@ -179,3 +207,22 @@ pub fn split_traceparent(tp: &str) -> (String, String) {
     let parts: Vec<&str> = tp.splitn(4, '-').collect();
     (parts[1].to_string(), parts[2].to_string())
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn url_targets_server_requires_url_boundary() {
+        let server_url = "http://127.0.0.1:1234";
+
+        assert!(url_targets_server(server_url, server_url));
+        assert!(url_targets_server("http://127.0.0.1:1234/", server_url));
+        assert!(url_targets_server(
+            "http://127.0.0.1:1234/?Action=PutTargets",
+            server_url
+        ));
+        assert!(!url_targets_server("http://127.0.0.1:12345/", server_url));
+        assert!(!url_targets_server("http://127.0.0.1:123", server_url));
+    }
+}

instrumentation/datadog-aws-core-test-utils/src/integration_test_helpers.rs

@@ -27,5 +27,4 @@ opentelemetry = { workspace = true }
 [dev-dependencies]
 datadog-aws-core-test-utils = { path = "../datadog-aws-core-test-utils" }
 aws-types = { workspace = true }
-serial_test = "3"
 tokio = { workspace = true, features = ["rt", "macros", "net", "io-util"] }

instrumentation/datadog-aws-eventbridge/Cargo.toml

@@ -2,17 +2,12 @@
 // SPDX-License-Identifier: Apache-2.0
 
 //! Integration tests for EventBridgeInterceptor.
-//!
-//! Tests are serialized with `#[serial]` because `init_test_tracer` sets the
-//! global OTel tracer provider and propagator. Concurrent tests would race on
-//! that global state, causing spans to land in the wrong exporter.
 
 use aws_sdk_eventbridge::types::PutEventsRequestEntry;
 use aws_types::SdkConfig;
 use datadog_aws_core_test_utils::integration_test_helpers::{
     span_attrs, split_traceparent, TestHarness,
 };
-use serial_test::serial;
 
 use datadog_aws_eventbridge::ConfigExt as _;
 
@@ -24,7 +19,6 @@ fn eventbridge_client(cfg: &SdkConfig) -> aws_sdk_eventbridge::Client {
 }
 
 #[tokio::test]
-#[serial]
 async fn eventbridge_put_events_creates_span_and_injects_detail() {
     let harness = TestHarness::ok().await;
     let client = eventbridge_client(&harness.sdk_config());
@@ -68,7 +62,6 @@ async fn eventbridge_put_events_creates_span_and_injects_detail() {
 }
 
 #[tokio::test]
-#[serial]
 async fn eventbridge_put_events_with_bus_name_creates_span() {
     let harness = TestHarness::ok().await;
     let client = eventbridge_client(&harness.sdk_config());
@@ -88,7 +81,6 @@ async fn eventbridge_put_events_with_bus_name_creates_span() {
 }
 
 #[tokio::test]
-#[serial]
 async fn eventbridge_put_events_multi_entry_creates_single_span() {
     let harness = TestHarness::ok().await;
     let client = eventbridge_client(&harness.sdk_config());
@@ -118,7 +110,6 @@ async fn eventbridge_put_events_multi_entry_creates_single_span() {
 }
 
 #[tokio::test]
-#[serial]
 async fn eventbridge_put_rule_creates_span_with_rulename() {
     let harness = TestHarness::ok().await;
     let client = eventbridge_client(&harness.sdk_config());
@@ -133,7 +124,6 @@ async fn eventbridge_put_rule_creates_span_with_rulename() {
 }
 
 #[tokio::test]
-#[serial]
 async fn eventbridge_put_targets_creates_span_with_rulename() {
     let harness = TestHarness::ok().await;
     let client = eventbridge_client(&harness.sdk_config());

instrumentation/datadog-aws-eventbridge/tests/integration.rs

@@ -26,5 +26,4 @@ opentelemetry = { workspace = true }
 [dev-dependencies]
 datadog-aws-core-test-utils = { path = "../datadog-aws-core-test-utils" }
 aws-types = { workspace = true }
-serial_test = "3"
 tokio = { workspace = true, features = ["rt", "macros", "net", "io-util"] }