New boto3 instrumentation (#267)

* Add shell files

* Base Boto3 instrumentation & tests

* Add new test dependencies

* Tests for s3, ses and sqs

* Updated tags, safeties & tests

* Remove unused mocks

* More tags & tests

* boto3 supported on Python > 3.5.3 only

* Properly log and report errors

* Dont run boto3 tests in unsupported versions

* Updated http tags

* Inject context when invoking lambdas

* Avoid CLI options as service names - not cool

* Partition boto3 dependencies

* To Mock Lambda requires docker; skip

Author: Peter Giacomo Lombardo

instana/__init__.py

@@ -130,6 +130,7 @@ def boot_agent():
             from .instrumentation.aiohttp import client
             from .instrumentation.aiohttp import server
             from .instrumentation import asynqp
+            from .instrumentation import boto3_inst
 
         if sys.version_info[0] < 3:
             from .instrumentation import mysqlpython

instana/instrumentation/boto3_inst.py

@@ -0,0 +1,129 @@
+from __future__ import absolute_import
+
+import json
+import wrapt
+import inspect
+
+from ..log import logger
+from ..singletons import tracer
+
+
+try:
+    import boto3
+    from boto3.s3 import inject
+
+    def lambda_inject_context(payload, scope):
+        """
+        When boto3 lambda client 'Invoke' is called, we want to inject the tracing context.
+        boto3/botocore has specific requirements:
+        https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/lambda.html#Lambda.Client.invoke
+        """
+        try:
+            invoke_payload = payload.get('Payload', {})
+
+            if not isinstance(invoke_payload, dict):
+                invoke_payload = json.loads(invoke_payload)
+
+            tracer.inject(scope.span.context, 'http_headers', invoke_payload)
+            payload['Payload'] = json.dumps(invoke_payload)
+        except Exception:
+            logger.debug("non-fatal lambda_inject_context: ", exc_info=True)
+
+
+    @wrapt.patch_function_wrapper('botocore.client', 'BaseClient._make_api_call')
+    def make_api_call_with_instana(wrapped, instance, arg_list, kwargs):
+        # pylint: disable=protected-access
+        parent_span = tracer.active_span
+
+        # If we're not tracing, just return
+        if parent_span is None:
+            return wrapped(*arg_list, **kwargs)
+
+        with tracer.start_active_span("boto3", child_of=parent_span) as scope:
+            try:
+                operation = arg_list[0]
+                payload = arg_list[1]
+
+                scope.span.set_tag('op', operation)
+                scope.span.set_tag('ep', instance._endpoint.host)
+                scope.span.set_tag('reg', instance._client_config.region_name)
+
+                scope.span.set_tag('http.url', instance._endpoint.host + ':443/' + arg_list[0])
+                scope.span.set_tag('http.method', 'POST')
+
+                # Don't collect payload for SecretsManager
+                if not hasattr(instance, 'get_secret_value'):
+                    scope.span.set_tag('payload', payload)
+
+                # Inject context when invoking lambdas
+                if 'lambda' in instance._endpoint.host and operation == 'Invoke':
+                    lambda_inject_context(payload, scope)
+
+
+            except Exception as exc:
+                logger.debug("make_api_call_with_instana: collect error", exc_info=True)
+
+            try:
+                result = wrapped(*arg_list, **kwargs)
+
+                if isinstance(result, dict):
+                    http_dict = result.get('ResponseMetadata')
+                    if isinstance(http_dict, dict):
+                        status = http_dict.get('HTTPStatusCode')
+                        if status is not None:
+                            scope.span.set_tag('http.status_code', status)
+
+                return result
+            except Exception as exc:
+                scope.span.mark_as_errored({'error': exc})
+                raise
+
+    def s3_inject_method_with_instana(wrapped, instance, arg_list, kwargs):
+        fas = inspect.getfullargspec(wrapped)
+        fas_args = fas.args
+        fas_args.remove('self')
+
+        # pylint: disable=protected-access
+        parent_span = tracer.active_span
+
+        # If we're not tracing, just return
+        if parent_span is None:
+            return wrapped(*arg_list, **kwargs)
+
+        with tracer.start_active_span("boto3", child_of=parent_span) as scope:
+            try:
+                operation = wrapped.__name__
+                scope.span.set_tag('op', operation)
+                scope.span.set_tag('ep', instance._endpoint.host)
+                scope.span.set_tag('reg', instance._client_config.region_name)
+
+                scope.span.set_tag('http.url', instance._endpoint.host + ':443/' + operation)
+                scope.span.set_tag('http.method', 'POST')
+
+                index = 1
+                payload = {}
+                arg_length = len(arg_list)
+
+                for arg_name in fas_args:
+                    payload[arg_name] = arg_list[index-1]
+
+                    index += 1
+                    if index > arg_length:
+                        break
+
+                scope.span.set_tag('payload', payload)
+            except Exception as exc:
+                logger.debug("s3_inject_method_with_instana: collect error", exc_info=True)
+
+            try:
+                return wrapped(*arg_list, **kwargs)
+            except Exception as exc:
+                scope.span.mark_as_errored({'error': exc})
+                raise
+
+    for method in ['upload_file', 'upload_fileobj', 'download_file', 'download_fileobj']:
+        wrapt.wrap_function_wrapper('boto3.s3.inject', method, s3_inject_method_with_instana)
+
+    logger.debug("Instrumenting boto3")
+except ImportError:
+    pass

instana/recorder.py

@@ -1,11 +1,11 @@
+# Accept, process and queue spans for eventual reporting.
 from __future__ import absolute_import
 
 import os
 import sys
 
 from basictracer import Sampler
 
-from .log import logger
 from .span import (RegisteredSpan, SDKSpan)
 
 if sys.version_info.major == 2:
@@ -17,7 +17,7 @@
 class StanRecorder(object):
     THREAD_NAME = "Instana Span Reporting"
 
-    REGISTERED_SPANS = ("aiohttp-client", "aiohttp-server", "aws.lambda.entry", "cassandra",
+    REGISTERED_SPANS = ("aiohttp-client", "aiohttp-server", "aws.lambda.entry", "boto3", "cassandra",
                         "celery-client", "celery-worker", "couchbase", "django", "gcs", "log",
                         "memcache", "mongo", "mysql", "postgres", "pymongo", "rabbitmq", "redis",
                         "render", "rpc-client", "rpc-server", "sqlalchemy", "soap", "tornado-client",

instana/span.py

@@ -230,7 +230,7 @@ class RegisteredSpan(BaseSpan):
     HTTP_SPANS = ("aiohttp-client", "aiohttp-server", "django", "http", "soap", "tornado-client",
                   "tornado-server", "urllib3", "wsgi")
 
-    EXIT_SPANS = ("aiohttp-client", "cassandra", "celery-client", "couchbase", "log", "memcache",
+    EXIT_SPANS = ("aiohttp-client", "boto3", "cassandra", "celery-client", "couchbase", "log", "memcache",
                   "mongo", "mysql", "postgres", "rabbitmq", "redis", "rpc-client", "sqlalchemy",
                   "soap", "tornado-client", "urllib3", "pymongo", "gcs")
 
@@ -337,6 +337,18 @@ def _populate_exit_span_data(self, span):
         if span.operation_name in self.HTTP_SPANS:
             self._collect_http_tags(span)
 
+        elif span.operation_name == "boto3":
+            # boto3 also sends http tags
+            self._collect_http_tags(span)
+
+            for tag in ['op', 'ep', 'reg', 'payload', 'error']:
+                value = span.tags.pop(tag, None)
+                if value is not None:
+                    if tag == 'payload':
+                        self.data["boto3"][tag] = self._validate_tags(value)
+                    else:
+                        self.data["boto3"][tag] = value
+
         elif span.operation_name == "cassandra":
             self.data["cassandra"]["cluster"] = span.tags.pop('cassandra.cluster', None)
             self.data["cassandra"]["query"] = span.tags.pop('cassandra.query', None)

instana/util.py

@@ -380,16 +380,29 @@ def determine_service_name():
     if "INSTANA_SERVICE_NAME" in os.environ:
         return os.environ["INSTANA_SERVICE_NAME"]
 
-    try:
-        # Now best effort in naming this process.  No nice package.json like in Node.js
-        # so we do best effort detection here.
-        app_name = "python"  # the default name
+    # Now best effort in naming this process.  No nice package.json like in Node.js
+    # so we do best effort detection here.
+    app_name = "python"  # the default name
+    basename = None
 
+    try:
         if not hasattr(sys, 'argv'):
             proc_cmdline = get_proc_cmdline(as_string=False)
             return os.path.basename(proc_cmdline[0])
 
-        basename = os.path.basename(sys.argv[0])
+        # Get first argument that is not an CLI option
+        for candidate in sys.argv:
+            if candidate[0] != '-':
+                basename = candidate
+                break
+
+        # If nothing found, fall back to executable
+        if basename is None:
+            basename = os.path.basename(sys.executable)
+        else:
+            # Assure leading paths are stripped
+            basename = os.path.basename(basename)
+
         if basename == "gunicorn":
             if 'setproctitle' in sys.modules:
                 # With the setproctitle package, gunicorn renames their processes
@@ -435,9 +448,9 @@ def determine_service_name():
                 app_name = uwsgi_type % app_name
             except ImportError:
                 pass
-        return app_name
     except Exception:
         logger.debug("get_application_name: ", exc_info=True)
+    finally:
         return app_name
 
 

pytest.ini

@@ -1,5 +1,5 @@
 [pytest]
 log_cli = 1
-log_cli_level = DEBUG
+log_cli_level = WARN
 log_cli_format = %(asctime)s %(levelname)s %(message)s
 log_cli_date_format = %H:%M:%S

setup.py

@@ -93,16 +93,18 @@ def check_setuptools():
           'test': [
               'aiohttp>=3.5.4;python_version>="3.5"',
               'asynqp>=0.4;python_version>="3.5"',
+              'boto3>=1.10.0',
               'celery>=4.1.1',
               'django>=1.11,<2.2',
-              'nose>=1.0',
               'flask>=0.12.2',
               'grpcio>=1.18.0',
               'google-cloud-storage>=1.24.0;python_version>="3.5"',
               'lxml>=3.4',
               'mock>=2.0.0',
+              'moto>=1.3.16',
               'mysqlclient>=1.3.14;python_version>="3.5"',
               'MySQL-python>=1.2.5;python_version<="2.7"',
+              'nose>=1.0',
               'PyMySQL[rsa]>=0.9.1',
               'pyOpenSSL>=16.1.0;python_version<="2.7"',
               'psycopg2>=2.7.1',

tests/apps/flask_app/app.py

@@ -1,11 +1,20 @@
 #!/usr/bin/env python
 # -*- coding: utf-8 -*-
+import os
 import logging
 import opentracing.ext.tags as ext
 from flask import jsonify, Response
 from wsgiref.simple_server import make_server
 from flask import Flask, redirect, render_template, render_template_string
 
+try:
+    import boto3
+    from moto import mock_sqs
+except ImportError:
+    # Doesn't matter.  We won't call routes using boto3
+    # in test sets that don't install/test for it.
+    pass
+
 from ...helpers import testenv
 from instana.singletons import tracer
 
@@ -153,6 +162,38 @@ def response_headers():
     return resp
 
 
+@app.route("/boto3/sqs")
+def boto3_sqs():
+    os.environ['AWS_ACCESS_KEY_ID'] = 'testing'
+    os.environ['AWS_SECRET_ACCESS_KEY'] = 'testing'
+    os.environ['AWS_SECURITY_TOKEN'] = 'testing'
+    os.environ['AWS_SESSION_TOKEN'] = 'testing'
+
+    with mock_sqs():
+        boto3_client = boto3.client('sqs', region_name='us-east-1')
+        response = boto3_client.create_queue(
+            QueueName='SQS_QUEUE_NAME',
+            Attributes={
+                'DelaySeconds': '60',
+                'MessageRetentionPeriod': '600'
+            }
+        )
+
+        queue_url = response['QueueUrl']
+        response = boto3_client.send_message(
+                QueueUrl=queue_url,
+                DelaySeconds=10,
+                MessageAttributes={
+                    'Website': {
+                        'DataType': 'String',
+                        'StringValue': 'https://www.instana.com'
+                    },
+                },
+                MessageBody=('Monitor any application, service, or request '
+                            'with Instana Application Performance Monitoring')
+            )
+        return Response(response)
+
 @app.errorhandler(InvalidUsage)
 def handle_invalid_usage(error):
     logger.error("InvalidUsage error handler invoked")

tests/clients/boto3/README.md

@@ -0,0 +1,29 @@
+If you would like to run this test server manually from an ipython console:
+
+```
+import os
+import urllib3
+
+from moto import mock_sqs
+import tests.apps.flask_app
+from tests.helpers import testenv
+from instana.singletons import tracer
+
+http_client = urllib3.PoolManager()
+
+os.environ['AWS_ACCESS_KEY_ID'] = 'testing'
+os.environ['AWS_SECRET_ACCESS_KEY'] = 'testing'
+os.environ['AWS_SECURITY_TOKEN'] = 'testing'
+os.environ['AWS_SESSION_TOKEN'] = 'testing'
+
+@mock_sqs
+def test_app_boto3_sqs():
+    with tracer.start_active_span('wsgi') as scope:
+        scope.span.set_tag('span.kind', 'entry')
+        scope.span.set_tag('http.host', 'localhost:80')
+        scope.span.set_tag('http.path', '/')
+        scope.span.set_tag('http.method', 'GET')
+        scope.span.set_tag('http.status_code', 200)
+        response = http_client.request('GET', testenv["wsgi_server"] + '/boto3/sqs')
+
+```