AWS specific instrumentations (#1054)

* add localstack setup for enhanced botocore tests

* add specific instrumentation data for S3

note: this will break current botocore tests. These will get fixed
at a later point.

* use correct localstack URL

* added support for SNS and DynamoDB

* added some more testing for SNS and fixed an issue with default handlers

* updated docs

* add DB context fields to dynamodb

* add WAIT_FOR config for localstack

* don't use bridged network for localstack

.editorconfig

@@ -14,7 +14,6 @@ trim_trailing_whitespace = false
 
 [Makefile]
 indent_style = tab
-indent_size = 4
 
 [Jenkinsfile]
 indent_size = 2
@@ -24,3 +23,6 @@ indent_size = 2
 
 [*.feature]
 indent_size = 2
+
+[*.yml]
+indent_size = 2

docs/supported-technologies.asciidoc

@@ -516,13 +516,32 @@ Instrumented methods:
 
  * `botocore.client.BaseClient._make_api_call`
 
-Collected trace data:
+Collected trace data for all services:
 
  * AWS region (e.g. `eu-central-1`)
  * AWS service name (e.g. `s3`)
  * operation name (e.g. `ListBuckets`)
 
+Additionally, some services collect more specific data
+
+[float]
+[[automatic-instrumentation-s3]]
+===== S3
+
+ * Bucket name
+
+[float]
+[[automatic-instrumentation-dynamodb]]
+===== DynamoDB
+
+ * Table name
+
+
+[float]
+[[automatic-instrumentation-sns]]
+===== SNS
 
+ * Topic name
 
 [float]
 [[automatic-instrumentation-template-engines]]

elasticapm/instrumentation/packages/botocore.py

@@ -28,10 +28,14 @@
 #  OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 #  OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
+from collections import namedtuple
+
 from elasticapm.instrumentation.packages.base import AbstractInstrumentedModule
 from elasticapm.traces import capture_span
 from elasticapm.utils.compat import urlparse
 
+HandlerInfo = namedtuple("HandlerInfo", ("signature", "span_type", "span_subtype", "span_action", "context"))
+
 
 class BotocoreInstrumentation(AbstractInstrumentedModule):
     name = "botocore"
@@ -44,14 +48,109 @@ def call(self, module, method, wrapped, instance, args, kwargs):
         else:
             operation_name = args[0]
 
-        target_endpoint = instance._endpoint.host
-        parsed_url = urlparse.urlparse(target_endpoint)
-        if "." in parsed_url.hostname:
-            service = parsed_url.hostname.split(".", 2)[0]
-        else:
-            service = parsed_url.hostname
+        service = instance._service_model.service_id
 
-        signature = "{}:{}".format(service, operation_name)
+        parsed_url = urlparse.urlparse(instance._endpoint.host)
+        context = {
+            "destination": {
+                "address": parsed_url.hostname,
+                "port": parsed_url.port,
+                "cloud": {"region": instance.meta.region_name},
+            }
+        }
 
-        with capture_span(signature, "aws", leaf=True, span_subtype=service, span_action=operation_name):
+        handler_info = None
+        handler = handlers.get(service, False)
+        if handler:
+            handler_info = handler(operation_name, service, instance, args, kwargs, context)
+        if not handler_info:
+            handler_info = handle_default(operation_name, service, instance, args, kwargs, context)
+
+        with capture_span(
+            handler_info.signature,
+            span_type=handler_info.span_type,
+            leaf=True,
+            span_subtype=handler_info.span_subtype,
+            span_action=handler_info.span_action,
+            extra=handler_info.context,
+        ):
             return wrapped(*args, **kwargs)
+
+
+def handle_s3(operation_name, service, instance, args, kwargs, context):
+    span_type = "storage"
+    span_subtype = "s3"
+    span_action = operation_name
+    if len(args) > 1 and "Bucket" in args[1]:
+        bucket = args[1]["Bucket"]
+    else:
+        # TODO handle Access Points
+        bucket = ""
+    signature = f"S3 {operation_name} {bucket}"
+
+    context["destination"]["name"] = span_subtype
+    context["destination"]["resource"] = bucket
+    context["destination"]["service"] = {"type": span_type}
+
+    return HandlerInfo(signature, span_type, span_subtype, span_action, context)
+
+
+def handle_dynamodb(operation_name, service, instance, args, kwargs, context):
+    span_type = "db"
+    span_subtype = "dynamodb"
+    span_action = "query"
+    if len(args) > 1 and "TableName" in args[1]:
+        table = args[1]["TableName"]
+    else:
+        table = ""
+    signature = f"DynamoDB {operation_name} {table}".rstrip()
+
+    context["db"] = {"type": "dynamodb", "instance": instance.meta.region_name}
+    if operation_name == "Query" and len(args) > 1 and "KeyConditionExpression" in args[1]:
+        context["db"]["statement"] = args[1]["KeyConditionExpression"]
+
+    context["destination"]["name"] = span_subtype
+    context["destination"]["resource"] = table
+    context["destination"]["service"] = {"type": span_type}
+    return HandlerInfo(signature, span_type, span_subtype, span_action, context)
+
+
+def handle_sns(operation_name, service, instance, args, kwargs, context):
+    if operation_name != "Publish":
+        # only "publish" is handled specifically, other endpoints get the default treatment
+        return False
+    span_type = "messaging"
+    span_subtype = "sns"
+    span_action = "send"
+    topic_name = ""
+    if len(args) > 1:
+        if "Name" in args[1]:
+            topic_name = args[1]["Name"]
+        if "TopicArn" in args[1]:
+            topic_name = args[1]["TopicArn"].rsplit(":", maxsplit=1)[-1]
+    signature = f"SNS {operation_name} {topic_name}".rstrip()
+    context["destination"]["name"] = span_subtype
+    context["destination"]["resource"] = f"{span_subtype}/{topic_name}" if topic_name else span_subtype
+    context["destination"]["type"] = span_type
+    return HandlerInfo(signature, span_type, span_subtype, span_action, context)
+
+
+def handle_sqs(operation_name, service, instance, args, kwargs, destination):
+    pass
+
+
+def handle_default(operation_name, service, instance, args, kwargs, destination):
+    span_type = "aws"
+    span_subtype = service.lower()
+    span_action = operation_name
+
+    signature = f"{service}:{operation_name}"
+    return HandlerInfo(signature, span_type, span_subtype, span_action, destination)
+
+
+handlers = {
+    "S3": handle_s3,
+    "DynamoDB": handle_dynamodb,
+    "SNS": handle_sns,
+    "default": handle_default,
+}

tests/docker-compose.yml

@@ -122,6 +122,23 @@ services:
     volumes:
       - mysqldata:/var/lib/mysql
 
+  localstack:
+    image: localstack/localstack
+    ports:
+      - "4566:4566"
+      - "4571:4571"
+    environment:
+      - HOSTNAME=localstack
+      - HOSTNAME_EXTERNAL=localstack
+      - SERVICES=sns,sqs,s3,dynamodb,ec2
+      - DEBUG=${DEBUG- }
+      - DOCKER_HOST=unix:///var/run/docker.sock
+      - HOST_TMP_FOLDER=${TMPDIR}
+      - START_WEB=0
+    volumes:
+      - "${TMPDIR:-/tmp/localstack}:/tmp/localstack"
+      - "/var/run/docker.sock:/var/run/docker.sock"
+
   run_tests:
     image: apm-agent-python:${PYTHON_VERSION}
     environment: