feat: add kinesis batch trigger (#257)

Co-authored-by: CircleCI <[email protected]>

Author: saartochner-lumigo

.secrets.baseline

@@ -168,14 +168,14 @@
         "filename": "src/test/unit/event/test_event_trigger.py",
         "hashed_secret": "885bb9903f72e004ff2974807b70e7c970d3e6d5",
         "is_verified": false,
-        "line_number": 451
+        "line_number": 476
       },
       {
         "type": "Hex High Entropy String",
         "filename": "src/test/unit/event/test_event_trigger.py",
         "hashed_secret": "3fae06dc55a618caed1d794dcd512bfe7e76c9f1",
         "is_verified": false,
-        "line_number": 485
+        "line_number": 510
       }
     ],
     "src/test/unit/test_lumigo_utils.py": [
@@ -220,5 +220,5 @@
       }
     ]
   },
-  "generated_at": "2022-08-17T15:30:21Z"
+  "generated_at": "2022-08-22T12:01:52Z"
 }

src/lumigo_tracer/event/event_trigger.py

@@ -227,10 +227,7 @@ def _parse_streams(event: dict) -> Dict[str, str]:
     if triggered_by == "sqs":
         result.update(_parse_sqs_event(event))
     elif triggered_by == "kinesis":
-        result[MESSAGE_ID_KEY] = safe_get(event, ["Records", 0, "kinesis", "sequenceNumber"])
-        event_id = safe_get(event, ["Records", 0, "eventID"])
-        if isinstance(event_id, str):
-            result["shardId"] = event_id.split(":", 1)[0]
+        result.update(_parse_kinesis_event(event))
     elif triggered_by == "dynamodb":
         result.update(_parse_dynamomdb_event(event))
     return result
@@ -258,6 +255,25 @@ def _parse_dynamomdb_event(event) -> Dict[str, Union[int, List[str]]]:
     }
 
 
+def _parse_kinesis_event(event) -> Dict[str, Union[int, str, List[str], List[Dict[str, str]]]]:
+    result = {}
+    message_ids = []
+    records = safe_get(event, ["Records"], default=[])
+    for record in records:
+        message_id = safe_get(record, ["kinesis", "sequenceNumber"])
+        if message_id:
+            message_ids.append(message_id)
+    if message_ids:
+        if len(message_ids) == 1:
+            result[MESSAGE_ID_KEY] = message_ids[0]
+        else:
+            result[MESSAGE_IDS_KEY] = message_ids
+    event_id = safe_get(event, ["Records", 0, "eventID"])
+    if isinstance(event_id, str):
+        result["shardId"] = event_id.split(":", 1)[0]
+    return result
+
+
 def _parse_sqs_event(event) -> Dict[str, Union[int, str, List[str], List[Dict[str, str]]]]:
     message_ids = []
     chained_resources: List[Dict[str, str]] = []

src/test/unit/event/test_event_trigger.py

@@ -100,6 +100,31 @@
                 "shardId": "shardId-000000000006",
             },
         ),
+        (  # kinesis example trigger - multiple records
+            {
+                "Records": [
+                    {
+                        "eventSourceARN": "arn:aws:kinesis:us-east-1:123456789:stream/kinesis-stream-name",
+                        "eventSource": "aws:kinesis",
+                        "eventID": "shardId-000000000006:49590338271490256608559692538361571095921575989136588898",
+                        "kinesis": {"sequenceNumber": "12"},
+                    },
+                    {
+                        "eventSourceARN": "arn:aws:kinesis:us-east-1:123456789:stream/kinesis-stream-name",
+                        "eventSource": "aws:kinesis",
+                        "eventID": "shardId-000000000006:49590338271490256608559692538361571095921575989136588898",
+                        "kinesis": {"sequenceNumber": "34"},
+                    },
+                ]
+            },
+            {
+                "triggeredBy": "kinesis",
+                "arn": "arn:aws:kinesis:us-east-1:123456789:stream/kinesis-stream-name",
+                "messageIds": ["12", "34"],
+                "recordsNum": 2,
+                "shardId": "shardId-000000000006",
+            },
+        ),
         (  # SQS example trigger
             {
                 "Records": [