New grpcio Instrumentation (#170)

* Initial RPC test app and pkg dependencies

* Propagator Improvements:

- New binary propagator for upcoming grpc instrumentation
- Fixed handling/processing of level header

* Register new Binary propagator

* Updated GRPC application for the test suite

* Initial GRPC instrumentation and tests

* Update tests to follow InstanaSpanContext changes

* Update MariaDB to fix py2.7 build issue with Mysql-python

* Updated GRPC app with various streaming rpc call support

* GRPC instrumentation: streaming call suport

* Streaming GRPC tests

* GRPC Async/Futures support with tests

* Improved exception logging

* Test cases for error reporting

* Attempt to work-around build bug

* Limit grpcio tests to python >3.5

Author: pglombardo

.circleci/config.yml

@@ -6,13 +6,13 @@ version: 2
 jobs:
   python27:
     docker:
-      - image: circleci/python:2.7.16
+      - image: circleci/python:2.7.15
 
       # Specify service dependencies here if necessary
       # CircleCI maintains a library of pre-built images
       # documented at https://circleci.com/docs/2.0/circleci-images/
       - image: circleci/postgres:9.6.5-alpine-ram
-      - image: circleci/mariadb:10-ram
+      - image: circleci/mariadb:10.1-ram
       - image: circleci/redis:5.0.4
       - image: rabbitmq:3.5.4
 

instana/__init__.py

@@ -69,6 +69,7 @@ def boot_agent():
             from .instrumentation import mysqlclient
 
         from .instrumentation import flask
+        from .instrumentation import grpcio
         from .instrumentation.tornado import client
         from .instrumentation.tornado import server
         from .instrumentation import logging

instana/binary_propagator.py

@@ -0,0 +1,83 @@
+from __future__ import absolute_import
+
+import opentracing as ot
+
+from .log import logger
+from .util import header_to_id
+from .span_context import InstanaSpanContext
+
+
+class BinaryPropagator():
+    """
+    A Propagator for TEXT_MAP.
+    """
+    HEADER_KEY_T = b'x-instana-t'
+    HEADER_KEY_S = b'x-instana-s'
+    HEADER_KEY_L = b'x-instana-l'
+
+    def inject(self, span_context, carrier):
+        try:
+            trace_id = str.encode(span_context.trace_id)
+            span_id = str.encode(span_context.span_id)
+            level = str.encode("1")
+
+            if type(carrier) is dict or hasattr(carrier, "__dict__"):
+                carrier[self.HEADER_KEY_T] = trace_id
+                carrier[self.HEADER_KEY_S] = span_id
+                carrier[self.HEADER_KEY_L] = level
+            elif type(carrier) is list:
+                carrier.append((self.HEADER_KEY_T, trace_id))
+                carrier.append((self.HEADER_KEY_S, span_id))
+                carrier.append((self.HEADER_KEY_L, level))
+            elif type(carrier) is tuple:
+                carrier = carrier.__add__(((self.HEADER_KEY_T, trace_id),))
+                carrier = carrier.__add__(((self.HEADER_KEY_S, span_id),))
+                carrier = carrier.__add__(((self.HEADER_KEY_L, level),))
+            elif hasattr(carrier, '__setitem__'):
+                carrier.__setitem__(self.HEADER_KEY_T, trace_id)
+                carrier.__setitem__(self.HEADER_KEY_S, span_id)
+                carrier.__setitem__(self.HEADER_KEY_L, level)
+            else:
+                raise Exception("Unsupported carrier type", type(carrier))
+
+            return carrier
+        except Exception:
+            logger.debug("inject error:", exc_info=True)
+
+    def extract(self, carrier):  # noqa
+        trace_id = None
+        span_id = None
+        level = None
+
+        try:
+            if type(carrier) is dict or hasattr(carrier, "__getitem__"):
+                dc = carrier
+            elif hasattr(carrier, "__dict__"):
+                dc = carrier.__dict__
+            elif type(carrier) is list:
+                dc = dict(carrier)
+            else:
+                raise ot.SpanContextCorruptedException()
+
+            for key, value in dc.items():
+                if type(key) is str:
+                    key = str.encode(key)
+
+                if self.HEADER_KEY_T == key:
+                    trace_id = header_to_id(value)
+                elif self.HEADER_KEY_S == key:
+                    span_id = header_to_id(value)
+                elif self.HEADER_KEY_L == key:
+                    level = value
+
+            ctx = None
+            if trace_id is not None and span_id is not None:
+                ctx = InstanaSpanContext(span_id=span_id,
+                                         trace_id=trace_id,
+                                         level=level,
+                                         baggage={},
+                                         sampled=True)
+            return ctx
+
+        except Exception:
+            logger.debug("extract error:", exc_info=True)

instana/http_propagator.py

@@ -1,7 +1,7 @@
 from __future__ import absolute_import
 
 import opentracing as ot
-from basictracer.context import SpanContext
+from .span_context import InstanaSpanContext
 
 from .log import logger
 from .util import header_to_id
@@ -62,6 +62,7 @@ def inject(self, span_context, carrier):
     def extract(self, carrier):  # noqa
         trace_id = None
         span_id = None
+        level = 1
 
         try:
             if type(carrier) is dict or hasattr(carrier, "__getitem__"):
@@ -82,18 +83,23 @@ def extract(self, carrier):  # noqa
                     trace_id = header_to_id(dc[key])
                 elif self.LC_HEADER_KEY_S == lc_key:
                     span_id = header_to_id(dc[key])
+                elif self.LC_HEADER_KEY_L == lc_key:
+                    level = dc[key]
 
                 elif self.ALT_LC_HEADER_KEY_T == lc_key:
                     trace_id = header_to_id(dc[key])
                 elif self.ALT_LC_HEADER_KEY_S == lc_key:
                     span_id = header_to_id(dc[key])
+                elif self.ALT_LC_HEADER_KEY_L == lc_key:
+                    level = dc[key]
 
             ctx = None
             if trace_id is not None and span_id is not None:
-                ctx = SpanContext(span_id=span_id,
-                                  trace_id=trace_id,
-                                  baggage={},
-                                  sampled=True)
+                ctx = InstanaSpanContext(span_id=span_id,
+                                         trace_id=trace_id,
+                                         level=level,
+                                         baggage={},
+                                         sampled=True)
             return ctx
 
         except Exception: