elasticapm/traces.py

#  BSD 3-Clause License
#
#  Copyright (c) 2019, Elasticsearch BV
#  All rights reserved.
#
#  Redistribution and use in source and binary forms, with or without
#  modification, are permitted provided that the following conditions are met:
#
#  * Redistributions of source code must retain the above copyright notice, this
#    list of conditions and the following disclaimer.
#
#  * Redistributions in binary form must reproduce the above copyright notice,
#    this list of conditions and the following disclaimer in the documentation
#    and/or other materials provided with the distribution.
#
#  * Neither the name of the copyright holder nor the names of its
#    contributors may be used to endorse or promote products derived from
#    this software without specific prior written permission.
#
#  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
#  AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
#  IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
#  DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
#  FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
#  DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
#  SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
#  CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
#  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.

import functools
import random
import re
import threading
import time
import timeit
import urllib.parse
import warnings
from collections import defaultdict
from datetime import timedelta
from types import TracebackType
from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple, Type, TypeVar, Union

import elasticapm
from elasticapm.conf import constants
from elasticapm.conf.constants import LABEL_RE, SPAN, TRANSACTION
from elasticapm.context import init_execution_context
from elasticapm.metrics.base_metrics import Timer
from elasticapm.utils import encoding, get_name_from_func, nested_key, url_to_destination_resource
from elasticapm.utils.disttracing import TraceParent
from elasticapm.utils.logging import get_logger
from elasticapm.utils.time import time_to_perf_counter

__all__ = ("capture_span", "label", "set_transaction_name", "set_custom_context", "set_user_context")

error_logger = get_logger("elasticapm.errors")
logger = get_logger("elasticapm.traces")

_time_func = timeit.default_timer


execution_context = init_execution_context()

SpanType = Union["Span", "DroppedSpan"]

FuncType = Callable[..., Any]
_AnnotatedFunctionT = TypeVar("_AnnotatedFunctionT", bound=FuncType)


class ChildDuration(object):
    __slots__ = ("obj", "_nesting_level", "_start", "_duration", "_lock")

    def __init__(self, obj: "BaseSpan") -> None:
        self.obj = obj
        self._nesting_level: int = 0
        self._start: float = 0
        self._duration: timedelta = timedelta(seconds=0)
        self._lock = threading.Lock()

    def start(self, timestamp: float) -> None:
        with self._lock:
            self._nesting_level += 1
            if self._nesting_level == 1:
                self._start = timestamp

    def stop(self, timestamp: float) -> None:
        with self._lock:
            self._nesting_level -= 1
            if self._nesting_level == 0:
                self._duration += timedelta(seconds=timestamp - self._start)

    @property
    def duration(self) -> timedelta:
        return self._duration


class BaseSpan(object):
    def __init__(self, labels=None, start=None, links: Optional[Sequence[TraceParent]] = None) -> None:
        self._child_durations = ChildDuration(self)
        self.labels = {}
        self.outcome: Optional[str] = None
        self.compression_buffer: Optional[Union[Span, DroppedSpan]] = None
        self.compression_buffer_lock = threading.Lock()
        self.start_time: float = time_to_perf_counter(start) if start is not None else _time_func()
        self.ended_time: Optional[float] = None
        self.duration: Optional[timedelta] = None
        self.links: Optional[List[Dict[str, str]]] = None
        if links:
            for trace_parent in links:
                self.add_link(trace_parent)
        if labels:
            self.label(**labels)

    def child_started(self, timestamp) -> None:
        self._child_durations.start(timestamp)

    def child_ended(self, child: SpanType) -> None:
        with self.compression_buffer_lock:
            if not child.is_compression_eligible():
                if self.compression_buffer:
                    self.compression_buffer.report()
                    self.compression_buffer = None
                child.report()
            elif self.compression_buffer is None:
                self.compression_buffer = child
            elif not self.compression_buffer.try_to_compress(child):
                self.compression_buffer.report()
                self.compression_buffer = child

    def end(self, skip_frames: int = 0, duration: Optional[timedelta] = None) -> None:
        self.ended_time = _time_func()
        self.duration = duration if duration is not None else timedelta(seconds=self.ended_time - self.start_time)
        if self.compression_buffer:
            self.compression_buffer.report()
            self.compression_buffer = None

    def to_dict(self) -> dict:
        raise NotImplementedError()

    def label(self, **labels) -> None:
        """
        Label this span with one or multiple key/value labels. Keys should be strings, values can be strings, booleans,
        or numerical values (int, float, Decimal)

            span_obj.label(key1="value1", key2=True, key3=42)

        Note that keys will be dedotted, replacing dot (.), star (*) and double quote (") with an underscore (_)

        :param labels: key/value pairs of labels
        :return: None
        """
        labels = encoding.enforce_label_format(labels)
        self.labels.update(labels)

    def add_link(self, trace_parent: TraceParent) -> None:
        """
        Causally link this span/transaction to another span/transaction
        """
        if self.links is None:
            self.links = []
        self.links.append({"trace_id": trace_parent.trace_id, "span_id": trace_parent.span_id})

    def set_success(self) -> None:
        self.outcome = constants.OUTCOME.SUCCESS

    def set_failure(self) -> None:
        self.outcome = constants.OUTCOME.FAILURE

    @staticmethod
    def get_dist_tracing_id() -> str:
        return "%016x" % random.getrandbits(64)

    @property
    def tracer(self) -> "Tracer":
        raise NotImplementedError()


