MySQLdb Instrumentation (mysql-python) (#87)

pglombardo · web-flow · commit 240077aaec79 · 2018-07-13T15:52:15.000+02:00
* Fix background server names.

* More attempts to quiet spyne background server

* Initial MySQLdb instrumentation and basic tests.

* Add Mysql-python to test bundle; fix python 3 test runs

* Fix Python version test

* Update import to use new locations

* Make sure pip is updated before package installation

* Instrumentation safeties; executemany impl

* Centralize KV collection; callproc support

* Upgrade setuptools for python 3.3 issue

* For Travis, use a blank password

* Create mysql db before test run

* formatting

* formatting

* Trigger Travis build

* Trigger Travis build

* Trigger Travis build

* Make sure message has a message

* Add test to validate error capture/logging

* Fix .travis.yml syntax error

* linter fixups; test touchups

* Work-around for "command out of sync on Travis" (but not locally)

* 2nd Work-around for "command out of sync on Travis" (but not locally)

* 3rd Work-around for "command out of sync on Travis" (but not locally)

* 4th Work-around for "command out of sync on Travis" (but not locally)

* Break out test proc removal

* Use env specific vars depending on where tests run

* Use env var db name

* Linter fixups; fix wrapped calls

* Linter fixups; fix error count check

* Update to use mysql span KVs
diff --git a/.travis.yml b/.travis.yml
@@ -1,9 +1,18 @@
 language: python
+
 python:
   - "2.7"
   - "3.3"
   - "3.4"
   - "3.5"
   - "3.6"
+
+before_install:
+  - "pip install --upgrade pip"
+  - "pip install --upgrade setuptools"
+  - "mysql -e 'CREATE DATABASE travis_ci_test;'"
+
 install: "pip install -r requirements-test.txt"
+
+
 script: nosetests -v
diff --git a/instana/__init__.py b/instana/__init__.py
@@ -56,4 +56,5 @@ def load(module):
     # noqa: ignore=W0611
     from .instrumentation import urllib3  # noqa
     from .instrumentation import sudsjurko  # noqa
+    from .instrumentation import mysqlpython  # noqa
     from .instrumentation.django import middleware  # noqa
diff --git a/instana/instrumentation/mysqlpython.py b/instana/instrumentation/mysqlpython.py
@@ -0,0 +1,17 @@
+from __future__ import absolute_import
+
+from ..log import logger
+from .pep0249 import ConnectionFactory
+
+try:
+    import MySQLdb # noqa
+
+    cf = ConnectionFactory(connect_func=MySQLdb.connect, module_name='mysql')
+
+    setattr(MySQLdb, 'connect', cf)
+    if hasattr(MySQLdb, 'Connect'):
+        setattr(MySQLdb, 'Connect', cf)
+
+    logger.debug("Instrumenting mysql-python")
+except ImportError:
+    pass
diff --git a/instana/instrumentation/pep0249.py b/instana/instrumentation/pep0249.py
@@ -0,0 +1,141 @@
+# This is a wrapper for PEP-0249: Python Database API Specification v2.0
+import opentracing.ext.tags as ext
+import wrapt
+
+from ..tracer import internal_tracer
+from ..log import logger
+
+
+class CursorWrapper(wrapt.ObjectProxy):
+    __slots__ = ('_module_name', '_connect_params', '_cursor_params')
+
+    def __init__(self, cursor, module_name,
+                 connect_params=None, cursor_params=None):
+        super(CursorWrapper, self).__init__(wrapped=cursor)
+        self._module_name = module_name
+        self._connect_params = connect_params
+        self._cursor_params = cursor_params
+
+    def _collect_kvs(self, span, sql):
+        try:
+            span.set_tag(ext.SPAN_KIND, 'exit')
+            span.set_tag(ext.DATABASE_INSTANCE, self._connect_params[1]['db'])
+            span.set_tag(ext.DATABASE_STATEMENT, sql)
+            span.set_tag(ext.DATABASE_TYPE, 'mysql')
+            span.set_tag(ext.DATABASE_USER, self._connect_params[1]['user'])
+            span.set_tag('host', "%s:%s" %
+                         (self._connect_params[1]['host'],
+                          self._connect_params[1]['port']))
+        except Exception as e:
+            logger.debug(e)
+        finally:
+            return span
+
+    def execute(self, sql, params=None):
+        try:
+            span = None
+            context = internal_tracer.current_context()
+
+            # If we're not tracing, just return
+            if context is None:
+                return self.__wrapped__.execute(sql, params)
+
+            span = internal_tracer.start_span(self._module_name, child_of=context)
+            span = self._collect_kvs(span, sql)
+
+            result = self.__wrapped__.execute(sql, params)
+        except Exception as e:
+            if span:
+                span.log_exception(e)
+            raise
+        else:
+            return result
+        finally:
+            if span:
+                span.finish()
+
+    def executemany(self, sql, seq_of_parameters):
+        try:
+            span = None
+            context = internal_tracer.current_context()
+
+            # If we're not tracing, just return
+            if context is None:
+                return self.__wrapped__.executemany(sql, seq_of_parameters)
+
+            span = internal_tracer.start_span(self._module_name, child_of=context)
+            span = self._collect_kvs(span, sql)
+
+            result = self.__wrapped__.executemany(sql, seq_of_parameters)
+        except Exception as e:
+            if span:
+                span.log_exception(e)
+            raise
+        else:
+            return result
+        finally:
+            if span:
+                span.finish()
+
+    def callproc(self, proc_name, params):
+        try:
+            span = None
+            context = internal_tracer.current_context()
+
+            # If we're not tracing, just return
+            if context is None:
+                return self.__wrapped__.execute(proc_name, params)
+
+            span = internal_tracer.start_span(self._module_name, child_of=context)
+            span = self._collect_kvs(span, proc_name)
+
+            result = self.__wrapped__.callproc(proc_name, params)
+        except Exception as e:
+            if span:
+                span.log_exception(e)
+            raise
+        else:
+            return result
+        finally:
+            if span:
+                span.finish()
+
+
+class ConnectionWrapper(wrapt.ObjectProxy):
+    __slots__ = ('_module_name', '_connect_params')
+
+    def __init__(self, connection, module_name, connect_params):
+        super(ConnectionWrapper, self).__init__(wrapped=connection)
+        self._module_name = module_name
+        self._connect_params = connect_params
+
+    def cursor(self, *args, **kwargs):
+        return CursorWrapper(
+            cursor=self.__wrapped__.cursor(*args, **kwargs),
+            module_name=self._module_name,
+            connect_params=self._connect_params,
+            cursor_params=(args, kwargs) if args or kwargs else None)
+
+    def begin(self):
+        return self.__wrapped__.begin()
+
+    def commit(self):
+        return self.__wrapped__.commit()
+
+    def rollback(self):
+        return self.__wrapped__.rollback()
+
+
+class ConnectionFactory(object):
+    def __init__(self, connect_func, module_name):
+        self._connect_func = connect_func
+        self._module_name = module_name
+        self._wrapper_ctor = ConnectionWrapper
+
+    def __call__(self, *args, **kwargs):
+        connect_params = (args, kwargs) if args or kwargs else None
+
+        return self._wrapper_ctor(
+            connection=self._connect_func(*args, **kwargs),
+            module_name=self._module_name,
+            connect_params=connect_params)
diff --git a/instana/json_span.py b/instana/json_span.py
@@ -28,6 +28,17 @@ def __init__(self, **kwds):
         self.__dict__.update(kwds)
 
 
+class MySQLData(object):
+    db = None
+    host = None
+    user = None
+    stmt = None
+    error = None
+
+    def __init__(self, **kwds):
+        self.__dict__.update(kwds)
+
+
 class HttpData(object):
     host = None
     url = None
@@ -37,6 +48,7 @@ class HttpData(object):
     def __init__(self, **kwds):
         self.__dict__.update(kwds)
 
+
 class SoapData(object):
     action = None
 
diff --git a/instana/recorder.py b/instana/recorder.py
@@ -10,7 +10,7 @@
 import instana
 
 from .agent_const import AGENT_TRACES_URL
-from .json_span import CustomData, Data, HttpData, JsonSpan, SDKData, SoapData
+from .json_span import CustomData, Data, HttpData, JsonSpan, MySQLData, SDKData, SoapData
 
 if sys.version_info.major is 2:
     import Queue as queue
@@ -20,8 +20,9 @@
 
 class InstanaRecorder(SpanRecorder):
     sensor = None
-    registered_spans = ("django", "memcache", "rpc-client",
+    registered_spans = ("django", "memcache", "mysql", "rpc-client",
                         "rpc-server", "soap", "urllib3", "wsgi")
+    http_spans = ("django", "wsgi", "urllib3", "soap")
     entry_kind = ["entry", "server", "consumer"]
     exit_kind = ["exit", "client", "producer"]
     queue = queue.Queue()
@@ -83,14 +84,27 @@ def record_span(self, span):
 
     def build_registered_span(self, span):
         """ Takes a BasicSpan and converts it into a registered JsonSpan """
-        data = Data(http=HttpData(host=self.get_host_name(span),
-                                  url=self.get_string_tag(span, ext.HTTP_URL),
-                                  method=self.get_string_tag(span, ext.HTTP_METHOD),
-                                  status=self.get_tag(span, ext.HTTP_STATUS_CODE)),
-                    soap=SoapData(action=self.get_tag(span, 'soap.action')),
-                    baggage=span.context.baggage,
+        data = Data(baggage=span.context.baggage,
                     custom=CustomData(tags=span.tags,
-                    logs=self.collect_logs(span)))
+                                      logs=self.collect_logs(span)))
+
+        if span.operation_name in self.http_spans:
+            data.http = HttpData(host=self.get_host_name(span),
+                                 url=self.get_string_tag(span, ext.HTTP_URL),
+                                 method=self.get_string_tag(span, ext.HTTP_METHOD),
+                                 status=self.get_tag(span, ext.HTTP_STATUS_CODE))
+
+        if span.operation_name == "soap":
+            data.soap = SoapData(action=self.get_tag(span, 'soap.action'))
+
+        if span.operation_name == "mysql":
+            data.mysql = MySQLData(host=self.get_tag(span, 'host'),
+                                   db=self.get_tag(span, ext.DATABASE_INSTANCE),
+                                   user=self.get_tag(span, ext.DATABASE_USER),
+                                   stmt=self.get_tag(span, ext.DATABASE_STATEMENT))
+            if len(data.custom.logs.keys()):
+                tskey = list(data.custom.logs.keys())[0]
+                data.mysql.error = data.custom.logs[tskey]['message']
 
         entityFrom = {'e': self.sensor.agent.from_.pid,
                       'h': self.sensor.agent.from_.agentUuid}
@@ -103,7 +117,7 @@ def build_registered_span(self, span):
                     ts=int(round(span.start_time * 1000)),
                     d=int(round(span.duration * 1000)),
                     f=entityFrom,
-                    ec=self.get_tag(span, "ec"),
+                    ec=self.get_tag(span, "ec", 0),
                     error=self.get_tag(span, "error"),
                     data=data)
 
