MySQLdb Instrumentation (mysql-python) (#84)

* 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

Author: pglombardo

.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

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

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='MySQLdb')
+
+    setattr(MySQLdb, 'connect', cf)
+    if hasattr(MySQLdb, 'Connect'):
+        setattr(MySQLdb, 'Connect', cf)
+
+    logger.debug("Instrumenting mysql-python")
+except ImportError:
+    pass

instana/instrumentation/pep0249.py

@@ -0,0 +1,147 @@
+# 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(ext.PEER_ADDRESS, "mysql://%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)
+            span.set_tag('op', 'execute')
+
+            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__.execute(sql, params)
+
+            span = internal_tracer.start_span(self._module_name, child_of=context)
+            span = self._collect_kvs(span, sql)
+            span.set_tag('op', 'executemany')
+            span.set_tag('count', len(seq_of_parameters))
+
+            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(sql, params)
+
+            span = internal_tracer.start_span(self._module_name, child_of=context)
+            span = self._collect_kvs(span, proc_name)
+            span.set_tag('op', 'callproc')
+
+            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)

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__()})

setup.py

@@ -26,10 +26,11 @@
         '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'
         ],
       },

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)

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()