capture destination host for instrumentations of services (#618)

Implementation of span destination information as per elastic/apm#180

Author: beniwohli

.ci/.jenkins_framework.yml

@@ -33,6 +33,7 @@ FRAMEWORK:
   - zerorpc-0.4
   - mysql_connector-newest
   - pymysql-newest
+  - mysqlclient-newest
   - aiohttp-newest
   - aiopg-newest
   - tornado-newest

CHANGELOG.asciidoc

@@ -22,6 +22,9 @@ endif::[]
 
 // Unreleased changes go here
 // When the next release happens, nest these changes under the "Python Agent version 5.x" heading
+[float]
+===== Features
+* Added destination information to database/HTTP spans, used for service maps {pull}618[#618]
 
 [float]
 ===== Deprecations

elasticapm/instrumentation/packages/asyncio/aiohttp_client.py

@@ -32,7 +32,7 @@
 from elasticapm.conf import constants
 from elasticapm.instrumentation.packages.asyncio.base import AsyncAbstractInstrumentedModule
 from elasticapm.traces import DroppedSpan, execution_context
-from elasticapm.utils import get_host_from_url, sanitize_url
+from elasticapm.utils import get_host_from_url, sanitize_url, url_to_destination
 from elasticapm.utils.disttracing import TracingOptions
 
 
@@ -45,13 +45,18 @@ async def call(self, module, method, wrapped, instance, args, kwargs):
         method = kwargs["method"] if "method" in kwargs else args[0]
         url = kwargs["url"] if "url" in kwargs else args[1]
         url = str(url)
+        destination = url_to_destination(url)
 
         signature = " ".join([method.upper(), get_host_from_url(url)])
         url = sanitize_url(url)
         transaction = execution_context.get_transaction()
 
         async with async_capture_span(
-            signature, span_type="external", span_subtype="http", extra={"http": {"url": url}}, leaf=True
+            signature,
+            span_type="external",
+            span_subtype="http",
+            extra={"http": {"url": url}, "destination": destination},
+            leaf=True,
         ) as span:
             leaf_span = span
             while isinstance(leaf_span, DroppedSpan):

elasticapm/instrumentation/packages/asyncio/elasticsearch.py

@@ -40,7 +40,7 @@ class ElasticSearchAsyncConnection(ElasticSearchConnectionMixin, AsyncAbstractIn
 
     async def call(self, module, method, wrapped, instance, args, kwargs):
         signature = self.get_signature(args, kwargs)
-        context = self.get_context(args, kwargs)
+        context = self.get_context(instance, args, kwargs)
 
         async with elasticapm.async_capture_span(
             signature,

elasticapm/instrumentation/packages/cassandra.py

@@ -41,10 +41,14 @@ class CassandraInstrumentation(AbstractInstrumentedModule):
 
     def call(self, module, method, wrapped, instance, args, kwargs):
         name = self.get_wrapped_name(wrapped, instance, method)
-        context = None
+        context = {}
         if method == "Cluster.connect":
             span_action = "connect"
+            host = instance.endpoints_resolved[0].address
+            port = instance.endpoints_resolved[0].port
         else:
+            host = instance.hosts[0].endpoint.address
+            port = instance.hosts[0].endpoint.port
             span_action = "query"
             query = args[0] if args else kwargs.get("query")
             if hasattr(query, "query_string"):
@@ -57,7 +61,12 @@ def call(self, module, method, wrapped, instance, args, kwargs):
                 query_str = None
             if query_str:
                 name = extract_signature(query_str)
-                context = {"db": {"type": "sql", "statement": query_str}}
+                context["db"] = {"type": "sql", "statement": query_str}
+        context["destination"] = {
+            "address": host,
+            "port": port,
+            "service": {"name": "cassandra", "resource": "cassandra", "type": "db"},
+        }
 
         with capture_span(name, span_type="db", span_subtype="cassandra", span_action=span_action, extra=context):
             return wrapped(*args, **kwargs)

elasticapm/instrumentation/packages/dbapi2.py

@@ -1,7 +1,3 @@
-"""Provides classes to instrument dbapi2 providers
-
-https://www.python.org/dev/peps/pep-0249/
-"""
 #  BSD 3-Clause License
 #
 #  Copyright (c) 2019, Elasticsearch BV
@@ -32,6 +28,11 @@
 #  OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 #  OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
+"""Provides classes to instrument dbapi2 providers
+
+https://www.python.org/dev/peps/pep-0249/
+"""
+
 import re
 
 from elasticapm.instrumentation.packages.base import AbstractInstrumentedModule
@@ -197,6 +198,10 @@ def extract_signature(sql):
 class CursorProxy(wrapt.ObjectProxy):
     provider_name = None
 
+    def __init__(self, wrapped, destination_info=None):
+        super(CursorProxy, self).__init__(wrapped)
+        self._self_destination_info = destination_info
+
     def callproc(self, procname, params=None):
         return self._trace_sql(self.__wrapped__.callproc, procname, params, action=EXEC_ACTION)
 
@@ -224,7 +229,7 @@ def _trace_sql(self, method, sql, params, action=QUERY_ACTION):
             span_type="db",
             span_subtype=self.provider_name,
             span_action=action,
-            extra={"db": {"type": "sql", "statement": sql_string}},
+            extra={"db": {"type": "sql", "statement": sql_string}, "destination": self._self_destination_info},
             skip_frames=1,
         ):
             if params is None:
@@ -239,8 +244,12 @@ def extract_signature(self, sql):
 class ConnectionProxy(wrapt.ObjectProxy):
     cursor_proxy = CursorProxy
 
+    def __init__(self, wrapped, destination_info=None):
+        super(ConnectionProxy, self).__init__(wrapped)
+        self._self_destination_info = destination_info
+
     def cursor(self, *args, **kwargs):
-        return self.cursor_proxy(self.__wrapped__.cursor(*args, **kwargs))
+        return self.cursor_proxy(self.__wrapped__.cursor(*args, **kwargs), self._self_destination_info)
 
 
 class DbApi2Instrumentation(AbstractInstrumentedModule):

elasticapm/instrumentation/packages/elasticsearch.py

@@ -54,7 +54,7 @@ def get_signature(self, args, kwargs):
 
         return "ES %s %s" % (http_method, http_path)
 
-    def get_context(self, args, kwargs):
+    def get_context(self, instance, args, kwargs):
         args_len = len(args)
         params = args[2] if args_len > 2 else kwargs.get("params")
         body = params.pop(BODY_REF_NAME, None) if params else None
@@ -78,7 +78,10 @@ def get_context(self, args, kwargs):
             if isinstance(body, dict) and "script" in body:
                 # only get the `script` field from the body
                 context["db"]["statement"] = json.dumps({"script": body["script"]})
-        # TODO: add instance.base_url to context once we agreed on a format
+        context["destination"] = {
+            "address": instance.host,
+            "service": {"name": "elasticsearch", "resource": "elasticsearch", "type": "db"},
+        }
         return context
 
 
@@ -92,7 +95,7 @@ class ElasticsearchConnectionInstrumentation(ElasticSearchConnectionMixin, Abstr
 
     def call(self, module, method, wrapped, instance, args, kwargs):
         signature = self.get_signature(args, kwargs)
-        context = self.get_context(args, kwargs)
+        context = self.get_context(instance, args, kwargs)
 
         with elasticapm.capture_span(
             signature,

elasticapm/instrumentation/packages/mysql.py

@@ -34,6 +34,7 @@
     DbApi2Instrumentation,
     extract_signature,
 )
+from elasticapm.utils import default_ports
 
 
 class MySQLCursorProxy(CursorProxy):
@@ -53,4 +54,9 @@ class MySQLInstrumentation(DbApi2Instrumentation):
     instrument_list = [("MySQLdb", "connect")]
 
     def call(self, module, method, wrapped, instance, args, kwargs):
-        return MySQLConnectionProxy(wrapped(*args, **kwargs))
+        destination_info = {
+            "address": args[0] if len(args) else kwargs.get("host", "localhost"),
+            "port": args[4] if len(args) > 4 else int(kwargs.get("port", default_ports.get("mysql"))),
+            "service": {"name": "mysql", "resource": "mysql", "type": "db"},
+        }
+        return MySQLConnectionProxy(wrapped(*args, **kwargs), destination_info=destination_info)

elasticapm/instrumentation/packages/mysql_connector.py

@@ -35,6 +35,7 @@
     DbApi2Instrumentation,
     extract_signature,
 )
+from elasticapm.utils import default_ports
 
 
 class MySQLCursorProxy(CursorProxy):
@@ -54,4 +55,9 @@ class MySQLConnectorInstrumentation(DbApi2Instrumentation):
     instrument_list = [("mysql.connector", "connect")]
 
     def call(self, module, method, wrapped, instance, args, kwargs):
-        return MySQLConnectionProxy(wrapped(*args, **kwargs))
+        destination_info = {
+            "address": kwargs.get("host", "localhost"),
+            "port": int(kwargs.get("port", default_ports.get("mysql"))),
+            "service": {"name": "mysql", "resource": "mysql", "type": "db"},
+        }
+        return MySQLConnectionProxy(wrapped(*args, **kwargs), destination_info=destination_info)

elasticapm/instrumentation/packages/psycopg2.py

@@ -35,7 +35,7 @@
     extract_signature,
 )
 from elasticapm.traces import capture_span
-from elasticapm.utils import default_ports
+from elasticapm.utils import compat, default_ports
 
 
 class PGCursorProxy(CursorProxy):
@@ -52,14 +52,14 @@ def extract_signature(self, sql):
         return extract_signature(sql)
 
     def __enter__(self):
-        return PGCursorProxy(self.__wrapped__.__enter__())
+        return PGCursorProxy(self.__wrapped__.__enter__(), destination_info=self._self_destination_info)
 
 
 class PGConnectionProxy(ConnectionProxy):
     cursor_proxy = PGCursorProxy
 
     def __enter__(self):
-        return PGConnectionProxy(self.__wrapped__.__enter__())
+        return PGConnectionProxy(self.__wrapped__.__enter__(), destination_info=self._self_destination_info)
 
 
 class Psycopg2Instrumentation(DbApi2Instrumentation):
@@ -72,19 +72,30 @@ def call(self, module, method, wrapped, instance, args, kwargs):
 
         host = kwargs.get("host")
         if host:
-            signature += " " + str(host)
+            signature += " " + compat.text_type(host)
 
             port = kwargs.get("port")
             if port:
                 port = str(port)
                 if int(port) != default_ports.get("postgresql"):
-                    signature += ":" + port
+                    host += ":" + port
+            signature += " " + compat.text_type(host)
         else:
             # Parse connection string and extract host/port
             pass
-
-        with capture_span(signature, span_type="db", span_subtype="postgresql", span_action="connect"):
-            return PGConnectionProxy(wrapped(*args, **kwargs))
+        destination_info = {
+            "address": kwargs.get("host", "localhost"),
+            "port": int(kwargs.get("port", default_ports.get("postgresql"))),
+            "service": {"name": "postgresql", "resource": "postgresql", "type": "db"},
+        }
+        with capture_span(
+            signature,
+            span_type="db",
+            span_subtype="postgresql",
+            span_action="connect",
+            extra={"destination": destination_info},
+        ):
+            return PGConnectionProxy(wrapped(*args, **kwargs), destination_info=destination_info)
 
 
 class Psycopg2ExtensionsInstrumentation(DbApi2Instrumentation):