ref(event tracker) use consumer_type (#81080)

use consumer_type which is more aligned with pipeline names
`data.get("event_type")` has a bunch of other values other than the
general pipeline names

src/sentry/ingest/consumer/processors.py

@@ -203,9 +203,9 @@ def process_event(
         else:
             with metrics.timer("ingest_consumer._store_event"):
                 cache_key = processing_store.store(data)
-            if data.get("type") == "transaction":
+            if consumer_type == ConsumerType.Transactions:
                 track_sampled_event(
-                    data["event_id"], "transaction", TransactionStageStatus.REDIS_PUT
+                    data["event_id"], ConsumerType.Transactions, TransactionStageStatus.REDIS_PUT
                 )
 
             save_attachments(attachments, cache_key)

src/sentry/ingest/types.py

@@ -1,4 +1,7 @@
-class ConsumerType:
+from enum import StrEnum
+
+
+class ConsumerType(StrEnum):
     """
     Defines the types of ingestion consumers
     """

src/sentry/options/defaults.py

@@ -2916,7 +2916,7 @@
     flags=FLAG_ALLOW_EMPTY | FLAG_AUTOMATOR_MODIFIABLE,
 )
 register(
-    "performance.event-tracker.sample-rate.transaction",
+    "performance.event-tracker.sample-rate.transactions",
     default=0.0,
     flags=FLAG_AUTOMATOR_MODIFIABLE,
 )

src/sentry/utils/event_tracker.py

@@ -2,22 +2,18 @@
 from enum import StrEnum
 
 from sentry import options
-
-
-class EventType(StrEnum):
-    TRANSACTION = "transaction"
-    ERROR = "error"
+from sentry.ingest.types import ConsumerType
 
 
 class TransactionStageStatus(StrEnum):
     # the transaction is stored to rc-transactions
     REDIS_PUT = "redis_put"
 
     # a save_transactions task is kicked off
-    SAVE_TRX_STARTED = "save_trx_started"
+    SAVE_TXN_STARTED = "save_txn_started"
 
     # a save_transactions task is finished
-    SAVE_TRX_FINISHED = "save_trx_finished"
+    SAVE_TXN_FINISHED = "save_txn_finished"
 
     # the transaction is published to the `events` topic for snuba/sbc consumers to consume
     SNUBA_TOPIC_PUT = "snuba_topic_put"
@@ -35,21 +31,23 @@ class TransactionStageStatus(StrEnum):
 logger = logging.getLogger("EventTracker")
 
 
-def track_sampled_event(event_id: str, event_type: str, status: TransactionStageStatus) -> None:
+def track_sampled_event(
+    event_id: str, consumer_type: ConsumerType, status: TransactionStageStatus
+) -> None:
     """
     Records how far an event has made it through the ingestion pipeline.
     Each event type will pick up its sampling rate from its registered option.
     """
 
-    sample_rate = options.get(f"performance.event-tracker.sample-rate.{event_type}")
+    sample_rate = options.get(f"performance.event-tracker.sample-rate.{consumer_type}")
     if sample_rate == 0:
         return
 
     event_float = (int(event_id, 16) % 10000) / 10000
     if event_float < sample_rate:
         extra = {
             "event_id": event_id,
-            "event_type": getattr(EventType, event_type.upper(), None),
+            "consumer_type": consumer_type,
             "status": status,
         }
         _do_record(extra)

tests/sentry/utils/test_event_tracker.py

@@ -1,31 +1,30 @@
 import unittest
 from unittest.mock import patch
 
+from sentry.ingest.types import ConsumerType
 from sentry.testutils.cases import TestCase
 from sentry.testutils.helpers.options import override_options
-from sentry.utils.event_tracker import EventType, TransactionStageStatus, track_sampled_event
+from sentry.utils.event_tracker import TransactionStageStatus, track_sampled_event
 
 EVENT_ID = "9cdc4c32dff14fbbb012b0aa9e908126"
-EVENT_TYPE_STR = "transaction"
+CONSUMER_TYPE = ConsumerType.Transactions
 STATUS = TransactionStageStatus.REDIS_PUT
 
-EXPECTED_EVENT_TYPE = EventType.TRANSACTION
-
 
 class TestEventTracking(TestCase):
 
     @patch("sentry.utils.event_tracker._do_record")
     def test_track_sampled_event_logs_event(self, mock_do_record):
-        with override_options({"performance.event-tracker.sample-rate.transaction": 1.0}):
-            track_sampled_event(EVENT_ID, EVENT_TYPE_STR, STATUS)
+        with override_options({"performance.event-tracker.sample-rate.transactions": 1.0}):
+            track_sampled_event(EVENT_ID, CONSUMER_TYPE, STATUS)
             mock_do_record.assert_called_once_with(
-                {"event_id": EVENT_ID, "event_type": EXPECTED_EVENT_TYPE, "status": STATUS}
+                {"event_id": EVENT_ID, "consumer_type": CONSUMER_TYPE, "status": STATUS}
             )
 
     @patch("sentry.utils.event_tracker._do_record")
     def test_track_sampled_event_does_not_log_event(self, mock_do_record):
-        with override_options({"performance.event-tracker.sample-rate.transaction": 0.0}):
-            track_sampled_event(EVENT_ID, EVENT_TYPE_STR, STATUS)
+        with override_options({"performance.event-tracker.sample-rate.transactions": 0.0}):
+            track_sampled_event(EVENT_ID, CONSUMER_TYPE, STATUS)
             mock_do_record.assert_not_called()