metrics.py

import io
import os
import random
import re
import sys
import threading
import time
import zlib
from contextlib import contextmanager
from datetime import datetime
from functools import wraps, partial

import sentry_sdk
from sentry_sdk._compat import PY2, text_type, utc_from_timestamp, iteritems
from sentry_sdk.utils import (
    ContextVar,
    now,
    nanosecond_time,
    to_timestamp,
    serialize_frame,
    json_dumps,
    is_gevent,
)
from sentry_sdk.envelope import Envelope, Item
from sentry_sdk.tracing import (
    TRANSACTION_SOURCE_ROUTE,
    TRANSACTION_SOURCE_VIEW,
    TRANSACTION_SOURCE_COMPONENT,
    TRANSACTION_SOURCE_TASK,
)
from sentry_sdk._types import TYPE_CHECKING

if TYPE_CHECKING:
    from typing import Any
    from typing import Callable
    from typing import Dict
    from typing import Generator
    from typing import Iterable
    from typing import List
    from typing import Optional
    from typing import Set
    from typing import Tuple
    from typing import Union

    from sentry_sdk._types import BucketKey
    from sentry_sdk._types import DurationUnit
    from sentry_sdk._types import FlushedMetricValue
    from sentry_sdk._types import MeasurementUnit
    from sentry_sdk._types import MetricMetaKey
    from sentry_sdk._types import MetricTagValue
    from sentry_sdk._types import MetricTags
    from sentry_sdk._types import MetricTagsInternal
    from sentry_sdk._types import MetricType
    from sentry_sdk._types import MetricValue


try:
    from gevent.monkey import get_original  # type: ignore
    from gevent.threadpool import ThreadPool  # type: ignore
except ImportError:
    import importlib

    def get_original(module, name):
        # type: (str, str) -> Any
        return getattr(importlib.import_module(module), name)


_in_metrics = ContextVar("in_metrics")
_sanitize_key = partial(re.compile(r"[^a-zA-Z0-9_/.-]+").sub, "_")
_sanitize_value = partial(re.compile(r"[^\w\d_:/@\.{}\[\]$-]+", re.UNICODE).sub, "_")
_set = set  # set is shadowed below

GOOD_TRANSACTION_SOURCES = frozenset(
    [
        TRANSACTION_SOURCE_ROUTE,
        TRANSACTION_SOURCE_VIEW,
        TRANSACTION_SOURCE_COMPONENT,
        TRANSACTION_SOURCE_TASK,
    ]
)


def get_code_location(stacklevel):
    # type: (int) -> Optional[Dict[str, Any]]
    try:
        frm = sys._getframe(stacklevel)
    except Exception:
        return None

    return serialize_frame(
        frm, include_local_variables=False, include_source_context=True
    )


@contextmanager
def recursion_protection():
    # type: () -> Generator[bool, None, None]
    """Enters recursion protection and returns the old flag."""
    old_in_metrics = _in_metrics.get(False)
    _in_metrics.set(True)
    try:
        yield old_in_metrics
    finally:
        _in_metrics.set(old_in_metrics)


def metrics_noop(func):
    # type: (Any) -> Any
    """Convenient decorator that uses `recursion_protection` to
    make a function a noop.
    """

    @wraps(func)
    def new_func(*args, **kwargs):
        # type: (*Any, **Any) -> Any
        with recursion_protection() as in_metrics:
            if not in_metrics:
                return func(*args, **kwargs)

    return new_func


class Metric(object):
    __slots__ = ()

    @property
    def weight(self):
        # type: (...) -> int
        raise NotImplementedError()

    def add(
        self, value  # type: MetricValue
    ):
        # type: (...) -> None
        raise NotImplementedError()

    def serialize_value(self):
        # type: (...) -> Iterable[FlushedMetricValue]
        raise NotImplementedError()