class Transaction(BaseSpan):
    def __init__(
        self,
        tracer: "Tracer",
        transaction_type: str = "custom",
        trace_parent: Optional[TraceParent] = None,
        is_sampled: bool = True,
        start: Optional[float] = None,
        sample_rate: Optional[float] = None,
        links: Optional[Sequence[TraceParent]] = None,
    ) -> None:
        """
        tracer
            Tracer object
        transaction_type
            Transaction type
        trace_parent
            TraceParent object representing the parent trace and trace state
        is_sampled
            Whether or not this transaction is sampled
        start
            Optional start timestamp. This is expected to be an epoch timestamp
            in seconds (such as from `time.time()`). If it is not, it's recommended
            that a `duration` is passed into the `end()` method.
        sample_rate
            Sample rate which was used to decide whether to sample this transaction.
            This is reported to the APM server so that unsampled transactions can
            be extrapolated.
        links:
            A list of traceparents to link this transaction causally
        """
        self.id = self.get_dist_tracing_id()
        if not trace_parent:
            trace_parent = TraceParent.new(self.id, is_sampled)

        # While this is set to True, underlying instrumentations will not create
        # spans. The same outcome could be reached via `is_sampled`, but
        # that requires a user to correctly restore the previous value. This
        # setting works independent of the is_sampled setting, and is assumed to
        # be False unless a user sets it to True.
        self.pause_sampling = False

        self.trace_parent: TraceParent = trace_parent
        self.timestamp = start if start is not None else time.time()
        self.name: Optional[str] = None
        self.result: Optional[str] = None
        self.transaction_type = transaction_type
        self._tracer = tracer
        # The otel bridge uses Transactions/Spans interchangeably -- storing
        # a reference to the Transaction in the Transaction simplifies things.
        self.transaction = self
        self.config_span_compression_enabled = tracer.config.span_compression_enabled
        self.config_span_compression_exact_match_max_duration = tracer.config.span_compression_exact_match_max_duration
        self.config_span_compression_same_kind_max_duration = tracer.config.span_compression_same_kind_max_duration
        self.config_exit_span_min_duration = tracer.config.exit_span_min_duration
        self.config_transaction_max_spans = tracer.config.transaction_max_spans

        self.dropped_spans: int = 0
        self.context: Dict[str, Any] = {}

        self._is_sampled = is_sampled
        self.sample_rate = sample_rate
        self._span_counter: int = 0
        self._span_timers: Dict[Tuple[str, str], Timer] = defaultdict(Timer)
        self._span_timers_lock = threading.Lock()
        self._dropped_span_statistics = defaultdict(lambda: {"count": 0, "duration.sum.us": 0})
        try:
            self._breakdown = self.tracer._agent.metrics.get_metricset(
                "elasticapm.metrics.sets.breakdown.BreakdownMetricSet"
            )
        except (LookupError, AttributeError):
            self._breakdown = None
        super().__init__(start=start)
        if links:
            for trace_parent in links:
                self.add_link(trace_parent)

    def end(self, skip_frames: int = 0, duration: Optional[timedelta] = None) -> None:
        super().end(skip_frames, duration)
        if self._breakdown:
            for (span_type, span_subtype), timer in self._span_timers.items():
                labels = {
                    "span.type": span_type,
                    "transaction.name": self.name,
                    "transaction.type": self.transaction_type,
                }
                if span_subtype:
                    labels["span.subtype"] = span_subtype
                val = timer.val
                self._breakdown.timer("span.self_time", reset_on_collect=True, unit="us", **labels).update(
                    val[0], val[1]
                )
            if self.is_sampled:
                self._breakdown.timer(
                    "span.self_time",
                    reset_on_collect=True,
                    unit="us",
                    **{"span.type": "app", "transaction.name": self.name, "transaction.type": self.transaction_type},
                ).update((self.duration - self._child_durations.duration).total_seconds() * 1_000_000)

    def _begin_span(
        self,
        name,
        span_type,
        context=None,
        leaf=False,
        labels=None,
        parent_span_id=None,
        span_subtype=None,
        span_action=None,
        sync=None,
        start=None,
        auto_activate=True,
        links: Optional[Sequence[TraceParent]] = None,
    ):
        parent_span = execution_context.get_span()
        tracer = self.tracer
        if parent_span and parent_span.leaf:
            span = DroppedSpan(parent_span, leaf=True)
        elif self.config_transaction_max_spans and self._span_counter > self.config_transaction_max_spans - 1:
            self.dropped_spans += 1
            span = DroppedSpan(parent_span, context=context)
        else:
            span = Span(
                transaction=self,
                name=name,
                span_type=span_type or "code.custom",
                context=context,
                leaf=leaf,
                labels=labels,
                parent=parent_span,
                parent_span_id=parent_span_id,
                span_subtype=span_subtype,
                span_action=span_action,
                sync=sync,
                start=start,
                links=links,
            )
            span.frames = tracer.frames_collector_func()
            self._span_counter += 1
        if auto_activate:
            execution_context.set_span(span)
        return span

    def begin_span(
        self,
        name,
        span_type,
        context=None,
        leaf=False,
        labels=None,
        span_subtype=None,
        span_action=None,
        sync=None,
        start=None,
        auto_activate=True,
        links: Optional[Sequence[TraceParent]] = None,
    ):
        """
        Begin a new span
        :param name: name of the span
        :param span_type: type of the span
        :param context: a context dict
        :param leaf: True if this is a leaf span
        :param labels: a flat string/string dict of labels
        :param span_subtype: sub type of the span, e.g. "postgresql"
        :param span_action: action of the span , e.g. "query"
        :param sync: indicate if the span is synchronous or not. In most cases, `None` should be used
        :param start: timestamp, mostly useful for testing
        :param auto_activate: whether to set this span in execution_context
        :param links: an optional list of traceparents to link this span with
        :return: the Span object
        """
        return self._begin_span(
            name,
            span_type,
            context=context,
            leaf=leaf,
            labels=labels,
            parent_span_id=None,
            span_subtype=span_subtype,
            span_action=span_action,
            sync=sync,
            start=start,
            auto_activate=auto_activate,
            links=links,
        )

    def end_span(self, skip_frames: int = 0, duration: Optional[float] = None, outcome: str = "unknown"):
        """
        End the currently active span
        :param skip_frames: numbers of frames to skip in the stack trace
        :param duration: override duration, mostly useful for testing
        :param outcome: outcome of the span, either success, failure or unknown
        :return: the ended span
        """
        span = execution_context.get_span()
        if span is None:
            raise LookupError()

        # only overwrite span outcome if it is still unknown
        if not span.outcome or span.outcome == "unknown":
            span.outcome = outcome

        span.end(skip_frames=skip_frames, duration=duration)
        return span

    def ensure_parent_id(self) -> str:
        """If current trace_parent has no span_id, generate one, then return it

        This is used to generate a span ID which the RUM agent will use to correlate
        the RUM transaction with the backend transaction.
        """
        if self.trace_parent.span_id == self.id:
            self.trace_parent.span_id = "%016x" % random.getrandbits(64)
            logger.debug("Set parent id to generated %s", self.trace_parent.span_id)
        return self.trace_parent.span_id

    def to_dict(self) -> dict:
        # This copy() only covers top level `.pop()` calls, so if we ever start
        # modifying nested data, we'll need to do a deep copy.
        context = self.context.copy()
        context["tags"] = self.labels
        result = {
            "id": self.id,
            "trace_id": self.trace_parent.trace_id,
            "name": encoding.keyword_field(self.name or ""),
            "type": encoding.keyword_field(self.transaction_type),
            "duration": self.duration.total_seconds() * 1000 if self.duration else None,
            "result": encoding.keyword_field(str(self.result)),
            "timestamp": int(self.timestamp * 1_000_000),  # microseconds
            "outcome": self.outcome,
            "sampled": self.is_sampled,
            "span_count": {"started": self._span_counter, "dropped": self.dropped_spans},
        }
        if self._dropped_span_statistics:
            result["dropped_spans_stats"] = [
                {
                    "destination_service_resource": resource,
                    "service_target_type": target_type,
                    "service_target_name": target_name,
                    "outcome": outcome,
                    "duration": {"count": v["count"], "sum": {"us": int(v["duration.sum.us"])}},
                }
                for (resource, outcome, target_type, target_name), v in self._dropped_span_statistics.items()
            ]
        if self.sample_rate is not None:
            result["sample_rate"] = float(self.sample_rate)
        if self.trace_parent:
            result["trace_id"] = self.trace_parent.trace_id
            # only set parent_id if this transaction isn't the root
            if self.trace_parent.span_id and self.trace_parent.span_id != self.id:
                result["parent_id"] = self.trace_parent.span_id
        if self.links:
            result["links"] = self.links
        # faas context belongs top-level on the transaction
        if "faas" in context:
            result["faas"] = context.pop("faas")
        # otel attributes and spankind need to be top-level
        if "otel_spankind" in context:
            result["otel"] = {"span_kind": context.pop("otel_spankind")}
        # Some transaction_store_tests use the Tracer without a Client -- the
        # extra check against `get_client()` is here to make those tests pass
        if elasticapm.get_client() and elasticapm.get_client().check_server_version(gte=(7, 16)):
            if "otel_attributes" in context:
                if "otel" not in result:
                    result["otel"] = {"attributes": context.pop("otel_attributes")}
                else:
                    result["otel"]["attributes"] = context.pop("otel_attributes")
        else:
            # Attributes map to labels for older versions
            attributes = context.pop("otel_attributes", {})
            for key, value in attributes.items():
                result["context"]["tags"][key] = value
        if self.is_sampled:
            result["context"] = context
        return result

    def track_span_duration(self, span_type, span_subtype, self_duration) -> None:
        # TODO: once asynchronous spans are supported, we should check if the transaction is already finished
        # TODO: and, if it has, exit without tracking.
        with self._span_timers_lock:
            self._span_timers[(span_type, span_subtype)].update(self_duration.total_seconds() * 1_000_000)

    @property
    def is_sampled(self) -> bool:
        return self._is_sampled

    @is_sampled.setter
    def is_sampled(self, is_sampled: bool) -> None:
        """
        This should never be called in normal operation, but often is used
        for testing. We just want to make sure our sample_rate comes out correctly
        in tracestate if we set is_sampled to False.
        """
        self._is_sampled = is_sampled
        if not is_sampled:
            if self.sample_rate:
                self.sample_rate = "0"
                self.trace_parent.add_tracestate(constants.TRACESTATE.SAMPLE_RATE, self.sample_rate)

    @property
    def tracer(self) -> "Tracer":
        return self._tracer

    def track_dropped_span(self, span: SpanType) -> None:
        with self._span_timers_lock:
            try:
                resource = span.context["destination"]["service"]["resource"]
                target_type = nested_key(span.context, "service", "target", "type")
                target_name = nested_key(span.context, "service", "target", "name")
                stats = self._dropped_span_statistics[(resource, span.outcome, target_type, target_name)]
                stats["count"] += 1
                stats["duration.sum.us"] += int(span.duration.total_seconds() * 1_000_000)
            except KeyError:
                pass


