feat(post-process-forwarder) Allow different types of post process forwarders

src/sentry/eventstream/base.py

@@ -120,6 +120,7 @@ def requires_post_process_forwarder(self):
 
     def run_post_process_forwarder(
         self,
+        entity,
         consumer_group,
         commit_log_topic,
         synchronize_commit_group,

src/sentry/eventstream/kafka/backend.py

@@ -10,7 +10,10 @@
 from sentry.eventstream.kafka.consumer import SynchronizedConsumer
 from sentry.eventstream.kafka.postprocessworker import (
     _CONCURRENCY_OPTION,
+    ErrorsPostProcessForwarderWorker,
+    PostProcessForwarderType,
     PostProcessForwarderWorker,
+    TransactionsPostProcessForwarderWorker,
     _sampled_eventstream_timer,
 )
 from sentry.eventstream.kafka.protocol import (
@@ -151,6 +154,7 @@ def requires_post_process_forwarder(self):
 
     def _build_consumer(
         self,
+        entity,
         consumer_group,
         commit_log_topic,
         synchronize_commit_group,
@@ -169,7 +173,16 @@ def _build_consumer(
         )
 
         concurrency = options.get(_CONCURRENCY_OPTION)
-        worker = PostProcessForwarderWorker(concurrency=concurrency)
+        logger.info(f"Starting post process forwrader to consume {entity} messages")
+        if entity == PostProcessForwarderType.TRANSACTIONS:
+            worker = TransactionsPostProcessForwarderWorker(concurrency=concurrency)
+        elif entity == PostProcessForwarderType.ERRORS:
+            worker = ErrorsPostProcessForwarderWorker(concurrency=concurrency)
+        else:
+            # Default implementation which processes both errors and transactions
+            # irrespective of values in the header. This would most likely be the case
+            # for development environments.
+            worker = PostProcessForwarderWorker(concurrency=concurrency)
 
         consumer = BatchingKafkaConsumer(
             topics=self.topic,
@@ -183,6 +196,7 @@ def _build_consumer(
 
     def run_batched_consumer(
         self,
+        entity,
         consumer_group,
         commit_log_topic,
         synchronize_commit_group,
@@ -191,6 +205,7 @@ def run_batched_consumer(
         initial_offset_reset="latest",
     ):
         consumer = self._build_consumer(
+            entity,
             consumer_group,
             commit_log_topic,
             synchronize_commit_group,
@@ -404,6 +419,7 @@ def _get_task_kwargs_and_dispatch(self, message) -> None:
 
     def run_post_process_forwarder(
         self,
+        entity,
         consumer_group,
         commit_log_topic,
         synchronize_commit_group,
@@ -416,6 +432,7 @@ def run_post_process_forwarder(
         if settings.SENTRY_POST_PROCESS_FORWARDER_BATCHING:
             logger.info("Starting batching consumer")
             self.run_batched_consumer(
+                entity,
                 consumer_group,
                 commit_log_topic,
                 synchronize_commit_group,

src/sentry/eventstream/kafka/postprocessworker.py

@@ -2,6 +2,7 @@
 import random
 from concurrent.futures import Future, ThreadPoolExecutor, as_completed
 from contextlib import contextmanager
+from enum import Enum
 from typing import Any, Generator, Mapping, Optional, Sequence
 
 from sentry import options
@@ -25,6 +26,12 @@
 _TRANSACTION_FORWARDER_HEADER = "transaction_forwarder"
 
 
+class PostProcessForwarderType(str, Enum):
+    ERRORS = "errors"
+    TRANSACTIONS = "transactions"
+    ALL = "all"
+
+
 @contextmanager
 def _sampled_eventstream_timer(instance: str) -> Generator[None, None, None]:
     record_metric = random.random() < 0.1

src/sentry/runner/commands/devserver.py

@@ -13,6 +13,7 @@
         "sentry",
         "run",
         "post-process-forwarder",
+        "--entity=all",
         "--loglevel=debug",
         "--commit-batch-size=1",
     ],

src/sentry/runner/commands/run.py

@@ -324,6 +324,12 @@ def cron(**options):
     type=click.Choice(["earliest", "latest"]),
     help="Position in the commit log topic to begin reading from when no prior offset has been recorded.",
 )
+@click.option(
+    "--entity",
+    default="all",
+    type=click.Choice(["all", "errors", "transactions"]),
+    help="The type of entity to process (all, errors, transactions).",
+)
 @log_options()
 @configuration
 def post_process_forwarder(**options):
@@ -332,6 +338,7 @@ def post_process_forwarder(**options):
 
     try:
         eventstream.run_post_process_forwarder(
+            entity=options["entity"],
             consumer_group=options["consumer_group"],
             commit_log_topic=options["commit_log_topic"],
             synchronize_commit_group=options["synchronize_commit_group"],

tests/sentry/eventstream/kafka/test_consumer.py

@@ -691,6 +691,7 @@ def test_post_process_forwarder_batch_consumer(self, dispatch_post_process_group
 
         eventstream = KafkaEventStream()
         consumer = eventstream._build_consumer(
+            entity="all",
             consumer_group=consumer_group,
             commit_log_topic=self.commit_log_topic,
             synchronize_commit_group=synchronize_commit_group,