ref(arroyo): Arroyo metrics in every Sentry process (#94163)

Initialize arroyo's metrics backend in every sentry process. Due to
existing metrics, all arroyo metrics are namespaced under
sentry.consumer even though it doesn't make sense for producers.

Author: untitaker

.github/CODEOWNERS

@@ -13,8 +13,6 @@
 /src/sentry/post_process_forwarder/                      @getsentry/owners-snuba
 /src/sentry/utils/snuba.py                               @getsentry/owners-snuba @getsentry/visibility
 /src/sentry/utils/snql.py                                @getsentry/owners-snuba
-/src/sentry/utils/arroyo.py                              @getsentry/owners-snuba
-/src/sentry/utils/arroyo_producer.py                     @getsentry/owners-snuba
 /src/sentry/tsdb/snuba.py                                @getsentry/owners-snuba
 /src/sentry/tsdb/redissnuba.py                           @getsentry/owners-snuba
 /src/sentry/tagstore/snuba/                              @getsentry/owners-snuba
@@ -672,6 +670,10 @@ tests/sentry/api/endpoints/test_organization_dashboard_widget_details.py     @ge
 /src/sentry/spans/                                @getsentry/visibility @getsentry/streaming-platform
 /tests/sentry/spans/                              @getsentry/visibility @getsentry/streaming-platform
 
+# Streaming platform
+/src/sentry/utils/arroyo.py                              @getsentry/streaming-platform
+/src/sentry/utils/arroyo_producer.py                     @getsentry/streaming-platform
+
 # Workflow engine
 /src/sentry/workflow_engine/                               @getsentry/alerts-create-issues
 /tests/sentry/workflow_engine/                             @getsentry/alerts-create-issues

src/sentry/runner/commands/run.py

@@ -623,15 +623,12 @@ def basic_consumer(
     """
     from sentry.consumers import get_stream_processor
     from sentry.metrics.middleware import add_global_tags
-    from sentry.utils.arroyo import initialize_arroyo_main
 
     log_level = options.pop("log_level", None)
     if log_level is not None:
         logging.getLogger("arroyo").setLevel(log_level.upper())
 
     add_global_tags(kafka_topic=topic, consumer_group=options["group_id"])
-    initialize_arroyo_main()
-
     processor = get_stream_processor(consumer_name, consumer_args, topic=topic, **options)
 
     # for backwards compat: should eventually be removed
@@ -654,9 +651,6 @@ def dev_consumer(consumer_names: tuple[str, ...]) -> None:
     """
 
     from sentry.consumers import get_stream_processor
-    from sentry.utils.arroyo import initialize_arroyo_main
-
-    initialize_arroyo_main()
 
     processors = [
         get_stream_processor(

src/sentry/runner/initializer.py

@@ -11,6 +11,7 @@
 
 from sentry.silo.patches.silo_aware_transaction_patch import patch_silo_aware_atomic
 from sentry.utils import warnings
+from sentry.utils.arroyo import initialize_arroyo_main
 from sentry.utils.sdk import configure_sdk
 from sentry.utils.warnings import DeprecatedSettingWarning
 
@@ -387,6 +388,8 @@ def initialize_app(config: dict[str, Any], skip_service_validation: bool = False
 
     import_grouptype()
 
+    initialize_arroyo_main()
+
     # Hacky workaround to dynamically set the CSRF_TRUSTED_ORIGINS for self hosted
     if settings.SENTRY_SELF_HOSTED and not settings.CSRF_TRUSTED_ORIGINS:
         from sentry import options

src/sentry/utils/arroyo.py

@@ -133,6 +133,10 @@ def initialize_arroyo_main() -> None:
 
     from sentry.utils.metrics import backend
 
+    # XXX: we initially called this function only to initialize sentry consumer
+    # metrics, and namespaced arroyo metrics under sentry.consumer. Now we
+    # initialize arroyo metrics in every sentry process, and so even producer
+    # metrics are namespaced under sentry.consumer.
     metrics_wrapper = MetricsWrapper(backend, name="consumer")
     configure_metrics(metrics_wrapper)