[SLS-1671] Inferred Spans for Kineses, EventBridge, S3, DynamoDB

datadog_lambda/tracing.py

@@ -412,6 +412,18 @@ def create_inferred_span(event, context):
         elif event_source.equals(EventTypes.SNS):
             logger.debug("SNS event detected. Inferring a span")
             return create_inferred_span_from_sns_event(event, context)
+        elif event_source.equals(EventTypes.KINESIS):
+            logger.debug("Kinesis event detected. Inferring a span")
+            return create_inferred_span_from_kinesis_event(event, context)
+        elif event_source.equals(EventTypes.DYNAMODB):
+            logger.debug("Dynamodb event detected. Inferring a span")
+            return create_inferred_span_from_dynamodb_event(event, context)
+        elif event_source.equals(EventTypes.S3):
+            logger.debug("S3 event detected. Inferring a span")
+            return create_inferred_span_from_s3_event(event, context)
+        elif event_source.equals(EventTypes.EVENTBRIDGE):
+            logger.debug("Eventbridge event detected. Inferring a span")
+            return create_inferred_span_from_eventbridge_event(event, context)
 
     except Exception as e:
         logger.debug(
@@ -549,6 +561,101 @@ def create_inferred_span_from_sns_event(event, context):
     return span
 
 
+def create_inferred_span_from_kinesis_event(event, context):
+    event_record = get_first_record(event)
+    stream_name = event_record["eventSourceARN"].split(":")[-1]
+    tags = {
+        "operation_name": "aws.kinesis",
+        "service.name": "kinesis",
+        "resource_names": stream_name,
+        SPAN_TYPE_TAG: SPAN_TYPE_INFERRED,
+    }
+    request_time_epoch = event_record["kinesis"]["approximateArrivalTimestamp"]
+
+    args = {
+        "resource": stream_name,
+        "span_type": "web",
+    }
+    tracer.set_tags({"_dd.origin": "lambda"})
+    span = tracer.trace("aws.kinesis", **args)
+    if span:
+        span.set_tags(tags)
+    span.start = int(request_time_epoch)
+    return span
+
+
+def create_inferred_span_from_dynamodb_event(event, context):
+    event_record = get_first_record(event)
+    table_name = event_record["eventSourceARN"].split("/")[1]
+    tags = {
+        "operation_name": "aws.dynamodb",
+        "service.name": "dynamodb",
+        "resource_names": table_name,
+        SPAN_TYPE_TAG: SPAN_TYPE_INFERRED,
+    }
+    request_time_epoch = event_record["dynamodb"]["ApproximateCreationDateTime"]
+
+    args = {
+        "resource": table_name,
+        "span_type": "web",
+    }
+    tracer.set_tags({"_dd.origin": "lambda"})
+    span = tracer.trace("aws.dynamodb", **args)
+    if span:
+        span.set_tags(tags)
+    span.start = int(request_time_epoch)
+    return span
+
+
+def create_inferred_span_from_s3_event(event, context):
+    event_record = get_first_record(event)
+    bucket_name = event_record["s3"]["bucket"]["name"]
+    tags = {
+        "operation_name": "aws.s3",
+        "service.name": "s3",
+        "resource_names": bucket_name,
+        SPAN_TYPE_TAG: SPAN_TYPE_INFERRED,
+    }
+    dt_format = "%Y-%m-%dT%H:%M:%S.%fZ"
+    timestamp = event_record["eventTime"]
+    request_time_epoch = datetime.strptime(timestamp, dt_format)
+
+    args = {
+        "resource": bucket_name,
+        "span_type": "web",
+    }
+    tracer.set_tags({"_dd.origin": "lambda"})
+    span = tracer.trace("aws.s3", **args)
+    if span:
+        span.set_tags(tags)
+    span.start = int(request_time_epoch.strftime("%s"))
+    return span
+
+
+def create_inferred_span_from_eventbridge_event(event, context):
+    source = event["source"]
+    tags = {
+        "operation_name": "aws.eventbridge",
+        "service.name": "eventbridge",
+        "resource_names": source,
+        SPAN_TYPE_TAG: SPAN_TYPE_INFERRED,
+    }
+    dt_format = "%Y-%m-%dT%H:%M:%SZ"
+    timestamp = event["time"]
+    request_time_epoch = datetime.strptime(timestamp, dt_format)
+
+    args = {
+        "resource": source,
+        "span_type": "web",
+    }
+    tracer.set_tags({"_dd.origin": "lambda"})
+    span = tracer.trace("aws.eventbridge", **args)
+    if span:
+        span.set_tags(tags)
+    span.start = int(request_time_epoch.strftime("%s"))
+    return span
+
+
 def create_function_execution_span(
     context,
     function_name,

datadog_lambda/trigger.py

@@ -38,6 +38,7 @@ class EventTypes(_stringTypedEnum):
     S3 = "s3"
     SNS = "sns"
     SQS = "sqs"
+    EVENTBRIDGE = "eventbridge"
 
 
 class EventSubtypes(_stringTypedEnum):
@@ -128,6 +129,9 @@ def parse_event_source(event: dict) -> _EventSource:
     if event.get("awslogs"):
         event_source = _EventSource(EventTypes.CLOUDWATCH_LOGS)
 
+    if event.get("detail-type"):
+        event_source = _EventSource(EventTypes.EVENTBRIDGE)
+
     event_detail = event.get("detail")
     cw_event_categories = event_detail and event_detail.get("EventCategories")
     if event.get("source") == "aws.events" or cw_event_categories:

tests/event_samples/eventbridge-custom.json

@@ -0,0 +1,13 @@
+{
+  "version": "0",
+  "id": "fd03f394-e769-eff5-08a8-53c228933591",
+  "detail-type": "testdetail",
+  "source": "eventbridge.custom.event.sender",
+  "account": "601427279990",
+  "time": "2021-11-04T01:37:45Z",
+  "region": "sa-east-1",
+  "resources": [],
+  "detail": {
+    "foo": "bar"
+  }
+}

tests/integration/input_events/dynamodb.json

@@ -0,0 +1,93 @@
+{
+    "Records": [
+        {
+            "eventID": "c4ca4238a0b923820dcc509a6f75849b",
+            "eventName": "INSERT",
+            "eventVersion": "1.1",
+            "eventSource": "aws:dynamodb",
+            "awsRegion": "us-east-1",
+            "dynamodb": {
+                "Keys": {
+                    "Id": {
+                        "N": "101"
+                    }
+                },
+                "NewImage": {
+                    "Message": {
+                        "S": "New item!"
+                    },
+                    "Id": {
+                        "N": "101"
+                    }
+                },
+                "ApproximateCreationDateTime": 1428537600,
+                "SequenceNumber": "4421584500000000017450439091",
+                "SizeBytes": 26,
+                "StreamViewType": "NEW_AND_OLD_IMAGES"
+            },
+            "eventSourceARN": "arn:aws:dynamodb:us-east-1:123456789012:table/ExampleTableWithStream/stream/2015-06-27T00:48:05.899"
+        },
+        {
+            "eventID": "c81e728d9d4c2f636f067f89cc14862c",
+            "eventName": "MODIFY",
+            "eventVersion": "1.1",
+            "eventSource": "aws:dynamodb",
+            "awsRegion": "us-east-1",
+            "dynamodb": {
+                "Keys": {
+                    "Id": {
+                        "N": "101"
+                    }
+                },
+                "NewImage": {
+                    "Message": {
+                        "S": "This item has changed"
+                    },
+                    "Id": {
+                        "N": "101"
+                    }
+                },
+                "OldImage": {
+                    "Message": {
+                        "S": "New item!"
+                    },
+                    "Id": {
+                        "N": "101"
+                    }
+                },
+                "ApproximateCreationDateTime": 1428537600,
+                "SequenceNumber": "4421584500000000017450439092",
+                "SizeBytes": 59,
+                "StreamViewType": "NEW_AND_OLD_IMAGES"
+            },
+            "eventSourceARN": "arn:aws:dynamodb:us-east-1:123456789012:table/ExampleTableWithStream/stream/2015-06-27T00:48:05.899"
+        },
+        {
+            "eventID": "eccbc87e4b5ce2fe28308fd9f2a7baf3",
+            "eventName": "REMOVE",
+            "eventVersion": "1.1",
+            "eventSource": "aws:dynamodb",
+            "awsRegion": "us-east-1",
+            "dynamodb": {
+                "Keys": {
+                    "Id": {
+                        "N": "101"
+                    }
+                },
+                "OldImage": {
+                    "Message": {
+                        "S": "This item has changed"
+                    },
+                    "Id": {
+                        "N": "101"
+                    }
+                },
+                "ApproximateCreationDateTime": 1428537600,
+                "SequenceNumber": "4421584500000000017450439093",
+                "SizeBytes": 38,
+                "StreamViewType": "NEW_AND_OLD_IMAGES"
+            },
+            "eventSourceARN": "arn:aws:dynamodb:us-east-1:123456789012:table/ExampleTableWithStream/stream/2015-06-27T00:48:05.899"
+        }
+    ]
+}

tests/integration/input_events/eventbridge-custom.json

@@ -0,0 +1,13 @@
+{
+  "version": "0",
+  "id": "fd03f394-e769-eff5-08a8-53c228933591",
+  "detail-type": "testdetail",
+  "source": "eventbridge.custom.event.sender",
+  "account": "601427279990",
+  "time": "2021-11-04T01:37:45Z",
+  "region": "sa-east-1",
+  "resources": [],
+  "detail": {
+    "foo": "bar"
+  }
+}

tests/integration/input_events/kinesis.json

@@ -0,0 +1,20 @@
+{
+    "Records": [
+        {
+            "kinesis": {
+                "partitionKey": "partitionKey-03",
+                "kinesisSchemaVersion": "1.0",
+                "data": "SGVsbG8sIHRoaXMgaXMgYSB0ZXN0IDEyMy4=",
+                "sequenceNumber": "49545115243490985018280067714973144582180062593244200961",
+                "approximateArrivalTimestamp": 1428537600
+            },
+            "eventSource": "aws:kinesis",
+            "eventID": "shardId-000000000000:49545115243490985018280067714973144582180062593244200961",
+            "invokeIdentityArn": "arn:aws:iam::EXAMPLE",
+            "eventVersion": "1.0",
+            "eventName": "aws:kinesis:record",
+            "eventSourceARN": "arn:aws:kinesis:EXAMPLE",
+            "awsRegion": "us-east-1"
+        }
+    ]
+}

tests/integration/input_events/s3.json

@@ -0,0 +1,38 @@
+{
+    "Records": [
+        {
+            "eventVersion": "2.0",
+            "eventSource": "aws:s3",
+            "awsRegion": "us-east-1",
+            "eventTime": "1970-01-01T00:00:00.000Z",
+            "eventName": "ObjectCreated:Put",
+            "userIdentity": {
+                "principalId": "EXAMPLE"
+            },
+            "requestParameters": {
+                "sourceIPAddress": "127.0.0.1"
+            },
+            "responseElements": {
+                "x-amz-request-id": "EXAMPLE123456789",
+                "x-amz-id-2": "EXAMPLE123/5678abcdefghijklambdaisawesome/mnopqrstuvwxyzABCDEFGH"
+            },
+            "s3": {
+                "s3SchemaVersion": "1.0",
+                "configurationId": "testConfigRule",
+                "bucket": {
+                    "name": "example-bucket",
+                    "ownerIdentity": {
+                        "principalId": "EXAMPLE"
+                    },
+                    "arn": "arn:aws:s3:::example-bucket"
+                },
+                "object": {
+                    "key": "test/key",
+                    "size": 1024,
+                    "eTag": "0123456789abcdef0123456789abcdef",
+                    "sequencer": "0A1B2C3D4E5F678901"
+                }
+            }
+        }
+    ]
+}