class Span(BaseSpan):
    __slots__ = (
        "id",
        "transaction",
        "name",
        "type",
        "subtype",
        "action",
        "context",
        "leaf",
        "dist_tracing_propagated",
        "timestamp",
        "start_time",
        "ended_time",
        "duration",
        "parent",
        "parent_span_id",
        "frames",
        "labels",
        "sync",
        "outcome",
        "_child_durations",
        "_cancelled",
    )

    def __init__(
        self,
        transaction: Transaction,
        name: str,
        span_type: str,
        context: Optional[dict] = None,
        leaf: bool = False,
        labels: Optional[dict] = None,
        parent: Optional["Span"] = None,
        parent_span_id: Optional[str] = None,
        span_subtype: Optional[str] = None,
        span_action: Optional[str] = None,
        sync: Optional[bool] = None,
        start: Optional[int] = None,
        links: Optional[Sequence[TraceParent]] = None,
    ) -> None:
        """
        Create a new Span

        :param transaction: transaction object that this span relates to
        :param name: Generic name of the span
        :param span_type: type of the span, e.g. db
        :param context: context dictionary
        :param leaf: is this span a leaf span?
        :param labels: a dict of labels
        :param parent_span_id: override of the span ID
        :param span_subtype: sub type of the span, e.g. mysql
        :param span_action: sub type of the span, e.g. query
        :param sync: indicate if the span was executed synchronously or asynchronously
        :param start: timestamp, mostly useful for testing
        """
        self.id = self.get_dist_tracing_id()
        self.transaction = transaction
        self.name = name or "unnamed"
        self.context = context if context is not None else {}
        self.leaf = leaf
        # timestamp is bit of a mix of monotonic and non-monotonic time sources.
        # we take the (non-monotonic) transaction timestamp, and add the (monotonic) difference of span
        # start time and transaction start time. In this respect, the span timestamp is guaranteed to grow
        # monotonically with respect to the transaction timestamp
        self.parent = parent
        self.parent_span_id = parent_span_id
        self.frames = None
        self.sync = sync
        self.type = span_type
        self.subtype = span_subtype
        self.action = span_action
        self.dist_tracing_propagated = False
        self.composite: Dict[str, Any] = {}
        self._cancelled: bool = False
        super().__init__(labels=labels, start=start, links=links)
        self.timestamp = transaction.timestamp + (self.start_time - transaction.start_time)
        if self.transaction._breakdown:
            p = self.parent if self.parent else self.transaction
            p.child_started(self.start_time)

    def to_dict(self) -> dict:
        if (
            self.composite
            and self.composite["compression_strategy"] == "same_kind"
            and nested_key(self.context, "destination", "service", "resource")
        ):
            name = "Calls to " + self.context["destination"]["service"]["resource"]
        else:
            name = self.name
        result = {
            "id": self.id,
            "transaction_id": self.transaction.id,
            "trace_id": self.transaction.trace_parent.trace_id,
            # use either the explicitly set parent_span_id, or the id of the parent, or finally the transaction id
            "parent_id": self.parent_span_id or (self.parent.id if self.parent else self.transaction.id),
            "name": encoding.keyword_field(name),
            "type": encoding.keyword_field(self.type),
            "subtype": encoding.keyword_field(self.subtype),
            "action": encoding.keyword_field(self.action),
            "timestamp": int(self.timestamp * 1000000),  # microseconds
            "duration": self.duration.total_seconds() * 1000,
            "outcome": self.outcome,
        }
        if self.transaction.sample_rate is not None:
            result["sample_rate"] = float(self.transaction.sample_rate)
        if self.sync is not None:
            result["sync"] = self.sync
        if self.labels:
            if self.context is None:
                self.context = {}
            self.context["tags"] = self.labels
        if self.links:
            result["links"] = self.links
        if self.context:
            self.autofill_resource_context()
            # otel attributes and spankind need to be top-level
            if "otel_spankind" in self.context:
                result["otel"] = {"span_kind": self.context.pop("otel_spankind")}
            if self.tracer._agent.check_server_version(gte=(7, 16)):
                if "otel_attributes" in self.context:
                    if "otel" not in result:
                        result["otel"] = {"attributes": self.context.pop("otel_attributes")}
                    else:
                        result["otel"]["attributes"] = self.context.pop("otel_attributes")
            else:
                # Attributes map to labels for older versions
                attributes = self.context.pop("otel_attributes", {})
                if attributes and ("tags" not in self.context):
                    self.context["tags"] = {}
                for key, value in attributes.items():
                    self.context["tags"][key] = value
            result["context"] = self.context
        if self.frames:
            result["stacktrace"] = self.frames
        if self.composite:
            result["composite"] = {
                "compression_strategy": self.composite["compression_strategy"],
                "sum": self.composite["sum"].total_seconds() * 1000,
                "count": self.composite["count"],
            }
        return result

    def is_same_kind(self, other_span: SpanType) -> bool:
        """
        For compression purposes, two spans are considered to be of the same kind if they have the same
        values for type, subtype, and destination.service.resource
        :param other_span: another span object
        :return: bool
        """
        target_type = nested_key(self.context, "service", "target", "type")
        target_name = nested_key(self.context, "service", "target", "name")
        return bool(
            self.type == other_span.type
            and self.subtype == other_span.subtype
            and (target_type or target_name)
            and target_type == nested_key(other_span.context, "service", "target", "type")
            and target_name == nested_key(other_span.context, "service", "target", "name")
        )

    def is_exact_match(self, other_span: SpanType) -> bool:
        """
        For compression purposes, two spans are considered to be an exact match if the have the same
        name and are of the same kind.

        :param other_span: another span object
        :return: bool
        """
        return bool(self.name == other_span.name and self.is_same_kind(other_span))

    def is_compression_eligible(self) -> bool:
        """
        Determine if this span is eligible for compression.
        """
        if self.transaction.config_span_compression_enabled:
            return self.leaf and not self.dist_tracing_propagated and self.outcome in (None, constants.OUTCOME.SUCCESS)
        return False

    @property
    def discardable(self) -> bool:
        return self.leaf and not self.dist_tracing_propagated and self.outcome == constants.OUTCOME.SUCCESS

    def end(self, skip_frames: int = 0, duration: Optional[float] = None) -> None:
        """
        End this span and queue it for sending.

        :param skip_frames: amount of frames to skip from the beginning of the stack trace
        :param duration: override duration, mostly useful for testing
        :return: None
        """
        self.autofill_resource_context()
        self.autofill_service_target()
        super().end(skip_frames, duration)
        tracer = self.transaction.tracer
        if (
            tracer.span_stack_trace_min_duration >= timedelta(seconds=0)
            and self.duration >= tracer.span_stack_trace_min_duration
            and self.frames
        ):
            self.frames = tracer.frames_processing_func(self.frames)[skip_frames:]
        else:
            self.frames = None
        current_span = execution_context.get_span()
        # Because otel can detach context without ending the span, we need to
        # make sure we only unset the span if it's currently set.
        if current_span is self:
            execution_context.unset_span()

        p = self.parent if self.parent else self.transaction
        if self.transaction._breakdown:
            p._child_durations.stop(self.start_time + self.duration.total_seconds())
            self.transaction.track_span_duration(
                self.type, self.subtype, self.duration - self._child_durations.duration
            )
        p.child_ended(self)

    def report(self) -> None:
        if self.discardable and self.duration < self.transaction.config_exit_span_min_duration:
            self.transaction.track_dropped_span(self)
            self.transaction.dropped_spans += 1
        elif self._cancelled:
            self.transaction._span_counter -= 1
        else:
            self.tracer.queue_func(SPAN, self.to_dict())

    def try_to_compress(self, sibling: SpanType) -> bool:
        compression_strategy = (
            self._try_to_compress_composite(sibling) if self.composite else self._try_to_compress_regular(sibling)
        )
        if not compression_strategy:
            return False

        if not self.composite:
            self.composite = {"compression_strategy": compression_strategy, "count": 1, "sum": self.duration}
        self.composite["count"] += 1
        self.composite["sum"] += sibling.duration
        self.duration = timedelta(seconds=sibling.ended_time - self.start_time)
        self.transaction._span_counter -= 1
        return True

    def _try_to_compress_composite(self, sibling: SpanType) -> Optional[str]:
        if self.composite["compression_strategy"] == "exact_match":
            return (
                "exact_match"
                if (
                    self.is_exact_match(sibling)
                    and sibling.duration <= self.transaction.config_span_compression_exact_match_max_duration
                )
                else None
            )
        elif self.composite["compression_strategy"] == "same_kind":
            return (
                "same_kind"
                if (
                    self.is_same_kind(sibling)
                    and sibling.duration <= self.transaction.config_span_compression_same_kind_max_duration
                )
                else None
            )
        return None

    def _try_to_compress_regular(self, sibling: SpanType) -> Optional[str]:
        if not self.is_same_kind(sibling):
            return None
        if self.name == sibling.name:
            max_duration = self.transaction.config_span_compression_exact_match_max_duration
            if self.duration <= max_duration and sibling.duration <= max_duration:
                return "exact_match"
            return None
        max_duration = self.transaction.config_span_compression_same_kind_max_duration
        if self.duration <= max_duration and sibling.duration <= max_duration:
            return "same_kind"
        return None

    def update_context(self, key, data) -> None:
        """
        Update the context data for given key
        :param key: the key, e.g. "db"
        :param data: a dictionary
        :return: None
        """
        current = self.context.get(key, {})
        current.update(data)
        self.context[key] = current

    def autofill_resource_context(self) -> None:
        """Automatically fills "resource" fields based on other fields"""
        if self.context:
            resource = nested_key(self.context, "destination", "service", "resource")
            if not resource and (self.leaf or any(k in self.context for k in ("destination", "db", "message", "http"))):
                type_info = self.subtype or self.type
                instance = nested_key(self.context, "db", "instance")
                queue_name = nested_key(self.context, "message", "queue", "name")
                http_url = nested_key(self.context, "http", "url")
                if instance:
                    resource = f"{type_info}/{instance}"
                elif queue_name:
                    resource = f"{type_info}/{queue_name}"
                elif http_url:
                    resource = url_to_destination_resource(http_url)
                else:
                    resource = type_info
                if "destination" not in self.context:
                    self.context["destination"] = {}
                if "service" not in self.context["destination"]:
                    self.context["destination"]["service"] = {}
                self.context["destination"]["service"]["resource"] = resource
                # set fields that are deprecated, but still required by APM Server API
                if "name" not in self.context["destination"]["service"]:
                    self.context["destination"]["service"]["name"] = ""
                if "type" not in self.context["destination"]["service"]:
                    self.context["destination"]["service"]["type"] = ""

    def autofill_service_target(self) -> None:
        if self.leaf:
            service_target = nested_key(self.context, "service", "target") or {}

            if "type" not in service_target:  # infer type from span type & subtype
                # use sub-type if provided, fallback on type othewise
                service_target["type"] = self.subtype or self.type

            if "name" not in service_target:  # infer name from span attributes
                if nested_key(self.context, "db", "instance"):  # database spans
                    service_target["name"] = self.context["db"]["instance"]
                elif "message" in self.context:  # messaging spans
                    service_target["name"] = self.context["message"]["queue"]["name"]
                elif nested_key(self.context, "http", "url"):  # http spans
                    url = self.context["http"]["url"]
                    parsed_url = urllib.parse.urlparse(url)
                    service_target["name"] = parsed_url.hostname
                    if parsed_url.port:
                        service_target["name"] += f":{parsed_url.port}"
            if "service" not in self.context:
                self.context["service"] = {}
            self.context["service"]["target"] = service_target
        elif nested_key(self.context, "service", "target"):
            # non-exit spans should not have service.target.* fields
            del self.context["service"]["target"]

    def cancel(self) -> None:
        """
        Mark span as cancelled. Cancelled spans don't count towards started spans nor dropped spans.

        No checks are made to ensure that spans which already propagated distributed context are not
        cancelled.
        """
        self._cancelled = True

    def __str__(self):
        return "{}/{}/{}".format(self.name, self.type, self.subtype)

    @property
    def tracer(self) -> "Tracer":
        return self.transaction.tracer


