feat(bottlecap): add EventBridge-SNS inferred spans

bottlecap/src/lifecycle/invocation/span_inferrer.rs

@@ -106,6 +106,27 @@ impl SpanInferrer {
             if let Some(t) = SnsRecord::new(payload_value.clone()) {
                 t.enrich_span(&mut inferred_span);
 
+                if let Some(message) = &t.sns.message {
+                    if let Ok(event_bridge_wrapper_message) =
+                        serde_json::from_str::<EventBridgeEvent>(message)
+                    {
+                        let mut wrapped_inferred_span = Span {
+                            span_id: Self::generate_span_id(),
+                            ..Default::default()
+                        };
+
+                        event_bridge_wrapper_message.enrich_span(&mut wrapped_inferred_span);
+                        inferred_span
+                            .meta
+                            .extend(event_bridge_wrapper_message.get_tags());
+
+                        wrapped_inferred_span.duration =
+                            inferred_span.start - wrapped_inferred_span.start;
+
+                        self.wrapped_inferred_span = Some(wrapped_inferred_span);
+                    }
+                }
+
                 trigger = Some(Box::new(t));
             }
         } else if DynamoDbRecord::is_match(payload_value) {

bottlecap/src/lifecycle/invocation/triggers/dynamodb_event.rs

@@ -142,7 +142,7 @@ mod tests {
 
         let expected = DynamoDbRecord {
             dynamodb: DynamoDbEntity {
-                approximate_creation_date_time: 1428537600.0,
+                approximate_creation_date_time: 1_428_537_600.0,
                 size_bytes: 26,
                 stream_view_type: String::from("NEW_AND_OLD_IMAGES"),
             },

bottlecap/src/lifecycle/invocation/triggers/event_bridge_event.rs

@@ -200,7 +200,7 @@ mod tests {
 
         assert_eq!(span.resource, "testBus");
         // Seconds resolution
-        assert_eq!(span.start, 1731140535000000000);
+        assert_eq!(span.start, 1_731_140_535_000_000_000);
     }
 
     #[test]

bottlecap/src/lifecycle/invocation/triggers/sns_event.rs

@@ -8,7 +8,10 @@ use tracing::debug;
 use crate::lifecycle::invocation::{
     base64_to_string,
     processor::MS_TO_NS,
-    triggers::{Trigger, DATADOG_CARRIER_KEY, FUNCTION_TRIGGER_EVENT_SOURCE_TAG},
+    triggers::{
+        event_bridge_event::EventBridgeEvent, Trigger, DATADOG_CARRIER_KEY,
+        FUNCTION_TRIGGER_EVENT_SOURCE_TAG,
+    },
 };
 
 #[derive(Debug, Serialize, Deserialize, Clone, PartialEq)]
@@ -39,6 +42,8 @@ pub struct SnsEntity {
     pub timestamp: DateTime<Utc>,
     #[serde(rename = "Subject")]
     pub subject: Option<String>,
+    #[serde(rename = "Message")]
+    pub message: Option<String>,
 }
 
 #[derive(Debug, Serialize, Deserialize, Clone, PartialEq)]
@@ -50,7 +55,7 @@ pub struct MessageAttribute {
 }
 
 impl Trigger for SnsRecord {
-    fn new(payload: serde_json::Value) -> Option<Self> {
+    fn new(payload: Value) -> Option<Self> {
         match payload.get("Records").and_then(Value::as_array) {
             Some(records) => match serde_json::from_value::<SnsRecord>(records[0].clone()) {
                 Ok(record) => Some(record),
@@ -145,6 +150,10 @@ impl Trigger for SnsRecord {
                     debug!("Unsupported type in SNS message attribute");
                 }
             }
+        } else if let Some(event_bridge_message) = &self.sns.message {
+            if let Ok(event) = serde_json::from_str::<EventBridgeEvent>(event_bridge_message) {
+                return event.get_carrier();
+            }
         }
 
         HashMap::new()
@@ -187,6 +196,7 @@ mod tests {
                     .unwrap()
                     .with_timezone(&Utc),
                 subject: None,
+                message: Some("Asynchronously invoking a Lambda function with SNS.".to_string()),
             },
         };
 
@@ -302,4 +312,33 @@ mod tests {
 
         assert_eq!(carrier, expected);
     }
+
+    #[test]
+    fn test_get_carrier_from_event_bridge() {
+        let json = read_json_file("eventbridge_sns_event.json");
+        let payload = serde_json::from_str(&json).expect("Failed to deserialize into Value");
+        println!("{payload:?}");
+        let event = SnsRecord::new(payload).expect("Failed to deserialize SnsRecord");
+        let carrier = event.get_carrier();
+
+        let expected = HashMap::from([
+            (
+                "x-datadog-resource-name".to_string(),
+                "test-bus".to_string(),
+            ),
+            ("x-datadog-trace-id".to_string(), "12345".to_string()),
+            (
+                "x-datadog-start-time".to_string(),
+                "1726515840997".to_string(),
+            ),
+            ("x-datadog-sampling-priority".to_string(), "1".to_string()),
+            ("x-datadog-parent-id".to_string(), "67890".to_string()),
+            (
+                "x-datadog-tags".to_string(),
+                "_dd.p.dm=-1,_dd.p.tid=123567890".to_string(),
+            ),
+        ]);
+
+        assert_eq!(carrier, expected);
+    }
 }

bottlecap/tests/payloads/eventbridge_sns_event.json

@@ -0,0 +1,17 @@
+{
+  "Records":[
+    {
+      "Sns":{
+        "MessageId":"12345678-90abc-def-1234-567890abcdef",
+        "Type":"Notification",
+        "TopicArn":"arn:aws:sns:us-east-1:123456789012:test-notifier",
+        "MessageAttributes":{
+
+        },
+        "Timestamp":"2024-09-16T19:44:01.713Z",
+        "Subject":"",
+        "Message":"{\"version\":\"0\",\"id\":\"12345678-90abc-def-1234-567890abcdef\",\"detail-type\":\"TestDetail\",\"source\":\"com.test.source\",\"account\":\"12345667890\",\"time\":\"2024-09-16T19:44:01Z\",\"region\":\"us-east-1\",\"resources\":[],\"detail\":{\"foo\":\"bar\",\"_datadog\":{\"x-datadog-trace-id\":\"12345\",\"x-datadog-parent-id\":\"67890\",\"x-datadog-sampling-priority\":\"1\",\"x-datadog-start-time\":\"1726515840997\",\"x-datadog-resource-name\":\"test-bus\",\"x-datadog-tags\":\"_dd.p.dm=-1,_dd.p.tid=123567890\"}}}"
+      }
+    }
+  ]
+}