@@ -129,15 +143,15 @@ def build_sdk_span(self, span):
                     d=int(round(span.duration * 1000)),
                     n="sdk",
                     f=entityFrom,
-                    # ec=self.get_tag(span, "ec"),
-                    # error=self.get_tag(span, "error"),
+                    ec=self.get_tag(span, "ec"),
+                    error=self.get_tag(span, "error"),
                     data=data)
 
-    def get_tag(self, span, tag):
+    def get_tag(self, span, tag, default=None):
         if tag in span.tags:
             return span.tags[tag]
 
-        return None
+        return default
 
     def get_string_tag(self, span, tag):
         ret = self.get_tag(span, tag)
diff --git a/instana/span.py b/instana/span.py
@@ -16,7 +16,7 @@ def finish(self, finish_time=None):
         super(InstanaSpan, self).finish(finish_time)
 
     def log_exception(self, e):
-        if hasattr(e, 'message'):
+        if hasattr(e, 'message') and len(e.message):
             self.log_kv({'message': e.message})
         elif hasattr(e, '__str__'):
             self.log_kv({'message': e.__str__()})
diff --git a/setup.py b/setup.py
@@ -18,18 +18,19 @@
       entry_points={
                     'instana':  ['string = instana:load'],
                     'flask':    ['flask = instana.flaskana:hook'],
-                    'runtime':  ['string = instana:load'], # deprecated: use same as 'instana'
-                    'django':   ['string = instana:load'], # deprecated: use same as 'instana'
-                    'django19': ['string = instana:load'], # deprecated: use same as 'instana'
+                    'runtime':  ['string = instana:load'],  # deprecated: use same as 'instana'
+                    'django':   ['string = instana:load'],  # deprecated: use same as 'instana'
+                    'django19': ['string = instana:load'],  # deprecated: use same as 'instana'
                     },
       extras_require={
         'test': [
             'nose>=1.0',
             'flask>=0.12.2',
+            'lxml>=3.4',
+            'MySQL-python>=1.2.5;python_version<="2.7"',
             'requests>=2.17.1',
             'urllib3[secure]>=1.15',
             'spyne>=2.9',
-            'lxml>=3.4',
             'suds-jurko>=0.6'
         ],
       },
