feat(forwarder): add SQS support as event source for S3 notifications

Support S3 event notifications delivered via SQS queues (S3 -> SQS -> Lambda
and S3 -> SNS -> SQS -> Lambda). The forwarder detects SQS event records,
unwraps the inner S3 event from the SQS body, and delegates to the existing
S3EventHandler. Each SQS record gets fresh metadata to avoid cross-contamination
across batch items.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: LorisFriedel

aws/logs_monitoring/steps/enums.py

@@ -62,6 +62,7 @@ class AwsEventType(Enum):
     KINESIS = "kinesis"
     S3 = "s3"
     SNS = "sns"
+    SQS = "sqs"
     UNKNOWN = "unknown"
 
     def __str__(self):

aws/logs_monitoring/steps/parsing.py

@@ -39,6 +39,9 @@ def parse(event, context, cache_layer):
             case AwsEventType.S3:
                 s3_handler = S3EventHandler(context, metadata, cache_layer)
                 events = s3_handler.handle(event)
+            case AwsEventType.SQS:
+                events = sqs_handler(event, context, cache_layer)
+                return collect_and_count(events)
             case AwsEventType.EVENTBRIDGE_S3:
                 events = eventbridge_s3_handler(event, context, metadata, cache_layer)
             case AwsEventType.EVENTS:
@@ -78,6 +81,8 @@ def parse_event_type(event):
             return AwsEventType.SNS
         elif str(AwsEventType.KINESIS) in record:
             return AwsEventType.KINESIS
+        elif record.get("eventSource") == "aws:sqs":
+            return AwsEventType.SQS
     elif str(AwsEventType.AWSLOGS) in event:
         return AwsEventType.AWSLOGS
     elif "detail" in event:
@@ -90,6 +95,52 @@ def parse_event_type(event):
     raise Exception("Event type not supported (see #Event supported section)")
 
 
