Allow multiple dogstatsd instances, record precise timing distributions (#97662)

This fixes the `DogStatsdMetricsBackend` so that it does not depend on
the globally configured `initialize/statsd` singleton, but rather
configures its own statsd instance.

Also, this is now capturing precise timings as `distribution`, as the
existing `timing` type has special handling within datadog/agent such
that it again splits the metric into multiple pre-aggregations.

Author: Swatinem

src/sentry/metrics/base.py

@@ -52,6 +52,7 @@ def timing(
         tags: Tags | None = None,
         sample_rate: float = 1,
         stacklevel: int = 0,
+        precise: bool = False,
     ) -> None:
         raise NotImplementedError
 

src/sentry/metrics/datadog.py

@@ -64,6 +64,7 @@ def timing(
         tags: Tags | None = None,
         sample_rate: float = 1,
         stacklevel: int = 0,
+        precise: bool = False,
     ) -> None:
         tags = dict(tags or ())
 
@@ -73,9 +74,14 @@ def timing(
             tags["instance"] = instance
 
         tags_list = [f"{k}:{v}" for k, v in tags.items()]
-        self.stats.timing(
-            self._get_key(key), value, sample_rate=sample_rate, tags=tags_list, host=self.host
-        )
+        if not precise:
+            self.stats.timing(
+                self._get_key(key), value, sample_rate=sample_rate, tags=tags_list, host=self.host
+            )
+        else:
+            self.stats.distribution(
+                self._get_key(key), value, sample_rate=sample_rate, tags=tags_list, host=self.host
+            )
 
     def gauge(
         self,

src/sentry/metrics/dogstatsd.py

@@ -1,45 +1,41 @@
 import atexit
 from typing import Any
 
-from datadog import initialize
-from datadog.dogstatsd.base import statsd
+from datadog.dogstatsd.base import DogStatsd
 
 from .base import MetricsBackend, Tags
 
 __all__ = ["DogStatsdMetricsBackend"]
 
-# Set the maximum number of packets to queue for the sender.
-# How may packets to queue before blocking or dropping the packet if the packet queue is already full.
-# 0 means unlimited.
-SENDER_QUEUE_SIZE = 0
-
-# Set timeout for packet queue operations, in seconds
-# How long the application thread is willing to wait for the queue clear up before dropping the metric packet.
-# If set to None, wait forever.
-# If set to zero drop the packet immediately if the queue is full.
-SENDER_QUEUE_TIMEOUT = 0
-
 
 class DogStatsdMetricsBackend(MetricsBackend):
     def __init__(self, prefix: str | None = None, **kwargs: Any) -> None:
-        # TODO(dcramer): it'd be nice if the initialize call wasn't a global
         self.tags = kwargs.pop("tags", None)
-        kwargs["statsd_disable_buffering"] = False
-
-        initialize(**kwargs)
-        statsd.disable_telemetry()
 
-        # When enabled, a background thread will be used to send metric payloads to the Agent.
-        statsd.enable_background_sender(
-            sender_queue_size=SENDER_QUEUE_SIZE, sender_queue_timeout=SENDER_QUEUE_TIMEOUT
-        )
-        # Applications should call wait_for_pending() before exiting to make sure all pending payloads are sent.
-        atexit.register(statsd.wait_for_pending)
+        instance_kwargs: dict[str, Any] = {
+            "disable_telemetry": True,
+            "disable_buffering": False,
+            # When enabled, a background thread will be used to send metric payloads to the Agent.
+            "disable_background_sender": False,
+        }
+        if socket_path := kwargs.get("statsd_socket_path"):
+            instance_kwargs["socket_path"] = socket_path
+        else:
+            if host := kwargs.get("statsd_host"):
+                instance_kwargs["host"] = host
+            if port := kwargs.get("statsd_port"):
+                instance_kwargs["port"] = int(port)
+
+        self.statsd = DogStatsd(**instance_kwargs)
 
         # Origin detection is enabled after 0.45 by default.
         # Disable it since it silently fails.
         # Ref: https://github.com/DataDog/datadogpy/issues/764
-        statsd._container_id = None
+        self.statsd._container_id = None
+
+        # Applications should call wait_for_pending() before exiting to make sure all pending payloads are sent.
+        atexit.register(self.statsd.wait_for_pending)
+
         super().__init__(prefix=prefix)
 
     def incr(
@@ -60,7 +56,7 @@ def incr(
             tags["instance"] = instance
 
         tags_list = [f"{k}:{v}" for k, v in tags.items()]
-        statsd.increment(self._get_key(key), amount, sample_rate=sample_rate, tags=tags_list)
+        self.statsd.increment(self._get_key(key), amount, sample_rate=sample_rate, tags=tags_list)
 
     def timing(
         self,
@@ -70,6 +66,7 @@ def timing(
         tags: Tags | None = None,
         sample_rate: float = 1,
         stacklevel: int = 0,
+        precise: bool = False,
     ) -> None:
         tags = dict(tags or ())
 
@@ -79,7 +76,12 @@ def timing(
             tags["instance"] = instance
 
         tags_list = [f"{k}:{v}" for k, v in tags.items()]
-        statsd.timing(self._get_key(key), value, sample_rate=sample_rate, tags=tags_list)
+        if not precise:
+            self.statsd.timing(self._get_key(key), value, sample_rate=sample_rate, tags=tags_list)
+        else:
+            self.statsd.distribution(
+                self._get_key(key), value, sample_rate=sample_rate, tags=tags_list
+            )
 
     def gauge(
         self,
@@ -99,7 +101,7 @@ def gauge(
             tags["instance"] = instance
 
         tags_list = [f"{k}:{v}" for k, v in tags.items()]
-        statsd.gauge(self._get_key(key), value, sample_rate=sample_rate, tags=tags_list)
+        self.statsd.gauge(self._get_key(key), value, sample_rate=sample_rate, tags=tags_list)
 
     def distribution(
         self,
@@ -124,7 +126,7 @@ def distribution(
             tags["instance"] = instance
 
         tags_list = [f"{k}:{v}" for k, v in tags.items()]
-        statsd.distribution(self._get_key(key), value, sample_rate=sample_rate, tags=tags_list)
+        self.statsd.distribution(self._get_key(key), value, sample_rate=sample_rate, tags=tags_list)
 
     def event(
         self,
@@ -146,7 +148,7 @@ def event(
             tags["instance"] = instance
 
         tags_list = [f"{k}:{v}" for k, v in tags.items()]
-        statsd.event(
+        self.statsd.event(
             title=title,
             message=message,
             alert_type=alert_type,

src/sentry/metrics/dummy.py

@@ -24,6 +24,7 @@ def timing(
         tags: Tags | None = None,
         sample_rate: float = 1,
         stacklevel: int = 0,
+        precise: bool = False,
     ) -> None:
         pass
 

src/sentry/metrics/logging.py

@@ -26,6 +26,7 @@ def timing(
         tags: Tags | None = None,
         sample_rate: float = 1,
         stacklevel: int = 0,
+        precise: bool = False,
     ) -> None:
         logger.debug(
             "%r: %g ms", key, value * 1000, extra={"instance": instance, "tags": tags or {}}

src/sentry/metrics/middleware.py

@@ -144,13 +144,16 @@ def timing(
         tags: Tags | None = None,
         sample_rate: float = 1,
         stacklevel: int = 0,
+        precise: bool = False,
     ) -> None:
         current_tags = get_current_global_tags()
         if tags is not None:
             current_tags.update(tags)
         current_tags = _filter_tags(key, current_tags)
 
-        return self.inner.timing(key, value, instance, current_tags, sample_rate, stacklevel + 1)
+        return self.inner.timing(
+            key, value, instance, current_tags, sample_rate, stacklevel + 1, precise
+        )
 
     def gauge(
         self,

src/sentry/metrics/statsd.py

@@ -37,6 +37,7 @@ def timing(
         tags: Tags | None = None,
         sample_rate: float = 1,
         stacklevel: int = 0,
+        precise: bool = False,
     ) -> None:
         self.client.timing(self._full_key(self._get_key(key)), value, sample_rate)
 

src/sentry/utils/metrics.py

@@ -178,7 +178,9 @@ def timing(
 
     if precise and precise_backend:
         try:
-            precise_backend.timing(key, value, instance, tags, sample_rate, stacklevel + 1)
+            precise_backend.timing(
+                key, value, instance, tags, sample_rate, stacklevel + 1, precise=True
+            )
         except Exception:
             logger = logging.getLogger("sentry.errors")
             logger.exception("Unable to record precise metric")