class DroppedSpan(BaseSpan):
    __slots__ = ("leaf", "parent", "id", "context", "outcome", "dist_tracing_propagated")

    def __init__(self, parent, leaf=False, start=None, context=None) -> None:
        self.parent = parent
        self.leaf = leaf
        self.id = None
        self.dist_tracing_propagated = False
        self.context = context
        self.outcome = constants.OUTCOME.UNKNOWN
        super(DroppedSpan, self).__init__(start=start)

    def end(self, skip_frames: int = 0, duration: Optional[float] = None) -> None:
        super().end(skip_frames, duration)
        execution_context.unset_span()

    def child_started(self, timestamp) -> None:
        pass

    def child_ended(self, child: SpanType) -> None:
        pass

    def update_context(self, key, data) -> None:
        pass

    def report(self) -> None:
        pass

    def try_to_compress(self, sibling: SpanType) -> bool:
        return False

    def is_compression_eligible(self) -> bool:
        return False

    @property
    def name(self):
        return "DroppedSpan"

    @property
    def type(self):
        return None

    @property
    def subtype(self):
        return None

    @property
    def action(self):
        return None


class Tracer(object):
    def __init__(
        self, frames_collector_func, frames_processing_func, queue_func, config, agent: "elasticapm.Client"
    ) -> None:
        self.config = config
        self.queue_func = queue_func
        self.frames_processing_func = frames_processing_func
        self.frames_collector_func = frames_collector_func
        self._agent = agent
        self._ignore_patterns = [re.compile(p) for p in config.transactions_ignore_patterns or []]

    @property
    def span_stack_trace_min_duration(self) -> timedelta:
        if self.config.span_stack_trace_min_duration != timedelta(
            seconds=0.005
        ) or self.config.span_frames_min_duration == timedelta(seconds=0.005):
            # No need to check span_frames_min_duration
            return self.config.span_stack_trace_min_duration
        else:
            # span_stack_trace_min_duration is default value and span_frames_min_duration is non-default.
            # warn and use span_frames_min_duration
            warnings.warn(
                "`span_frames_min_duration` is deprecated. Please use `span_stack_trace_min_duration`.",
                DeprecationWarning,
            )
            if self.config.span_frames_min_duration < timedelta(seconds=0):
                return timedelta(seconds=0)
            elif self.config.span_frames_min_duration == timedelta(seconds=0):
                return timedelta(seconds=-1)
            else:
                return self.config.span_frames_min_duration

    def begin_transaction(
        self,
        transaction_type: str,
        trace_parent: Optional[TraceParent] = None,
        start: Optional[float] = None,
        auto_activate: bool = True,
        links: Optional[Sequence[TraceParent]] = None,
    ) -> Transaction:
        """
        Start a new transactions and bind it in a thread-local variable

        :param transaction_type: type of the transaction, e.g. "request"
        :param trace_parent: an optional TraceParent object
        :param start: override the start timestamp, mostly useful for testing
        :param auto_activate: whether to set this transaction in execution_context
        :param links: list of traceparents to causally link this transaction to
        :returns the Transaction object
        """
        links = links if links else []
        continuation_strategy = self.config.trace_continuation_strategy

        # we restart the trace if continuation strategy is "restart", or if it is "restart_external" and our
        # "es" key is not in the tracestate header. In both cases, the original TraceParent is added to trace links.
        if trace_parent and continuation_strategy != constants.TRACE_CONTINUATION_STRATEGY.CONTINUE:
            if continuation_strategy == constants.TRACE_CONTINUATION_STRATEGY.RESTART or (
                continuation_strategy == constants.TRACE_CONTINUATION_STRATEGY.RESTART_EXTERNAL
                and not trace_parent.tracestate_dict
            ):
                links.append(trace_parent)
                trace_parent = None
        if trace_parent:
            is_sampled = bool(trace_parent.trace_options.recorded)
            sample_rate = trace_parent.tracestate_dict.get(constants.TRACESTATE.SAMPLE_RATE)
        else:
            is_sampled = (
                self.config.transaction_sample_rate == 1.0 or self.config.transaction_sample_rate > random.random()
            )
            if not is_sampled:
                sample_rate = "0"
            else:
                sample_rate = str(self.config.transaction_sample_rate)

        transaction = Transaction(
            self,
            transaction_type,
            trace_parent=trace_parent,
            is_sampled=is_sampled,
            start=start,
            sample_rate=sample_rate,
            links=links,
        )
        if trace_parent is None:
            transaction.trace_parent.add_tracestate(constants.TRACESTATE.SAMPLE_RATE, sample_rate)
        if auto_activate:
            execution_context.set_transaction(transaction)
        return transaction

    def end_transaction(self, result=None, transaction_name=None, duration=None):
        """
        End the current transaction and queue it for sending
        :param result: result of the transaction, e.g. "OK" or 200
        :param transaction_name: name of the transaction
        :param duration: override duration, mostly useful for testing
        :return:
        """
        transaction = execution_context.get_transaction(clear=True)
        if transaction:
            if transaction.name is None:
                transaction.name = str(transaction_name) if transaction_name is not None else ""
            transaction.end(duration=duration)
            if self._should_ignore(transaction.name):
                return
            if not transaction.is_sampled and self._agent.check_server_version(gte=(8, 0)):
                return
            if transaction.result is None:
                transaction.result = result
            self.queue_func(TRANSACTION, transaction.to_dict())
        return transaction

    def _should_ignore(self, transaction_name):
        for pattern in self._ignore_patterns:
            if pattern.search(transaction_name):
                return True
        return False


