fix(issue detectors): Fix missing nodestore data on segment-derived occurrences (#123288)

NOTE: This is heavily based on the work done in #123147, but includes a number of additional fixes identified by Claude as needing to be part of that change. Because the final result differs quite a bit from that PR, I created a new one which supersedes it. H/t to Claude for a great deal of help with the reasoning and rough drafts of all of the code included here, as well as all the tests.

------------------

Currently, occurrences from the span segment pipeline set `is_buffered_spans`, which builds an `Event` backed only by `snuba_data` and never writes to nodestore. The eventstream payload is therefore just `{"received": ...}`, so occurrences are reaching Snuba and showing up in the issue feed, but issue details is broken because there's no corresponding event JSON.

To fix this problem, this PR drops the special case and sends segment-derived occurrences through the same path that every other occurrence takes. Each detected problem now carries its own event holding only the spans its evidence points at, filtered down to the fields issue details and Seer actually read, and trimmed to safely fit within the occurrence producer's limits.

Notes:

- Since the `event` we create is entirely synthetic, the occurrence id is reused for the event id.

- Trimming happens both in terms of how many spans we keep and in terms of the data in those spans. Offender spans are given priority over parent and cause spans, since offender spans are the ones actually at the root of whatever problem we're reporting.

- Any span whose id is referenced in one of the span id lists (parent, cause, or offender) but which isn't present in the segment is dropped, so we never error out trying to pull data from a span we don't have.

- This fixes new occurrences only. Ones already written by the old path still point at nodestore keys that were never created, and stay broken until they age out.

- There are a number of other issues Claude noted when working on this, but for ease of review those have been split off into follow-up PRs.

Author: lobsterkatie

src/sentry/issues/occurrence_consumer.py

@@ -109,51 +109,6 @@ def lookup_event(project_id: int, event_id: str) -> Event:
     return event
 
 
-@trace
-def create_event(project_id: int, event_id: str, event_data: dict[str, Any]) -> Event:
-    return Event(
-        event_id=event_id,
-        project_id=project_id,
-        # `snuba_data` only backs the `Event` property accessors. The eventstream serializes
-        # `event.data`, where the `generic-events` schema requires `received`.
-        data={"received": event_data["received"]},
-        snuba_data={
-            "event_id": event_data["event_id"],
-            "project_id": event_data["project_id"],
-            "timestamp": event_data["timestamp"],
-            "release": event_data.get("release"),
-            "environment": event_data.get("environment"),
-            "platform": event_data.get("platform"),
-            "tags.key": [tag[0] for tag in event_data.get("tags") or []],
-            "tags.value": [tag[1] for tag in event_data.get("tags") or []],
-        },
-    )
-
-
-@trace
-def create_event_and_issue_occurrence(
-    occurrence_data: IssueOccurrenceData, event_data: dict[str, Any]
-) -> tuple[IssueOccurrence, GroupInfo | None]:
-    """With standalone span ingestion, we won't be storing events in
-    nodestore, so instead we create a light-weight event with a small
-    set of fields that lets us create occurrences.
-    """
-    project_id = occurrence_data["project_id"]
-    event_id = occurrence_data["event_id"]
-    if occurrence_data["event_id"] != event_data["event_id"]:
-        raise ValueError(
-            f"event_id in occurrence({occurrence_data['event_id']}) is different from event_id in event_data({event_data['event_id']})"
-        )
-
-    event = create_event(project_id, event_id, event_data)
-
-    with metrics.timer(
-        "occurrence_consumer._process_message.save_issue_occurrence",
-        tags={"method": "create_event_and_issue_occurrence"},
-    ):
-        return save_issue_occurrence(occurrence_data, event)
-
-
 @trace
 def process_event_and_issue_occurrence(
     occurrence_data: IssueOccurrenceData, event_data: dict[str, Any]
@@ -278,6 +233,7 @@ def _get_kwargs(payload: Mapping[str, Any]) -> Mapping[str, Any]:
                     "request",
                     "sdk",
                     "server_name",
+                    "spans",
                     "stacktrace",
                     "trace_id",
                     "transaction",
@@ -335,11 +291,7 @@ def _get_kwargs(payload: Mapping[str, Any]) -> Mapping[str, Any]:
                     "title": occurrence_data["issue_title"],
                 }
 
-                return {
-                    "occurrence_data": occurrence_data,
-                    "event_data": event_data,
-                    "is_buffered_spans": payload.get("is_buffered_spans") is True,
-                }
+                return {"occurrence_data": occurrence_data, "event_data": event_data}
             else:
                 if not payload.get("event_id"):
                     raise InvalidEventPayloadError(
@@ -363,8 +315,6 @@ def process_occurrence_message(
         kwargs = _get_kwargs(message)
     occurrence_data = kwargs["occurrence_data"]
     metric_tags = {"occurrence_type": occurrence_data["type"]}
-    is_buffered_spans = kwargs.get("is_buffered_spans", False)
-
     metrics.incr(
         "occurrence_ingest.messages",
         sample_rate=1.0,
@@ -399,9 +349,7 @@ def process_occurrence_message(
         set_span_tag(span, "result", "dropped_rate_limited")
         return None
 
-    if "event_data" in kwargs and is_buffered_spans:
-        return create_event_and_issue_occurrence(kwargs["occurrence_data"], kwargs["event_data"])
-    elif "event_data" in kwargs:
+    if "event_data" in kwargs:
         set_span_tag(span, "result", "success")
         with metrics.timer(
             "occurrence_consumer._process_message.process_event_and_issue_occurrence",

src/sentry/issues/producer.py

@@ -55,10 +55,9 @@ def produce_occurrence_to_kafka(
     occurrence: IssueOccurrence | None = None,
     status_change: StatusChangeMessage | None = None,
     event_data: dict[str, Any] | None = None,
-    is_buffered_spans: bool | None = False,
 ) -> None:
     if payload_type == PayloadType.OCCURRENCE:
-        payload_data = _prepare_occurrence_message(occurrence, event_data, is_buffered_spans)
+        payload_data = _prepare_occurrence_message(occurrence, event_data)
     elif payload_type == PayloadType.STATUS_CHANGE:
         payload_data = _prepare_status_change_message(status_change)
     else:
@@ -96,7 +95,6 @@ def produce_occurrence_to_kafka(
 def _prepare_occurrence_message(
     occurrence: IssueOccurrence | None,
     event_data: dict[str, Any] | None,
-    is_buffered_spans: bool | None = False,
 ) -> MutableMapping[str, Any] | None:
     if not occurrence:
         raise ValueError("occurrence must be provided")
@@ -131,9 +129,6 @@ def _prepare_occurrence_message(
 
         payload_data["event"] = event_data
 
-    if is_buffered_spans:
-        payload_data["is_buffered_spans"] = True
-
     return payload_data
 
 

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

@@ -34,7 +34,7 @@
     get_detection_settings,
 )
 from sentry.issue_detection.performance_problem import PerformanceProblem
-from sentry.issues.issue_occurrence import IssueOccurrence
+from sentry.issues.issue_occurrence import IssueEvidence, IssueOccurrence
 from sentry.issues.producer import PayloadType, produce_occurrence_to_kafka
 from sentry.killswitches import killswitch_matches_context
 from sentry.models.environment import Environment
@@ -62,6 +62,31 @@
 
 outcome_aggregator = OutcomeAggregator()
 
+# This set of constants helps us keep occurrence data within the occurrence consumer's limits.
+# Limits on span count are split into overall and context (parent and cause) span limits in order to
+# prioritize offender spans, because they're the ones containing the actual problems we report.
+OVERALL_MAX_EVIDENCE_SPANS = 100
+MAX_EVIDENCE_CONTEXT_SPANS = 10
+MAX_SPAN_DESCRIPTION_LENGTH = 2048
+MAX_SPAN_DATA_VALUE_LENGTH = 500
+MAX_EVIDENCE_VALUE_LENGTH = 500
+MAX_EVIDENCE_LIST_ITEMS = 100
+
+# The `evidence_data` values we actually use for issue details and Seer - all others are dropped
+# when we produce the occurrence
+EVIDENCE_SPAN_DATA_KEYS = frozenset(
+    (
+        "code.filepath",
+        "code.function",
+        "code.lineno",
+        "http.query",
+        "http.request.request_start",
+        "http.request.response_start",
+        "http.response_content_length",
+        "url",
+    )
+)
+
 
 @metrics.wraps("spans.consumers.process_segments.process_segment")
 def process_segment(
@@ -422,39 +447,153 @@ def _run_legacy_detectors(
     ):
         return detected_problems
 
-    # Prepare a slimmer event payload for the occurrence consumer. This event will be persisted
-    # by the consumer. Once issue detectors can run on standalone spans, we should directly
-    # build a minimal occurrence event payload here, instead.
-    event_data["spans"] = []
-    event_data["timestamp"] = event_data["datetime"]
-
+    # Produce an occurrence for each problem, first filtering and trimming data to stay within the
+    # occurrence consumer's limits
+    spans_by_id = {span["span_id"]: span for span in event_data["spans"]}
     for problem in detected_problems:
+        evidence_display = [
+            IssueEvidence(
+                evidence.name,
+                _truncate_value_for_occurrence(evidence.value, MAX_EVIDENCE_VALUE_LENGTH),
+                evidence.important,
+            )
+            for evidence in problem.evidence_display
+        ]
+        evidence_data = _get_evidence_data_for_occurrence(problem, spans_by_id)
+
+        occurrence_id = uuid.uuid4().hex
+        occurrence_spans = [
+            _get_evidence_span_for_occurrence(spans_by_id[id])
+            # We use `dict.fromkeys` here to preserve ordering
+            for id in dict.fromkeys(
+                (
+                    *evidence_data["parent_span_ids"],
+                    *evidence_data["cause_span_ids"],
+                    *evidence_data["offender_span_ids"],
+                )
+            )
+        ]
+        occurrence_event_data = {
+            **event_data,
+            "event_id": occurrence_id,
+            "spans": occurrence_spans,
+        }
+
         occurrence = IssueOccurrence(
-            id=uuid.uuid4().hex,
+            id=occurrence_id,
             resource_id=None,
             project_id=project.id,
-            event_id=event_data["event_id"],
+            event_id=occurrence_id,
             fingerprint=[problem.fingerprint],
             type=problem.type,
             issue_title=problem.title,
-            subtitle=problem.desc,
+            subtitle=_truncate_value_for_occurrence(problem.desc, MAX_EVIDENCE_VALUE_LENGTH),
             culprit=event_data["transaction"],
-            evidence_data=problem.evidence_data or {},
-            evidence_display=problem.evidence_display,
+            evidence_data=evidence_data,
+            evidence_display=evidence_display,
             detection_time=to_datetime(segment_span["end_timestamp"]),
             level="info",
         )
 
         produce_occurrence_to_kafka(
             payload_type=PayloadType.OCCURRENCE,
             occurrence=occurrence,
-            event_data=event_data,
-            is_buffered_spans=True,
+            event_data=occurrence_event_data,
         )
 
     return detected_problems
 
 
+def _truncate_span_id_list(
+    raw_span_ids: Sequence[str], spans_by_id: dict[str, Any], max_span_ids: int
+) -> list[str]:
+    """
+    Drop ids the segment has no span for (so performance problem evidence never references a missing
+    span), and then truncate the list of ids to the given max length.
+    """
+    ids_for_existing_spans = [span_id for span_id in raw_span_ids if span_id in spans_by_id]
+    return ids_for_existing_spans[:max_span_ids]
+
+
+def _get_evidence_data_for_occurrence(
+    problem: PerformanceProblem, spans_by_id: dict[str, Any]
+) -> dict[str, Any]:
+    # For the three lists of span ids, remove any which points to spans we don't have and then cap
+    # the list length. The three lists can overlap, so the cap on offender span ids may be more
+    # conservative than necessary, but better that than create an occurrence which gets rejected.
+    parent_span_ids = _truncate_span_id_list(
+        problem.parent_span_ids, spans_by_id, MAX_EVIDENCE_CONTEXT_SPANS
+    )
+    cause_span_ids = _truncate_span_id_list(
+        problem.cause_span_ids, spans_by_id, MAX_EVIDENCE_CONTEXT_SPANS
+    )
+    max_offender_spans = OVERALL_MAX_EVIDENCE_SPANS - len(parent_span_ids) - len(cause_span_ids)
+    offender_span_ids = _truncate_span_id_list(
+        problem.offender_span_ids, spans_by_id, max_offender_spans
+    )
+
+    # Now trim the rest of the evidence data
+    evidence_data = problem.evidence_data or {}
+    span_id_keys = {"parent_span_ids", "cause_span_ids", "offender_span_ids"}
+    non_span_id_evidence_data = {
+        key: value for key, value in evidence_data.items() if key not in span_id_keys
+    }
+    trimmed_non_span_id_evidence_data = _truncate_value_for_occurrence(
+        non_span_id_evidence_data, MAX_EVIDENCE_VALUE_LENGTH
+    )
+
+    return {
+        **trimmed_non_span_id_evidence_data,
+        "parent_span_ids": parent_span_ids,
+        "cause_span_ids": cause_span_ids,
+        "offender_span_ids": offender_span_ids,
+    }
+
+
+def _truncate_value_for_occurrence(
+    value: Any, max_chars: int, max_items: int = MAX_EVIDENCE_LIST_ITEMS
+) -> Any:
+    """
+    Create a recursively-trimmed copy of a value for use in issue occurrences.
+    """
+    if isinstance(value, str):
+        return value[:max_chars]
+    if isinstance(value, list | tuple):
+        return [
+            _truncate_value_for_occurrence(inner_value, max_chars, max_items)
+            for inner_value in value[:max_items]
+        ]
+    if isinstance(value, dict):
+        return {
+            key: _truncate_value_for_occurrence(inner_value, max_chars, max_items)
+            for key, inner_value in value.items()
+        }
+    return value
+
+
+def _get_evidence_span_for_occurrence(span: dict[str, Any]) -> dict[str, Any]:
+    trimmed_description = _truncate_value_for_occurrence(
+        span.get("description"), MAX_SPAN_DESCRIPTION_LENGTH
+    )
+    # Only keep the `data` entries we actually need for evidence data
+    filtered_data = {
+        key: _truncate_value_for_occurrence(value, MAX_SPAN_DATA_VALUE_LENGTH)
+        for key, value in (span.get("data") or {}).items()
+        if key in EVIDENCE_SPAN_DATA_KEYS
+    }
+
+    return {
+        "span_id": span["span_id"],
+        "trace_id": span.get("trace_id"),
+        "op": span.get("op"),
+        "description": trimmed_description,
+        "start_timestamp": span.get("start_timestamp"),
+        "timestamp": span.get("timestamp"),
+        "exclusive_time": span.get("exclusive_time"),
+        "data": filtered_data,
+    }
+
+
 def _maybe_run_span_first_detector_parity_check(
     segment_span: CompatibleSpan,
     segment: list[CompatibleSpan],

tests/sentry/issues/test_occurrence_consumer.py

@@ -24,7 +24,6 @@
 )
 from sentry.issues.producer import _prepare_status_change_message
 from sentry.issues.status_change_message import StatusChangeMessage
-from sentry.models.environment import Environment
 from sentry.models.group import Group, GroupStatus
 from sentry.models.groupassignee import GroupAssignee
 from sentry.ratelimits.sliding_windows import Quota
@@ -310,51 +309,6 @@ def test_occurrence_rate_limit_quota(self) -> None:
         assert rate_limit_quota.limit == 1000
 
 
-class IssueOccurrenceBufferedSpansTest(IssueOccurrenceTestBase):
-    """Occurrences from the span segment processor never get saved to nodestore, so the eventstream
-    payload has to be built entirely from the event data on the message.
-    """
-
-    def setUp(self) -> None:
-        super().setUp()
-        # Nothing on this path saves an event, so the default environment is never created for us
-        Environment.get_or_create(self.project, None)
-
-    def run_buffered_spans_message(self, **event_overrides: Any) -> dict[str, Any]:
-        message = get_test_message(self.project.id, is_buffered_spans=True)
-        message["event"].update(event_overrides)
-
-        with mock.patch("sentry.issues.ingest.eventstream.backend.insert") as mock_insert:
-            with self.feature("organizations:profile-file-io-main-thread-ingest"):
-                result = _process_message(message)
-
-        assert result is not None
-        assert mock_insert.call_count == 1
-        group_event = mock_insert.call_args.kwargs["event"]
-        return dict(group_event.get_raw_data(for_stream=True))
-
-    @django_db_all
-    def test_received_reaches_the_eventstream(self) -> None:
-        received = before_now(minutes=1)
-        stream_data = self.run_buffered_spans_message(received=received.isoformat())
-
-        assert stream_data["received"] == pytest.approx(received.timestamp())
-
-    @django_db_all
-    def test_received_reaches_the_eventstream_when_sent_as_a_timestamp(self) -> None:
-        received = before_now(minutes=1).timestamp()
-        stream_data = self.run_buffered_spans_message(received=received)
-
-        assert stream_data["received"] == pytest.approx(received)
-
-    @django_db_all
-    @freeze_time()
-    def test_null_received_falls_back_to_now(self) -> None:
-        stream_data = self.run_buffered_spans_message(received=None)
-
-        assert stream_data["received"] == pytest.approx(datetime.datetime.now().timestamp())
-
-
 class IssueOccurrenceLookupEventIdTest(IssueOccurrenceTestBase):
     def test_lookup_event_doesnt_exist(self) -> None:
         message = get_test_message(self.project.id, include_event=False)