create rc-processing-transaction

src/sentry/conf/server.py

@@ -1666,6 +1666,13 @@ def custom_parameter_sort(parameter: dict) -> tuple[str, int]:
 )
 SENTRY_EVENT_PROCESSING_STORE_OPTIONS: dict[str, str] = {}
 
+# Transactions processing backend
+# If these are set, transactions will be written to a different processing store
+# than errors. If these are set to none, Events(errors) and transactions will
+# both write to the EVENT_PROCESSING_STORE.
+SENTRY_TRANSACTION_PROCESSING_STORE: str | None = None
+SENTRY_TRANSACTION_PROCESSING_STORE_OPTIONS: dict[str, str] = {}
+
 # The internal Django cache is still used in many places
 # TODO(dcramer): convert uses over to Sentry's backend
 CACHES = {"default": {"BACKEND": "django.core.cache.backends.dummy.DummyCache"}}

src/sentry/consumers/__init__.py

@@ -23,6 +23,7 @@
     validate_consumer_definition,
 )
 from sentry.consumers.validate_schema import ValidateSchema
+from sentry.eventstream.types import EventStreamEventType
 from sentry.ingest.types import ConsumerType
 from sentry.utils.imports import import_string
 from sentry.utils.kafka_config import get_kafka_producer_cluster_options, get_topic_definition
@@ -373,20 +374,29 @@ def ingest_transactions_options() -> list[click.Option]:
         "synchronize_commit_log_topic_default": "snuba-generic-events-commit-log",
         "synchronize_commit_group_default": "generic_events_group",
         "click_options": _POST_PROCESS_FORWARDER_OPTIONS,
+        "static_args": {
+            "eventstream_type": EventStreamEventType.Generic.value,
+        },
     },
     "post-process-forwarder-transactions": {
         "topic": Topic.TRANSACTIONS,
         "strategy_factory": "sentry.eventstream.kafka.dispatch.EventPostProcessForwarderStrategyFactory",
         "synchronize_commit_log_topic_default": "snuba-transactions-commit-log",
         "synchronize_commit_group_default": "transactions_group",
         "click_options": _POST_PROCESS_FORWARDER_OPTIONS,
+        "static_args": {
+            "eventstream_type": EventStreamEventType.Transaction.value,
+        },
     },
     "post-process-forwarder-errors": {
         "topic": Topic.EVENTS,
         "strategy_factory": "sentry.eventstream.kafka.dispatch.EventPostProcessForwarderStrategyFactory",
         "synchronize_commit_log_topic_default": "snuba-commit-log",
         "synchronize_commit_group_default": "snuba-consumers",
         "click_options": _POST_PROCESS_FORWARDER_OPTIONS,
+        "static_args": {
+            "eventstream_type": EventStreamEventType.Error.value,
+        },
     },
     "process-spans": {
         "topic": Topic.SNUBA_SPANS,

src/sentry/eventstore/processing/__init__.py

@@ -9,5 +9,18 @@
     settings.SENTRY_EVENT_PROCESSING_STORE_OPTIONS,
 )
 
+transaction_processing_store = LazyServiceWrapper(
+    EventProcessingStore,
+    (
+        settings.SENTRY_EVENT_PROCESSING_STORE
+        if settings.SENTRY_TRANSACTION_PROCESSING_STORE is None
+        else settings.SENTRY_TRANSACTION_PROCESSING_STORE
+    ),
+    (
+        settings.SENTRY_EVENT_PROCESSING_STORE_OPTIONS
+        if settings.SENTRY_TRANSACTION_PROCESSING_STORE_OPTIONS is None
+        else settings.SENTRY_TRANSACTION_PROCESSING_STORE_OPTIONS
+    ),
+)
 
-__all__ = ["event_processing_store"]
+__all__ = ["event_processing_store", "transaction_processing_store"]

src/sentry/eventstream/base.py

@@ -3,7 +3,6 @@
 import logging
 from collections.abc import Collection, Mapping, MutableMapping, Sequence
 from datetime import datetime
-from enum import Enum
 from typing import TYPE_CHECKING, Any, Optional, TypedDict, cast
 
 from sentry.issues.issue_occurrence import IssueOccurrence
@@ -12,6 +11,8 @@
 from sentry.utils.cache import cache_key_for_event
 from sentry.utils.services import Service
 
+from .types import EventStreamEventType
+
 logger = logging.getLogger(__name__)
 
 
@@ -36,16 +37,6 @@ class GroupState(TypedDict):
 GroupStates = Sequence[GroupState]
 
 