class capture_span(object):
    __slots__ = (
        "name",
        "type",
        "subtype",
        "action",
        "extra",
        "skip_frames",
        "leaf",
        "labels",
        "duration",
        "start",
        "sync",
        "links",
    )

    def __init__(
        self,
        name: Optional[str] = None,
        span_type: str = "code.custom",
        extra: Optional[dict] = None,
        skip_frames: int = 0,
        leaf: bool = False,
        labels: Optional[dict] = None,
        span_subtype: Optional[str] = None,
        span_action: Optional[str] = None,
        start: Optional[int] = None,
        duration: Optional[Union[float, timedelta]] = None,
        sync: Optional[bool] = None,
        links: Optional[Sequence[TraceParent]] = None,
    ) -> None:
        self.name = name
        if span_subtype is None and "." in span_type:
            # old style dotted type, let's split it up
            type_bits = span_type.split(".")
            if len(type_bits) == 2:
                span_type, span_subtype = type_bits[:2]
            else:
                span_type, span_subtype, span_action = type_bits[:3]
        self.type = span_type
        self.subtype = span_subtype
        self.action = span_action
        self.extra = extra
        self.skip_frames = skip_frames
        self.leaf = leaf
        self.labels = labels
        self.start = start
        if duration and not isinstance(duration, timedelta):
            duration = timedelta(seconds=duration)
        self.duration = duration
        self.sync = sync
        self.links = links

    def __call__(self, func: _AnnotatedFunctionT) -> _AnnotatedFunctionT:
        self.name = self.name or get_name_from_func(func)

        @functools.wraps(func)
        def decorated(*args, **kwds):
            with self:
                return func(*args, **kwds)

        return decorated

    def __enter__(self) -> Optional[SpanType]:
        return self.handle_enter(self.sync)

    def __exit__(
        self, exc_type: Optional[Type[BaseException]], exc_val: Optional[BaseException], exc_tb: Optional[TracebackType]
    ) -> None:
        self.handle_exit(exc_type, exc_val, exc_tb)

    def handle_enter(self, sync: bool) -> Optional[SpanType]:
        transaction = execution_context.get_transaction()
        if transaction and transaction.is_sampled:
            return transaction.begin_span(
                self.name,
                self.type,
                context=self.extra,
                leaf=self.leaf,
                labels=self.labels,
                span_subtype=self.subtype,
                span_action=self.action,
                start=self.start,
                sync=sync,
                links=self.links,
            )
        return None

    def handle_exit(
        self, exc_type: Optional[Type[BaseException]], exc_val: Optional[BaseException], exc_tb: Optional[TracebackType]
    ) -> None:
        transaction = execution_context.get_transaction()

        if transaction and transaction.is_sampled:
            try:
                outcome = "failure" if exc_val else "success"
                span = transaction.end_span(self.skip_frames, duration=self.duration, outcome=outcome)
                should_track_dropped = (
                    transaction.tracer._agent.check_server_version(gte=(7, 16)) if transaction.tracer._agent else True
                )
                if should_track_dropped and isinstance(span, DroppedSpan) and span.context:
                    transaction.track_dropped_span(span)
                if exc_val and not isinstance(span, DroppedSpan):
                    try:
                        exc_val._elastic_apm_span_id = span.id
                    except AttributeError:
                        # could happen if the exception has __slots__
                        pass
            except LookupError:
                logger.debug("ended non-existing span %s of type %s", self.name, self.type)


