New Celery Instrumentation

.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/aws_lambda.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/base.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.py → instana/agent/host.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