tracer.py

# Copyright (c) 2016-2018 Uber Technologies, Inc.
#
# 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 socket

import logging
import os
import random
import sys
import time
import opentracing
from typing import Any, Dict, Optional, List, Union

from opentracing import Format, UnsupportedFormatException
from opentracing.ext import tags as ext_tags
from opentracing.scope_managers import ThreadLocalScopeManager, ScopeManager
from opentracing.tracer import Reference

from tornado.concurrent import Future

from . import constants
from .codecs import TextCodec, ZipkinCodec, ZipkinSpanFormat, BinaryCodec, Codec
from .span import Span, SAMPLED_FLAG, DEBUG_FLAG
from .span_context import SpanContext
from .metrics import Metrics, LegacyMetricsFactory, MetricsFactory
from .utils import local_ip
from .sampler import Sampler
from .reporter import BaseReporter
from .throttler import Throttler

logger = logging.getLogger('jaeger_tracing')


class Tracer(opentracing.Tracer):
    """
    N.B. metrics has been deprecated, use metrics_factory instead.
    """
    def __init__(
        self,
        service_name: str,
        reporter: BaseReporter,
        sampler: Sampler,
        metrics: Optional[Metrics] = None,
        metrics_factory: Optional[MetricsFactory] = None,
        trace_id_header: str = constants.TRACE_ID_HEADER,
        generate_128bit_trace_id: bool = False,
        baggage_header_prefix: str = constants.BAGGAGE_HEADER_PREFIX,
        debug_id_header: str = constants.DEBUG_ID_HEADER_KEY,
        one_span_per_rpc: bool = False,
        extra_codecs: Optional[Dict[str, Codec]] = None,
        tags: Optional[Dict[str, Any]] = None,
        max_tag_value_length: int = constants.MAX_TAG_VALUE_LENGTH,
        max_traceback_length: int = constants.MAX_TRACEBACK_LENGTH,
        throttler: Optional[Throttler] = None,
        scope_manager: Optional[ScopeManager] = None,
    ) -> None:
        self.service_name = service_name
        self.reporter = reporter
        self.sampler = sampler
        self.metrics_factory = metrics_factory or LegacyMetricsFactory(metrics or Metrics())
        self.metrics = TracerMetrics(self.metrics_factory)
        self.random = random.Random(time.time() * (os.getpid() or 1))
        self.debug_id_header = debug_id_header
        self.one_span_per_rpc = one_span_per_rpc
        self.max_tag_value_length = max_tag_value_length
        self.max_traceback_length = max_traceback_length
        self.max_trace_id_bits = constants._max_trace_id_bits if generate_128bit_trace_id \
            else constants._max_id_bits
        self.codecs = {
            Format.TEXT_MAP: TextCodec(
                url_encoding=False,
                trace_id_header=trace_id_header,
                baggage_header_prefix=baggage_header_prefix,
                debug_id_header=debug_id_header,
            ),
            Format.HTTP_HEADERS: TextCodec(
                url_encoding=True,
                trace_id_header=trace_id_header,
                baggage_header_prefix=baggage_header_prefix,
                debug_id_header=debug_id_header,
            ),
            Format.BINARY: BinaryCodec(),
            ZipkinSpanFormat: ZipkinCodec(),
        }
        if extra_codecs:
            self.codecs.update(extra_codecs)
        self.tags: dict = {
            constants.JAEGER_VERSION_TAG_KEY: constants.JAEGER_CLIENT_VERSION,
        }
        if tags:
            self.tags.update(tags)

        if self.tags.get(constants.JAEGER_IP_TAG_KEY) is None:
            self.tags[constants.JAEGER_IP_TAG_KEY] = local_ip()

        if self.tags.get(constants.JAEGER_HOSTNAME_TAG_KEY) is None:
            try:
                hostname = socket.gethostname()
                self.tags[constants.JAEGER_HOSTNAME_TAG_KEY] = hostname
            except socket.error:
                logger.exception('Unable to determine host name')

        self.throttler = throttler
        if self.throttler:
            client_id = random.randint(0, sys.maxsize)
            self.throttler.set_client_id(client_id)
            self.tags[constants.CLIENT_UUID_TAG_KEY] = client_id

        self.reporter.set_process(
            service_name=self.service_name,
            tags=self.tags,
            max_length=self.max_tag_value_length,
        )

        super(Tracer, self).__init__(
            scope_manager=scope_manager or ThreadLocalScopeManager()
        )

    def start_span(self,
                   operation_name: Optional[str] = None,
                   child_of: Union[None, Span, SpanContext] = None,
                   references: Union[List[Reference], None, Reference] = None,
                   tags: Union[dict, None] = None,
                   start_time: Optional[float] = None,
                   ignore_active_span: bool = False,
                   ) -> Span:
        """
        Start and return a new Span representing a unit of work.

        :param operation_name: name of the operation represented by the new
            span from the perspective of the current service.
        :param child_of: shortcut for 'child_of' reference
        :param references: (optional) either a single Reference object or a
            list of Reference objects that identify one or more parent
            SpanContexts. (See the opentracing.Reference documentation for detail)
        :param tags: optional dictionary of Span Tags. The caller gives up
            ownership of that dictionary, because the Tracer may use it as-is
            to avoid extra data copying.
        :param start_time: an explicit Span start time as a unix timestamp per
            time.time()
        :param ignore_active_span: an explicit flag that ignores the current
            active :class:`Scope` and creates a root :class:`Span`

        :return: Returns an already-started Span instance.
        """
        parent = child_of

        if self.active_span is not None \
                and not ignore_active_span \
                and not parent:
            parent = self.active_span

        # allow Span to be passed as reference, not just SpanContext
        if isinstance(parent, Span):
            parent = parent.context

        valid_references = None
        if references:
            valid_references = list()
            if not isinstance(references, list):
                references = [references]
            for reference in references:
                if reference.referenced_context is not None:
                    valid_references.append(reference)

        # setting first reference as parent
        if valid_references and (parent is None or not parent.has_trace):
            parent = valid_references[0].referenced_context

        rpc_server = bool(tags and tags.get(ext_tags.SPAN_KIND) == ext_tags.SPAN_KIND_RPC_SERVER)

        if parent is None or not parent.has_trace:
            trace_id = self._random_id(self.max_trace_id_bits)
            span_id = self._random_id(constants._max_id_bits)
            parent_id = None
            flags = 0
            baggage = None
            if parent is None:
                sampled, sampler_tags = \
                    self.sampler.is_sampled(trace_id, operation_name or '')
                if sampled:
                    flags = SAMPLED_FLAG
                    tags = tags or {}
                    for k, v in sampler_tags.items():
                        tags[k] = v
            elif parent.debug_id and self.is_debug_allowed(operation_name):
                flags = SAMPLED_FLAG | DEBUG_FLAG
                tags = tags or {}
                tags[self.debug_id_header] = parent.debug_id
            if parent and parent.baggage:
                baggage = dict(parent.baggage)  # TODO do we need to clone?
        else:
            trace_id = parent.trace_id
            if rpc_server and self.one_span_per_rpc:
                # Zipkin-style one-span-per-RPC
                span_id = parent.span_id
                parent_id = parent.parent_id
            else:
                span_id = self._random_id(constants._max_id_bits)
                parent_id = parent.span_id
            flags = parent.flags
            baggage = dict(parent.baggage)  # TODO do we need to clone?

        span_ctx = SpanContext(trace_id=trace_id, span_id=span_id,
                               parent_id=parent_id, flags=flags,
                               baggage=baggage)
        span = Span(context=span_ctx, tracer=self,
                    operation_name=operation_name or '',
                    tags=tags, start_time=start_time, references=valid_references)

        self._emit_span_metrics(span=span, join=rpc_server)

        return span

    def start_active_span(self,
                          operation_name: Optional[str] = None,
                          child_of: Union[None, Span, SpanContext] = None,
                          references: Union[None, Reference, List[Reference]] = None,
                          tags: Optional[dict] = None,
                          start_time: Optional[float] = None,
                          ignore_active_span: bool = False,
                          finish_on_close: bool = True,
                          ):
        """
        Returns a newly started and activated :class:`Scope`

        :param operation_name: name of the operation represented by the new
            span from the perspective of the current service.
        :param child_of: shortcut for 'child_of' reference
        :param references: (optional) either a single Reference object or a
            list of Reference objects that identify one or more parent
            SpanContexts. (See the Reference documentation for detail)
        :param tags: optional dictionary of Span Tags. The caller gives up
            ownership of that dictionary, because the Tracer may use it as-is
            to avoid extra data copying.
        :param start_time: an explicit Span start time as a unix timestamp per
            time.time()
        :param ignore_active_span: an explicit flag that ignores the current
            active :class:`Scope` and creates a root :class:`Span`
        :param finish_on_close: whether :class:`Span` should automatically be
            finished when :meth:`Scope.close()` is called.

        :return: a :class:`Scope`, already registered via the :class:`ScopeManager`.
        """
        span = self.start_span(
            operation_name=operation_name,
            child_of=child_of,
            references=references,
            tags=tags,
            start_time=start_time,
            ignore_active_span=ignore_active_span,
        )
        return self.scope_manager.activate(span, finish_on_close)

    def inject(self, span_context: Union[Span, SpanContext], format: str, carrier: dict) -> None:
        codec = self.codecs.get(format, None)
        if codec is None:
            raise UnsupportedFormatException(format)
        if isinstance(span_context, Span):
            # be flexible and allow Span as argument, not only SpanContext
            span_context = span_context.context
        if not isinstance(span_context, SpanContext):
            raise ValueError(
                'Expecting Jaeger SpanContext, not %s', type(span_context))
        codec.inject(span_context=span_context, carrier=carrier)

    def extract(self, format: str, carrier: dict) -> SpanContext:
        codec = self.codecs.get(format, None)
        if codec is None:
            raise UnsupportedFormatException(format)
        return codec.extract(carrier)

    def close(self) -> Future:
        """
        Perform a clean shutdown of the tracer, flushing any traces that
        may be buffered in memory.

        :return: Returns a tornado.concurrent.Future that indicates if the
            flush has been completed.
        """
        self.sampler.close()
        return self.reporter.close()

    def _emit_span_metrics(self, span: Span, join: Optional[bool] = False) -> Span:
        if span.is_sampled():
            self.metrics.spans_started_sampled(1)
        else:
            self.metrics.spans_started_not_sampled(1)
        if not span.context.parent_id:
            if span.is_sampled():
                if join:
                    self.metrics.traces_joined_sampled(1)
                else:
                    self.metrics.traces_started_sampled(1)
            else:
                if join:
                    self.metrics.traces_joined_not_sampled(1)
                else:
                    self.metrics.traces_started_not_sampled(1)
        return span

    def report_span(self, span: Span) -> None:
        self.reporter.report_span(span)
        self.metrics.spans_finished(1)

    def random_id(self) -> int:
        """
        DEPRECATED: use _random_id() instead
        """
        return self.random.getrandbits(constants.MAX_ID_BITS)

    def _random_id(self, bitsize: int) -> int:
        return self.random.getrandbits(bitsize)

    def is_debug_allowed(self, *args: Any, **kwargs: Any) -> bool:
        if not self.throttler:
            return True
        return self.throttler.is_allowed(*args, **kwargs)


class TracerMetrics(object):
    """Tracer specific metrics."""

    def __init__(self, metrics_factory: MetricsFactory) -> None:
        self.traces_started_sampled = \
            metrics_factory.create_counter(name='jaeger:traces',
                                           tags={'state': 'started', 'sampled': 'y'})
        self.traces_started_not_sampled = \
            metrics_factory.create_counter(name='jaeger:traces',
                                           tags={'state': 'started', 'sampled': 'n'})
        self.traces_joined_sampled = \
            metrics_factory.create_counter(name='jaeger:traces',
                                           tags={'state': 'joined', 'sampled': 'y'})
        self.traces_joined_not_sampled = \
            metrics_factory.create_counter(name='jaeger:traces',
                                           tags={'state': 'joined', 'sampled': 'n'})
        self.spans_started_sampled = \
            metrics_factory.create_counter(name='jaeger:started_spans', tags={'sampled': 'y'})
        self.spans_started_not_sampled = \
            metrics_factory.create_counter(name='jaeger:started_spans', tags={'sampled': 'n'})
        self.spans_finished = \
            metrics_factory.create_counter(name='jaeger:finished_spans')