def label(**labels) -> None:
    """
    Labels current transaction. Keys should be strings, values can be strings, booleans,
    or numerical values (int, float, Decimal)

    :param labels: key/value map of labels
    """
    transaction = execution_context.get_transaction()
    if not transaction:
        client = elasticapm.get_client()
        if not client or client.config.enabled:
            error_logger.warning("Ignored labels %s. No transaction currently active.", ", ".join(labels.keys()))
    else:
        transaction.label(**labels)


def set_transaction_name(name: str, override: bool = True) -> None:
    """
    Sets the name of the transaction

    :param name: the name of the transaction
    :param override: if set to False, the name is only set if no name has been set before
    :return: None
    """
    transaction = execution_context.get_transaction()
    if not transaction:
        return
    if transaction.name is None or override:
        transaction.name = str(name)


def set_transaction_result(result, override=True) -> None:
    """
    Sets the result of the transaction. The result could be e.g. the HTTP status class (e.g "HTTP 5xx") for
    HTTP requests, or "success"/"failure" for background tasks.

    :param result: Details of the transaction result that should be set
    :param override: if set to False, the name is only set if no name has been set before
    :return: None
    """

    transaction = execution_context.get_transaction()
    if not transaction:
        return
    if transaction.result is None or override:
        transaction.result = result


