Source code for osprofiler.drivers.jaeger

# Copyright 2018 Fujitsu Ltd.
# All Rights Reserved.
#
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
#    not use this file except in compliance with the License. You may obtain
#    a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#    License for the specific language governing permissions and limitations
#    under the License.

import collections
import datetime
import time
from urllib import parse as parser

from oslo_config import cfg
from oslo_serialization import jsonutils

from osprofiler import _utils as utils
from osprofiler.drivers import base
from osprofiler import exc


[docs]class Jaeger(base.Driver):
    def __init__(self, connection_str, project=None, service=None, host=None,
                 conf=cfg.CONF, **kwargs):
        """Jaeger driver for OSProfiler."""

        super(Jaeger, self).__init__(connection_str, project=project,
                                     service=service, host=host,
                                     conf=conf, **kwargs)
        try:
            import jaeger_client
            self.jaeger_client = jaeger_client
        except ImportError:
            raise exc.CommandError(
                "To use OSProfiler with Uber Jaeger tracer, "
                "please install `jaeger-client` library. "
                "To install with pip:\n `pip install jaeger-client`."
            )

        parsed_url = parser.urlparse(connection_str)
        cfg = {
            "local_agent": {
                "reporting_host": parsed_url.hostname,
                "reporting_port": parsed_url.port,
            }
        }

        # Initialize tracer for each profiler
        service_name = "{}-{}".format(project, service)
        config = jaeger_client.Config(cfg, service_name=service_name)
        self.tracer = config.initialize_tracer()

        self.spans = collections.deque()

[docs]    @classmethod
    def get_name(cls):
        return "jaeger"

[docs]    def notify(self, payload):
        if payload["name"].endswith("start"):
            timestamp = datetime.datetime.strptime(payload["timestamp"],
                                                   "%Y-%m-%dT%H:%M:%S.%f")
            epoch = datetime.datetime.utcfromtimestamp(0)
            start_time = (timestamp - epoch).total_seconds()

            # Create parent span
            child_of = self.jaeger_client.SpanContext(
                trace_id=utils.shorten_id(payload["base_id"]),
                span_id=utils.shorten_id(payload["parent_id"]),
                parent_id=None,
                flags=self.jaeger_client.span.SAMPLED_FLAG
            )

            # Create Jaeger Tracing span
            span = self.tracer.start_span(
                operation_name=payload["name"].rstrip("-start"),
                child_of=child_of,
                tags=self.create_span_tags(payload),
                start_time=start_time
            )

            # Replace Jaeger Tracing span_id (random id) to OSProfiler span_id
            span.context.span_id = utils.shorten_id(payload["trace_id"])
            self.spans.append(span)
        else:
            span = self.spans.pop()

            # Store result of db call and function call
            for call in ("db", "function"):
                if payload.get("info", {}).get(call) is not None:
                    span.set_tag("result", payload["info"][call]["result"])

            # Span error tag and log
            if payload["info"].get("etype") is not None:
                span.set_tag("error", True)
                span.log_kv({"error.kind": payload["info"]["etype"]})
                span.log_kv({"message": payload["info"]["message"]})

            span.finish(finish_time=time.time())

[docs]    def get_report(self, base_id):
        """Please use Jaeger Tracing UI for this task."""
        return self._parse_results()

[docs]    def list_traces(self, fields=None):
        """Please use Jaeger Tracing UI for this task."""
        return []

[docs]    def list_error_traces(self):
        """Please use Jaeger Tracing UI for this task."""
        return []

[docs]    def create_span_tags(self, payload):
        """Create tags for OpenTracing span.

        :param info: Information from OSProfiler trace.
        :returns tags: A dictionary contains standard tags
                       from OpenTracing sematic conventions,
                       and some other custom tags related to http, db calls.
        """
        tags = {}
        info = payload["info"]

        if info.get("db"):
            # DB calls
            tags["db.statement"] = info["db"]["statement"]
            tags["db.params"] = jsonutils.dumps(info["db"]["params"])
        elif info.get("request"):
            # WSGI call
            tags["http.path"] = info["request"]["path"]
            tags["http.query"] = info["request"]["query"]
            tags["http.method"] = info["request"]["method"]
            tags["http.scheme"] = info["request"]["scheme"]
        elif info.get("function"):
            # RPC, function calls
            if "args" in info["function"]:
                tags["args"] = info["function"]["args"]
            if "kwargs" in info["function"]:
                tags["kwargs"] = info["function"]["kwargs"]
            tags["name"] = info["function"]["name"]

        return tags