class CounterMetric(Metric):
    __slots__ = ("value",)

    def __init__(
        self, first  # type: MetricValue
    ):
        # type: (...) -> None
        self.value = float(first)

    @property
    def weight(self):
        # type: (...) -> int
        return 1

    def add(
        self, value  # type: MetricValue
    ):
        # type: (...) -> None
        self.value += float(value)

    def serialize_value(self):
        # type: (...) -> Iterable[FlushedMetricValue]
        return (self.value,)


class GaugeMetric(Metric):
    __slots__ = (
        "last",
        "min",
        "max",
        "sum",
        "count",
    )

    def __init__(
        self, first  # type: MetricValue
    ):
        # type: (...) -> None
        first = float(first)
        self.last = first
        self.min = first
        self.max = first
        self.sum = first
        self.count = 1

    @property
    def weight(self):
        # type: (...) -> int
        # Number of elements.
        return 5

    def add(
        self, value  # type: MetricValue
    ):
        # type: (...) -> None
        value = float(value)
        self.last = value
        self.min = min(self.min, value)
        self.max = max(self.max, value)
        self.sum += value
        self.count += 1

    def serialize_value(self):
        # type: (...) -> Iterable[FlushedMetricValue]
        return (
            self.last,
            self.min,
            self.max,
            self.sum,
            self.count,
        )


class DistributionMetric(Metric):
    __slots__ = ("value",)

    def __init__(
        self, first  # type: MetricValue
    ):
        # type(...) -> None
        self.value = [float(first)]

    @property
    def weight(self):
        # type: (...) -> int
        return len(self.value)

    def add(
        self, value  # type: MetricValue
    ):
        # type: (...) -> None
        self.value.append(float(value))

    def serialize_value(self):
        # type: (...) -> Iterable[FlushedMetricValue]
        return self.value


class SetMetric(Metric):
    __slots__ = ("value",)

    def __init__(
        self, first  # type: MetricValue
    ):
        # type: (...) -> None
        self.value = {first}

    @property
    def weight(self):
        # type: (...) -> int
        return len(self.value)

    def add(
        self, value  # type: MetricValue
    ):
        # type: (...) -> None
        self.value.add(value)

    def serialize_value(self):
        # type: (...) -> Iterable[FlushedMetricValue]
        def _hash(x):
            # type: (MetricValue) -> int
            if isinstance(x, str):
                return zlib.crc32(x.encode("utf-8")) & 0xFFFFFFFF
            return int(x)

        return (_hash(value) for value in self.value)


def _encode_metrics(flushable_buckets):
    # type: (Iterable[Tuple[int, Dict[BucketKey, Metric]]]) -> bytes
    out = io.BytesIO()
    _write = out.write

    # Note on sanetization: we intentionally sanetize in emission (serialization)
    # and not during aggregation for performance reasons.  This means that the
    # envelope can in fact have duplicate buckets stored.  This is acceptable for
    # relay side emission and should not happen commonly.

    for timestamp, buckets in flushable_buckets:
        for bucket_key, metric in iteritems(buckets):
            metric_type, metric_name, metric_unit, metric_tags = bucket_key
            metric_name = _sanitize_key(metric_name)
            _write(metric_name.encode("utf-8"))
            _write(b"@")
            _write(metric_unit.encode("utf-8"))

            for serialized_value in metric.serialize_value():
                _write(b":")
                _write(str(serialized_value).encode("utf-8"))

            _write(b"|")
            _write(metric_type.encode("ascii"))

            if metric_tags:
                _write(b"|#")
                first = True
                for tag_key, tag_value in metric_tags:
                    tag_key = _sanitize_key(tag_key)
                    if not tag_key:
                        continue
                    if first:
                        first = False
                    else:
                        _write(b",")
                    _write(tag_key.encode("utf-8"))
                    _write(b":")
                    _write(_sanitize_value(tag_value).encode("utf-8"))

            _write(b"|T")
            _write(str(timestamp).encode("ascii"))
            _write(b"\n")

    return out.getvalue()


def _encode_locations(timestamp, code_locations):
    # type: (int, Iterable[Tuple[MetricMetaKey, Dict[str, Any]]]) -> bytes
    mapping = {}  # type: Dict[str, List[Any]]

    for key, loc in code_locations:
        metric_type, name, unit = key
        mri = "{}:{}@{}".format(metric_type, _sanitize_key(name), unit)

        loc["type"] = "location"
        mapping.setdefault(mri, []).append(loc)

    return json_dumps({"timestamp": timestamp, "mapping": mapping})


