Add support for S3 Object Created events distributed using Eventbridge (#1052)

blockmar · domalessi · ge0Aja · web-flow · commit 2df78acec506 · 2026-02-10T15:57:07.000+01:00
* Add support for S3 Object Created events distrubuted using Eventbridge * Update aws/logs_monitoring/README.md Co-authored-by: domalessi <111786334+domalessi@users.noreply.github.com> * Update aws/logs_monitoring/README.md Co-authored-by: domalessi <111786334+domalessi@users.noreply.github.com> * Update aws/logs_monitoring/README.md Co-authored-by: domalessi <111786334+domalessi@users.noreply.github.com> * Update aws/logs_monitoring/README.md Co-authored-by: domalessi <111786334+domalessi@users.noreply.github.com> * Update aws/logs_monitoring/README.md Co-authored-by: domalessi <111786334+domalessi@users.noreply.github.com> * Update aws/logs_monitoring/README.md Co-authored-by: domalessi <111786334+domalessi@users.noreply.github.com> * Moved event parsing/rewrite login to parsing.py * Refactored tests not to be end-to-end * Refactored EventBridge S3 events to a new event type. Extracted handler, similar to kinesis. * Apply suggestion from @blockmar * Update test_s3_handler.py Removed added tests, since s3_handler.py is no longer modified. * Update aws/logs_monitoring/steps/parsing.py Co-authored-by: Georgi <georgi.ajaeiya@datadoghq.com> * Black code formatting --------- Co-authored-by: domalessi <111786334+domalessi@users.noreply.github.com> Co-authored-by: Georgi <georgi.ajaeiya@datadoghq.com>
diff --git a/aws/logs_monitoring/README.md b/aws/logs_monitoring/README.md
@@ -290,6 +290,88 @@ In case you encounter the following error when creating S3 triggers, we recommen
 An error occurred when creating the trigger: Configuration is ambiguously defined. Cannot have overlapping suffixes in two rules if the prefixes are overlapping for the same event type.
 ```
 
+### Using EventBridge for S3 event notifications
+
+The Datadog Forwarder supports S3 event notifications delivered through Amazon EventBridge, in addition to direct S3 event notifications. EventBridge provides advanced filtering and supports more complex integration patterns.
+
+#### Supported event types
+
+- `Object Created` events from S3 delivered through EventBridge
+
+#### Configuration
+
+To use EventBridge for S3 event notifications:
+
+1. **Enable EventBridge notifications on your S3 bucket:**
+
+   Using AWS CLI:
+   ```bash
+   aws s3api put-bucket-notification-configuration \
+       --bucket my-bucket \
+       --notification-configuration '{
+           "EventBridgeConfiguration": {}
+       }'
+   ```
+
+   Using the AWS Console:
+   - Navigate to your S3 bucket.
+   - Go to **Properties** > **Event notifications**.
+   - Enable **Amazon EventBridge**.
+
+2. **Create an EventBridge rule to trigger the Forwarder:**
+
+   Using AWS CLI:
+   ```bash
+   aws events put-rule \
+       --name s3-logs-to-datadog \
+       --event-pattern '{
+           "source": ["aws.s3"],
+           "detail-type": ["Object Created"]
+       }'
+
+   aws events put-targets \
+       --rule s3-logs-to-datadog \
+       --targets "Id"="1","Arn"="arn:aws:lambda:REGION:ACCOUNT:function:datadog-forwarder"
+   ```
+
+   Using the AWS Console:
+   - Navigate to Amazon EventBridge.
+   - Create a new rule.
+   - Set **Event source** to `aws.s3`.
+   - Set **Detail type** to `Object Created`.
+   - Add your Datadog Forwarder Lambda as a target.
+
+3. **Add advanced filtering (optional):**
+
+   You can filter events by bucket name or object key patterns:
+   ```json
+   {
+     "source": ["aws.s3"],
+     "detail-type": ["Object Created"],
+     "detail": {
+       "bucket": {
+         "name": ["my-logs-bucket"]
+       },
+       "object": {
+         "key": [{
+           "prefix": "logs/"
+         }]
+       }
+     }
+   }
+   ```
+
+#### Benefits of using EventBridge
+
+- **Advanced filtering**: Filter events based on bucket names, object key patterns, object size, and more
+- **Multiple targets**: Route the same S3 events to multiple Lambda functions or services
+- **Cross-account routing**: Forward events across AWS accounts
+- **Better delivery guarantees**: Use built-in retry and dead-letter queue capabilities
+
+#### Backward compatibility
+
+Direct S3 event notifications and SNS-wrapped S3 events continue to work without any changes. You can use EventBridge alongside existing triggers.
+
 ## Contributing
 
 We love pull requests. Here's a quick guide.
diff --git a/aws/logs_monitoring/steps/enums.py b/aws/logs_monitoring/steps/enums.py
@@ -58,6 +58,7 @@ def __str__(self):
 class AwsEventType(Enum):
     AWSLOGS = "awslogs"
     EVENTS = "events"
+    EVENTBRIDGE_S3 = "eventbridge_s3"
     KINESIS = "kinesis"
     S3 = "s3"
     SNS = "sns"
diff --git a/aws/logs_monitoring/steps/parsing.py b/aws/logs_monitoring/steps/parsing.py
@@ -39,6 +39,8 @@ def parse(event, context, cache_layer):
             case AwsEventType.S3:
                 s3_handler = S3EventHandler(context, metadata, cache_layer)
                 events = s3_handler.handle(event)
+            case AwsEventType.EVENTBRIDGE_S3:
+                events = eventbridge_s3_handler(event, context, metadata, cache_layer)
             case AwsEventType.EVENTS:
                 events = cwevent_handler(event, metadata)
             case AwsEventType.SNS:
@@ -79,10 +81,61 @@ def parse_event_type(event):
     elif str(AwsEventType.AWSLOGS) in event:
         return AwsEventType.AWSLOGS
     elif "detail" in event:
+        # Check if this is an EventBridge S3 event
+        if event.get("source", "") == "aws.s3" and "Object Created" in event.get(
+            "detail-type", ""
+        ):
+            return AwsEventType.EVENTBRIDGE_S3
         return AwsEventType.EVENTS
     raise Exception("Event type not supported (see #Event supported section)")
 
 
+# Handle S3 event over EventBridge
+def eventbridge_s3_handler(event, context, metadata, cache_layer):
+    """
+    Transform EventBridge S3 event to standard S3 event format.
+
+    EventBridge format:
+    {
+        "version": "0",
+        "detail-type": "Object Created",
+        "source": "aws.s3",
+        "detail": {
+            "bucket": {"name": "bucket-name"},
+            "object": {"key": "object-key", "size": 1234}
+        }
+    }
+
+    Standard S3 format:
+    {
+        "Records": [{
+            "s3": {
+                "bucket": {"name": "bucket-name"},
+                "object": {"key": "object-key"}
+            }
+        }]
+    }
+    """
+
+    def reformat_eventbridge_s3_event(event):
+        return {
+            "Records": [
+                {
+                    "s3": {
+                        "bucket": event["detail"]["bucket"],
+                        "object": {
+                            "key": event["detail"]["object"]["key"],
+                        },
+                    }
+                }
+            ]
+        }
+
+    event = reformat_eventbridge_s3_event(event)
+    s3_handler = S3EventHandler(context, metadata, cache_layer)
+    return s3_handler.handle(event)
+
+
 # Handle Cloudwatch Events
 def cwevent_handler(event, metadata):
     # Set the source on the log
diff --git a/aws/logs_monitoring/tests/test_parsing.py b/aws/logs_monitoring/tests/test_parsing.py
@@ -1,8 +1,10 @@
 import unittest
+from unittest.mock import MagicMock, patch
 
 from settings import DD_CUSTOM_TAGS, DD_SOURCE
 from steps.common import get_service_from_tags_and_remove_duplicates, parse_event_source
-from steps.enums import AwsEventSource
+from steps.enums import AwsEventSource, AwsEventType
+from steps.parsing import parse, parse_event_type
 
 
 class TestParseEventSource(unittest.TestCase):
@@ -183,5 +185,100 @@ 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_event = {
+            "version": "0",
+            "id": "test-event-id",
+            "detail-type": "Object Created",
+            "source": "aws.s3",
+            "account": "123456789012",
+            "time": "2024-01-15T12:00:00Z",
+            "region": "us-east-1",
+            "resources": ["arn:aws:s3:::my-bucket"],
+            "detail": {
+                "bucket": {"name": "my-bucket"},
+                "object": {"key": "my-key.log"},
+            },
+        }
+
+        event_type = parse_event_type(eventbridge_s3_event)
+        self.assertEqual(event_type, 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)
+
+    def test_parse_non_s3_eventbridge_event_type(self):
+        """Test that 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"},
+        }
+
+        event_type = parse_event_type(eventbridge_other_event)
+        self.assertEqual(event_type, AwsEventType.EVENTS)
+
+
+class TestEventBridgeS3Parsing(unittest.TestCase):
+    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"
+
+    @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"}])
+
+        eventbridge_event = {
+            "version": "0",
+            "detail-type": "Object Created",
+            "source": "aws.s3",
+            "detail": {
+                "bucket": {"name": "my-bucket"},
+                "object": {"key": "my-key.log", "size": 1234},
+            },
+        }
+
+        cache_layer = MagicMock()
+
+        # Act
+        _ = parse(eventbridge_event, self.Context(), cache_layer)
+
+        # 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
+
+        self.assertIn("Records", normalized_event)
+        self.assertEqual(
+            normalized_event["Records"][0]["s3"]["bucket"]["name"], "my-bucket"
+        )
+        self.assertEqual(
+            normalized_event["Records"][0]["s3"]["object"]["key"], "my-key.log"
+        )
+
+
 if __name__ == "__main__":
     unittest.main()