diff --git a/tests/__init__.py b/tests/__init__.py
@@ -13,23 +13,23 @@
 # Spawn our background Flask app that the tests will throw
 # requests at.  Don't continue until the test app is fully
 # up and running.
-timer = threading.Thread(target=flaskalino.run)
-timer.daemon = True
-timer.name = "Background Flask app"
+flask = threading.Thread(target=flaskalino.run)
+flask.daemon = True
+flask.name = "Background Flask app"
 print("Starting background Flask app...")
-timer.start()
+flask.start()
 
 
 # Background Soap Server
 #
 # Spawn our background Flask app that the tests will throw
 # requests at.  Don't continue until the test app is fully
 # up and running.
-timer = threading.Thread(target=soapserver.serve_forever)
-timer.daemon = True
-timer.name = "Background Soap server"
+soap = threading.Thread(target=soapserver.serve_forever)
+soap.daemon = True
+soap.name = "Background Soap server"
 print("Starting background Soap server...")
-timer.start()
+soap.start()
 
 
 time.sleep(1)
diff --git a/tests/apps/soapserver4132.py b/tests/apps/soapserver4132.py
@@ -41,18 +41,21 @@ def client_fault(ctx):
 
 
 
+# logging.basicConfig(level=logging.WARN)
+logging.getLogger('suds').setLevel(logging.WARN)
+logging.getLogger('suds.resolver').setLevel(logging.WARN)
+logging.getLogger('spyne.protocol.xml').setLevel(logging.WARN)
+logging.getLogger('spyne.model.complex').setLevel(logging.WARN)
+logging.getLogger('spyne.interface._base').setLevel(logging.WARN)
+logging.getLogger('spyne.interface.xml').setLevel(logging.WARN)
+logging.getLogger('spyne.util.appreg').setLevel(logging.WARN)
+
 app = Application([StanSoapService], 'instana.tests.app.ask_question',
                   in_protocol=Soap11(validator='lxml'), out_protocol=Soap11())
 
 # Use Instana middleware so we can test context passing and Soap server traces.
 wsgi_app = iWSGIMiddleware(WsgiApplication(app))
 soapserver = make_server('127.0.0.1', 4132, wsgi_app)
 
-logging.basicConfig(level=logging.WARN)
-logging.getLogger('suds').setLevel(logging.WARN)
-logging.getLogger('suds.resolver').setLevel(logging.WARN)
-logging.getLogger('spyne.protocol.xml').setLevel(logging.WARN)
-logging.getLogger('spyne.model.complex').setLevel(logging.WARN)
-
 if __name__ == '__main__':
     soapserver.serve_forever()
diff --git a/tests/test_mysql-python.py b/tests/test_mysql-python.py
diff --git a/tests/test_sudsjurko.py b/tests/test_sudsjurko.py
diff --git a/tests/test_urllib3.py b/tests/test_urllib3.py
