ref(consumers): Allow injecting StreamProcessor kwargs using options (#104244)

We have had situations where applying processor arguments using CLI
arguments was error prone and took a long time to revert if something is
wrong. Let's use dynamic cli flags instead.

currently you have to:
* add a feature to arroyo
* add a cli flag to sentry
* add cli in k8s

then the flag cleanup is all of that in reverse

with this pr you at least only have to bump arroyo in sentry, and
there's nothing to be done in sentry on cleanup

Author: untitaker

src/sentry/consumers/__init__.py

@@ -3,7 +3,7 @@
 import logging
 import uuid
 from collections.abc import Mapping, Sequence
-from typing import Any
+from typing import Any, Union, get_args, get_origin, get_type_hints
 
 import click
 from arroyo.backends.abstract import Consumer
@@ -27,12 +27,79 @@
 from sentry.consumers.validate_schema import ValidateSchema
 from sentry.eventstream.types import EventStreamEventType
 from sentry.ingest.types import ConsumerType
+from sentry.utils import json
 from sentry.utils.imports import import_string
 from sentry.utils.kafka_config import get_topic_definition
 
 logger = logging.getLogger(__name__)
 
 
+def apply_processor_args_overrides(
+    consumer_name: str, base_args: dict[str, Any], overrides: Sequence[str]
+) -> dict[str, Any]:
+    """
+    Apply processor args overrides from CLI strings to the base StreamProcessor arguments.
+
+    Args:
+        consumer_name: Name of the consumer
+        base_args: Base arguments dict for StreamProcessor
+        overrides: Raw CLI argument strings in format 'key:value'
+
+    Returns:
+        Updated arguments dict with overrides applied
+    """
+    # Get resolved type hints (handles __future__.annotations)
+    type_hints = get_type_hints(StreamProcessor.__init__)
+
+    for arg in overrides:
+        try:
+            key, value_str = arg.split(":", 1)
+            param_type = type_hints[key]
+
+            # Extract the actual type from Optional[T] (which is Union[T, None])
+            if get_origin(param_type) is Union:
+                # For Optional[T] (Union[T, None]), extract T
+                type_args = get_args(param_type)
+                param_type = next((t for t in type_args if t is not type(None)), str)
+
+            # Try to parse as JSON first, fallback to string
+            try:
+                value = json.loads(value_str)
+            except Exception:
+                value = value_str
+
+            # Validate the type matches what we expect
+            # Allow int->float coercion since JSON parses as int
+            if param_type == float and isinstance(value, int):
+                value = float(value)
+            elif not isinstance(value, param_type):
+                raise ValueError(f"Expected {param_type}, got {type(value)}")
+
+            if key in base_args:
+                logger.info(
+                    "overriding argument %s from CLI: %s -> %s",
+                    key,
+                    base_args[key],
+                    value,
+                    extra={
+                        "consumer_name": consumer_name,
+                    },
+                )
+            base_args[key] = value
+        except Exception as e:
+            logger.warning(
+                "skipping invalid argument %s from CLI: %s",
+                arg,
+                str(e),
+                extra={
+                    "consumer_name": consumer_name,
+                    "valid_params": sorted(type_hints.keys()),
+                },
+            )
+
+    return base_args
+
+
 def convert_max_batch_time(ctx, param, value):
     if value <= 0:
         raise click.BadParameter("--max-batch-time must be greater than 0")
@@ -467,6 +534,7 @@ def get_stream_processor(
     kafka_slice_id: int | None = None,
     add_global_tags: bool = False,
     profile_consumer_join: bool = False,
+    arroyo_args: Sequence[str] | None = None,
 ) -> StreamProcessor:
     from sentry.utils import kafka_config
 
@@ -623,14 +691,21 @@ def build_consumer_config(group_id: str):
     else:
         dlq_policy = None
 
-    return StreamProcessor(
-        consumer=consumer,
-        topic=ArroyoTopic(topic),
-        processor_factory=strategy_factory,
-        commit_policy=ONCE_PER_SECOND,
-        join_timeout=join_timeout,
-        dlq_policy=dlq_policy,
-    )
+    # Build base StreamProcessor arguments
+    processor_args = {
+        "consumer": consumer,
+        "topic": ArroyoTopic(topic),
+        "processor_factory": strategy_factory,
+        "commit_policy": ONCE_PER_SECOND,
+        "join_timeout": join_timeout,
+        "dlq_policy": dlq_policy,
+    }
+
+    # Apply CLI-provided overrides for StreamProcessor arguments
+    if arroyo_args:
+        processor_args = apply_processor_args_overrides(consumer_name, processor_args, arroyo_args)
+
+    return StreamProcessor(**processor_args)
 
 
 class ValidateSchemaStrategyFactoryWrapper(ProcessingStrategyFactory):

src/sentry/runner/commands/run.py

@@ -544,13 +544,20 @@ def taskbroker_send_tasks(
     default=False,
     help="Adds a ProcessingStrategy to the start of a consumer that records a transaction of the consumer's join() method.",
 )
+@click.option(
+    "--arroyo-arg",
+    "arroyo_args",
+    multiple=True,
+    help="Override StreamProcessor arguments. Format: --arroyo-arg='key:value'. Example: --arroyo-arg='join_timeout:60'",
+)
 @configuration
 def basic_consumer(
     consumer_name: str,
     consumer_args: tuple[str, ...],
     topic: str | None,
     kafka_slice_id: int | None,
     quantized_rebalance_delay_secs: int | None,
+    arroyo_args: tuple[str, ...],
     **options: Any,
 ) -> None:
     """
@@ -594,6 +601,7 @@ def basic_consumer(
         topic=topic,
         kafka_slice_id=kafka_slice_id,
         add_global_tags=True,
+        arroyo_args=arroyo_args,
         **options,
     )
 

tests/sentry/consumers/test_run.py

@@ -1,3 +1,5 @@
+from unittest.mock import patch
+
 import pytest
 from arroyo.processing.strategies.abstract import ProcessingStrategyFactory
 
@@ -58,3 +60,61 @@ def test_dlq(consumer_def) -> None:
         + consumers_that_should_have_dlq_but_dont
     ):
         assert defn.get("dlq_topic") is not None, f"{consumer_name} consumer is missing DLQ"
+
+
+def test_apply_processor_args_overrides() -> None:
+    """Test the apply_processor_args_overrides function."""
+    from sentry.consumers import apply_processor_args_overrides
+
+    # Test with CLI string overrides
+    result = apply_processor_args_overrides(
+        "ingest-monitors",
+        {"join_timeout": 10.0, "consumer": "mock_consumer", "topic": "mock_topic"},
+        ("join_timeout:123", "stuck_detector_timeout:456"),
+    )
+    assert result["join_timeout"] == 123
+    assert result["stuck_detector_timeout"] == 456
+    assert result["consumer"] == "mock_consumer"
+
+    # Test with empty overrides
+    result = apply_processor_args_overrides("ingest-monitors", {"join_timeout": 10.0}, ())
+    assert result["join_timeout"] == 10.0
+
+    # Test logging when overriding existing arg
+    with patch("sentry.consumers.logger") as mock_logger:
+        result = apply_processor_args_overrides(
+            "ingest-monitors",
+            {"join_timeout": 10.0},
+            ("join_timeout:999",),
+        )
+        assert result["join_timeout"] == 999
+        mock_logger.info.assert_called_once()
+        call_args = mock_logger.info.call_args
+        assert call_args[0][0] == "overriding argument %s from CLI: %s -> %s"
+        assert call_args[0][1] == "join_timeout"
+        assert call_args[0][2] == 10.0
+        assert call_args[0][3] == 999
+
+    # Test no logging when adding new arg
+    with patch("sentry.consumers.logger") as mock_logger:
+        result = apply_processor_args_overrides(
+            "ingest-monitors",
+            {"join_timeout": 10.0},
+            ("stuck_detector_timeout:456",),
+        )
+        assert result["stuck_detector_timeout"] == 456
+        mock_logger.info.assert_not_called()
+
+    # Test skipping invalid parameters and logging warning
+    with patch("sentry.consumers.logger") as mock_logger:
+        result = apply_processor_args_overrides(
+            "ingest-monitors",
+            {"join_timeout": 10.0},
+            ("invalid_param:789", "join_timeout:999"),
+        )
+        assert result["join_timeout"] == 999
+        assert "invalid_param" not in result
+        mock_logger.warning.assert_called_once()
+        call_args = mock_logger.warning.call_args
+        assert call_args[0][0] == "skipping invalid argument %s from CLI: %s"
+        assert call_args[0][1] == "invalid_param:789"