chore(spans): Reorder functions in the process-segments consumer (#87270)

Moves code around for readability, no functional changes. Public
functions are now at the top, and remaining functions in functional
decomposition order.

Author: jan-auer

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

@@ -26,38 +26,6 @@
 logger = logging.getLogger(__name__)
 
 
-def process_message(message: Message[KafkaPayload]) -> list[Span]:
-    value = message.payload.value
-    segment = BUFFERED_SEGMENT_SCHEMA.decode(value)
-    return process_segment(cast(list[Span], segment["spans"]))
-
-
-def _process_message(message: Message[KafkaPayload]) -> list[Span]:
-    if not options.get("standalone-spans.process-segments-consumer.enable"):
-        return []
-
-    try:
-        return process_message(message)
-    except Exception:  # NOQA
-        raise
-        # TODO: Implement error handling
-        # sentry_sdk.capture_exception()
-        # assert isinstance(message.value, BrokerValue)
-        # raise InvalidMessage(message.value.partition, message.value.offset)
-
-
-def explode_segment(message: tuple[list[Span], Mapping[Partition, int]]):
-    spans, committable = message
-    last = len(spans) - 1
-    for i, span in enumerate(spans):
-        if span is not None:
-            yield Value(
-                payload=KafkaPayload(key=None, value=orjson.dumps(span), headers=[]),
-                committable=committable if i == last else {},
-                timestamp=None,
-            )
-
-
 class DetectPerformanceIssuesStrategyFactory(ProcessingStrategyFactory[KafkaPayload]):
     def __init__(
         self,
@@ -104,7 +72,7 @@ def create_with_partitions(
         # the generator function.
         zip_commit = RunTask(
             function=lambda m: (m.payload, m.committable),
-            next_step=Unfold(generator=explode_segment, next_step=produce_step),
+            next_step=Unfold(generator=_unfold_segment, next_step=produce_step),
         )
 
         return run_task_with_multiprocessing(
@@ -119,3 +87,31 @@ def create_with_partitions(
 
     def shutdown(self):
         self.pool.close()
+
+
+def _process_message(message: Message[KafkaPayload]) -> list[Span]:
+    if not options.get("standalone-spans.process-segments-consumer.enable"):
+        return []
+
+    try:
+        value = message.payload.value
+        segment = BUFFERED_SEGMENT_SCHEMA.decode(value)
+        return process_segment(cast(list[Span], segment["spans"]))
+    except Exception:  # NOQA
+        raise
+        # TODO: Implement error handling
+        # sentry_sdk.capture_exception()
+        # assert isinstance(message.value, BrokerValue)
+        # raise InvalidMessage(message.value.partition, message.value.offset)
+
+
+def _unfold_segment(message: tuple[list[Span], Mapping[Partition, int]]):
+    spans, committable = message
+    last = len(spans) - 1
+    for i, span in enumerate(spans):
+        if span is not None:
+            yield Value(
+                payload=KafkaPayload(key=None, value=orjson.dumps(span), headers=[]),
+                committable=committable if i == last else {},
+                timestamp=None,
+            )

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

@@ -39,6 +39,24 @@ class Span(SchemaSpan, total=False):
     hash: str | None  # Added in enrichment
 
 
+def process_segment(spans: list[Span]) -> list[Span]:
+    segment_span = _find_segment_span(spans)
+    if segment_span is None:
+        # TODO: Handle segments without a defined segment span once all
+        # functions are refactored to a span interface.
+        return spans
+
+    with metrics.timer("spans.consumers.process_segments.get_project"):
+        project = Project.objects.get_from_cache(id=segment_span["project_id"])
+
+    _enrich_spans(segment_span, spans)
+    _create_models(segment_span, project)
+    _detect_performance_problems(segment_span, spans, project)
+    _record_signals(segment_span, spans, project)
+
+    return spans
+
+
 def _find_segment_span(spans: list[Span]) -> Span | None:
     """
     Finds the segment in the span in the list that has ``is_segment`` set to
@@ -237,21 +255,3 @@ def _record_signals(segment_span: Span, spans: list[Span], project: Project) ->
     for module, is_module in INSIGHT_MODULE_FILTERS.items():
         if not get_project_insight_flag(project, module) and is_module(spans):
             record_first_insight_span(project, module)
-
-
-def process_segment(spans: list[Span]) -> list[Span]:
-    segment_span = _find_segment_span(spans)
-    if segment_span is None:
-        # TODO: Handle segments without a defined segment span once all
-        # functions are refactored to a span interface.
-        return spans
-
-    with metrics.timer("spans.consumers.process_segments.get_project"):
-        project = Project.objects.get_from_cache(id=segment_span["project_id"])
-
-    _enrich_spans(segment_span, spans)
-    _create_models(segment_span, project)
-    _detect_performance_problems(segment_span, spans, project)
-    _record_signals(segment_span, spans, project)
-
-    return spans