def set_transaction_outcome(outcome=None, http_status_code=None, override=True) -> None:
    """
    Set the outcome of the transaction. This should only be done at the end of a transaction
    after the outcome is determined.

    If an invalid outcome is provided, an INFO level log message will be issued.

    :param outcome: the outcome of the transaction. Allowed values are "success", "failure", "unknown". None is
                    allowed if a http_status_code is provided.
    :param http_status_code: An integer value of the HTTP status code. If provided, the outcome will be determined
                             based on the status code: Success if the status is lower than 500, failure otherwise.
                             If both a valid outcome and an http_status_code is provided, the former is used
    :param override: If set to False, the outcome will only be updated if its current value is None

    :return: None
    """
    transaction = execution_context.get_transaction()
    if not transaction:
        return
    if http_status_code and outcome not in constants.OUTCOME:
        try:
            http_status_code = int(http_status_code)
            outcome = constants.OUTCOME.SUCCESS if http_status_code < 500 else constants.OUTCOME.FAILURE
        except ValueError:
            logger.info('Invalid HTTP status %r provided, outcome set to "unknown"', http_status_code)
            outcome = constants.OUTCOME.UNKNOWN
    elif outcome not in constants.OUTCOME:
        logger.info('Invalid outcome %r provided, outcome set to "unknown"', outcome)
        outcome = constants.OUTCOME.UNKNOWN
    if outcome and (transaction.outcome is None or override):
        transaction.outcome = outcome


