New Celery Instrumentation (#238)

* Exclude celery as parent

* Updated ports

* Add celery to test set

* Break agent out into its own package

* Update agent imports

* Move app init to app package

* Background Celery app; Clean up test apps

* Updated redis config

* Assure test env var is set

* Error handler unification

* No announce in tests

* Celery instrumentation & tests

* TestAgent override

* Parent span exclusion

* Pytest configuration file

* Redis config

* Update test imports

* Tests cleanup

* Test run with Pytest

* No unicode characters for py2

* Update pytest ignore globs

* Update min pytest version

* Rename app packages to avoid naming conflicts

* Update tests to run under Pytest

* Fix log formatting

* assertEquals instead of assert_equals

* Moar assertEquals

* Method docs and task_catalog_get

* Retry and failure hooks

* Update recorded tags

* Update instrumentation message

* Moar tests

* Add skiptest

Author: pglombardo

.circleci/config.yml

@@ -37,7 +37,7 @@ jobs:
           name: run tests
           command: |
             . venv/bin/activate
-            python runtests.py
+            pytest -v
 
   python38:
     docker:
@@ -69,7 +69,7 @@ jobs:
           name: run tests
           command: |
             . venv/bin/activate
-            python runtests.py
+            pytest -v
 
   py27cassandra:
     docker:
@@ -96,7 +96,7 @@ jobs:
           name: run tests
           command: |
             . venv/bin/activate
-            CASSANDRA_TEST=1 nosetests -v tests/test_cassandra-driver.py:TestCassandra
+            CASSANDRA_TEST=1 pytest -v tests/clients/test_cassandra-driver.py
 
   py36cassandra:
     docker:
@@ -120,7 +120,7 @@ jobs:
           name: run tests
           command: |
             . venv/bin/activate
-            CASSANDRA_TEST=1 nosetests -v tests/test_cassandra-driver.py:TestCassandra
+            CASSANDRA_TEST=1 pytest -v tests/clients/test_cassandra-driver.py
 
   gevent38:
     docker:
@@ -140,7 +140,7 @@ jobs:
           name: run tests
           command: |
             . venv/bin/activate
-            GEVENT_TEST=1 nosetests -v tests/test_gevent.py
+            GEVENT_TEST=1 pytest -v tests/frameworks/test_gevent.py
 workflows:
   version: 2
   build:

docker-compose.yml

@@ -1,11 +1,17 @@
 version: '2'
 services:
   redis:
-    image: 'bitnami/redis:latest'
-    environment:
-      - ALLOW_EMPTY_PASSWORD=yes
+    image: redis:4.0.6
+    #image: 'bitnami/redis:latest'
+    #environment:
+    #  - ALLOW_EMPTY_PASSWORD=yes
+    #volumes:
+    #  - ./tests/conf/redis.conf:/opt/bitnami/redis/mounted-etc/redis.conf
+    volumes:
+      - ./tests/conf/redis.conf:/usr/local/etc/redis/redis.conf
+    command: redis-server /usr/local/etc/redis/redis.conf
     ports:
-      - 6379:6379
+      - "0.0.0.0:6379:6379"
 
 #
 #  Dev: Optionally enable to validate Redis Sentinel

instana/__init__.py

@@ -85,14 +85,14 @@ def lambda_handler(event, context):
         # Import the module specified in module_name
         handler_module = importlib.import_module(module_name)
     except ImportError:
-        print("Couldn't determine and locate default module handler: %s.%s", module_name, function_name)
+        print("Couldn't determine and locate default module handler: %s.%s" % (module_name, function_name))
     else:
         # Now get the function and execute it
         if hasattr(handler_module, function_name):
             handler_function = getattr(handler_module, function_name)
             return handler_function(event, context)
         else:
-            print("Couldn't determine and locate default function handler: %s.%s", module_name, function_name)
+            print("Couldn't determine and locate default function handler: %s.%s" % (module_name, function_name))
 
 
 def boot_agent_later():
@@ -129,6 +129,8 @@ def boot_agent():
         else:
             from .instrumentation import mysqlclient
 
+        from .instrumentation.celery import hooks
+
         from .instrumentation import cassandra_inst
         from .instrumentation import couchbase_inst
         from .instrumentation import flask

instana/agent/__init__.py

@@ -0,0 +1,104 @@
+"""
+The Instana agent (for AWS Lambda functions) that manages
+monitoring state and reporting that data.
+"""
+import os
+import time
+from ..log import logger
+from ..util import to_json
+from .base import BaseAgent
+from instana.collector import Collector
+from instana.options import AWSLambdaOptions
+
+
+class AWSLambdaFrom(object):
+    """ The source identifier for AWSLambdaAgent """
+    hl = True
+    cp = "aws"
+    e = "qualifiedARN"
+
+    def __init__(self, **kwds):
+        self.__dict__.update(kwds)
+
+
+class AWSLambdaAgent(BaseAgent):
+    """ In-process agent for AWS Lambda """
+    def __init__(self):
+        super(AWSLambdaAgent, self).__init__()
+
+        self.from_ = AWSLambdaFrom()
+        self.collector = None
+        self.options = AWSLambdaOptions()
+        self.report_headers = None
+        self._can_send = False
+        self.extra_headers = self.options.extra_http_headers
+
+        if self._validate_options():
+            self._can_send = True
+            self.collector = Collector(self)
+            self.collector.start()
+        else:
+            logger.warning("Required INSTANA_AGENT_KEY and/or INSTANA_ENDPOINT_URL environment variables not set.  "
+                           "We will not be able monitor this function.")
+
+    def can_send(self):
+        """
+        Are we in a state where we can send data?
+        @return: Boolean
+        """
+        return self._can_send
+
+    def get_from_structure(self):
+        """
+        Retrieves the From data that is reported alongside monitoring data.
+        @return: dict()
+        """
+        return {'hl': True, 'cp': 'aws', 'e': self.collector.context.invoked_function_arn}
+
+    def report_data_payload(self, payload):
+        """
+        Used to report metrics and span data to the endpoint URL in self.options.endpoint_url
+        """
+        response = None
+        try:
+            if self.report_headers is None:
+                # Prepare request headers
+                self.report_headers = dict()
+                self.report_headers["Content-Type"] = "application/json"
+                self.report_headers["X-Instana-Host"] = self.collector.context.invoked_function_arn
+                self.report_headers["X-Instana-Key"] = self.options.agent_key
+                self.report_headers["X-Instana-Time"] = str(round(time.time() * 1000))
+
+            # logger.debug("using these headers: %s", self.report_headers)
+
+            if 'INSTANA_DISABLE_CA_CHECK' in os.environ:
+                ssl_verify = False
+            else:
+                ssl_verify = True
+
+            response = self.client.post(self.__data_bundle_url(),
+                                        data=to_json(payload),
+                                        headers=self.report_headers,
+                                        timeout=self.options.timeout,
+                                        verify=ssl_verify)
+
+            if 200 <= response.status_code < 300:
+                logger.debug("report_data_payload: Instana responded with status code %s", response.status_code)
+            else:
+                logger.info("report_data_payload: Instana responded with status code %s", response.status_code)
+        except Exception as e:
+            logger.debug("report_data_payload: connection error (%s)", type(e))
+        finally:
+            return response
+
+    def _validate_options(self):
+        """
+        Validate that the options used by this Agent are valid.  e.g. can we report data?
+        """
+        return self.options.endpoint_url is not None and self.options.agent_key is not None
+
+    def __data_bundle_url(self):
+        """
+        URL for posting metrics to the host agent.  Only valid when announced.
+        """
+        return "%s/bundle" % self.options.endpoint_url

instana/agent/aws_lambda.py

@@ -0,0 +1,15 @@
+import requests
+
+
+class BaseAgent(object):
+    """ Base class for all agent flavors """
+    client = None
+    sensor = None
+    secrets_matcher = 'contains-ignore-case'
+    secrets_list = ['key', 'pass', 'secret']
+    extra_headers = None
+    options = None
+
+    def __init__(self):
+        self.client = requests.Session()
+

instana/agent/base.py

@@ -1,21 +1,23 @@
-""" The in-process Instana agent that manages monitoring state and reporting that data. """
+"""
+The in-process Instana agent (for host based processes) that manages
+monitoring state and reporting that data.
+"""
 from __future__ import absolute_import
 
 import json
 import os
-import time
 from datetime import datetime
 import threading
-import requests
 
 import instana.singletons
-from instana.collector import Collector
 
-from .fsm import TheMachine
-from .log import logger
-from .sensor import Sensor
-from .util import to_json, get_py_source, package_version
-from .options import StandardOptions, AWSLambdaOptions
+from ..fsm import TheMachine
+from ..log import logger
+from ..sensor import Sensor
+from ..util import to_json, get_py_source, package_version
+from ..options import StandardOptions
+
+from .base import BaseAgent
 
 
 class AnnounceData(object):
@@ -27,30 +29,7 @@ def __init__(self, **kwds):
         self.__dict__.update(kwds)
 
 
-class AWSLambdaFrom(object):
-    """ The source identifier for AWSLambdaAgent """
-    hl = True
-    cp = "aws"
-    e = "qualifiedARN"
-
-    def __init__(self, **kwds):
-        self.__dict__.update(kwds)
-
-
-class BaseAgent(object):
-    """ Base class for all agent flavors """
-    client = None
-    sensor = None
-    secrets_matcher = 'contains-ignore-case'
-    secrets_list = ['key', 'pass', 'secret']
-    extra_headers = None
-    options = None
-
-    def __init__(self):
-        self.client = requests.Session()
-
-
-class StandardAgent(BaseAgent):
+class HostAgent(BaseAgent):
     """
     The Agent class is the central controlling entity for the Instana Python language sensor.  The key
     parts it handles are the announce state and the collection and reporting of metrics and spans to the
@@ -75,7 +54,7 @@ class StandardAgent(BaseAgent):
     should_threads_shutdown = threading.Event()
 
     def __init__(self):
-        super(StandardAgent, self).__init__()
+        super(HostAgent, self).__init__()
         logger.debug("initializing agent")
         self.sensor = Sensor(self)
         self.machine = TheMachine(self)
@@ -170,11 +149,7 @@ def get_from_structure(self):
         Retrieves the From data that is reported alongside monitoring data.
         @return: dict()
         """
-        if os.environ.get("INSTANA_TEST", False):
-            from_data = {'e': os.getpid(), 'h': 'fake'}
-        else:
-            from_data = {'e': self.announce_data.pid, 'h': self.announce_data.agentUuid}
-        return from_data
+        return {'e': self.announce_data.pid, 'h': self.announce_data.agentUuid}
 
     def is_agent_listening(self, host, port):
         """
@@ -342,86 +317,3 @@ def __response_url(self, message_id):
         """
         path = "com.instana.plugin.python/response.%d?messageId=%s" % (int(self.announce_data.pid), message_id)
         return "http://%s:%s/%s" % (self.options.agent_host, self.options.agent_port, path)
-
-
-class AWSLambdaAgent(BaseAgent):
-    """ In-process agent for AWS Lambda """
-    def __init__(self):
-        super(AWSLambdaAgent, self).__init__()
-
-        self.from_ = AWSLambdaFrom()
-        self.collector = None
-        self.options = AWSLambdaOptions()
-        self.report_headers = None
-        self._can_send = False
-        self.extra_headers = self.options.extra_http_headers
-
-        if self._validate_options():
-            self._can_send = True
-            self.collector = Collector(self)
-            self.collector.start()
-        else:
-            logger.warning("Required INSTANA_AGENT_KEY and/or INSTANA_ENDPOINT_URL environment variables not set.  "
-                           "We will not be able monitor this function.")
-
-    def can_send(self):
-        """
-        Are we in a state where we can send data?
-        @return: Boolean
-        """
-        return self._can_send
-
-    def get_from_structure(self):
-        """
-        Retrieves the From data that is reported alongside monitoring data.
-        @return: dict()
-        """
-        return {'hl': True, 'cp': 'aws', 'e': self.collector.context.invoked_function_arn}
-
-    def report_data_payload(self, payload):
-        """
-        Used to report metrics and span data to the endpoint URL in self.options.endpoint_url
-        """
-        response = None
-        try:
-            if self.report_headers is None:
-                # Prepare request headers
-                self.report_headers = dict()
-                self.report_headers["Content-Type"] = "application/json"
-                self.report_headers["X-Instana-Host"] = self.collector.context.invoked_function_arn
-                self.report_headers["X-Instana-Key"] = self.options.agent_key
-                self.report_headers["X-Instana-Time"] = str(round(time.time() * 1000))
-
-            # logger.debug("using these headers: %s", self.report_headers)
-
-            if 'INSTANA_DISABLE_CA_CHECK' in os.environ:
-                ssl_verify = False
-            else:
-                ssl_verify = True
-
-            response = self.client.post(self.__data_bundle_url(),
-                                        data=to_json(payload),
-                                        headers=self.report_headers,
-                                        timeout=self.options.timeout,
-                                        verify=ssl_verify)
-
-            if 200 <= response.status_code < 300:
-                logger.debug("report_data_payload: Instana responded with status code %s", response.status_code)
-            else:
-                logger.info("report_data_payload: Instana responded with status code %s", response.status_code)
-        except Exception as e:
-            logger.debug("report_data_payload: connection error (%s)", type(e))
-        finally:
-            return response
-
-    def _validate_options(self):
-        """
-        Validate that the options used by this Agent are valid.  e.g. can we report data?
-        """
-        return self.options.endpoint_url is not None and self.options.agent_key is not None
-
-    def __data_bundle_url(self):
-        """
-        URL for posting metrics to the host agent.  Only valid when announced.
-        """
-        return "%s/bundle" % self.options.endpoint_url