feat(spans): Skip enrichment when message contains skip_enrichment=True flag (#112024)

Going off of product requirements from
#111820 (comment),
we want to have a way to signal the process-segments consumer to skip
enrichment once a segment hits the size limit.

When this happens, the process-spans consumer would produce the segment
in chunks to the buffered-segments topic, where each message contains
one chunk along with the flag `skip_enrichment=True`.

Author: lvthanh03

src/sentry/spans/consumers/process_segments/factory.py

@@ -115,7 +115,10 @@ def _process_message(
     try:
         value = message.payload.value
         segment = orjson.loads(value)
-        processed = process_segment(segment["spans"], skip_produce=skip_produce)
+        skip_enrichment = segment.get("skip_enrichment", False)
+        processed = process_segment(
+            segment["spans"], skip_produce=skip_produce, skip_enrichment=skip_enrichment
+        )
         return [_serialize_payload(span, message.timestamp) for span in processed]
     except Exception:
         logger.exception("segments.invalid-message")

src/sentry/spans/consumers/process_segments/message.py

@@ -50,7 +50,7 @@
 
 @metrics.wraps("spans.consumers.process_segments.process_segment")
 def process_segment(
-    unprocessed_spans: list[SpanEvent], skip_produce: bool = False
+    unprocessed_spans: list[SpanEvent], skip_produce: bool = False, skip_enrichment: bool = False
 ) -> list[CompatibleSpan]:
     sample_rate = (
         settings.SENTRY_PROCESS_SEGMENTS_TRANSACTIONS_SAMPLE_RATE
@@ -62,14 +62,17 @@ def process_segment(
             "sample_rate": sample_rate,
         },
     ):
-        return _process_segment(unprocessed_spans, skip_produce)
+        return _process_segment(unprocessed_spans, skip_produce, skip_enrichment)
 
 
 def _process_segment(
-    unprocessed_spans: list[SpanEvent], skip_produce: bool
+    unprocessed_spans: list[SpanEvent], skip_produce: bool, skip_enrichment: bool
 ) -> list[CompatibleSpan]:
     _verify_compatibility(unprocessed_spans)
 
+    if skip_enrichment:
+        return [make_compatible(span) for span in unprocessed_spans]
+
     if unprocessed_spans:
         project_id = unprocessed_spans[0].get("project_id")
         if project_id in options.get("spans.process-segments.skip-enrichment-projects"):

tests/sentry/spans/consumers/process_segments/test_message.py

@@ -405,6 +405,45 @@ def test_no_skip_enrichment_for_other_project(self, mock_enrich: mock.MagicMock)
 
         mock_enrich.assert_called_once()
 
+    @mock.patch(
+        "sentry.spans.consumers.process_segments.message.TreeEnricher.enrich_spans",
+        wraps=None,
+    )
+    def test_skip_enrichment_flag(self, mock_enrich: mock.MagicMock) -> None:
+        """Test that enrichment is skipped when skip_enrichment=True is passed."""
+        segment_span = build_mock_span(
+            project_id=self.project.id,
+            is_segment=True,
+        )
+        child_span = build_mock_span(
+            project_id=self.project.id,
+            parent_span_id=segment_span["span_id"],
+        )
+
+        processed_spans = process_segment([child_span, segment_span], skip_enrichment=True)
+
+        mock_enrich.assert_not_called()
+        assert len(processed_spans) == 2
+
+    @mock.patch(
+        "sentry.spans.consumers.process_segments.message.TreeEnricher.enrich_spans",
+    )
+    def test_no_skip_enrichment_when_flag_is_false(self, mock_enrich: mock.MagicMock) -> None:
+        """Test that enrichment runs normally when skip_enrichment=False."""
+        mock_enrich.return_value = (None, [])
+        segment_span = build_mock_span(
+            project_id=self.project.id,
+            is_segment=True,
+        )
+        child_span = build_mock_span(
+            project_id=self.project.id,
+            parent_span_id=segment_span["span_id"],
+        )
+
+        process_segment([child_span, segment_span], skip_enrichment=False)
+
+        mock_enrich.assert_called_once()
+
 
 @exclude_experimental_detectors
 class TestSegmentDropKillswitch(TestCase):