METRIC_TYPES = {
    "c": CounterMetric,
    "g": GaugeMetric,
    "d": DistributionMetric,
    "s": SetMetric,
}

# some of these are dumb
TIMING_FUNCTIONS = {
    "nanosecond": nanosecond_time,
    "microsecond": lambda: nanosecond_time() / 1000.0,
    "millisecond": lambda: nanosecond_time() / 1000000.0,
    "second": now,
    "minute": lambda: now() / 60.0,
    "hour": lambda: now() / 3600.0,
    "day": lambda: now() / 3600.0 / 24.0,
    "week": lambda: now() / 3600.0 / 24.0 / 7.0,
}


class LocalAggregator(object):
    __slots__ = ("_measurements",)

    def __init__(self):
        # type: (...) -> None
        self._measurements = (
            {}
        )  # type: Dict[Tuple[str, MetricTagsInternal], Tuple[float, float, int, float]]

    def add(
        self,
        ty,  # type: MetricType
        key,  # type: str
        value,  # type: float
        unit,  # type: MeasurementUnit
        tags,  # type: MetricTagsInternal
    ):
        # type: (...) -> None
        export_key = "%s:%s@%s" % (ty, key, unit)
        bucket_key = (export_key, tags)

        old = self._measurements.get(bucket_key)
        if old is not None:
            v_min, v_max, v_count, v_sum = old
            v_min = min(v_min, value)
            v_max = max(v_max, value)
            v_count += 1
            v_sum += value
        else:
            v_min = v_max = v_sum = value
            v_count = 1
        self._measurements[bucket_key] = (v_min, v_max, v_count, v_sum)

    def to_json(self):
        # type: (...) -> Dict[str, Any]
        rv = {}  # type: Any
        for (export_key, tags), (
            v_min,
            v_max,
            v_count,
            v_sum,
        ) in self._measurements.items():
            rv.setdefault(export_key, []).append(
                {
                    "tags": _tags_to_dict(tags),
                    "min": v_min,
                    "max": v_max,
                    "count": v_count,
                    "sum": v_sum,
                }
            )
        return rv


