feat(tasks) Add taskbroker-client metrics for tasks

pyproject.toml

@@ -107,7 +107,7 @@ dependencies = [
   "statsd>=3.3.0",
   "structlog>=22.1.0",
   "symbolic>=13.1.1",
-  "taskbroker-client>=0.19.1",
+  "taskbroker-client>=0.19.2",
   "tiktoken>=0.8.0",
   "tokenizers>=0.22.0",
   "tldextract>=5.1.2",

src/sentry/taskworker/runtime.py

@@ -1,6 +1,10 @@
+import logging
+import os
+
 from django.conf import settings
 from django.core.cache import cache
 from taskbroker_client.app import TaskbrokerApp
+from taskbroker_client.metrics import DatadogMetrics, MetricsBackend
 
 from sentry.taskworker.adapters import (
     DjangoCacheAtMostOnceStore,
@@ -10,10 +14,47 @@
     make_producer,
 )
 
+logger = logging.getLogger(__name__)
+
+
+def _extract_metrics_config() -> tuple[str | None, int | None]:
+    host, port = None, None
+    metric_options = settings.SENTRY_METRICS_OPTIONS
+    try:
+        # Use the metrics settings options to infer the host/port.
+        # The metrics options have different structures depending on which backend is used.
+        if settings.SENTRY_METRICS_BACKEND == "sentry.metrics.dualwrite.DualWriteMetricsBackend":
+            metric_options = settings.SENTRY_METRICS_OPTIONS["primary_backend_args"]
+
+        # Some backends use `host` and others use `statsd_host`
+        host = metric_options.get("statsd_host", None) or metric_options.get("host", None)
+        port = metric_options.get("statsd_port", None) or metric_options.get("port", None)
+        port = int(port)
+    except Exception as e:
+        logger.warning("Could not extract metrics settings", extra={"error": str(e)})
+    return host, port
+
+
+metrics_class: MetricsBackend = SentryMetricsBackend()
+
+if os.getenv("USE_TASKWORKER_METRICS", None) == "1":
+    host, port = _extract_metrics_config()
+    if host and port:
+        # Metrics created by this interface will not
+        # have `sentry.` prefix, and will not have
+        # K8S_LABEL applied.
+        metrics_class = DatadogMetrics(
+            application="sentry",
+            statsd_host=host,
+            statsd_port=port,
+            sample_rate=settings.SENTRY_METRICS_SAMPLE_RATE,
+            enable_prefixed_metrics=True,
+        )
+
 app = TaskbrokerApp(
     name="sentry",
     producer_factory=make_producer,
-    metrics_class=SentryMetricsBackend(),
+    metrics_class=metrics_class,
     router_class=SentryRouter(),
     at_most_once_store=DjangoCacheAtMostOnceStore(cache),
     context_hooks=[ViewerContextHook()],