feat(spans-buffer): track pre-merge oversized parent sets (#108447)

Adds the `parent_span_set_already_oversized` metric to record when the
destination span set already exceeds `max_segment_bytes` before
sunionstore.

Author: lvthanh03

src/sentry/scripts/spans/add-buffer.lua

@@ -127,6 +127,12 @@ table.insert(latency_table, {"sunionstore_args_step_latency_ms", sunionstore_arg
 -- Used outside the if statement
 local spop_end_time_ms = -1
 if #sunionstore_args > 0 then
+    if (redis.call("memory", "usage", set_key) or 0) > max_segment_bytes then
+        table.insert(metrics_table, {"parent_span_set_already_oversized", 1})
+    else
+        table.insert(metrics_table, {"parent_span_set_already_oversized", 0})
+    end
+
     local start_output_size = redis.call("scard", set_key)
     local scard_end_time_ms = get_time_ms()
     table.insert(latency_table, {"scard_step_latency_ms", scard_end_time_ms - sunionstore_args_end_time_ms})

src/sentry/spans/buffer_logger.py

@@ -243,6 +243,7 @@ def emit_observability_metrics(
     gauge_metrics_dict: dict[
         str, tuple[float, float, float, float]
     ] = {}  # metric, min, max, sum, count
+    oversized_count = 0
 
     for evalsha_latency_metrics, evalsha_gauge_metrics in zip(latency_metrics, gauge_metrics):
         for raw_key, value in evalsha_latency_metrics:
@@ -267,6 +268,8 @@ def emit_observability_metrics(
                     gauge_metrics_dict[key][2] + value,
                     gauge_metrics_dict[key][3] + 1.0,
                 )
+            if raw_key == b"parent_span_set_already_oversized":
+                oversized_count += int(value)
 
     for metric, (min_value, max_value, sum_value, count) in latency_metrics_dict.items():
         tags = {"stage": metric}
@@ -295,3 +298,9 @@ def emit_observability_metrics(
             value,
             tags={"stage": key},
         )
+
+    if oversized_count > 0:
+        metrics.incr(
+            "spans.buffer.process_spans.parent_span_set_already_oversized.count",
+            amount=oversized_count,
+        )

tests/sentry/spans/test_buffer.py

@@ -292,6 +292,62 @@ def test_observability_metrics(
     emit_observability_metrics.assert_called_once()
 
 
+@mock.patch("sentry.spans.buffer.emit_observability_metrics")
+def test_observability_metrics_parent_span_already_oversized(
+    emit_observability_metrics: mock.MagicMock,
+    buffer: SpansBuffer,
+) -> None:
+    # Disable compression so payload size in Redis is predictable, then force a
+    # low max-segment-bytes threshold so the destination set is already too
+    # large before merge.
+    oversized_parent_payload = orjson.dumps({"span_id": "b" * 16, "blob": "x" * 2048})
+    spans = [
+        Span(
+            # payload=_payload("a" * 16),
+            payload=oversized_parent_payload,
+            trace_id="a" * 32,
+            span_id="a" * 16,
+            parent_span_id="b" * 16,
+            segment_id=None,
+            project_id=1,
+            end_timestamp=1700000000.0,
+        ),
+        Span(
+            # payload=oversized_parent_payload,
+            payload=_payload("a" * 16),
+            trace_id="a" * 32,
+            span_id="b" * 16,
+            parent_span_id=None,
+            segment_id=None,
+            is_segment_span=True,
+            project_id=1,
+            end_timestamp=1700000000.0,
+        ),
+    ]
+
+    with override_options(
+        {"spans.buffer.max-segment-bytes": 200, "spans.buffer.compression.level": -1}
+    ):
+        process_spans(spans, buffer, now=0)
+
+    emit_observability_metrics.assert_called_once()
+    args, _ = emit_observability_metrics.call_args
+    gauge_metrics = args[1]
+
+    oversized_metric_values = [
+        value
+        for evalsha_metrics in gauge_metrics
+        for metric_name, value in evalsha_metrics
+        if metric_name == b"parent_span_set_already_oversized"
+    ]
+    assert oversized_metric_values, (
+        "Expected parent_span_set_already_oversized metric to be emitted"
+    )
+    assert 1 in oversized_metric_values, (
+        "Expected at least one evalsha call with an already oversized parent set"
+    )
+
+
 @pytest.mark.parametrize(
     "spans",
     list(

tests/sentry/spans/test_buffer_logger.py

@@ -474,3 +474,21 @@ def t(stage):
                 call(LG, 55.0, tags=t("spopcalls")),
             ]
         )
+
+    @mock.patch("sentry.spans.buffer_logger.metrics.incr")
+    def test_emit_observability_metrics_oversized_count_metric(self, mock_incr):
+        emit_observability_metrics(
+            latency_metrics=[[]],
+            gauge_metrics=[
+                [
+                    (b"parent_span_set_already_oversized", 1.0),
+                    (b"redirect_depth", 1.0),
+                ],
+            ],
+            longest_evalsha_data=(0, [], []),
+        )
+
+        mock_incr.assert_called_once_with(
+            "spans.buffer.process_spans.parent_span_set_already_oversized.count",
+            amount=1,
+        )