feat: [SVLS-5677] DynamoDB Stream event span pointers

Author: apiarian-datadog

datadog_lambda/span_pointers.py

@@ -2,6 +2,9 @@
 import logging
 from typing import List
 
+from ddtrace._trace.utils_botocore.span_pointers.dynamodb import (
+    _aws_dynamodb_item_span_pointer_description,
+)
 from ddtrace._trace.utils_botocore.span_pointers.s3 import (
     _aws_s3_object_span_pointer_description,
 )
@@ -21,6 +24,9 @@ def calculate_span_pointers(
         if event_source.equals(EventTypes.S3):
             return _calculate_s3_span_pointers_for_event(event)
 
+        elif event_source.equals(EventTypes.DYNAMODB):
+            return _calculate_dynamodb_span_pointers_for_event(event)
+
     except Exception as e:
         logger.warning(
             "failed to calculate span pointers for event: %s",
@@ -89,3 +95,64 @@ def _calculate_s3_span_pointers_for_object_created_s3_information(
             str(e),
         )
         return []
+
+
+def _calculate_dynamodb_span_pointers_for_event(event) -> List[_SpanPointerDescription]:
+    # Example event:
+    # https://docs.aws.amazon.com/lambda/latest/dg/with-ddb.html
+
+    return list(
+        chain.from_iterable(
+            _calculate_dynamodb_span_pointers_for_event_record(record)
+            for record in event.get("Records", [])
+        )
+    )
+
+
+def _calculate_dynamodb_span_pointers_for_event_record(
+    record,
+) -> List[_SpanPointerDescription]:
+    try:
+        table_name = _extract_table_name_from_dynamodb_stream_record(record)
+        primary_key = record["dynamodb"]["Keys"]
+
+    except Exception as e:
+        logger.warning(
+            "missing DynamoDB information required to make a span pointer: %s",
+            str(e),
+        )
+        return []
+
+    try:
+        return [
+            _aws_dynamodb_item_span_pointer_description(
+                pointer_direction=_SpanPointerDirection.UPSTREAM,
+                table_name=table_name,
+                primary_key=primary_key,
+            )
+        ]
+
+    except Exception as e:
+        logger.warning(
+            "failed to generate DynamoDB span pointer: %s",
+            str(e),
+        )
+        return []
+
+
+def _extract_table_name_from_dynamodb_stream_record(record) -> str:
+    # Example eventSourceARN:
+    # arn:aws:dynamodb:us-east-2:123456789012:table/my-table/stream/2024-06-10T19:26:16.525
+    event_source_arn = record["eventSourceARN"]
+
+    [_arn, _aws, _dynamodb, _region, _account, dynamodb_info] = event_source_arn.split(
+        ":", maxsplit=5
+    )
+    if _arn != "arn" or _aws != "aws" or _dynamodb != "dynamodb":
+        raise ValueError(f"unexpected eventSourceARN format: {event_source_arn}")
+
+    [_table, table_name, _stream, _timestamp] = dynamodb_info.split("/")
+    if not _table == "table" or not _stream == "stream":
+        raise ValueError(f"unexpected eventSourceARN format: {event_source_arn}")
+
+    return table_name

tests/test_span_pointers.py

@@ -80,6 +80,67 @@ class SpanPointersCase(NamedTuple):
                 },
                 span_pointers=[],
             ),
+            SpanPointersCase(
+                name="empty dynamodb event",
+                event_source=_EventSource(EventTypes.DYNAMODB),
+                event={},
+                span_pointers=[],
+            ),
+            SpanPointersCase(
+                name="sensible dynamodb event",
+                event_source=_EventSource(EventTypes.DYNAMODB),
+                event={
+                    "Records": [
+                        {
+                            "eventSourceARN": "arn:aws:dynamodb:us-west-2:123456789012:table/some-table/stream/2015-06-27T00:48:05.899",
+                            "dynamodb": {
+                                "Keys": {
+                                    "some-key": {"S": "some-value"},
+                                },
+                            },
+                        },
+                        {
+                            "eventSourceARN": "arn:aws:dynamodb:us-west-2:123456789012:table/some-table/stream/2015-06-27T00:48:05.899",
+                            "dynamodb": {
+                                "Keys": {
+                                    "some-key": {"S": "some-other-value"},
+                                },
+                            },
+                        },
+                    ],
+                },
+                span_pointers=[
+                    _SpanPointerDescription(
+                        pointer_kind="aws.dynamodb.item",
+                        pointer_direction=_SpanPointerDirection.UPSTREAM,
+                        pointer_hash="7f1aee721472bcb48701d45c7c7f7821",
+                        extra_attributes={},
+                    ),
+                    _SpanPointerDescription(
+                        pointer_kind="aws.dynamodb.item",
+                        pointer_direction=_SpanPointerDirection.UPSTREAM,
+                        pointer_hash="36b820424312a6069bd3f2185f1af584",
+                        extra_attributes={},
+                    ),
+                ],
+            ),
+            SpanPointersCase(
+                name="malformed dynamodb event",
+                event_source=_EventSource(EventTypes.DYNAMODB),
+                event={
+                    "Records": [
+                        {
+                            "eventSourceARN": "arn:aws:dynamodb:us-west-2:123456789012:table/some-table",  # missing stream info
+                            "dynamodb": {
+                                "Keys": {
+                                    "some-key": {"S": "some-value"},
+                                },
+                            },
+                        },
+                    ],
+                },
+                span_pointers=[],
+            ),
         ],
         ids=lambda test_case: test_case.name,
     )