Add a new DualWriteMetricsBackend (#97754)

This is very similar to the previously removed
`CompositeExperimentalMetricsBackend`, and does what its name implies:

You can configure it with a primary and secondary backend/options, and
an allowlist of prefixes. All metrics that match the allowlist are
routed to the secondary backend in addition to the primary one.

Author: Swatinem

src/sentry/metrics/dualwrite.py

@@ -0,0 +1,130 @@
+from typing import Any
+
+from sentry.metrics.base import MetricsBackend, Tags
+from sentry.metrics.dummy import DummyMetricsBackend
+from sentry.utils.imports import import_string
+
+__all__ = ["DualWriteMetricsBackend"]
+
+
+def _initialize_backend(backend: str | None, backend_args: dict[str, Any]) -> MetricsBackend:
+    if backend is None:
+        return DummyMetricsBackend()
+    else:
+        cls: type[MetricsBackend] = import_string(backend)
+        return cls(**backend_args)
+
+
+class DualWriteMetricsBackend(MetricsBackend):
+    def __init__(self, **kwargs: Any):
+        super().__init__()
+        self._primary_backend = _initialize_backend(
+            kwargs.pop("primary_backend", None), kwargs.pop("primary_backend_args", {})
+        )
+        self._secondary_backend = _initialize_backend(
+            kwargs.pop("secondary_backend", None), kwargs.pop("secondary_backend_args", {})
+        )
+
+        self._allow_prefixes = tuple(kwargs.pop("allow_prefixes", []))
+
+    def _is_allowed(self, key: str) -> bool:
+        return key.startswith(self._allow_prefixes)
+
+    def incr(
+        self,
+        key: str,
+        instance: str | None = None,
+        tags: Tags | None = None,
+        amount: float | int = 1,
+        sample_rate: float = 1,
+        unit: str | None = None,
+        stacklevel: int = 0,
+    ) -> None:
+        self._primary_backend.incr(key, instance, tags, amount, sample_rate, unit, stacklevel + 1)
+        if self._is_allowed(key):
+            self._secondary_backend.incr(
+                key, instance, tags, amount, sample_rate, unit, stacklevel + 1
+            )
+
+    def timing(
+        self,
+        key: str,
+        value: float,
+        instance: str | None = None,
+        tags: Tags | None = None,
+        sample_rate: float = 1,
+        stacklevel: int = 0,
+    ) -> None:
+        self._primary_backend.timing(key, value, instance, tags, sample_rate, stacklevel + 1)
+        if self._is_allowed(key):
+            self._secondary_backend.timing(key, value, instance, tags, sample_rate, stacklevel + 1)
+
+    def gauge(
+        self,
+        key: str,
+        value: float,
+        instance: str | None = None,
+        tags: Tags | None = None,
+        sample_rate: float = 1,
+        unit: str | None = None,
+        stacklevel: int = 0,
+    ) -> None:
+        self._primary_backend.gauge(key, value, instance, tags, sample_rate, unit, stacklevel + 1)
+        if self._is_allowed(key):
+            self._secondary_backend.gauge(
+                key, value, instance, tags, sample_rate, unit, stacklevel + 1
+            )
+
+    def distribution(
+        self,
+        key: str,
+        value: float,
+        instance: str | None = None,
+        tags: Tags | None = None,
+        sample_rate: float = 1,
+        unit: str | None = None,
+        stacklevel: int = 0,
+    ) -> None:
+        self._primary_backend.distribution(
+            key, value, instance, tags, sample_rate, unit, stacklevel + 1
+        )
+        if self._is_allowed(key):
+            self._secondary_backend.distribution(
+                key, value, instance, tags, sample_rate, unit, stacklevel + 1
+            )
+
+    def event(
+        self,
+        title: str,
+        message: str,
+        alert_type: str | None = None,
+        aggregation_key: str | None = None,
+        source_type_name: str | None = None,
+        priority: str | None = None,
+        instance: str | None = None,
+        tags: Tags | None = None,
+        stacklevel: int = 0,
+    ) -> None:
+        self._primary_backend.event(
+            title,
+            message,
+            alert_type,
+            aggregation_key,
+            source_type_name,
+            priority,
+            instance,
+            tags,
+            stacklevel + 1,
+        )
+        if self._is_allowed(title):
+            self._secondary_backend.event(
+                title,
+                message,
+                alert_type,
+                aggregation_key,
+                source_type_name,
+                priority,
+                instance,
+                tags,
+                stacklevel + 1,
+            )

tests/sentry/metrics/test_dualwrite.py

@@ -0,0 +1,33 @@
+from unittest import mock
+
+from sentry.metrics.dualwrite import DualWriteMetricsBackend
+
+
+@mock.patch("datadog.threadstats.base.ThreadStats.timing")
+@mock.patch("datadog.dogstatsd.base.DogStatsd.distribution")
+def test_dualwrite_distribution(distribution, timing):
+    backend = DualWriteMetricsBackend(
+        primary_backend="sentry.metrics.datadog.DatadogMetricsBackend",
+        secondary_backend="sentry.metrics.precise_dogstatsd.PreciseDogStatsdMetricsBackend",
+        allow_prefixes=["foo"],
+    )
+
+    backend.distribution("foo", 100, tags={"some": "stuff"}, unit="byte")
+    # datadog treats distributions as timing
+    timing.assert_called_once()
+    distribution.assert_called_once()
+
+    timing.reset_mock()
+    distribution.reset_mock()
+
+    backend.timing("foo", 100, tags={"some": "stuff"})
+    # precise datadog treats timing as distribution
+    timing.assert_called_once()
+    distribution.assert_called_once()
+
+    timing.reset_mock()
+    distribution.reset_mock()
+
+    backend.timing("bar", 100, tags={"some": "stuff"})
+    timing.assert_called_once()
+    distribution.assert_not_called()