Merge pull request #3 from blockmar/log-forwarder-eventbridge-s3-support-handler-refactor

Refactored EventBridge S3 events to a new event type. Extracted handler, similar to kinesis.

Author: blockmar

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"

aws/logs_monitoring/steps/parsing.py

@@ -37,9 +37,10 @@ def parse(event, context, cache_layer):
                 events = aws_handler.handle(event)
                 return collect_and_count(events)
             case AwsEventType.S3:
-                event = _reformat_eventbridge_s3_event(event)
                 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:
@@ -82,12 +83,14 @@ def parse_event_type(event):
     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.S3
+            return AwsEventType.EVENTBRIDGE_S3
         return AwsEventType.EVENTS
     raise Exception("Event type not supported (see #Event supported section)")
 
 
-def _reformat_eventbridge_s3_event(event):
+# Handle S3 event over EventBridge
+def eventbridge_s3_handler(event, context, metadata, cache_layer):
+
     """
     Transform EventBridge S3 event to standard S3 event format.
 
@@ -112,7 +115,7 @@ def _reformat_eventbridge_s3_event(event):
         }]
     }
     """
-    if event.get("source") == "aws.s3" and "detail" in event:
+    def reformat_eventbridge_s3_event(event):
         return {
             "Records": [
                 {
@@ -125,7 +128,10 @@ def _reformat_eventbridge_s3_event(event):
                 }
             ]
         }
-    return event
+
+    event = reformat_eventbridge_s3_event(event)
+    s3_handler = S3EventHandler(context, metadata, cache_layer)
+    return s3_handler.handle(event)
 
 
 # Handle Cloudwatch Events

aws/logs_monitoring/tests/test_lambda_function.py

@@ -300,58 +300,5 @@ def _create_log_event_from_data(self, data):
         return encoded_data
 
 
-class TestParseEventType(unittest.TestCase):
-    def test_parse_eventbridge_s3_event_type(self):
-        """Test that EventBridge S3 events are correctly identified as 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.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)
-
-
 if __name__ == "__main__":
     unittest.main()

aws/logs_monitoring/tests/test_parsing.py

@@ -3,8 +3,8 @@
 
 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.parsing import parse
+from steps.enums import AwsEventSource, AwsEventType
+from steps.parsing import parse, parse_event_type
 
 
 class TestParseEventSource(unittest.TestCase):
@@ -185,6 +185,59 @@ 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"