749 lines
23 KiB
Python
749 lines
23 KiB
Python
import os
|
|
import io
|
|
import re
|
|
import sys
|
|
import threading
|
|
import random
|
|
import time
|
|
import zlib
|
|
from datetime import datetime
|
|
from functools import wraps, partial
|
|
from threading import Event, Lock, Thread
|
|
from contextlib import contextmanager
|
|
|
|
import sentry_sdk
|
|
from sentry_sdk._compat import text_type, utc_from_timestamp, iteritems
|
|
from sentry_sdk.utils import (
|
|
now,
|
|
nanosecond_time,
|
|
to_timestamp,
|
|
serialize_frame,
|
|
json_dumps,
|
|
)
|
|
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
|
|
|
|
|
|
_thread_local = threading.local()
|
|
_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 + 4)
|
|
except Exception:
|
|
return None
|
|
|
|
return serialize_frame(
|
|
frm, include_local_variables=False, include_source_context=False
|
|
)
|
|
|
|
|
|
@contextmanager
|
|
def recursion_protection():
|
|
# type: () -> Generator[bool, None, None]
|
|
"""Enters recursion protection and returns the old flag."""
|
|
try:
|
|
in_metrics = _thread_local.in_metrics
|
|
except AttributeError:
|
|
in_metrics = False
|
|
_thread_local.in_metrics = True
|
|
try:
|
|
yield in_metrics
|
|
finally:
|
|
_thread_local.in_metrics = 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 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._lock = Lock()
|
|
self._running = True
|
|
self._flush_event = Event()
|
|
self._force_flush = False
|
|
|
|
# The aggregator shifts it's 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[Thread]
|
|
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
|
|
self._flusher = Thread(target=self._flush_loop)
|
|
self._flusher.daemon = True
|
|
try:
|
|
self._flusher.start()
|
|
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
|
|
_thread_local.in_metrics = 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]]
|
|
stacklevel=0, # type: int
|
|
):
|
|
# type: (...) -> None
|
|
if not self._ensure_thread() or self._flusher is None:
|
|
return
|
|
|
|
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
|
|
)
|
|
bucket_key = (
|
|
ty,
|
|
key,
|
|
unit,
|
|
self._serialize_tags(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
|
|
|
|
self._buckets_total_weight += metric.weight - previous_weight
|
|
|
|
# Store code location once per metric and per day (of bucket timestamp)
|
|
if self._enable_code_locations:
|
|
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)
|
|
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)
|
|
)
|
|
|
|
# Given the new weight we consider whether we want to force flush.
|
|
self._consider_force_flush()
|
|
|
|
def kill(self):
|
|
# type: (...) -> None
|
|
if self._flusher is None:
|
|
return
|
|
|
|
self._running = False
|
|
self._flush_event.set()
|
|
self._flusher.join()
|
|
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(
|
|
self, 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 _get_aggregator_and_update_tags(key, tags):
|
|
# type: (str, Optional[MetricTags]) -> Tuple[Optional[MetricsAggregator], Optional[MetricTags]]
|
|
"""Returns the current metrics aggregator if there is one."""
|
|
hub = sentry_sdk.Hub.current
|
|
client = hub.client
|
|
if client is None or client.metrics_aggregator is None:
|
|
return None, tags
|
|
|
|
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
|
|
transaction_source = scope._transaction_info.get("source")
|
|
if transaction_source in GOOD_TRANSACTION_SOURCES:
|
|
transaction = scope._transaction
|
|
if transaction:
|
|
updated_tags.setdefault("transaction", transaction)
|
|
|
|
callback = client.options.get("_experiments", {}).get("before_emit_metric")
|
|
if callback is not None:
|
|
with recursion_protection() as in_metrics:
|
|
if not in_metrics:
|
|
if not callback(key, updated_tags):
|
|
return None, updated_tags
|
|
|
|
return client.metrics_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, tags = _get_aggregator_and_update_tags(key, tags)
|
|
if aggregator is not None:
|
|
aggregator.add("c", key, value, unit, tags, timestamp, 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.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")
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc_value, tb):
|
|
# type: (Any, Any, Any) -> None
|
|
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, self.stacklevel
|
|
)
|
|
|
|
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, tags = _get_aggregator_and_update_tags(key, tags)
|
|
if aggregator is not None:
|
|
aggregator.add("d", key, value, unit, tags, timestamp, 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, tags = _get_aggregator_and_update_tags(key, tags)
|
|
if aggregator is not None:
|
|
aggregator.add("d", key, value, unit, tags, timestamp, 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, tags = _get_aggregator_and_update_tags(key, tags)
|
|
if aggregator is not None:
|
|
aggregator.add("s", key, value, unit, tags, timestamp, stacklevel)
|
|
|
|
|
|
def gauge(
|
|
key, # type: str
|
|
value, # type: float
|
|
unit="none", # type: MetricValue
|
|
tags=None, # type: Optional[MetricTags]
|
|
timestamp=None, # type: Optional[Union[float, datetime]]
|
|
stacklevel=0, # type: int
|
|
):
|
|
# type: (...) -> None
|
|
"""Emits a gauge."""
|
|
aggregator, tags = _get_aggregator_and_update_tags(key, tags)
|
|
if aggregator is not None:
|
|
aggregator.add("g", key, value, unit, tags, timestamp, stacklevel)
|