-class EventStreamEventType(Enum):
-    """
-    We have 3 broad categories of event types that we care about in eventstream.
-    """
-
-    Error = "error"  # error, default, various security errors
-    Transaction = "transaction"  # transactions
-    Generic = "generic"  # generic events ingested via the issue platform
-
-
 class EventStream(Service):
     __all__ = (
         "insert",
@@ -69,6 +60,7 @@ def __init__(self, **options: Any) -> None:
 
     def _dispatch_post_process_group_task(
         self,
+        eventstream_type: str,
         event_id: str,
         project_id: int,
         group_id: int | None,
@@ -88,6 +80,7 @@ def _dispatch_post_process_group_task(
 
             post_process_group.apply_async(
                 kwargs={
+                    "eventstream_type": eventstream_type,
                     "is_new": is_new,
                     "is_regression": is_regression,
                     "is_new_group_environment": is_new_group_environment,
@@ -133,6 +126,7 @@ def insert(
         group_states: GroupStates | None = None,
     ) -> None:
         self._dispatch_post_process_group_task(
+            self._get_event_type(event).value,
             event.event_id,
             event.project_id,
             event.group_id,

src/sentry/eventstream/kafka/backend.py

@@ -12,8 +12,9 @@
 
 from sentry import options
 from sentry.conf.types.kafka_definition import Topic
-from sentry.eventstream.base import EventStreamEventType, GroupStates
+from sentry.eventstream.base import GroupStates
 from sentry.eventstream.snuba import KW_SKIP_SEMANTIC_PARTITIONING, SnubaProtocolEventStream
+from sentry.eventstream.types import EventStreamEventType
 from sentry.killswitches import killswitch_matches_context
 from sentry.utils import json
 from sentry.utils.kafka_config import get_kafka_producer_cluster_options, get_topic_definition

src/sentry/eventstream/kafka/dispatch.py

@@ -34,6 +34,7 @@ def _sampled_eventstream_timer(instance: str) -> Generator[None, None, None]:
 
 
 def dispatch_post_process_group_task(
+    eventstream_type: str,
     event_id: str,
     project_id: int,
     group_id: int | None,
@@ -53,6 +54,7 @@ def dispatch_post_process_group_task(
 
         post_process_group.apply_async(
             kwargs={
+                "eventstream_type": eventstream_type,
                 "is_new": is_new,
                 "is_regression": is_regression,
                 "is_new_group_environment": is_new_group_environment,
@@ -83,15 +85,15 @@ def _get_task_kwargs(message: Message[KafkaPayload]) -> Mapping[str, Any] | None
             return get_task_kwargs_for_message(message.payload.value)
 
 
-def _get_task_kwargs_and_dispatch(message: Message[KafkaPayload]) -> None:
+def _get_task_kwargs_and_dispatch(eventstream_type: str, message: Message[KafkaPayload]) -> None:
     task_kwargs = _get_task_kwargs(message)
     if not task_kwargs:
         return None
 
-    dispatch_post_process_group_task(**task_kwargs)
+    dispatch_post_process_group_task(eventstream_type=eventstream_type, **task_kwargs)
 
 
 class EventPostProcessForwarderStrategyFactory(PostProcessForwarderStrategyFactory):
     @staticmethod
-    def _dispatch_function(message: Message[KafkaPayload]) -> None:
-        return _get_task_kwargs_and_dispatch(message)
+    def _dispatch_function(eventstream_type: str, message: Message[KafkaPayload]) -> None:
+        return _get_task_kwargs_and_dispatch(eventstream_type, message)

src/sentry/eventstream/snuba.py

@@ -11,7 +11,8 @@
 from sentry import quotas
 from sentry.conf.types.kafka_definition import Topic, get_topic_codec
 from sentry.eventstore.models import GroupEvent
-from sentry.eventstream.base import EventStream, EventStreamEventType, GroupStates
+from sentry.eventstream.base import EventStream, GroupStates
+from sentry.eventstream.types import EventStreamEventType
 from sentry.utils import json, snuba
 from sentry.utils.safe import get_path
 from sentry.utils.sdk import set_current_event_project
@@ -469,6 +470,7 @@ def insert(
             **kwargs,
         )
         self._dispatch_post_process_group_task(
+            self._get_event_type(event).value,
             event.event_id,
             event.project_id,
             event.group_id,

src/sentry/eventstream/types.py

@@ -0,0 +1,11 @@
+from enum import Enum
+
+
+class EventStreamEventType(Enum):
+    """
+    We have 3 broad categories of event types that we care about in eventstream.
+    """
+
+    Error = "error"  # error, default, various security errors
+    Transaction = "transaction"  # transactions
+    Generic = "generic"  # generic events ingested via the issue platform

src/sentry/ingest/consumer/attachment_event.py

@@ -7,6 +7,7 @@
 from arroyo.dlq import InvalidMessage
 from arroyo.types import BrokerValue, Message
 
+from sentry.ingest.types import ConsumerType
 from sentry.models.project import Project
 from sentry.utils import metrics
 
@@ -23,7 +24,9 @@
 
 
 def decode_and_process_chunks(
-    raw_message: Message[KafkaPayload], consumer_type: str, reprocess_only_stuck_events: bool
+    raw_message: Message[KafkaPayload],
+    consumer_type: str,
+    reprocess_only_stuck_events: bool,
 ) -> IngestMessage | None:
     """
     The first pass for the `attachments` topic:
@@ -90,7 +93,7 @@ def process_attachments_and_events(
             if not reprocess_only_stuck_events:
                 process_individual_attachment(message, project)
         elif message_type == "event":
-            process_event(message, project, reprocess_only_stuck_events)
+            process_event(ConsumerType.Events, message, project, reprocess_only_stuck_events)
         elif message_type == "user_report":
             process_userreport(message, project)
         else:

src/sentry/ingest/consumer/processors.py

@@ -12,8 +12,9 @@
 from sentry import eventstore, features
 from sentry.attachments import CachedAttachment, attachment_cache
 from sentry.event_manager import EventManager, save_attachment
-from sentry.eventstore.processing import event_processing_store
+from sentry.eventstore.processing import event_processing_store, transaction_processing_store
 from sentry.feedback.usecases.create_feedback import FeedbackCreationSource, is_in_feedback_denylist
+from sentry.ingest.types import ConsumerType
 from sentry.ingest.userreport import Conflict, save_userreport
 from sentry.killswitches import killswitch_matches_context
 from sentry.models.project import Project
@@ -73,13 +74,14 @@ def process_transaction_no_celery(
         data = dict(data.items())
 
     with sentry_sdk.start_span(op="event_processing_store.store"):
-        cache_key = event_processing_store.store(data)
+        cache_key = transaction_processing_store.store(data)
     save_attachments(attachments, cache_key)
 
 
 @trace_func(name="ingest_consumer.process_event")
 @metrics.wraps("ingest_consumer.process_event")
 def process_event(
+    consumer_type: str,
     message: IngestMessage,
     project: Project,
     reprocess_only_stuck_events: bool = False,
@@ -95,6 +97,11 @@ def process_event(
     remote_addr = message.get("remote_addr")
     attachments = message.get("attachments") or ()
 
+    if consumer_type == ConsumerType.Transactions:
+        processing_store = transaction_processing_store
+    else:
+        processing_store = event_processing_store
+
     sentry_sdk.set_extra("event_id", event_id)
     sentry_sdk.set_extra("len_attachments", len(attachments))
 
@@ -183,7 +190,7 @@ def process_event(
         # process and consume the event from the `processing_store`, whereby getting it "unstuck".
         if reprocess_only_stuck_events:
             with sentry_sdk.start_span(op="event_processing_store.exists"):
-                if not event_processing_store.exists(data):
+                if not processing_store.exists(data):
                     return
 
         # The no_celery_mode version of the transactions consumer skips one trip to rc-processing
@@ -193,7 +200,7 @@ def process_event(
             cache_key = None
         else:
             with metrics.timer("ingest_consumer._store_event"):
-                cache_key = event_processing_store.store(data)
+                cache_key = processing_store.store(data)
             save_attachments(attachments, cache_key)
 
         try:

src/sentry/ingest/consumer/simple_event.py

@@ -5,6 +5,7 @@
 from arroyo.dlq import InvalidMessage
 from arroyo.types import BrokerValue, Message
 
+from sentry.ingest.types import ConsumerType
 from sentry.models.project import Project
 from sentry.utils import metrics
 
@@ -59,7 +60,9 @@ def process_simple_event_message(
             logger.exception("Project for ingested event does not exist: %s", project_id)
             return
 
-        return process_event(message, project, reprocess_only_stuck_events, no_celery_mode)
+        return process_event(
+            ConsumerType.Events, message, project, reprocess_only_stuck_events, no_celery_mode
+        )
 
     except Exception as exc:
         # If the retriable exception was raised, we should not DLQ

src/sentry/post_process_forwarder/post_process_forwarder.py

@@ -1,6 +1,7 @@
 import logging
 from abc import ABC, abstractmethod
 from collections.abc import Mapping
+from functools import partial
 
 from arroyo.backends.kafka import KafkaPayload
 from arroyo.processing.strategies import (
@@ -19,11 +20,12 @@
 class PostProcessForwarderStrategyFactory(ProcessingStrategyFactory[KafkaPayload], ABC):
     @staticmethod
     @abstractmethod
-    def _dispatch_function(message: Message[KafkaPayload]) -> None:
+    def _dispatch_function(eventstream_type: str, message: Message[KafkaPayload]) -> None:
         raise NotImplementedError()
 
     def __init__(
         self,
+        eventstream_type: str,
         mode: str,
         num_processes: int,
         input_block_size: int,
@@ -32,6 +34,7 @@ def __init__(
         max_batch_time: int,
         concurrency: int,
     ) -> None:
+        self.eventstream_type = eventstream_type
         self.mode = mode
         self.input_block_size = input_block_size
         self.output_block_size = output_block_size
@@ -49,15 +52,15 @@ def create_with_partitions(
         if self.mode == "multithreaded":
             logger.info("Starting multithreaded post process forwarder")
             return RunTaskInThreads(
-                processing_function=self._dispatch_function,
+                processing_function=partial(self._dispatch_function, self.eventstream_type),
                 concurrency=self.concurrency,
                 max_pending_futures=self.max_pending_futures,
                 next_step=CommitOffsets(commit),
             )
         elif self.mode == "multiprocess":
             logger.info("Starting multiprocess post process forwarder")
             return run_task_with_multiprocessing(
-                function=self._dispatch_function,
+                function=partial(self._dispatch_function, self.eventstream_type),
                 next_step=CommitOffsets(commit),
                 max_batch_size=self.max_batch_size,
                 max_batch_time=self.max_batch_time,