New Cassandra Instrumentation & Tests (#213)

pglombardo · web-flow · commit b6413ea63c4e · 2020-02-04T11:32:44.000+01:00
* Cassandra Instrumentation &amp; Tests

* Add cassandra docker image to CircleCI

* Make cassandra first to load

* New KVs, tests, cleanup and optimizations

* Attempted dedicated Cassandra run

* CircleCI: Manually set Cassandra HEAP

* CircleCI official cassandra (albeit outdated)

* CircleCI: Increase the memory!

* Add a dedicated cassandra job

* Add Py27 dedicated cassandra run

* Remove unnecessary couchbase steps from cassandra jobs

* Cleanup and fix py27 install

* Add deps for reqs.txt

* Remove unused store_artifacts
diff --git a/.circleci/config.yml b/.circleci/config.yml
@@ -58,10 +58,6 @@ jobs:
             . venv/bin/activate
             python runtests.py
 
-      - store_artifacts:
-          path: test-reports
-          destination: test-reports
-
   python35:
     docker:
       - image: circleci/python:3.5.6
@@ -114,10 +110,6 @@ jobs:
             . venv/bin/activate
             python runtests.py
 
-      - store_artifacts:
-          path: test-reports
-          destination: test-reports
-
   python36:
     docker:
       - image: circleci/python:3.6.8
@@ -170,13 +162,104 @@ jobs:
             . venv/bin/activate
             python runtests.py
 
-      - store_artifacts:
-          path: test-reports
-          destination: test-reports
+  py27cassandra:
+    docker:
+      - image: circleci/python:2.7.15
+      - image: circleci/cassandra:3.10
+        environment:
+          MAX_HEAP_SIZE: 2048m
+          HEAP_NEWSIZE: 512m
+
+    working_directory: ~/repo
+
+    steps:
+      - checkout
+
+      # Download and cache dependencies
+      - restore_cache:
+          keys:
+            - v1-dependencies-{{ checksum "requirements.txt" }}
+            # fallback to using the latest cache if no exact match is found
+            - v1-dependencies-
+
+      - run:
+          name: install dependencies
+          command: |
+            sudo apt-get update
+            sudo apt install lsb-release -y
+            curl -O https://packages.couchbase.com/releases/couchbase-release/couchbase-release-1.0-6-amd64.deb
+            sudo dpkg -i ./couchbase-release-1.0-6-amd64.deb
+            sudo apt-get update
+            sudo apt install libcouchbase-dev -y
+            rm -rf venv
+            export PATH=/home/circleci/.local/bin:$PATH
+            pip install --user -U pip setuptools virtualenv
+            virtualenv --python=python2.7 --always-copy venv
+            . venv/bin/activate
+            pip install -U pip
+            python setup.py install_egg_info
+            pip install -r requirements-test.txt
+
+      - save_cache:
+          paths:
+            - ./venv
+          key: v1-dependencies-{{ checksum "requirements.txt" }}
+
+      - run:
+          name: run tests
+          command: |
+            . venv/bin/activate
+            nosetests -v tests/test_cassandra-driver.py:TestCassandra
+
+  py36cassandra:
+    docker:
+      - image: circleci/python:3.6.8
+      - image: circleci/cassandra:3.10
+        environment:
+          MAX_HEAP_SIZE: 2048m
+          HEAP_NEWSIZE: 512m
+
+    working_directory: ~/repo
+
+    steps:
+      - checkout
+
+      # Download and cache dependencies
+      - restore_cache:
+          keys:
+            - v1-dependencies-{{ checksum "requirements.txt" }}
+            # fallback to using the latest cache if no exact match is found
+            - v1-dependencies-
+
+      - run:
+          name: install dependencies
+          command: |
+            sudo apt-get update
+            sudo apt install lsb-release -y
+            python -m venv venv
+            . venv/bin/activate
+            pip install -U pip
+            python setup.py install_egg_info
+            pip install -r requirements.txt
+            pip install -r requirements-test.txt
+
+      - save_cache:
+          paths:
+            - ./venv
+          key: v1-dependencies-{{ checksum "requirements.txt" }}
+
+      - run:
+          name: run tests
+          command: |
+            . venv/bin/activate
+            nosetests -v tests/test_cassandra-driver.py:TestCassandra
+
 workflows:
   version: 2
   build:
     jobs:
       - python27
       - python35
       - python36
+      - py27cassandra
+      - py36cassandra
diff --git a/instana/__init__.py b/instana/__init__.py
@@ -68,6 +68,7 @@ def boot_agent():
         else:
             from .instrumentation import mysqlclient
 
+        from .instrumentation import cassandra_inst
         from .instrumentation import couchbase_inst
         from .instrumentation import flask
         from .instrumentation import grpcio
diff --git a/instana/instrumentation/cassandra_inst.py b/instana/instrumentation/cassandra_inst.py
@@ -0,0 +1,85 @@
+"""
+cassandra instrumentation
+https://docs.datastax.com/en/developer/python-driver/3.20/
+https://github.com/datastax/python-driver
+"""
+from __future__ import absolute_import
+
+from distutils.version import LooseVersion
+import wrapt
+
+from ..log import logger
+from ..singletons import tracer
+
+try:
+    import cassandra
+
+    consistency_levels = dict({0: "ANY",
+                               1: "ONE",
+                               2: "TWO",
+                               3: "THREE",
+                               4: "QUORUM",
+                               5: "ALL",
+                               6: "LOCAL_QUORUM",
+                               7: "EACH_QUORUM",
+                               8: "SERIAL",
+                               9: "LOCAL_SERIAL",
+                               10: "LOCAL_ONE"})
+
+    def collect_response(span, fn):
+        tried_hosts = list()
+        for host in fn.attempted_hosts:
+            tried_hosts.append("%s:%d" % (host.endpoint.address, host.endpoint.port))
+
+        span.set_tag("cassandra.triedHosts", tried_hosts)
+        span.set_tag("cassandra.coordHost", fn.coordinator_host)
+
+        cl = fn.query.consistency_level
+        if cl and cl in consistency_levels:
+            span.set_tag("cassandra.achievedConsistency", consistency_levels[cl])
+
+
+    def cb_request_finish(results, span, fn):
+        collect_response(span, fn)
+        span.finish()
+
+    def cb_request_error(results, span, fn):
+        collect_response(span, fn)
+
+        span.set_tag("error", True)
+        ec = span.tags.get('ec', 0)
+        span.set_tag("ec", ec + 1)
+        span.set_tag("cassandra.error", results.message)
+        span.finish()
+
+    def request_init_with_instana(fn):
+        parent_span = tracer.active_span
+
+        if parent_span is not None:
+            ctags = dict()
+            if isinstance(fn.query, cassandra.query.SimpleStatement):
+                ctags["cassandra.query"] = fn.query.query_string
+            elif isinstance(fn.query, cassandra.query.BoundStatement):
+                ctags["cassandra.query"] = fn.query.prepared_statement.query_string
+
+            ctags["cassandra.keyspace"] = fn.session.keyspace
+            ctags["cassandra.cluster"] = fn.session.cluster.metadata.cluster_name
+
+            span = tracer.start_span(
+                operation_name="cassandra",
+                child_of=parent_span,
+                tags=ctags)
+
+            fn.add_callback(cb_request_finish, span, fn)
+            fn.add_errback(cb_request_error, span, fn)
+
+    @wrapt.patch_function_wrapper('cassandra.cluster', 'Session.__init__')
+    def init_with_instana(wrapped, instance, args, kwargs):
+        session =  wrapped(*args, **kwargs)
+        instance.add_request_init_listener(request_init_with_instana)
+        return session
+
+    logger.debug("Instrumenting cassandra")
+
+except ImportError:
+    pass
diff --git a/instana/json_span.py b/instana/json_span.py
@@ -26,13 +26,26 @@ class JsonSpan(BaseSpan):
     stack = None
 
 
+class CassandraData(BaseSpan):
+    cluster = None
+    query = None
+    keyspace = None
+    fetchSize = None
+    achievedConsistency = None
+    triedHosts = None
+    fullyFetched = None
+    error = None
+
+
 class CustomData(BaseSpan):
     tags = None
     logs = None
 
 
 class Data(BaseSpan):
     baggage = None
+    cassandra = None
+    couchbase = None
     custom = None
     http = None
     log = None
diff --git a/instana/recorder.py b/instana/recorder.py
@@ -9,9 +9,10 @@
 
 import instana.singletons
 
-from .json_span import (CouchbaseData, CustomData, Data, HttpData, JsonSpan, LogData, MySQLData, PostgresData,
-                        RabbitmqData, RedisData, RenderData, RPCData, SDKData, SoapData,
-                        SQLAlchemyData)
+from .json_span import (CassandraData, CouchbaseData, CustomData, Data, HttpData, JsonSpan, LogData,
+                        MySQLData, PostgresData, RabbitmqData, RedisData, RenderData,
+                        RPCData, SDKData, SoapData, SQLAlchemyData)
+
 from .log import logger
 from .util import every
 
@@ -23,15 +24,16 @@
 
 class InstanaRecorder(SpanRecorder):
     THREAD_NAME = "Instana Span Reporting"
-    registered_spans = ("aiohttp-client", "aiohttp-server", "couchbase", "django", "log", "memcache", "mysql",
-                        "postgres", "rabbitmq", "redis", "render", "rpc-client", "rpc-server", "sqlalchemy", "soap",
-                        "tornado-client", "tornado-server", "urllib3", "wsgi")
+    registered_spans = ("aiohttp-client", "aiohttp-server", "cassandra", "couchbase", "django", "log",
+                        "memcache", "mysql", "postgres", "rabbitmq", "redis", "render", "rpc-client",
+                        "rpc-server", "sqlalchemy", "soap", "tornado-client", "tornado-server",
+                        "urllib3", "wsgi")
 
     http_spans = ("aiohttp-client", "aiohttp-server", "django", "http", "soap", "tornado-client",
                   "tornado-server", "urllib3", "wsgi")
 
-    exit_spans = ("aiohttp-client", "couchbase", "log", "memcache", "mysql", "postgres", "rabbitmq", "redis", "rpc-client",
-                  "sqlalchemy", "soap", "tornado-client", "urllib3")
+    exit_spans = ("aiohttp-client", "cassandra", "couchbase", "log", "memcache", "mysql", "postgres",
+                  "rabbitmq", "redis", "rpc-client", "sqlalchemy", "soap", "tornado-client", "urllib3")
 
     entry_spans = ("aiohttp-server", "django", "wsgi", "rabbitmq", "rpc-server", "tornado-server")
 
@@ -164,22 +166,32 @@ def build_registered_span(self, span):
             if data.rabbitmq.sort == 'consume':
                 kind = 1 # entry
 
-        if span.operation_name == "couchbase":
+        elif span.operation_name == "cassandra":
+            data.cassandra = CassandraData(cluster=span.tags.pop('cassandra.cluster', None),
+                                           query=span.tags.pop('cassandra.query', None),
+                                           keyspace=span.tags.pop('cassandra.keyspace', None),
+                                           fetchSize=span.tags.pop('cassandra.fetchSize', None),
+                                           achievedConsistency=span.tags.pop('cassandra.achievedConsistency', None),
+                                           triedHosts=span.tags.pop('cassandra.triedHosts', None),
+                                           fullyFetched=span.tags.pop('cassandra.fullyFetched', None),
+                                           error=span.tags.pop('cassandra.error', None))
+
+        elif span.operation_name == "couchbase":
             data.couchbase = CouchbaseData(hostname=span.tags.pop('couchbase.hostname', None),
                                            bucket=span.tags.pop('couchbase.bucket', None),
                                            type=span.tags.pop('couchbase.type', None),
                                            error=span.tags.pop('couchbase.error', None),
                                            error_type=span.tags.pop('couchbase.error_type', None),
                                            sql=span.tags.pop('couchbase.sql', None))
 
-        if span.operation_name == "redis":
+        elif span.operation_name == "redis":
             data.redis = RedisData(connection=span.tags.pop('connection', None),
                                    driver=span.tags.pop('driver', None),
                                    command=span.tags.pop('command', None),
                                    error=span.tags.pop('redis.error', None),
                                    subCommands=span.tags.pop('subCommands', None))
 
-        if span.operation_name == "rpc-client" or span.operation_name == "rpc-server":
+        elif span.operation_name == "rpc-client" or span.operation_name == "rpc-server":
             data.rpc = RPCData(flavor=span.tags.pop('rpc.flavor', None),
                                host=span.tags.pop('rpc.host', None),
                                port=span.tags.pop('rpc.port', None),
@@ -189,22 +201,22 @@ def build_registered_span(self, span):
                                baggage=span.tags.pop('rpc.baggage', None),
                                error=span.tags.pop('rpc.error', None))
 
-        if span.operation_name == "render":
+        elif span.operation_name == "render":
             data.render = RenderData(name=span.tags.pop('name', None),
                                      type=span.tags.pop('type', None))
             data.log = LogData(message=span.tags.pop('message', None),
                                parameters=span.tags.pop('parameters', None))
 
-        if span.operation_name == "sqlalchemy":
+        elif span.operation_name == "sqlalchemy":
             data.sqlalchemy = SQLAlchemyData(sql=span.tags.pop('sqlalchemy.sql', None),
                                              eng=span.tags.pop('sqlalchemy.eng', None),
                                              url=span.tags.pop('sqlalchemy.url', None),
                                              err=span.tags.pop('sqlalchemy.err', None))
 
-        if span.operation_name == "soap":
+        elif span.operation_name == "soap":
             data.soap = SoapData(action=span.tags.pop('soap.action', None))
 
-        if span.operation_name == "mysql":
+        elif span.operation_name == "mysql":
             data.mysql = MySQLData(host=span.tags.pop('host', None),
                                    db=span.tags.pop(ext.DATABASE_INSTANCE, None),
                                    user=span.tags.pop(ext.DATABASE_USER, None),
@@ -213,7 +225,7 @@ def build_registered_span(self, span):
                 tskey = list(data.custom.logs.keys())[0]
                 data.mysql.error = data.custom.logs[tskey]['message']
 
-        if span.operation_name == "postgres":
+        elif span.operation_name == "postgres":
             data.pg = PostgresData(host=span.tags.pop('host', None),
                                    db=span.tags.pop(ext.DATABASE_INSTANCE, None),
                                    user=span.tags.pop(ext.DATABASE_USER, None),
@@ -223,7 +235,7 @@ def build_registered_span(self, span):
                 tskey = list(data.custom.logs.keys())[0]
                 data.pg.error = data.custom.logs[tskey]['message']
 
-        if span.operation_name == "log":
+        elif span.operation_name == "log":
             data.log = {}
             # use last special key values
             # TODO - logic might need a tweak here
diff --git a/runtests.py b/runtests.py
@@ -4,6 +4,10 @@
 
 command_line = [__file__, '--verbose']
 
+# Cassandra tests are run in dedicated jobs on CircleCI and will
+# be run explicitly.  (So always exclude them here)
+command_line.extend(['-e', 'cassandra'])
+
 if LooseVersion(sys.version) < LooseVersion('3.5.3'):
     command_line.extend(['-e', 'asynqp', '-e', 'aiohttp',
                          '-e', 'async', '-e', 'tornado',
diff --git a/setup.py b/setup.py
@@ -70,6 +70,7 @@ def check_setuptools():
               'aiohttp>=3.5.4;python_version>="3.5"',
               'asynqp>=0.4;python_version>="3.5"',
               'couchbase==2.5.9',
+              'cassandra-driver==3.20.2',
               'django>=1.11,<2.2',
               'nose>=1.0',
               'flask>=0.12.2',
diff --git a/tests/helpers.py b/tests/helpers.py
diff --git a/tests/test_cassandra-driver.py b/tests/test_cassandra-driver.py
