ref(profiles): Remove the ingest-profiles Arroyo consumer

src/sentry/consumers/__init__.py

@@ -303,10 +303,6 @@ def ingest_events_options() -> list[click.Option]:
 
 # consumer name -> consumer definition
 KAFKA_CONSUMERS: Mapping[str, ConsumerDefinition] = {
-    "ingest-profiles": {
-        "topic": Topic.PROFILES,
-        "strategy_factory": "sentry.profiles.consumers.process.factory.ProcessProfileStrategyFactory",
-    },
     "ingest-replay-recordings": {
         "topic": Topic.INGEST_REPLAYS_RECORDINGS,
         "strategy_factory": "sentry.replays.consumers.recording.ProcessReplayRecordingStrategyFactory",

src/sentry/killswitches.py

@@ -230,7 +230,7 @@ class KillswitchInfo:
     ),
     "profiling.killswitch.ingest-profiles": KillswitchInfo(
         description="""
-        Drop profiles in the ingest-profiles consumer.
+        Drop profiles in the sentry.profiles.task.process_profile_from_kafka task.
 
         This happens after relay produces profiles to the topic but before a task
         is started to process/ingest to profile.

src/sentry/profiles/task.py

@@ -28,6 +28,7 @@
 from sentry import features, options, quotas
 from sentry.conf.types.kafka_definition import Topic
 from sentry.constants import DataCategory
+from sentry.killswitches import killswitch_matches_context
 from sentry.lang.javascript.processing import _handles_frame as is_valid_javascript_frame
 from sentry.lang.native.processing import _merge_image
 from sentry.lang.native.symbolicator import (
@@ -63,7 +64,7 @@
 from sentry.signals import first_profile_received
 from sentry.silo.base import SiloMode
 from sentry.tasks.base import instrumented_task
-from sentry.taskworker.namespaces import ingest_profiling_passthrough_tasks, ingest_profiling_tasks
+from sentry.taskworker.namespaces import ingest_profiling_passthrough_tasks
 from sentry.taskworker.producer import get_task_producer
 from sentry.utils import json, metrics
 from sentry.utils.arroyo_producer import SingletonProducer, get_arroyo_producer
@@ -160,19 +161,29 @@ def process_profile_from_kafka(
     headers: dict[str, str],
 ) -> None:
     """Process a profile from raw Kafka message bytes (taskbroker passthrough mode)."""
-    from sentry.profiles.consumers.process.factory import _process_profile_message
+    if _should_drop(headers):
+        return
 
-    _process_profile_message(message_bytes, headers, inline=True)
+    sampled = _is_sampled(headers)
+
+    if not sampled and not options.get("profiling.profile_metrics.unsampled_profiles.enabled"):
+        return
+
+    process_profile_task(payload=message_bytes, sampled=sampled)
+
+
+def _is_sampled(headers: dict[str, str]) -> bool:
+    return headers.get("sampled", "true") == "true"
+
+
+def _should_drop(headers: dict[str, str]) -> bool:
+    context = {"project_id": headers["project_id"]} if "project_id" in headers else {}
+
+    return bool(context) and killswitch_matches_context(
+        "profiling.killswitch.ingest-profiles", context
+    )
 
 
-@instrumented_task(
-    name="sentry.profiles.task.process_profile",
-    namespace=ingest_profiling_tasks,
-    processing_deadline_duration=60,
-    retry=Retry(times=2, delay=5),
-    compression_type=CompressionType.ZSTD,
-    silo_mode=SiloMode.CELL,
-)
 def process_profile_task(
     profile: Profile | None = None,
     payload: bytes | str | None = None,

src/sentry/runner/commands/run.py

@@ -544,9 +544,9 @@ def basic_consumer(
 
     Example:
 
-        sentry run consumer ingest-profiles --consumer-group ingest-profiles
+        sentry run consumer ingest-occurrences --consumer-group ingest-occurrences
 
-    runs the ingest-profiles consumer with the consumer group ingest-profiles.
+    runs the ingest-occurrences consumer with the consumer group ingest-occurrences.
 
     Consumers are defined in 'sentry.consumers'. Each consumer can take
     additional CLI options. Those can be passed after '--':

src/sentry/taskworker/namespaces.py

@@ -82,11 +82,6 @@
     app_feature="hybrid_cloud",
 )
 
-ingest_profiling_tasks = app.taskregistry.create_namespace(
-    "ingest.profiling",
-    app_feature="profiles",
-)
-
 ingest_profiling_passthrough_tasks = app.taskregistry.create_namespace(
     "ingest.profiling.passthrough",
     app_feature="profiles",

tests/sentry/consumers/test_run.py

@@ -47,7 +47,6 @@ def test_dlq(consumer_def) -> None:
         "metrics-last-seen-updater",
         "generic-metrics-last-seen-updater",
         "billing-metrics-consumer",
-        "ingest-profiles",
         "ingest-occurrences",
         "ingest-replay-recordings",
         "ingest-replay-recordings-two-step",

tests/sentry/processing/backpressure/test_checking.py

@@ -11,11 +11,9 @@
 from sentry.ingest.consumer.factory import IngestStrategyFactory
 from sentry.ingest.types import ConsumerType
 from sentry.processing.backpressure.health import record_consumer_health
-from sentry.profiles.consumers.process.factory import ProcessProfileStrategyFactory
 from sentry.testutils.helpers.options import override_options
 from sentry.utils import json
 
-PROFILES_MSG = json.dumps({"platform": "android", "profile": ""})
 EVENTS_MSG = json.dumps(
     {
         "message": "test-event",
@@ -24,45 +22,6 @@
 )
 
 
-@override_options(
-    {
-        "backpressure.checking.enabled": True,
-        "backpressure.checking.interval": 5,
-        "backpressure.monitoring.enabled": False,
-        "backpressure.status_ttl": 60,
-        "profiling.process.raw_bytes_payload.enabled": False,
-    }
-)
-def test_bad_config() -> None:
-    with raises(MessageRejected):
-        process_one_message(consumer_type="profiles", topic="profiles", payload=PROFILES_MSG)
-
-
-@patch("sentry.profiles.consumers.process.factory.process_profile_task.delay")
-@override_options(
-    {
-        "backpressure.checking.enabled": True,
-        "backpressure.checking.interval": 5,
-        "backpressure.monitoring.enabled": True,
-        "backpressure.status_ttl": 60,
-        "profiling.process.raw_bytes_payload.enabled": False,
-    }
-)
-def test_backpressure_healthy_profiles(process_profile_task: MagicMock) -> None:
-    record_consumer_health(
-        {
-            "attachments-store": [],
-            "processing-store": [],
-            "processing-store-transactions": [],
-            "processing-locks": [],
-            "post-process-locks": [],
-        }
-    )
-    process_one_message(consumer_type="profiles", topic="profiles", payload=PROFILES_MSG)
-
-    process_profile_task.assert_called_once()
-
-
 @override_options(
     {
         "backpressure.checking.enabled": True,
@@ -82,7 +41,7 @@ def test_backpressure_unhealthy_events() -> None:
         }
     )
     with raises(MessageRejected):
-        process_one_message(consumer_type="ingest", topic="ingest-events", payload=EVENTS_MSG)
+        process_one_message(payload=EVENTS_MSG)
 
 
 @patch("sentry.ingest.consumer.factory.maybe_multiprocess_step")
@@ -104,41 +63,22 @@ def test_backpressure_healthy_events(preprocess_event: MagicMock) -> None:
             "post-process-locks": [],
         }
     )
-    process_one_message(consumer_type="ingest", topic="ingest-events", payload=EVENTS_MSG)
+    process_one_message(payload=EVENTS_MSG)
 
     preprocess_event.assert_called_once()
 
 
-@patch("sentry.profiles.consumers.process.factory.process_profile_task.delay")
-@override_options(
-    {
-        "backpressure.checking.enabled": False,
-        "backpressure.checking.interval": 5,
-        "profiling.process.raw_bytes_payload.enabled": False,
-    }
-)
-def test_backpressure_not_enabled(process_profile_task: MagicMock) -> None:
-    process_one_message(consumer_type="profiles", topic="profiles", payload=PROFILES_MSG)
-
-    process_profile_task.assert_called_once()
-
-
-def process_one_message(consumer_type: str, topic: str, payload: str) -> None:
-    if consumer_type == "profiles":
-        processing_strategy = ProcessProfileStrategyFactory().create_with_partitions(
-            commit=Mock(), partitions={}
-        )
-    elif consumer_type == "ingest":
-        processing_strategy = IngestStrategyFactory(
-            consumer_type=ConsumerType.Events,
-            reprocess_only_stuck_events=False,
-            stop_at_timestamp=None,
-            num_processes=1,
-            max_batch_size=10,
-            max_batch_time=10,
-            input_block_size=None,
-            output_block_size=None,
-        ).create_with_partitions(commit=Mock(), partitions={})
+def process_one_message(payload: str) -> None:
+    processing_strategy = IngestStrategyFactory(
+        consumer_type=ConsumerType.Events,
+        reprocess_only_stuck_events=False,
+        stop_at_timestamp=None,
+        num_processes=1,
+        max_batch_size=10,
+        max_batch_time=10,
+        input_block_size=None,
+        output_block_size=None,
+    ).create_with_partitions(commit=Mock(), partitions={})
     message_dict = {
         "organization_id": 1,
         "project_id": 1,
@@ -156,7 +96,7 @@ def process_one_message(consumer_type: str, topic: str, payload: str) -> None:
                     msgpack_payload,
                     [],
                 ),
-                Partition(Topic(topic), 1),
+                Partition(Topic("ingest-events"), 1),
                 1,
                 datetime.now(),
             )