RD-3862 - feat: support eventBridge (#158)

* feat: support eventBridge

Author: saartochner-lumigo

src/lumigo_tracer/event/event_trigger.py

@@ -37,6 +37,8 @@ def parse_triggered_by(event: dict):
             return _parse_cw(event)
         elif _is_step_function(event):
             return _parse_step_function(event)
+        elif _is_event_bridge(event):
+            return _parse_event_bridge(event)
 
     return _parse_unknown(event)
 
@@ -118,6 +120,23 @@ def _parse_sns(event: dict):
     }
 
 
+def _is_event_bridge(event: dict):
+    return (
+        isinstance(event.get("version"), str)
+        and isinstance(event.get("id"), str)  # noqa: W503
+        and isinstance(event.get("detail-type"), str)  # noqa: W503
+        and isinstance(event.get("source"), str)  # noqa: W503
+        and isinstance(event.get("time"), str)  # noqa: W503
+        and isinstance(event.get("region"), str)  # noqa: W503
+        and isinstance(event.get("resources"), list)  # noqa: W503
+        and isinstance(event.get("detail"), dict)  # noqa: W503
+    )
+
+
+def _parse_event_bridge(event: dict):
+    return {"triggeredBy": "eventBridge", "messageId": event["id"]}
+
+
 def _is_supported_cw(event: dict):
     return event.get("detail-type") == "Scheduled Event" and "source" in event and "time" in event
 

src/lumigo_tracer/wrappers/http/http_parser.py

@@ -236,6 +236,40 @@ def parse_response(self, url: str, status_code: int, headers, body: bytes) -> di
         )
 
 
+class EventBridgeParser(Parser):
+    def parse_request(self, parse_params: HttpRequest) -> dict:
+        try:
+            parsed_body = json.loads(parse_params.body)
+        except json.JSONDecodeError as e:
+            get_logger().exception(
+                "Error while trying to parse eventBridge request body", exc_info=e
+            )
+            parsed_body = {}
+        resource_names = set()
+        if isinstance(parsed_body.get("Entries"), list):
+            resource_names = {
+                e["EventBusName"] for e in parsed_body["Entries"] if e.get("EventBusName")
+            }
+        return recursive_json_join(
+            {"info": {"resourceNames": list(resource_names) or None}},
+            super().parse_request(parse_params),
+        )
+
+    def parse_response(self, url: str, status_code: int, headers, body: bytes) -> dict:
+        try:
+            parsed_body = json.loads(body)
+        except json.JSONDecodeError as e:
+            get_logger().debug("Error while trying to parse eventBridge request body", exc_info=e)
+            parsed_body = {}
+        message_ids = []
+        if isinstance(parsed_body.get("Entries"), list):
+            message_ids = [e["EventId"] for e in parsed_body["Entries"] if e.get("EventId")]
+        return recursive_json_join(
+            {"info": {"messageIds": message_ids}},
+            super().parse_response(url, status_code, headers, body),
+        )
+
+
 class ApiGatewayV2Parser(ServerlessAWSParser):
     # API-GW V1 covered by ServerlessAWSParser
 
@@ -259,6 +293,8 @@ def get_parser(url: str, headers: Optional[dict] = None) -> Type[Parser]:
         return LambdaParser
     elif service == "kinesis":
         return KinesisParser
+    elif service == "events":
+        return EventBridgeParser
     elif safe_split_get(url, ".", 1) == "s3":
         return S3Parser
     # SQS Legacy Endpoints: https://docs.aws.amazon.com/general/latest/gr/rande.html

src/test/unit/event/test_event_trigger.py

@@ -165,6 +165,19 @@
             },
             {"triggeredBy": "unknown"},
         ),
+        (  # EventBridge - happy flow
+            {
+                "version": "0",
+                "id": "f0f73aaa-e64f-a550-5be2-850898090583",
+                "detail-type": "string",
+                "source": "source_lambda",
+                "time": "2020-10-19T13:34:29Z",
+                "region": "us-west-2",
+                "resources": [],
+                "detail": {"a": 0.024995371455989845},
+            },
+            {"triggeredBy": "eventBridge", "messageId": "f0f73aaa-e64f-a550-5be2-850898090583"},
+        ),
         (  # cloudwatch
             {
                 "id": "cdc73f9d-aea9-11e3-9d5a-835b769c0d9c",

src/test/unit/wrappers/http/test_http_parser.py

@@ -10,6 +10,7 @@
     get_parser,
     ApiGatewayV2Parser,
     DynamoParser,
+    EventBridgeParser,
 )
 
 
@@ -24,7 +25,7 @@ def test_serverless_aws_parser_fallback_doesnt_change():
 
 
 def test_get_parser_check_headers():
-    url = "api.rti.dev.toyota.com"
+    url = "api.dev.com"
     headers = {"x-amzn-requestid": "1234"}
     assert get_parser(url, headers) == ServerlessAWSParser
 
@@ -164,3 +165,68 @@ def test_double_response_size_limit_on_error_status_code():
     assert response_with_error["headers"] == json.dumps(d)
     assert len(response_with_error["body"]) > len(response_no_error["body"])
     assert response_with_error["body"] == json.dumps(d)
+
+
+def test_event_bridge_parser_request_happy_flow():
+    parser = EventBridgeParser()
+    params = HttpRequest(
+        host="",
+        method="POST",
+        uri="",
+        headers={},
+        body=json.dumps(
+            {
+                "Entries": [
+                    {
+                        "Source": "source_lambda",
+                        "Resources": [],
+                        "DetailType": "string",
+                        "Detail": '{"a": 1}',
+                        "EventBusName": "name1",
+                    },
+                    {
+                        "Source": "source_lambda",
+                        "Resources": [],
+                        "DetailType": "string",
+                        "Detail": '{"a": 2}',
+                        "EventBusName": "name1",
+                    },
+                    {
+                        "Source": "source_lambda",
+                        "Resources": [],
+                        "DetailType": "string",
+                        "Detail": '{"a": 3}',
+                        "EventBusName": "name2",
+                    },
+                ]
+            }
+        ),
+    )
+    response = parser.parse_request(params)
+    assert set(response["info"]["resourceNames"]) == {"name2", "name1"}
+
+
+def test_event_bridge_parser_request_sad_flow():
+    parser = EventBridgeParser()
+    params = HttpRequest(host="", method="POST", uri="", headers={}, body="not a json")
+    response = parser.parse_request(params)
+    assert response["info"]["resourceNames"] is None
+
+
+def test_event_bridge_parser_response_happy_flow():
+    parser = EventBridgeParser()
+    response = parser.parse_response(
+        "",
+        200,
+        {},
+        body=json.dumps(
+            {"Entries": [{"EventId": "1-2-3-4"}, {"EventId": "6-7-8-9"}], "FailedEntryCount": 0}
+        ).encode(),
+    )
+    assert response["info"]["messageIds"] == ["1-2-3-4", "6-7-8-9"]
+
+
+def test_event_bridge_parser_response_sad_flow():
+    parser = EventBridgeParser()
+    response = parser.parse_response("", 200, {}, body=b"not a json")
+    assert not response["info"]["messageIds"]