+# Handle S3 events delivered via SQS (S3 -> SQS or S3 -> SNS -> SQS)
+def sqs_handler(event, context, cache_layer):
+    for record in event["Records"]:
+        inner_event = _extract_inner_event_from_sqs(record)
+        if inner_event is None:
+            continue
+        # Fresh metadata per SQS record: S3EventHandler mutates metadata
+        # (DD_SOURCE, tags, service), so each record needs its own copy.
+        metadata = generate_metadata(context)
+        s3_handler = S3EventHandler(context, metadata, cache_layer)
+        for log_event in s3_handler.handle(inner_event):
+            if isinstance(log_event, dict):
+                yield merge_dicts(log_event, metadata)
+            elif isinstance(log_event, str):
+                yield merge_dicts({"message": log_event}, metadata)
+
+
+def _extract_inner_event_from_sqs(sqs_record):
+    try:
+        body = json.loads(sqs_record["body"])
+    except (json.JSONDecodeError, KeyError, TypeError):
+        logger.warning("SQS record has missing or malformed body, skipping")
+        return None
+
+    # Direct S3 event: body contains Records[0].s3
+    if _contains_s3_records(body):
+        return body
+
+    # SNS-wrapped S3 event: body.Type == "Notification" and body.Message contains S3 event
+    if body.get("Type") == "Notification":
+        try:
+            message = json.loads(body.get("Message", ""))
+            if _contains_s3_records(message):
+                return message
+        except (json.JSONDecodeError, TypeError):
+            pass
+
+    logger.warning("SQS record body does not contain a recognized S3 event, skipping")
+    return None
+
+
+def _contains_s3_records(event):
+    records = event.get("Records")
+    return isinstance(records, list) and len(records) > 0 and records[0].get("s3")
+
+
 # Handle S3 event over EventBridge
 def eventbridge_s3_handler(event, context, metadata, cache_layer):
     """

aws/logs_monitoring/template.yaml

@@ -754,6 +754,13 @@ Resources:
                   Resource: "*"
                   Effect: Allow
                 - !Ref AWS::NoValue
+              # Required for Lambda event source mappings that poll SQS queues
+              - Action:
+                  - sqs:ReceiveMessage
+                  - sqs:DeleteMessage
+                  - sqs:GetQueueAttributes
+                Resource: "*"
+                Effect: Allow
       Tags:
         - Value: !FindInMap [Constants, DdForwarder, Version]
           Key: dd_forwarder_version

aws/logs_monitoring/tests/events/sqs_s3.json

@@ -0,0 +1,20 @@
+{
+  "Records": [
+    {
+      "messageId": "059f36b4-87a3-44ab-83d2-661975830a7d",
+      "receiptHandle": "AQEBwJnKyrHigUMZj6rYigCgxlaS3SLy0a...",
+      "body": "{\"Records\":[{\"eventVersion\":\"2.1\",\"eventSource\":\"aws:s3\",\"awsRegion\":\"us-east-1\",\"eventTime\":\"2024-01-15T12:00:00.000Z\",\"eventName\":\"ObjectCreated:Put\",\"s3\":{\"bucket\":{\"name\":\"my-bucket\",\"arn\":\"arn:aws:s3:::my-bucket\"},\"object\":{\"key\":\"my-key.log\",\"size\":1234}}}]}",
+      "attributes": {
+        "ApproximateReceiveCount": "1",
+        "SentTimestamp": "1545082649636",
+        "SenderId": "AIDAIENQZJOLO23YVJ4VO",
+        "ApproximateFirstReceiveTimestamp": "1545082649636"
+      },
+      "messageAttributes": {},
+      "md5OfBody": "e4e68fb7bd0e697a0ae8f1bb342846b3",
+      "eventSource": "aws:sqs",
+      "eventSourceARN": "arn:aws:sqs:us-east-1:123456789012:my-queue",
+      "awsRegion": "us-east-1"
+    }
+  ]
+}

aws/logs_monitoring/tests/events/sqs_sns_s3.json

@@ -0,0 +1,20 @@
+{
+  "Records": [
+    {
+      "messageId": "059f36b4-87a3-44ab-83d2-661975830a7d",
+      "receiptHandle": "AQEBwJnKyrHigUMZj6rYigCgxlaS3SLy0a...",
+      "body": "{\"Type\":\"Notification\",\"MessageId\":\"a1b2c3d4-e5f6-7890-abcd-ef1234567890\",\"TopicArn\":\"arn:aws:sns:us-east-1:123456789012:my-topic\",\"Subject\":\"Amazon S3 Notification\",\"Message\":\"{\\\"Records\\\":[{\\\"eventVersion\\\":\\\"2.1\\\",\\\"eventSource\\\":\\\"aws:s3\\\",\\\"awsRegion\\\":\\\"us-east-1\\\",\\\"eventTime\\\":\\\"2024-01-15T12:00:00.000Z\\\",\\\"eventName\\\":\\\"ObjectCreated:Put\\\",\\\"s3\\\":{\\\"bucket\\\":{\\\"name\\\":\\\"my-bucket\\\",\\\"arn\\\":\\\"arn:aws:s3:::my-bucket\\\"},\\\"object\\\":{\\\"key\\\":\\\"my-key.log\\\",\\\"size\\\":1234}}}]}\",\"Timestamp\":\"2024-01-15T12:00:00.000Z\",\"SignatureVersion\":\"1\"}",
+      "attributes": {
+        "ApproximateReceiveCount": "1",
+        "SentTimestamp": "1545082649636",
+        "SenderId": "AIDAIENQZJOLO23YVJ4VO",
+        "ApproximateFirstReceiveTimestamp": "1545082649636"
+      },
+      "messageAttributes": {},
+      "md5OfBody": "e4e68fb7bd0e697a0ae8f1bb342846b3",
+      "eventSource": "aws:sqs",
+      "eventSourceARN": "arn:aws:sqs:us-east-1:123456789012:my-queue",
+      "awsRegion": "us-east-1"
+    }
+  ]
+}

aws/logs_monitoring/tests/test_parsing.py

@@ -1,3 +1,4 @@
+import json
 import unittest
 from unittest.mock import MagicMock, patch
 
@@ -7,6 +8,28 @@
 from steps.parsing import parse, parse_event_type
 
 
+class Context:
+    function_version = "$LATEST"
+    invoked_function_arn = (
+        "arn:aws:lambda:us-east-1:123456789012:function:datadog-forwarder"
+    )
+    function_name = "datadog-forwarder"
+    memory_limit_in_mb = "128"
+
+
+def _make_s3_event(bucket, key):
+    return {"Records": [{"s3": {"bucket": {"name": bucket}, "object": {"key": key}}}]}
+
+
+def _make_sqs_record(body, message_id="msg-1"):
+    return {
+        "messageId": message_id,
+        "body": body if isinstance(body, str) else json.dumps(body),
+        "eventSource": "aws:sqs",
+        "eventSourceARN": "arn:aws:sqs:us-east-1:123456789012:q",
+    }
+
+
 class TestParseEventSource(unittest.TestCase):
     def test_aws_source_if_none_found(self):
         self.assertEqual(parse_event_source({}, "asdfalsfhalskjdfhalsjdf"), "aws")
@@ -187,7 +210,7 @@ def test_get_service_from_tags_removing_duplicates(self):
 
 class TestParseEventType(unittest.TestCase):
     def test_parse_eventbridge_s3_event_type(self):
-        """Test that EventBridge S3 events are correctly identified as EventBridge S3 type"""
+        """EventBridge S3 events are correctly identified"""
         eventbridge_s3_event = {
             "version": "0",
             "id": "test-event-id",
@@ -202,53 +225,147 @@ def test_parse_eventbridge_s3_event_type(self):
                 "object": {"key": "my-key.log"},
             },
         }
-
-        event_type = parse_event_type(eventbridge_s3_event)
-        self.assertEqual(event_type, AwsEventType.EVENTBRIDGE_S3)
+        self.assertEqual(parse_event_type(eventbridge_s3_event), AwsEventType.EVENTBRIDGE_S3)
 
     def test_parse_direct_s3_event_type(self):
-        """Test that direct S3 events are still correctly identified as S3 type"""
-        direct_s3_event = {
-            "Records": [
-                {
-                    "s3": {
-                        "bucket": {"name": "my-bucket"},
-                        "object": {"key": "my-key"},
-                    }
-                }
-            ]
-        }
-
-        event_type = parse_event_type(direct_s3_event)
-        self.assertEqual(event_type, AwsEventType.S3)
+        """Direct S3 events are correctly identified"""
+        self.assertEqual(
+            parse_event_type(_make_s3_event("my-bucket", "my-key")), AwsEventType.S3
+        )
 
     def test_parse_non_s3_eventbridge_event_type(self):
-        """Test that non-S3 EventBridge events are identified as EVENTS type"""
+        """Non-S3 EventBridge events are identified as EVENTS type"""
         eventbridge_other_event = {
             "version": "0",
             "detail-type": "EC2 Instance State-change Notification",
             "source": "aws.ec2",
             "detail": {"instance-id": "i-1234567890abcdef0", "state": "terminated"},
         }
+        self.assertEqual(parse_event_type(eventbridge_other_event), AwsEventType.EVENTS)
 
-        event_type = parse_event_type(eventbridge_other_event)
-        self.assertEqual(event_type, AwsEventType.EVENTS)
+    def test_parse_sqs_event_type(self):
+        """SQS events are correctly identified"""
+        sqs_event = {"Records": [_make_sqs_record(_make_s3_event("b", "k"))]}
+        self.assertEqual(parse_event_type(sqs_event), AwsEventType.SQS)
 
+    def test_direct_s3_event_not_detected_as_sqs(self):
+        """Direct S3 events must still be detected as S3, not SQS"""
+        self.assertEqual(
+            parse_event_type(_make_s3_event("my-bucket", "my-key")), AwsEventType.S3
+        )
 
-class TestEventBridgeS3Parsing(unittest.TestCase):
-    class Context:
-        function_version = "$LATEST"
-        invoked_function_arn = (
-            "arn:aws:lambda:us-east-1:123456789012:function:datadog-forwarder"
+    def test_sns_event_not_detected_as_sqs(self):
+        """SNS events must still be detected as SNS, not SQS"""
+        sns_event = {"Records": [{"Sns": {"Message": "hello"}}]}
+        self.assertEqual(parse_event_type(sns_event), AwsEventType.SNS)
+
+    def test_kinesis_event_not_detected_as_sqs(self):
+        """Kinesis events must still be detected as Kinesis, not SQS"""
+        kinesis_event = {"Records": [{"kinesis": {"data": "base64data"}}]}
+        self.assertEqual(parse_event_type(kinesis_event), AwsEventType.KINESIS)
+
+
+class TestSQSEventParsing(unittest.TestCase):
+    @patch("steps.parsing.S3EventHandler")
+    def test_parse_sqs_s3_event(self, mock_s3_handler_cls):
+        """S3 event delivered via SQS is unwrapped and forwarded to S3EventHandler"""
+        mock_s3_handler = mock_s3_handler_cls.return_value
+        mock_s3_handler.handle.return_value = iter([{"message": "log line"}])
+
+        sqs_event = {
+            "Records": [_make_sqs_record(_make_s3_event("my-bucket", "my-key.log"))]
+        }
+
+        result = parse(sqs_event, Context(), MagicMock())
+
+        mock_s3_handler.handle.assert_called_once()
+        inner_event = mock_s3_handler.handle.call_args.args[0]
+        self.assertEqual(inner_event["Records"][0]["s3"]["bucket"]["name"], "my-bucket")
+        self.assertEqual(len(result), 1)
+        self.assertIn("ddsourcecategory", result[0])
+        self.assertIn("aws", result[0])
+        self.assertIn("invoked_function_arn", result[0]["aws"])
+
+    @patch("steps.parsing.S3EventHandler")
+    def test_parse_sqs_sns_s3_event(self, mock_s3_handler_cls):
+        """S3 event delivered via SNS -> SQS is unwrapped and forwarded to S3EventHandler"""
+        mock_s3_handler = mock_s3_handler_cls.return_value
+        mock_s3_handler.handle.return_value = iter([{"message": "log line"}])
+
+        sns_body = {
+            "Type": "Notification",
+            "MessageId": "a1b2c3d4",
+            "TopicArn": "arn:aws:sns:us-east-1:123456789012:my-topic",
+            "Message": json.dumps(_make_s3_event("sns-bucket", "sns-key.log")),
+        }
+        sqs_event = {"Records": [_make_sqs_record(sns_body)]}
+
+        result = parse(sqs_event, Context(), MagicMock())
+
+        mock_s3_handler.handle.assert_called_once()
+        inner_event = mock_s3_handler.handle.call_args.args[0]
+        self.assertEqual(
+            inner_event["Records"][0]["s3"]["bucket"]["name"], "sns-bucket"
         )
-        function_name = "datadog-forwarder"
-        memory_limit_in_mb = "128"
+        self.assertEqual(len(result), 1)
+
+    @patch("steps.parsing.S3EventHandler")
+    def test_parse_sqs_batch_multiple_records(self, mock_s3_handler_cls):
+        """Multiple SQS records in a single batch are all processed"""
+        mock_s3_handler = mock_s3_handler_cls.return_value
+        mock_s3_handler.handle.side_effect = [
+            iter([{"message": "line1"}]),
+            iter([{"message": "line2"}]),
+        ]
+
+        sqs_event = {
+            "Records": [
+                _make_sqs_record(_make_s3_event("b1", "k1"), message_id="msg-1"),
+                _make_sqs_record(_make_s3_event("b2", "k2"), message_id="msg-2"),
+            ]
+        }
+
+        result = parse(sqs_event, Context(), MagicMock())
+
+        self.assertEqual(mock_s3_handler.handle.call_count, 2)
+        self.assertEqual(len(result), 2)
+
+    @patch("steps.parsing.S3EventHandler")
+    def test_parse_sqs_malformed_body_skipped(self, mock_s3_handler_cls):
+        """SQS records with malformed body are skipped without crashing"""
+        mock_s3_handler = mock_s3_handler_cls.return_value
+        mock_s3_handler.handle.return_value = iter([{"message": "ok"}])
 
+        sqs_event = {
+            "Records": [
+                _make_sqs_record("not valid json", message_id="bad"),
+                _make_sqs_record(_make_s3_event("b", "k"), message_id="good"),
+            ]
+        }
+
+        result = parse(sqs_event, Context(), MagicMock())
+
+        mock_s3_handler.handle.assert_called_once()
+        self.assertEqual(len(result), 1)
+
+    @patch("steps.parsing.S3EventHandler")
+    def test_parse_sqs_unrecognized_body_skipped(self, mock_s3_handler_cls):
+        """SQS records with valid JSON but unrecognized content are skipped"""
+        mock_s3_handler = mock_s3_handler_cls.return_value
+
+        sqs_event = {"Records": [_make_sqs_record({"foo": "bar"})]}
+
+        result = parse(sqs_event, Context(), MagicMock())
+
+        mock_s3_handler.handle.assert_not_called()
+        self.assertEqual(len(result), 0)
+
+
+class TestEventBridgeS3Parsing(unittest.TestCase):
     @patch("steps.parsing.S3EventHandler")
     def test_parse_normalizes_eventbridge_s3_event_before_s3_handler(
         self, mock_s3_handler_cls
     ):
-        # Arrange: handler yields one log line; we only care about the input event it received
         mock_s3_handler = mock_s3_handler_cls.return_value
         mock_s3_handler.handle.return_value = iter([{"message": "ok"}])
 
@@ -262,12 +379,8 @@ def test_parse_normalizes_eventbridge_s3_event_before_s3_handler(
             },
         }
 
-        cache_layer = MagicMock()
-
-        # Act
-        _ = parse(eventbridge_event, self.Context(), cache_layer)
+        parse(eventbridge_event, Context(), MagicMock())
 
-        # Assert: parse() passed a canonical S3-shaped event into the S3 handler
         mock_s3_handler.handle.assert_called_once()
         (normalized_event,) = mock_s3_handler.handle.call_args.args