ref: Move shared arroyo metrics wrapper (#40814)

The MetricsWrapper is a compatibility layer for Arroyo that allows all Arroyo instrumented consumer metrics to be automatically sent to Sentry's backend. This code was originally part of the `sentry_metrics` product but is now moved to a shared utility as this can be reused to instrument all Arroyo consumers going forward.

mypy.ini

@@ -125,6 +125,7 @@ files = fixtures/mypy-stubs,
         src/sentry/types/region.py,
         src/sentry/unmerge.py,
         src/sentry/utils/appleconnect/,
+        src/sentry/utils/arroyo.py,
         src/sentry/utils/assets.py,
         src/sentry/utils/hashlib.py,
         src/sentry/utils/suspect_resolutions/commit_correlation.py,

src/sentry/eventstream/kafka/backend.py

@@ -22,8 +22,8 @@ from sentry.eventstream.kafka.postprocessworker import (
 from sentry.eventstream.kafka.synchronized import SynchronizedConsumer as ArroyoSynchronizedConsumer
 from sentry.eventstream.snuba import KW_SKIP_SEMANTIC_PARTITIONING, SnubaProtocolEventStream
 from sentry.killswitches import killswitch_matches_context
-from sentry.sentry_metrics.metrics_wrapper import MetricsWrapper
 from sentry.utils import json, kafka, metrics
+from sentry.utils.arroyo import MetricsWrapper
 from sentry.utils.batching_kafka_consumer import BatchingKafkaConsumer
 from sentry.utils.kafka_config import get_kafka_consumer_cluster_options
 

src/sentry/sentry_metrics/configuration.py

@@ -179,7 +179,7 @@ def initialize_global_consumer_state(config: MetricsIngestConfiguration) -> None
 
     add_global_tags(_all_threads=True, **global_tag_map)
 
-    from sentry.sentry_metrics.metrics_wrapper import MetricsWrapper
+    from sentry.utils.arroyo import MetricsWrapper
 
     metrics_wrapper = MetricsWrapper(backend, name="sentry_metrics.indexer", tags=global_tag_map)
     configure_metrics(metrics_wrapper)

src/sentry/sentry_metrics/metrics_wrapper.py → src/sentry/utils/arroyo.py

@@ -1,11 +1,19 @@
 from typing import Mapping, Optional, Union
 
+from arroyo.utils.metrics import Metrics
+
 from sentry.metrics.base import MetricsBackend
 
 Tags = Mapping[str, str]
 
 
-class MetricsWrapper:
+class MetricsWrapper(Metrics):
+    """
+    Metrics adapter for use with the Arroyo library. This allows consumer
+    metrics instrumented via the Arroyo library to be automatically recorded
+    and sent to Sentry's configured metrics backend.
+    """
+
     def __init__(
         self,
         backend: MetricsBackend,