feat(issues): make group attributes consumer enabled by default (#68633)

This PR enables the group attributes consumer by default. There is still
the ability to shut it off, however. We can do this now because we
enabled the consumer in Self Hosted
(getsentry/self-hosted#2927). In a future PR, I
intend to remove all the options and settings related to this and force
it to be on all the time.

src/sentry/conf/server.py

@@ -2753,7 +2753,7 @@ def custom_parameter_sort(parameter: dict) -> tuple[str, int]:
 SENTRY_USE_ISSUE_OCCURRENCE = False
 
 # This flag activates consuming GroupAttribute messages in the development environment
-SENTRY_USE_GROUP_ATTRIBUTES = False
+SENTRY_USE_GROUP_ATTRIBUTES = True
 
 # This flag activates code paths that are specific for customer domains
 SENTRY_USE_CUSTOMER_DOMAINS = False
@@ -2951,6 +2951,7 @@ def build_cdc_postgres_init_db_volume(settings: Any) -> dict[str, dict[str, str]
                     "1" if settings.SENTRY_USE_ISSUE_OCCURRENCE else ""
                 ),
                 "ENABLE_AUTORUN_MIGRATION_SEARCH_ISSUES": "1",
+                # TODO: remove setting
                 "ENABLE_GROUP_ATTRIBUTES_CONSUMER": (
                     "1" if settings.SENTRY_USE_GROUP_ATTRIBUTES else ""
                 ),

src/sentry/issues/attributes.py

@@ -4,7 +4,6 @@
 from enum import Enum
 from typing import cast
 
-import requests
 import urllib3
 from arroyo import Topic as ArroyoTopic
 from arroyo.backends.kafka import KafkaPayload, KafkaProducer, build_kafka_configuration
@@ -24,6 +23,7 @@
 from sentry.utils import json, metrics, snuba
 from sentry.utils.arroyo_producer import SingletonProducer
 from sentry.utils.kafka_config import get_kafka_producer_cluster_options, get_topic_definition
+from sentry.utils.snuba import _snuba_pool
 
 logger = logging.getLogger(__name__)
 
@@ -110,13 +110,15 @@ def produce_snapshot_to_kafka(snapshot: GroupAttributesSnapshot) -> None:
         # If we're not running Kafka then we're just in dev. Skip producing to Kafka and just
         # write to snuba directly
         try:
-            resp = requests.post(
-                f"{settings.SENTRY_SNUBA}/tests/entities/group_attributes/insert",
-                data=json.dumps([snapshot]),
+            resp = _snuba_pool.urlopen(
+                "POST",
+                "/tests/entities/group_attributes/insert",
+                body=json.dumps([snapshot]),
+                headers={},
             )
-            if resp.status_code != 200:
+            if resp.status != 200:
                 raise snuba.SnubaError(
-                    f"HTTP {resp.status_code} response from Snuba! {resp.content.decode('utf-8')}"
+                    f"HTTP {resp.status} response from Snuba! {resp.data.decode('utf-8')}"
                 )
             return None
         except urllib3.exceptions.HTTPError as err:

src/sentry/options/defaults.py

@@ -1892,7 +1892,7 @@
 # The flag activates whether to send group attributes messages to kafka
 register(
     "issues.group_attributes.send_kafka",
-    default=False,
+    default=True,
     flags=FLAG_MODIFIABLE_BOOL | FLAG_AUTOMATOR_MODIFIABLE,
 )
 

src/sentry/testutils/cases.py

@@ -1320,10 +1320,12 @@ def build_session(self, **kwargs):
     def store_group(self, group):
         data = [self.__wrap_group(group)]
         assert (
-            requests.post(
-                settings.SENTRY_SNUBA + "/tests/entities/groupedmessage/insert",
-                data=json.dumps(data),
-            ).status_code
+            _snuba_pool.urlopen(
+                "POST",
+                "/tests/entities/groupedmessage/insert",
+                body=json.dumps(data),
+                headers={},
+            ).status
             == 200
         )
 

src/sentry/testutils/pytest/sentry.py

@@ -270,7 +270,8 @@ def pytest_configure(config: pytest.Config) -> None:
 
     settings.SENTRY_USE_ISSUE_OCCURRENCE = True
 
-    settings.SENTRY_USE_GROUP_ATTRIBUTES = True
+    # TODO: enable this during tests
+    settings.SENTRY_OPTIONS["issues.group_attributes.send_kafka"] = False
 
     # For now, multiprocessing does not work in tests.
     settings.KAFKA_CONSUMER_FORCE_DISABLE_MULTIPROCESSING = True