class MetricsAggregator(object):
    ROLLUP_IN_SECONDS = 10.0
    MAX_WEIGHT = 100000
    FLUSHER_SLEEP_TIME = 5.0

    def __init__(
        self,
        capture_func,  # type: Callable[[Envelope], None]
        enable_code_locations=False,  # type: bool
    ):
        # type: (...) -> None
        self.buckets = {}  # type: Dict[int, Any]
        self._enable_code_locations = enable_code_locations
        self._seen_locations = _set()  # type: Set[Tuple[int, MetricMetaKey]]
        self._pending_locations = {}  # type: Dict[int, List[Tuple[MetricMetaKey, Any]]]
        self._buckets_total_weight = 0
        self._capture_func = capture_func
        self._running = True
        self._lock = threading.Lock()

        if is_gevent() and PY2:
            # get_original on threading.Event in Python 2 incorrectly returns
            # the gevent-patched class. Luckily, threading.Event is just an alias
            # for threading._Event in Python 2, and get_original on
            # threading._Event correctly gets us the stdlib original.
            event_cls = get_original("threading", "_Event")
        else:
            event_cls = get_original("threading", "Event")
        self._flush_event = event_cls()  # type: threading.Event

        self._force_flush = False

        # The aggregator shifts its flushing by up to an entire rollup window to
        # avoid multiple clients trampling on end of a 10 second window as all the
        # buckets are anchored to multiples of ROLLUP seconds.  We randomize this
        # number once per aggregator boot to achieve some level of offsetting
        # across a fleet of deployed SDKs.  Relay itself will also apply independent
        # jittering.
        self._flush_shift = random.random() * self.ROLLUP_IN_SECONDS

        self._flusher = None  # type: Optional[Union[threading.Thread, ThreadPool]]
        self._flusher_pid = None  # type: Optional[int]
        self._ensure_thread()

    def _ensure_thread(self):
        # type: (...) -> bool
        """For forking processes we might need to restart this thread.
        This ensures that our process actually has that thread running.
        """
        if not self._running:
            return False

        pid = os.getpid()
        if self._flusher_pid == pid:
            return True

        with self._lock:
            self._flusher_pid = pid

            if not is_gevent():
                self._flusher = threading.Thread(target=self._flush_loop)
                self._flusher.daemon = True
                start_flusher = self._flusher.start
            else:
                self._flusher = ThreadPool(1)
                start_flusher = partial(self._flusher.spawn, func=self._flush_loop)

            try:
                start_flusher()
            except RuntimeError:
                # Unfortunately at this point the interpreter is in a state that no
                # longer allows us to spawn a thread and we have to bail.
                self._running = False
                return False

        return True

    def _flush_loop(self):
        # type: (...) -> None
        _in_metrics.set(True)
        while self._running or self._force_flush:
            self._flush()
            if self._running:
                self._flush_event.wait(self.FLUSHER_SLEEP_TIME)

    def _flush(self):
        # type: (...) -> None
        self._emit(self._flushable_buckets(), self._flushable_locations())

    def _flushable_buckets(self):
        # type: (...) -> (Iterable[Tuple[int, Dict[BucketKey, Metric]]])
        with self._lock:
            force_flush = self._force_flush
            cutoff = time.time() - self.ROLLUP_IN_SECONDS - self._flush_shift
            flushable_buckets = ()  # type: Iterable[Tuple[int, Dict[BucketKey, Metric]]]
            weight_to_remove = 0

            if force_flush:
                flushable_buckets = self.buckets.items()
                self.buckets = {}
                self._buckets_total_weight = 0
                self._force_flush = False
            else:
                flushable_buckets = []
                for buckets_timestamp, buckets in iteritems(self.buckets):
                    # If the timestamp of the bucket is newer that the rollup we want to skip it.
                    if buckets_timestamp <= cutoff:
                        flushable_buckets.append((buckets_timestamp, buckets))

                # We will clear the elements while holding the lock, in order to avoid requesting it downstream again.
                for buckets_timestamp, buckets in flushable_buckets:
                    for _, metric in iteritems(buckets):
                        weight_to_remove += metric.weight
                    del self.buckets[buckets_timestamp]

                self._buckets_total_weight -= weight_to_remove

        return flushable_buckets

    def _flushable_locations(self):
        # type: (...) -> Dict[int, List[Tuple[MetricMetaKey, Dict[str, Any]]]]
        with self._lock:
            locations = self._pending_locations
            self._pending_locations = {}
        return locations

    @metrics_noop
    def add(
        self,
        ty,  # type: MetricType
        key,  # type: str
        value,  # type: MetricValue
        unit,  # type: MeasurementUnit
        tags,  # type: Optional[MetricTags]
        timestamp=None,  # type: Optional[Union[float, datetime]]
        local_aggregator=None,  # type: Optional[LocalAggregator]
        stacklevel=0,  # type: Optional[int]
    ):
        # type: (...) -> None
        if not self._ensure_thread() or self._flusher is None:
            return None

        if timestamp is None:
            timestamp = time.time()
        elif isinstance(timestamp, datetime):
            timestamp = to_timestamp(timestamp)

        bucket_timestamp = int(
            (timestamp // self.ROLLUP_IN_SECONDS) * self.ROLLUP_IN_SECONDS
        )
        serialized_tags = _serialize_tags(tags)
        bucket_key = (
            ty,
            key,
            unit,
            serialized_tags,
        )

        with self._lock:
            local_buckets = self.buckets.setdefault(bucket_timestamp, {})
            metric = local_buckets.get(bucket_key)
            if metric is not None:
                previous_weight = metric.weight
                metric.add(value)
            else:
                metric = local_buckets[bucket_key] = METRIC_TYPES[ty](value)
                previous_weight = 0

            added = metric.weight - previous_weight

            if stacklevel is not None:
                self.record_code_location(ty, key, unit, stacklevel + 2, timestamp)

        # Given the new weight we consider whether we want to force flush.
        self._consider_force_flush()

        if local_aggregator is not None:
            local_value = float(added if ty == "s" else value)
            local_aggregator.add(ty, key, local_value, unit, serialized_tags)

    def record_code_location(
        self,
        ty,  # type: MetricType
        key,  # type: str
        unit,  # type: MeasurementUnit
        stacklevel,  # type: int
        timestamp=None,  # type: Optional[float]
    ):
        # type: (...) -> None
        if not self._enable_code_locations:
            return
        if timestamp is None:
            timestamp = time.time()
        meta_key = (ty, key, unit)
        start_of_day = utc_from_timestamp(timestamp).replace(
            hour=0, minute=0, second=0, microsecond=0, tzinfo=None
        )
        start_of_day = int(to_timestamp(start_of_day))

        if (start_of_day, meta_key) not in self._seen_locations:
            self._seen_locations.add((start_of_day, meta_key))
            loc = get_code_location(stacklevel + 3)
            if loc is not None:
                # Group metadata by day to make flushing more efficient.
                # There needs to be one envelope item per timestamp.
                self._pending_locations.setdefault(start_of_day, []).append(
                    (meta_key, loc)
                )

    @metrics_noop
    def need_code_loation(
        self,
        ty,  # type: MetricType
        key,  # type: str
        unit,  # type: MeasurementUnit
        timestamp,  # type: float
    ):
        # type: (...) -> bool
        if self._enable_code_locations:
            return False
        meta_key = (ty, key, unit)
        start_of_day = utc_from_timestamp(timestamp).replace(
            hour=0, minute=0, second=0, microsecond=0, tzinfo=None
        )
        start_of_day = int(to_timestamp(start_of_day))
        return (start_of_day, meta_key) not in self._seen_locations

    def kill(self):
        # type: (...) -> None
        if self._flusher is None:
            return

        self._running = False
        self._flush_event.set()
        self._flusher = None

    @metrics_noop
    def flush(self):
        # type: (...) -> None
        self._force_flush = True
        self._flush()

    def _consider_force_flush(self):
        # type: (...) -> None
        # It's important to acquire a lock around this method, since it will touch shared data structures.
        total_weight = len(self.buckets) + self._buckets_total_weight
        if total_weight >= self.MAX_WEIGHT:
            self._force_flush = True
            self._flush_event.set()

    def _emit(
        self,
        flushable_buckets,  # type: (Iterable[Tuple[int, Dict[BucketKey, Metric]]])
        code_locations,  # type: Dict[int, List[Tuple[MetricMetaKey, Dict[str, Any]]]]
    ):
        # type: (...) -> Optional[Envelope]
        envelope = Envelope()

        if flushable_buckets:
            encoded_metrics = _encode_metrics(flushable_buckets)
            envelope.add_item(Item(payload=encoded_metrics, type="statsd"))

        for timestamp, locations in iteritems(code_locations):
            encoded_locations = _encode_locations(timestamp, locations)
            envelope.add_item(Item(payload=encoded_locations, type="metric_meta"))

        if envelope.items:
            self._capture_func(envelope)
            return envelope
        return None


def _serialize_tags(
    tags,  # type: Optional[MetricTags]
):
    # type: (...) -> MetricTagsInternal
    if not tags:
        return ()

    rv = []
    for key, value in iteritems(tags):
        # If the value is a collection, we want to flatten it.
        if isinstance(value, (list, tuple)):
            for inner_value in value:
                if inner_value is not None:
                    rv.append((key, text_type(inner_value)))
        elif value is not None:
            rv.append((key, text_type(value)))

    # It's very important to sort the tags in order to obtain the
    # same bucket key.
    return tuple(sorted(rv))


def _tags_to_dict(tags):
    # type: (MetricTagsInternal) -> Dict[str, Any]
    rv = {}  # type: Dict[str, Any]
    for tag_name, tag_value in tags:
        old_value = rv.get(tag_name)
        if old_value is not None:
            if isinstance(old_value, list):
                old_value.append(tag_value)
            else:
                rv[tag_name] = [old_value, tag_value]
        else:
            rv[tag_name] = tag_value
    return rv


def _get_aggregator():
    # type: () -> Optional[MetricsAggregator]
    hub = sentry_sdk.Hub.current
    client = hub.client
    return (
        client.metrics_aggregator
        if client is not None and client.metrics_aggregator is not None
        else None
    )


def _get_aggregator_and_update_tags(key, tags):
    # type: (str, Optional[MetricTags]) -> Tuple[Optional[MetricsAggregator], Optional[LocalAggregator], Optional[MetricTags]]
    hub = sentry_sdk.Hub.current
    client = hub.client
    if client is None or client.metrics_aggregator is None:
        return None, None, tags

    experiments = client.options.get("_experiments", {})

    updated_tags = dict(tags or ())  # type: Dict[str, MetricTagValue]
    updated_tags.setdefault("release", client.options["release"])
    updated_tags.setdefault("environment", client.options["environment"])

    scope = hub.scope
    local_aggregator = None

    # We go with the low-level API here to access transaction information as
    # this one is the same between just errors and errors + performance
    transaction_source = scope._transaction_info.get("source")
    if transaction_source in GOOD_TRANSACTION_SOURCES:
        transaction_name = scope._transaction
        if transaction_name:
            updated_tags.setdefault("transaction", transaction_name)
        if scope._span is not None:
            sample_rate = experiments.get("metrics_summary_sample_rate")
            # We default the sample rate of metrics summaries to 1.0 only when the sample rate is `None` since we
            # want to honor the user's decision if they pass a valid float.
            if sample_rate is None:
                sample_rate = 1.0
            should_summarize_metric_callback = experiments.get(
                "should_summarize_metric"
            )
            if random.random() < sample_rate and (
                should_summarize_metric_callback is None
                or should_summarize_metric_callback(key, updated_tags)
            ):
                local_aggregator = scope._span._get_local_aggregator()

    before_emit_callback = experiments.get("before_emit_metric")
    if before_emit_callback is not None:
        with recursion_protection() as in_metrics:
            if not in_metrics:
                if not before_emit_callback(key, updated_tags):
                    return None, None, updated_tags

    return client.metrics_aggregator, local_aggregator, updated_tags


def incr(
    key,  # type: str
    value=1.0,  # type: float
    unit="none",  # type: MeasurementUnit
    tags=None,  # type: Optional[MetricTags]
    timestamp=None,  # type: Optional[Union[float, datetime]]
    stacklevel=0,  # type: int
):
    # type: (...) -> None
    """Increments a counter."""
    aggregator, local_aggregator, tags = _get_aggregator_and_update_tags(key, tags)
    if aggregator is not None:
        aggregator.add(
            "c", key, value, unit, tags, timestamp, local_aggregator, stacklevel
        )


class _Timing(object):
    def __init__(
        self,
        key,  # type: str
        tags,  # type: Optional[MetricTags]
        timestamp,  # type: Optional[Union[float, datetime]]
        value,  # type: Optional[float]
        unit,  # type: DurationUnit
        stacklevel,  # type: int
    ):
        # type: (...) -> None
        self.key = key
        self.tags = tags
        self.timestamp = timestamp
        self.value = value
        self.unit = unit
        self.entered = None  # type: Optional[float]
        self._span = None  # type: Optional[sentry_sdk.tracing.Span]
        self.stacklevel = stacklevel

    def _validate_invocation(self, context):
        # type: (str) -> None
        if self.value is not None:
            raise TypeError(
                "cannot use timing as %s when a value is provided" % context
            )

    def __enter__(self):
        # type: (...) -> _Timing
        self.entered = TIMING_FUNCTIONS[self.unit]()
        self._validate_invocation("context-manager")
        self._span = sentry_sdk.start_span(op="metric.timing", description=self.key)
        if self.tags:
            for key, value in self.tags.items():
                if isinstance(value, (tuple, list)):
                    value = ",".join(sorted(map(str, value)))
                self._span.set_tag(key, value)
        self._span.__enter__()

        # report code locations here for better accuracy
        aggregator = _get_aggregator()
        if aggregator is not None:
            aggregator.record_code_location("d", self.key, self.unit, self.stacklevel)

        return self

    def __exit__(self, exc_type, exc_value, tb):
        # type: (Any, Any, Any) -> None
        assert self._span, "did not enter"
        aggregator, local_aggregator, tags = _get_aggregator_and_update_tags(
            self.key, self.tags
        )
        if aggregator is not None:
            elapsed = TIMING_FUNCTIONS[self.unit]() - self.entered  # type: ignore
            aggregator.add(
                "d",
                self.key,
                elapsed,
                self.unit,
                tags,
                self.timestamp,
                local_aggregator,
                None,  # code locations are reported in __enter__
            )

        self._span.__exit__(exc_type, exc_value, tb)
        self._span = None

    def __call__(self, f):
        # type: (Any) -> Any
        self._validate_invocation("decorator")

        @wraps(f)
        def timed_func(*args, **kwargs):
            # type: (*Any, **Any) -> Any
            with timing(
                key=self.key,
                tags=self.tags,
                timestamp=self.timestamp,
                unit=self.unit,
                stacklevel=self.stacklevel + 1,
            ):
                return f(*args, **kwargs)

        return timed_func


def timing(
    key,  # type: str
    value=None,  # type: Optional[float]
    unit="second",  # type: DurationUnit
    tags=None,  # type: Optional[MetricTags]
    timestamp=None,  # type: Optional[Union[float, datetime]]
    stacklevel=0,  # type: int
):
    # type: (...) -> _Timing
    """Emits a distribution with the time it takes to run the given code block.

    This method supports three forms of invocation:

    - when a `value` is provided, it functions similar to `distribution` but with
    - it can be used as a context manager
    - it can be used as a decorator
    """
    if value is not None:
        aggregator, local_aggregator, tags = _get_aggregator_and_update_tags(key, tags)
        if aggregator is not None:
            aggregator.add(
                "d", key, value, unit, tags, timestamp, local_aggregator, stacklevel
            )
    return _Timing(key, tags, timestamp, value, unit, stacklevel)


def distribution(
    key,  # type: str
    value,  # type: float
    unit="none",  # type: MeasurementUnit
    tags=None,  # type: Optional[MetricTags]
    timestamp=None,  # type: Optional[Union[float, datetime]]
    stacklevel=0,  # type: int
):
    # type: (...) -> None
    """Emits a distribution."""
    aggregator, local_aggregator, tags = _get_aggregator_and_update_tags(key, tags)
    if aggregator is not None:
        aggregator.add(
            "d", key, value, unit, tags, timestamp, local_aggregator, stacklevel
        )


def set(
    key,  # type: str
    value,  # type: MetricValue
    unit="none",  # type: MeasurementUnit
    tags=None,  # type: Optional[MetricTags]
    timestamp=None,  # type: Optional[Union[float, datetime]]
    stacklevel=0,  # type: int
):
    # type: (...) -> None
    """Emits a set."""
    aggregator, local_aggregator, tags = _get_aggregator_and_update_tags(key, tags)
    if aggregator is not None:
        aggregator.add(
            "s", key, value, unit, tags, timestamp, local_aggregator, stacklevel
        )


def gauge(
    key,  # type: str
    value,  # type: float
    unit="none",  # type: MeasurementUnit
    tags=None,  # type: Optional[MetricTags]
    timestamp=None,  # type: Optional[Union[float, datetime]]
    stacklevel=0,  # type: int
):
    # type: (...) -> None
    """Emits a gauge."""
    aggregator, local_aggregator, tags = _get_aggregator_and_update_tags(key, tags)
    if aggregator is not None:
        aggregator.add(
            "g", key, value, unit, tags, timestamp, local_aggregator, stacklevel
        )