ref(metrics): Refactor minimetrics (#55768)

src/minimetrics/__init__.py

@@ -0,0 +1,3 @@
+from .core import MetricTagsExternal, MiniMetricsClient
+
+__all__ = ["MiniMetricsClient", "MetricTagsExternal"]

src/minimetrics/core.py

@@ -0,0 +1,446 @@
+import threading
+import time
+import zlib
+from threading import Event, Lock, Thread
+from typing import (
+    Any,
+    Callable,
+    Dict,
+    Generic,
+    List,
+    Literal,
+    Mapping,
+    NamedTuple,
+    Optional,
+    Set,
+    Tuple,
+    TypedDict,
+    TypeVar,
+    Union,
+)
+
+from typing_extensions import NotRequired
+
+from sentry.utils import metrics
+
+# The thread local instance must be initialized globally in order to correctly use the state.
+thread_local = threading.local()
+
+
+T = TypeVar("T")
+
+# Unit of the metrics.
+MetricUnit = Literal[
+    None,
+    "nanosecond",
+    "microsecond",
+    "millisecond",
+    "second",
+    "minute",
+    "hour",
+    "day",
+    "week",
+    "bit",
+    "byte",
+    "kilobyte",
+    "kibibyte",
+    "mebibyte",
+    "gigabyte",
+    "terabyte",
+    "tebibyte",
+    "petabyte",
+    "pebibyte",
+    "exabyte",
+    "exbibyte",
+    "ratio",
+    "percent",
+]
+# Type of the metric.
+MetricType = Literal["d", "s", "g", "c"]
+# Value of the metric.
+MetricValue = Union[int, float, str]
+# Tag key of a metric.
+MetricTagKey = str
+
+# Internal representation of tags as a tuple of tuples (this is done in order to allow for the same key to exist
+# multiple times).
+MetricTagValueInternal = str
+MetricTagsInternal = Tuple[Tuple[MetricTagKey, MetricTagValueInternal], ...]
+
+# External representation of tags as a dictionary.
+MetricTagValueExternal = Union[str, List[str], Tuple[str, ...]]
+MetricTagsExternal = Mapping[MetricTagKey, MetricTagValueExternal]
+
+# Value of a metric that was extracted after bucketing.
+ExtractedMetricValue = Union[int, float, List[Union[int, float]]]
+
+
+class ExtractedMetric(TypedDict):
+    """
+    Metric extracted from a bucket.
+    """
+
+    type: MetricType
+    name: str
+    value: ExtractedMetricValue
+    timestamp: int
+    width: int
+    unit: NotRequired[MetricUnit]
+    tags: NotRequired[MetricTagsInternal]
+
+
+class BucketKey(NamedTuple):
+    """
+    Key of the bucket.
+    """
+
+    timestamp: int
+    metric_type: MetricType
+    metric_key: str
+    metric_unit: MetricUnit
+    metric_tags: MetricTagsInternal
+
+
+class Metric(Generic[T]):
+    @property
+    def weight(self) -> int:
+        return 1
+
+    def add(self, value: T) -> None:
+        raise NotImplementedError()
+
+    def serialize_value(self) -> Any:
+        raise NotImplementedError()
+
+
+class CounterMetric(Metric[float]):
+    __slots__ = ("value",)
+
+    def __init__(self, first: float) -> None:
+        self.value = first
+
+    def add(self, value: float) -> None:
+        self.value += value
+
+    def serialize_value(self) -> Any:
+        return self.value
+
+
+class GaugeMetric(Metric[float]):
+    __slots__ = (
+        "last",
+        "min",
+        "max",
+        "sum",
+        "count",
+    )
+
+    def __init__(self, first: float) -> None:
+        self.last = first
+        self.min = first
+        self.max = first
+        self.sum = first
+        self.count = 1
+
+    @property
+    def weight(self) -> int:
+        # Number of elements.
+        return 5
+
+    def add(self, value: float) -> None:
+        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) -> Any:
+        return {
+            "last": self.last,
+            "min": self.min,
+            "max": self.max,
+            "sum": self.sum,
+            "count": self.count,
+        }
+
+
+class DistributionMetric(Metric[float]):
+    __slots__ = ("value",)
+
+    def __init__(self, first: float) -> None:
+        self.value: List[float] = [first]
+
+    @property
+    def weight(self) -> int:
+        return len(self.value)
+
+    def add(self, value: float) -> None:
+        self.value.append(float(value))
+
+    def serialize_value(self) -> Any:
+        return self.value
+
+
+class SetMetric(Metric[Union[str, int]]):
+    __slots__ = ("value",)
+
+    def __init__(self, first: Union[str, int]) -> None:
+        self.value: Set[Union[str, int]] = {first}
+
+    @property
+    def weight(self) -> int:
+        return len(self.value)
+
+    def add(self, value: Union[str, int]) -> None:
+        self.value.add(value)
+
+    def serialize_value(self) -> Any:
+        def _hash(x: Any) -> int:
+            if isinstance(x, str):
+                return zlib.crc32(x.encode("utf-8")) & 0xFFFFFFFF
+            return int(x)
+
+        return [_hash(x) for x in self.value]
+
+
+METRIC_TYPES: Dict[str, Callable[[Any], Metric[Any]]] = {
+    "c": CounterMetric,
+    "g": GaugeMetric,
+    "d": DistributionMetric,
+    "s": SetMetric,
+}
+
+
+class Aggregator:
+    ROLLUP_IN_SECONDS = 10.0
+    MAX_WEIGHT = 100000
+
+    def __init__(self) -> None:
+        self.buckets: Dict[BucketKey, Metric[Any]] = {}
+        # Stores the total weight of the in-memory buckets. Weight is determined on a per metric type basis and
+        # represents how much weight is there to represent the metric (e.g., counter = 1, distribution = n).
+        self._buckets_total_weight: int = 0
+        # Lock protecting concurrent access to variables by the flusher and the calling threads that call add or stop.
+        self._lock: Lock = Lock()
+        # Signals whether the loop of the flusher is running.
+        self._running: bool = True
+        # Used to maintain synchronization between the flusher and external callers.
+        self._flush_event: Event = Event()
+        # Use to signal whether we want to flush the buckets in the next loop iteration, irrespectively of the cutoff.
+        self._force_flush: bool = False
+        # Thread handling the flushing loop.
+        self._flusher: Optional[Thread] = Thread(target=self._flush_loop)
+        self._flusher.daemon = True
+        self._flusher.start()
+
+    def _flush_loop(self) -> None:
+        while self._running or self._force_flush:
+            self._flush()
+            self._flush_event.wait(2.0)
+
+    def _flush(self):
+        with self._lock:
+            cutoff = time.time() - self.ROLLUP_IN_SECONDS
+            weight_to_remove = 0
+            buckets = self.buckets
+            force_flush = self._force_flush
+            flushed_buckets = set()
+            extracted_metrics = []
+
+            for bucket_key, metric in buckets.items():
+                if not force_flush and bucket_key.timestamp > cutoff:
+                    continue
+
+                extracted_metrics.append((bucket_key, metric))
+                flushed_buckets.add(bucket_key)
+                weight_to_remove += metric.weight
+
+            # We remove all flushed buckets, in order to avoid memory leaks.
+            for bucket_key in flushed_buckets:
+                buckets.pop(bucket_key)
+
+            self._force_flush = False
+            self._buckets_total_weight -= weight_to_remove
+
+        if extracted_metrics:
+            # You should emit metrics to `metrics` only inside this method, since we know that if we received metrics
+            # the `sentry.utils.metrics` file was initialized. If we do it before, it will likely cause a circular
+            # dependency since the methods in the `sentry.utils.metrics` depend on the backend initialization, thus
+            # if you emit metrics when a backend is initialized Python will throw an error.
+            self._emit(extracted_metrics, force_flush)
+
+    def add(
+        self,
+        ty: MetricType,
+        key: str,
+        value: MetricValue,
+        unit: MetricUnit,
+        tags: Optional[MetricTagsExternal],
+        timestamp: Optional[float],
+    ) -> None:
+        if self._flusher is None:
+            return
+
+        if timestamp is None:
+            timestamp = time.time()
+
+        bucket_key = BucketKey(
+            timestamp=int((timestamp // self.ROLLUP_IN_SECONDS) * self.ROLLUP_IN_SECONDS),
+            metric_type=ty,
+            metric_key=key,
+            metric_unit=unit,
+            # We have to convert tags into our own internal format, since we don't support lists as
+            # tag values.
+            metric_tags=self._to_internal_metric_tags(tags),
+        )
+
+        with self._lock:
+            metric = self.buckets.get(bucket_key)
+            if metric is not None:
+                metric.add(value)
+            else:
+                metric = self.buckets[bucket_key] = METRIC_TYPES[ty](value)
+
+            # We first change the weight by taking the old one and the new one.
+            previous_weight = metric.weight
+            self._buckets_total_weight += metric.weight - previous_weight
+            # Given the new weight we consider whether we want to force flush.
+            self.consider_force_flush()
+
+    def stop(self):
+        if self._flusher is None:
+            return
+
+        # Firstly we tell the flusher that we want to force flush.
+        with self._lock:
+            self._force_flush = True
+            self._running = False
+
+        # Secondly we notify the flusher to move on and we wait for its completion.
+        self._flush_event.set()
+        self._flusher.join()
+        self._flusher = None
+
+    def consider_force_flush(self):
+        # 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()
+
+    @classmethod
+    def _to_internal_metric_tags(cls, tags: Optional[MetricTagsExternal]) -> MetricTagsInternal:
+        rv = []
+        for key, value in (tags or {}).items():
+            # If the value is a collection, we want to flatten it.
+            if isinstance(value, (list, tuple)):
+                for inner_value in value:
+                    rv.append((key, inner_value))
+            else:
+                rv.append((key, value))
+
+        # It's very important to sort the tags in order to obtain the same bucket key.
+        return tuple(sorted(rv))
+
+    @classmethod
+    def _emit(cls, extracted_metrics: List[Tuple[BucketKey, Metric]], force_flush: bool) -> Any:
+        # We obtain the counts for each metric type of how many buckets we have and how much weight is in each
+        # bucket.
+        stats_by_type: Dict[MetricType, Tuple[int, int]] = {}
+
+        for bucket_key, metric in extracted_metrics:
+            (prev_buckets_count, prev_buckets_weight) = stats_by_type.get(
+                bucket_key.metric_type, (0, 0)
+            )
+            stats_by_type[bucket_key.metric_type] = (
+                prev_buckets_count + 1,
+                prev_buckets_weight + metric.weight,
+            )
+
+        for metric_type, (buckets_count, buckets_weight) in stats_by_type.items():
+            # We want to emit a metric on how many buckets and weight there was for a metric type.
+            cls._safe_emit_distribution_metric(
+                key="minimetrics.flushed_buckets",
+                value=buckets_count,
+                tags={"metric_type": metric_type, "force_flush": force_flush},
+            )
+            cls._safe_emit_distribution_metric(
+                key="minimetrics.flushed_buckets_weight",
+                value=buckets_weight,
+                tags={"metric_type": metric_type, "force_flush": force_flush},
+            )
+
+    @classmethod
+    def _safe_emit_count_metric(cls, key: str, amount: int, tags: Optional[Dict[str, Any]] = None):
+        cls._safe_run(lambda: metrics.incr(key, amount=amount, tags=tags))
+
+    @classmethod
+    def _safe_emit_distribution_metric(
+        cls, key: str, value: int, tags: Optional[Dict[str, Any]] = None
+    ):
+        cls._safe_run(lambda: metrics.timing(key, value=value, tags=tags))
+
+    @classmethod
+    def _safe_run(cls, block: Callable[[], None]):
+        # In order to avoid an infinite recursion for metrics, we want to use a thread local variable that will
+        # signal the downstream calls to only propagate the metric to the primary backend, otherwise if propagated to
+        # minimetrics, it will cause unbounded recursion.
+        thread_local.in_minimetrics = True
+        block()
+        thread_local.in_minimetrics = False
+
+
+class MiniMetricsClient:
+    def __init__(self) -> None:
+        self.aggregator = Aggregator()
+
+    @staticmethod
+    def _is_in_minimetrics():
+        try:
+            return thread_local.in_minimetrics
+        except AttributeError:
+            return False
+
+    def incr(
+        self,
+        key: str,
+        value: float,
+        unit: MetricUnit = "nanosecond",
+        tags: Optional[MetricTagsExternal] = None,
+        timestamp: Optional[float] = None,
+    ) -> None:
+        if not self._is_in_minimetrics():
+            self.aggregator.add("c", key, value, unit, tags, timestamp)
+
+    def timing(
+        self,
+        key: str,
+        value: float,
+        unit: MetricUnit = "second",
+        tags: Optional[MetricTagsExternal] = None,
+        timestamp: Optional[float] = None,
+    ) -> None:
+        if not self._is_in_minimetrics():
+            self.aggregator.add("d", key, value, unit, tags, timestamp)
+
+    def set(
+        self,
+        key: str,
+        value: Union[str, int],
+        unit: MetricUnit = None,
+        tags: Optional[MetricTagsExternal] = None,
+        timestamp: Optional[float] = None,
+    ) -> None:
+        if not self._is_in_minimetrics():
+            self.aggregator.add("s", key, value, unit, tags, timestamp)
+
+    def gauge(
+        self,
+        key: str,
+        value: float,
+        unit: MetricUnit = "second",
+        tags: Optional[MetricTagsExternal] = None,
+        timestamp: Optional[float] = None,
+    ) -> None:
+        if not self._is_in_minimetrics():
+            self.aggregator.add("g", key, value, unit, tags, timestamp)

src/sentry/metrics/minimetrics.py

@@ -1,437 +1,27 @@
 import random
-import threading
-import time
-import zlib
-from threading import Event, Lock, Thread
-from typing import (
-    Any,
-    Callable,
-    Dict,
-    Generic,
-    List,
-    Literal,
-    Mapping,
-    Optional,
-    Set,
-    Tuple,
-    TypeVar,
-    Union,
-)
+from typing import Optional, Union, cast
 
 import sentry_sdk
 
+from minimetrics import MetricTagsExternal, MiniMetricsClient
 from sentry.metrics.base import MetricsBackend, Tags
-from sentry.utils import metrics
 
-__all__ = ["MiniMetricsMetricsBackend"]
 
-# The thread local instance must be initialized globally in order to correctly use the state.
-thread_local = threading.local()
-
-
-T = TypeVar("T")
-
-
-MetricUnit = Literal[
-    None,
-    "nanosecond",
-    "microsecond",
-    "millisecond",
-    "second",
-    "minute",
-    "hour",
-    "day",
-    "week",
-    "bit",
-    "byte",
-    "kilobyte",
-    "kibibyte",
-    "mebibyte",
-    "gigabyte",
-    "terabyte",
-    "tebibyte",
-    "petabyte",
-    "pebibyte",
-    "exabyte",
-    "exbibyte",
-    "ratio",
-    "percent",
-]
-
-
-def _flatten_tags(tags: Optional[Mapping[str, Any]]) -> Tuple[Tuple[str, str], ...]:
-    rv = []
-    for key, value in (tags or {}).items():
-        if isinstance(value, (list, tuple)):
-            for inner_value in value:
-                rv.append((key, inner_value))
-        else:
-            rv.append((key, value))
-
-    return tuple(sorted(rv))
-
-
-class Metric(Generic[T]):
-    @property
-    def current_complexity(self) -> int:
-        return 1
-
-    def add(self, value: T) -> None:
-        raise NotImplementedError()
-
-    def serialize_value(self) -> Any:
-        raise NotImplementedError()
-
-
-class CounterMetric(Metric[float]):
-    __slots__ = ("value",)
-
-    def __init__(self) -> None:
-        self.value = 0.0
-
-    def add(self, value: float) -> None:
-        self.value += value
-
-    def serialize_value(self) -> Any:
-        return self.value
-
-
-class GaugeMetric(Metric[float]):
-    __slots__ = ("min", "max", "sum", "count", "last")
-
-    def __init__(self) -> None:
-        self.min = float("inf")
-        self.max = float("-inf")
-        self.sum = 0.0
-        self.count = 0.0
-        self.last = float("nan")
-
-    def add(self, value: float) -> None:
-        self.min = min(self.min, value)
-        self.max = max(self.max, value)
-        self.last = value
-        self.count += 1
-        self.sum += value
-
-    def serialize_value(self) -> Any:
-        return {
-            "min": self.min,
-            "max": self.max,
-            "last": self.last,
-            "sum": self.sum,
-            "count": self.count,
+def _to_minimetrics_external_metric_tags(tags: Optional[Tags]) -> Optional[MetricTagsExternal]:
+    # We remove all `None` values, since then the types will be compatible.
+    casted_tags = None
+    if tags is not None:
+        casted_tags = {
+            tag_key: str(tag_value) for tag_key, tag_value in tags.items() if tag_value is not None
         }
 
+    return cast(Optional[MetricTagsExternal], casted_tags)
 
-class DistributionMetric(Metric[float]):
-    __slots__ = ("value",)
-
-    def __init__(self) -> None:
-        self.value: List[float] = []
-
-    @property
-    def current_complexity(self) -> int:
-        return len(self.value)
-
-    def add(self, value: float) -> None:
-        self.value.append(value)
-
-    def serialize_value(self) -> Any:
-        return self.value
-
-
-class SetMetric(Metric[Union[str, int]]):
-    __slots__ = ("value",)
-
-    def __init__(self) -> None:
-        self.value: Set[Union[str, int]] = set()
-
-    @property
-    def current_complexity(self) -> int:
-        return len(self.value)
-
-    def add(self, value: Union[str, int]) -> None:
-        self.value.add(value)
-
-    def serialize_value(self) -> Any:
-        def _hash(x: Any) -> int:
-            if isinstance(x, str):
-                return zlib.crc32(x.encode("utf-8")) & 0xFFFFFFFF
-            return int(x)
-
-        return [_hash(x) for x in self.value]
-
-
-METRIC_TYPES: Dict[str, Callable[[], Metric[Any]]] = {
-    "c": CounterMetric,
-    "g": GaugeMetric,
-    "d": DistributionMetric,
-    "s": SetMetric,
-}
-
-ComposedKey = Tuple[int, str, str, MetricUnit, Tuple[Tuple[str, str], ...]]
-
-
-class Aggregator:
-    ROLLUP_IN_SECONDS = 10.0
-    MAX_COMPLEXITY = 100000
-
-    def __init__(self) -> None:
-        self.buckets: Dict[ComposedKey, Metric[Any]] = {}
-        self._bucket_complexity: int = 0
-        self._lock: Lock = Lock()
-        self._running: bool = True
-        self._flush_event: Event = Event()
-        self._force_flush: bool = False
-        # Thread handling the flushing loop.
-        self._flusher: Optional[Thread] = Thread(target=self._flush_loop)
-        self._flusher.daemon = True
-        self._flusher.start()
-
-    def _flush_loop(self) -> None:
-        # We check without locking these variables, such racy check can lead to problems if we are not careful. The most
-        # important invariant of the system that needs to be maintained is that if running and force_flush are false,
-        # the number of buckets is equal to 0.
-        while self._running or self._force_flush:
-            self._flush()
-            self._flush_event.wait(2.0)
-
-    def _flush(self):
-        with self._lock:
-            cutoff = time.time() - self.ROLLUP_IN_SECONDS
-            complexity_to_remove = 0
-            buckets = self.buckets
-            force_flush = self._force_flush
-            flushed_buckets = set()
-            extracted_metrics = []
-
-            for bucket_key, metric in buckets.items():
-                ts, ty, name, unit, tags = bucket_key
-                if not force_flush and ts > cutoff:
-                    continue
-
-                extracted_metric = {
-                    "timestamp": ts,
-                    "width": int(self.ROLLUP_IN_SECONDS),
-                    "name": name,
-                    "type": ty,
-                    "value": metric.serialize_value(),
-                }
-                if unit:
-                    extracted_metric["unit"] = unit
-                if tags:
-                    # We need to be careful here, since we have a list of tuples where the first element of tuples
-                    # can be duplicated, thus converting to a dict will end up compressing and losing data.
-                    extracted_metric["tags"] = tags
-
-                extracted_metrics.append((extracted_metric, metric.current_complexity))
-                flushed_buckets.add(bucket_key)
-                complexity_to_remove += metric.current_complexity
-
-            # We remove all flushed buckets, in order to avoid memory leaks.
-            for bucket_key in flushed_buckets:
-                buckets.pop(bucket_key)
-
-            self._force_flush = False
-            self._bucket_complexity -= complexity_to_remove
-
-        if extracted_metrics:
-            self._emit(extracted_metrics, force_flush)
-
-    def add(
-        self,
-        ty: str,
-        key: str,
-        value: Any,
-        unit: MetricUnit,
-        tags: Optional[Tags],
-        timestamp: Optional[float],
-    ) -> None:
-        if self._flusher is None:
-            return
-
-        if timestamp is None:
-            timestamp = time.time()
-
-        bucket_key: ComposedKey = (
-            int((timestamp // self.ROLLUP_IN_SECONDS) * self.ROLLUP_IN_SECONDS),
-            ty,
-            key,
-            unit,
-            _flatten_tags(tags),
-        )
-
-        with self._lock:
-            metric = self.buckets.get(bucket_key)
-            if metric is None:
-                metric = METRIC_TYPES[ty]()
-                self.buckets[bucket_key] = metric
 
-            # We first change the complexity by taking the old one and the new one.
-            previous_complexity = metric.current_complexity
-            metric.add(value)
-            self._bucket_complexity += metric.current_complexity - previous_complexity
-            # Given the new complexity we consider whether we want to force flush.
-            self.consider_force_flush()
-
-    def stop(self):
-        if self._flusher is None:
-            return
-
-        # Firstly we tell the flusher that we want to force flush.
-        with self._lock:
-            self._force_flush = True
-            self._running = False
-
-        # Secondly we notify the flusher to move on and we wait for its completion.
-        self._flush_event.set()
-        # Checking also here because of mypy.
-        self._flusher.join()
-        self._flusher = None
-
-    def consider_force_flush(self):
-        total_complexity = len(self.buckets) + self._bucket_complexity
-        if total_complexity >= self.MAX_COMPLEXITY:
-            self._force_flush = True
-            self._flush_event.set()
-
-    @classmethod
-    def _emit(cls, extracted_metrics: List[Tuple[Any, int]], force_flush: bool) -> Any:
-        # We obtain the counts for each metric type of how many buckets we have and how much complexity is in each
-        # bucket.
-        complexities_by_type: Dict[str, Tuple[int, int]] = {}
-        # We obtain the counts for each metric type, since we want to know how many by type we have.
-        counts_by_type: Dict[str, float] = {}
-        for metric, metric_complexity in extracted_metrics:
-            metric_type = metric["type"]
-            metric_value = metric["value"]
-
-            value: float = 0.0
-            if metric_type == "c":
-                # For counters, we want to sum the count value.
-                value = metric_value
-            elif metric_type == "d":
-                # For distributions, we want to track the size of the distribution.
-                value = len(metric_value)
-            elif metric_type == "g":
-                # For gauges, we will emit a count of 1.
-                value = metric_value.get("count", 1)
-            elif metric_type == "s":
-                # For sets, we want to track the cardinality of the set.
-                value = len(metric_value)
-
-            counts_by_type[metric_type] = counts_by_type.get(metric_type, 0) + value
-
-            (prev_buckets_count, prev_buckets_complexity) = complexities_by_type.get(
-                metric_type, (0, 0)
-            )
-            complexities_by_type[metric_type] = (
-                prev_buckets_count + 1,
-                prev_buckets_complexity + metric_complexity,
-            )
-
-        # For each type and count we want to emit a metric.
-        for metric_type, metric_count in counts_by_type.items():
-            # We want to emit a metric on how many metrics we would technically emit if we were to use minimetrics.
-            cls._safe_emit_count_metric(
-                key="minimetrics.emit",
-                amount=int(metric_count),
-                tags={"metric_type": metric_type, "force_flush": force_flush},
-            )
-
-        for metric_type, (buckets_count, buckets_complexity) in complexities_by_type.items():
-            # We want to emit a metric on how many buckets and complexity there was for a metric type.
-            cls._safe_emit_count_metric(
-                key="minimetrics.flushed_buckets_count",
-                amount=buckets_count,
-                tags={"metric_type": metric_type, "force_flush": force_flush},
-            )
-            cls._safe_emit_count_metric(
-                key="minimetrics.flushed_buckets_complexity",
-                amount=buckets_complexity,
-                tags={"metric_type": metric_type, "force_flush": force_flush},
-            )
-
-    @classmethod
-    def _safe_emit_count_metric(cls, key: str, amount: int, tags: Optional[Tags] = None):
-        cls._safe_run(lambda: metrics.incr(key, amount=amount, tags=tags))
-
-    @classmethod
-    def _safe_emit_distribution_metric(cls, key: str, value: int, tags: Optional[Tags] = None):
-        cls._safe_run(lambda: metrics.timing(key, value=value, tags=tags))
-
-    @classmethod
-    def _safe_run(cls, block: Callable[[], None]):
-        # In order to avoid an infinite recursion for metrics, we want to use a thread local variable that will
-        # signal the downstream calls to only propagate the metric to the primary backend, otherwise if propagated to
-        # minimetrics, it will cause unbounded recursion.
-        thread_local.in_minimetrics = True
-        block()
-        thread_local.in_minimetrics = False
-
-
-class Client:
-    def __init__(self) -> None:
-        self.aggregator = Aggregator()
-
-    @staticmethod
-    def _is_in_minimetrics():
-        try:
-            return thread_local.in_minimetrics
-        except AttributeError:
-            return False
-
-    def incr(
-        self,
-        key: str,
-        value: float,
-        unit: MetricUnit = "nanosecond",
-        tags: Optional[Tags] = None,
-        timestamp: Optional[float] = None,
-    ) -> None:
-        if not self._is_in_minimetrics():
-            self.aggregator.add("c", key, value, unit, tags, timestamp)
-
-    def timing(
-        self,
-        key: str,
-        value: float,
-        unit: MetricUnit = "second",
-        tags: Optional[Tags] = None,
-        timestamp: Optional[float] = None,
-    ) -> None:
-        if not self._is_in_minimetrics():
-            self.aggregator.add("d", key, value, unit, tags, timestamp)
-
-    def set(
-        self,
-        key: str,
-        value: Union[str, int],
-        tags: Optional[Tags] = None,
-        timestamp: Optional[float] = None,
-    ) -> None:
-        if not self._is_in_minimetrics():
-            self.aggregator.add("s", key, value, None, tags, timestamp)
-
-    def gauge(
-        self,
-        key: str,
-        value: float,
-        unit: MetricUnit = "second",
-        tags: Optional[Tags] = None,
-        timestamp: Optional[float] = None,
-    ) -> None:
-        if not self._is_in_minimetrics():
-            self.aggregator.add("g", key, value, unit, tags, timestamp)
-
-
-# TODO:
-#   * Check how to use units
-#   * Check usage of instance
 class MiniMetricsMetricsBackend(MetricsBackend):
     def __init__(self, prefix: Optional[str] = None):
         super().__init__(prefix=prefix)
-        self.client = Client()
+        self.client = MiniMetricsClient()
 
     def _patch_sdk(self):
         client = sentry_sdk.Hub.main.client
@@ -465,7 +55,11 @@ class MiniMetricsMetricsBackend(MetricsBackend):
         sample_rate: float = 1,
     ) -> None:
         if self._keep_metric(sample_rate):
-            self.client.incr(key=self._get_key(key), value=amount, tags=tags)
+            self.client.incr(
+                key=self._get_key(key),
+                value=amount,
+                tags=_to_minimetrics_external_metric_tags(tags),
+            )
 
     def timing(
         self,
@@ -476,7 +70,9 @@ class MiniMetricsMetricsBackend(MetricsBackend):
         sample_rate: float = 1,
     ) -> None:
         if self._keep_metric(sample_rate):
-            self.client.timing(key=self._get_key(key), value=value, tags=tags)
+            self.client.timing(
+                key=self._get_key(key), value=value, tags=_to_minimetrics_external_metric_tags(tags)
+            )
 
     def gauge(
         self,
@@ -487,4 +83,6 @@ class MiniMetricsMetricsBackend(MetricsBackend):
         sample_rate: float = 1,
     ) -> None:
         if self._keep_metric(sample_rate):
-            self.client.gauge(key=self._get_key(key), value=value, tags=tags)
+            self.client.gauge(
+                key=self._get_key(key), value=value, tags=_to_minimetrics_external_metric_tags(tags)
+            )

tests/minimetrics/test_core.py

@@ -0,0 +1,167 @@
+from unittest.mock import patch
+
+from freezegun import freeze_time
+
+from minimetrics import MiniMetricsClient
+from minimetrics.core import BucketKey, CounterMetric, DistributionMetric, GaugeMetric, SetMetric
+
+
+def test_simple():
+    client = MiniMetricsClient()
+    client.incr("button_clicked", 1.0)
+    client.aggregator.stop()
+
+    assert len(client.aggregator.buckets) == 0
+
+
+@freeze_time("2023-09-06 10:00:00")
+@patch("minimetrics.core.Aggregator._emit")
+def test_client_incr(_emit):
+    tags = {
+        "browser": "Chrome",
+        "browser.version": "1.0",
+        "user.orgs": ["sentry", "google", "apple"],
+        "user.classes": ["1", "2", "3"],
+    }
+    client = MiniMetricsClient()
+    client.incr("button_clicked", 1.0, tags=tags)  # type:ignore
+    client.aggregator.stop()
+
+    assert len(client.aggregator.buckets) == 0
+    extracted_metrics_arg = _emit.call_args.args[0]
+    assert len(extracted_metrics_arg) == 1
+    assert extracted_metrics_arg[0][0] == BucketKey(
+        timestamp=1693994400,
+        metric_type="c",
+        metric_key="button_clicked",
+        metric_unit="nanosecond",
+        metric_tags=(
+            ("browser", "Chrome"),
+            ("browser.version", "1.0"),
+            ("user.classes", "1"),
+            ("user.classes", "2"),
+            ("user.classes", "3"),
+            ("user.orgs", "apple"),
+            ("user.orgs", "google"),
+            ("user.orgs", "sentry"),
+        ),
+    )
+    assert isinstance(extracted_metrics_arg[0][1], CounterMetric)
+    assert extracted_metrics_arg[0][1].serialize_value() == 1
+
+
+@freeze_time("2023-09-06 10:00:00")
+@patch("minimetrics.core.Aggregator._emit")
+def test_client_timing(_emit):
+    tags = {
+        "browser": "Chrome",
+        "browser.version": "1.0",
+        "user.orgs": ["sentry", "google", "apple"],
+        "user.classes": ["1", "2", "3"],
+    }
+    client = MiniMetricsClient()
+    client.timing("execution_time", 1.0, tags=tags)  # type:ignore
+    client.aggregator.stop()
+
+    assert len(client.aggregator.buckets) == 0
+    extracted_metrics_arg = _emit.call_args.args[0]
+    assert len(extracted_metrics_arg) == 1
+    assert extracted_metrics_arg[0][0] == BucketKey(
+        timestamp=1693994400,
+        metric_type="d",
+        metric_key="execution_time",
+        metric_unit="second",
+        metric_tags=(
+            ("browser", "Chrome"),
+            ("browser.version", "1.0"),
+            ("user.classes", "1"),
+            ("user.classes", "2"),
+            ("user.classes", "3"),
+            ("user.orgs", "apple"),
+            ("user.orgs", "google"),
+            ("user.orgs", "sentry"),
+        ),
+    )
+    assert isinstance(extracted_metrics_arg[0][1], DistributionMetric)
+    assert extracted_metrics_arg[0][1].serialize_value() == [1.0]
+    assert len(client.aggregator.buckets) == 0
+
+
+@freeze_time("2023-09-06 10:00:00")
+@patch("minimetrics.core.Aggregator._emit")
+def test_client_set(_emit):
+    tags = {
+        "browser": "Chrome",
+        "browser.version": "1.0",
+        "user.orgs": ["sentry", "google", "apple"],
+        "user.classes": ["1", "2", "3"],
+    }
+    client = MiniMetricsClient()
+    client.set("user", "riccardo", tags=tags)  # type:ignore
+    client.aggregator.stop()
+
+    assert len(client.aggregator.buckets) == 0
+    extracted_metrics_arg = _emit.call_args.args[0]
+    assert len(extracted_metrics_arg) == 1
+    assert extracted_metrics_arg[0][0] == BucketKey(
+        timestamp=1693994400,
+        metric_type="s",
+        metric_key="user",
+        metric_unit=None,
+        metric_tags=(
+            ("browser", "Chrome"),
+            ("browser.version", "1.0"),
+            ("user.classes", "1"),
+            ("user.classes", "2"),
+            ("user.classes", "3"),
+            ("user.orgs", "apple"),
+            ("user.orgs", "google"),
+            ("user.orgs", "sentry"),
+        ),
+    )
+    assert isinstance(extracted_metrics_arg[0][1], SetMetric)
+    assert extracted_metrics_arg[0][1].serialize_value() == [3455635177]
+    assert len(client.aggregator.buckets) == 0
+
+
+@freeze_time("2023-09-06 10:00:00")
+@patch("minimetrics.core.Aggregator._emit")
+def test_client_gauge(_emit):
+    tags = {
+        "browser": "Chrome",
+        "browser.version": "1.0",
+        "user.orgs": ["sentry", "google", "apple"],
+        "user.classes": ["1", "2", "3"],
+    }
+    client = MiniMetricsClient()
+    client.gauge("frontend_time", 15.0, tags=tags)  # type:ignore
+    client.aggregator.stop()
+
+    assert len(client.aggregator.buckets) == 0
+    extracted_metrics_arg = _emit.call_args.args[0]
+    assert len(extracted_metrics_arg) == 1
+    assert extracted_metrics_arg[0][0] == BucketKey(
+        timestamp=1693994400,
+        metric_type="g",
+        metric_key="frontend_time",
+        metric_unit="second",
+        metric_tags=(
+            ("browser", "Chrome"),
+            ("browser.version", "1.0"),
+            ("user.classes", "1"),
+            ("user.classes", "2"),
+            ("user.classes", "3"),
+            ("user.orgs", "apple"),
+            ("user.orgs", "google"),
+            ("user.orgs", "sentry"),
+        ),
+    )
+    assert isinstance(extracted_metrics_arg[0][1], GaugeMetric)
+    assert extracted_metrics_arg[0][1].serialize_value() == {
+        "last": 15.0,
+        "min": 15.0,
+        "max": 15.0,
+        "sum": 15.0,
+        "count": 1,
+    }
+    assert len(client.aggregator.buckets) == 0

tests/sentry/metrics/test_minimetrics.py

@@ -8,39 +8,39 @@ class MiniMetricsMetricsBackendTest(TestCase):
     def setUp(self):
         self.backend = MiniMetricsMetricsBackend(prefix="sentrytest.")
 
-    @patch("sentry.metrics.minimetrics.Aggregator.ROLLUP_IN_SECONDS", 1.0)
-    @patch("sentry.metrics.minimetrics.metrics.incr")
-    def test_incr_called_with_no_tags(self, metrics_incr):
+    @patch("minimetrics.core.Aggregator.ROLLUP_IN_SECONDS", 1.0)
+    @patch("minimetrics.core.metrics.timing")
+    def test_incr_called_with_no_tags(self, metrics_call):
         self.backend.incr(key="foo")
         self.backend.client.aggregator.stop()
 
         assert len(self.backend.client.aggregator.buckets) == 0
-        assert metrics_incr.call_count == 3
+        assert metrics_call.call_count == 2
 
-    @patch("sentry.metrics.minimetrics.Aggregator.ROLLUP_IN_SECONDS", 1.0)
-    @patch("sentry.metrics.minimetrics.metrics.incr")
-    def test_incr_called_with_tag_value_as_list(self, metrics_incr):
+    @patch("minimetrics.core.Aggregator.ROLLUP_IN_SECONDS", 1.0)
+    @patch("minimetrics.core.metrics.timing")
+    def test_incr_called_with_tag_value_as_list(self, metrics_call):
         # The minimetrics backend supports the list type.
         self.backend.incr(key="foo", tags={"foo": ["bar", "baz"]})  # type:ignore
         self.backend.client.aggregator.stop()
 
         assert len(self.backend.client.aggregator.buckets) == 0
-        assert metrics_incr.call_count == 3
+        assert metrics_call.call_count == 2
 
-    @patch("sentry.metrics.minimetrics.Aggregator.ROLLUP_IN_SECONDS", 1.0)
-    @patch("sentry.metrics.minimetrics.metrics.incr")
-    def test_incr_not_called_after_flusher_stopped(self, metrics_incr):
+    @patch("minimetrics.core.Aggregator.ROLLUP_IN_SECONDS", 1.0)
+    @patch("minimetrics.core.metrics.timing")
+    def test_incr_not_called_after_flusher_stopped(self, metrics_call):
         self.backend.client.aggregator.stop()
         self.backend.incr(key="foo")
 
         assert len(self.backend.client.aggregator.buckets) == 0
-        assert metrics_incr.call_count == 0
+        assert metrics_call.call_count == 0
 
-    @patch("sentry.metrics.minimetrics.Aggregator.ROLLUP_IN_SECONDS", 1.0)
-    @patch("sentry.metrics.minimetrics.metrics.incr")
-    def test_stop_called_twice(self, metrics_incr):
+    @patch("minimetrics.core.Aggregator.ROLLUP_IN_SECONDS", 1.0)
+    @patch("minimetrics.core.metrics.timing")
+    def test_stop_called_twice(self, metrics_call):
         self.backend.client.aggregator.stop()
         self.backend.client.aggregator.stop()
 
         assert len(self.backend.client.aggregator.buckets) == 0
-        assert metrics_incr.call_count == 0
+        assert metrics_call.call_count == 0