feat(spans): Add timers for process_spans and process_batch (#94159)

Adds more timings to give additional insights into the overheads for
loading payloads, killswitches, and batch processing.

Author: jan-auer

src/sentry/spans/buffer.py

@@ -170,6 +170,7 @@ def __reduce__(self):
     def _get_span_key(self, project_and_trace: str, span_id: str) -> bytes:
         return f"span-buf:z:{{{project_and_trace}}}:{span_id}".encode("ascii")
 
+    @metrics.wraps("spans.buffer.process_spans")
     def process_spans(self, spans: Sequence[Span], now: int):
         """
         :param spans: List of to-be-ingested spans.

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

@@ -18,6 +18,7 @@
 from sentry import killswitches
 from sentry.spans.buffer import Span, SpansBuffer
 from sentry.spans.consumers.process.flusher import SpanFlusher
+from sentry.utils import metrics
 from sentry.utils.arroyo import MultiprocessingPool, run_task_with_multiprocessing
 
 logger = logging.getLogger(__name__)
@@ -122,6 +123,7 @@ def shutdown(self) -> None:
             self.__pool.close()
 
 
+@metrics.wraps("spans.buffer.process_batch")
 def process_batch(
     buffer: SpansBuffer,
     values: Message[ValuesBatch[tuple[int, KafkaPayload]]],
@@ -136,17 +138,16 @@ def process_batch(
             if min_timestamp is None or timestamp < min_timestamp:
                 min_timestamp = timestamp
 
-            val = cast(SpanEvent, rapidjson.loads(payload.value))
-
-            partition_id = value.partition.index
+            with metrics.timer("spans.buffer.process_batch.decode"):
+                val = cast(SpanEvent, rapidjson.loads(payload.value))
 
             if killswitches.killswitch_matches_context(
                 "spans.drop-in-buffer",
                 {
                     "org_id": val.get("organization_id"),
                     "project_id": val.get("project_id"),
                     "trace_id": val.get("trace_id"),
-                    "partition_id": partition_id,
+                    "partition_id": value.partition.index,
                 },
             ):
                 continue