def get_transaction_id():
    """
    Returns the current transaction ID
    """
    transaction = execution_context.get_transaction()
    if not transaction:
        return
    return transaction.id


def get_trace_parent_header():
    """
    Return the trace parent header for the current transaction.
    """
    transaction = execution_context.get_transaction()
    if not transaction or not transaction.trace_parent:
        return
    return transaction.trace_parent.to_string()


def get_trace_id():
    """
    Returns the current trace ID
    """
    transaction = execution_context.get_transaction()
    if not transaction:
        return
    return transaction.trace_parent.trace_id if transaction.trace_parent else None


def get_span_id():
    """
    Returns the current span ID
    """
    span = execution_context.get_span()
    if not span:
        return
    return span.id


def set_context(data, key="custom") -> None:
    """
    Attach contextual data to the current transaction and errors that happen during the current transaction.

    If the transaction is not sampled, this function becomes a no-op.

    :param data: a dictionary, or a callable that returns a dictionary
    :param key: the namespace for this data
    """
    transaction = execution_context.get_transaction()
    if not (transaction and transaction.is_sampled):
        return
    if callable(data):
        data = data()

    # remove invalid characters from key names
    for k in list(data.keys()):
        if LABEL_RE.search(k):
            data[LABEL_RE.sub("_", k)] = data.pop(k)

    if key in transaction.context:
        transaction.context[key].update(data)
    else:
        transaction.context[key] = data


set_custom_context = functools.partial(set_context, key="custom")


def set_user_context(username=None, email=None, user_id=None) -> None:
    data = {}
    if username is not None:
        data["username"] = encoding.keyword_field(username)
    if email is not None:
        data["email"] = encoding.keyword_field(email)
    if user_id is not None:
        data["id"] = encoding.keyword_field(user_id)
    set_context(data, "user")