feat(native): Symbolicate flamegraph attachments

fixtures/stubs-for-mypy/rapidjson.pyi

@@ -2,5 +2,5 @@ from typing import Any
 
 class JSONDecodeError(ValueError): ...
 
-def dumps(o: object) -> str: ...
-def loads(s: str | bytes) -> Any: ...
+def dumps(o: object, *, ensure_ascii: bool = True, allow_nan: bool = True) -> str: ...
+def loads(s: str | bytes, *, allow_nan: bool = True) -> Any: ...

src/sentry/features/temporary.py

@@ -446,6 +446,8 @@ def register_temporary_features(manager: FeatureManager) -> None:
     manager.add("organizations:unified-table-controls", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=True)
     # Enables view hierarchy attachment scrubbing
     manager.add("organizations:view-hierarchy-scrubbing", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=False)
+    # Enables symbolication for flamegraph attachments
+    manager.add("organizations:flamegraph-attachments", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=True)
     # Enable AI-powered assertion suggestions for uptime monitors
     manager.add("organizations:uptime-ai-assertion-suggestions", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=True)
     # Enable auto spam classification at User Feedback ingest time

src/sentry/ingest/consumer/__init__.py

@@ -0,0 +1 @@
+CACHE_TIMEOUT = 3600

src/sentry/ingest/consumer/processors.py

@@ -18,6 +18,7 @@
 from sentry.event_manager import save_attachment
 from sentry.feedback.lib.utils import FeedbackCreationSource, is_in_feedback_denylist
 from sentry.feedback.usecases.ingest.userreport import Conflict, save_userreport
+from sentry.ingest.consumer import CACHE_TIMEOUT
 from sentry.ingest.event_payload import get_event_payload_transport
 from sentry.ingest.types import ConsumerType
 from sentry.killswitches import killswitch_matches_context
@@ -41,8 +42,6 @@
 
 logger = logging.getLogger(__name__)
 
-CACHE_TIMEOUT = 3600
-
 IngestMessage = Mapping[str, Any]
 
 

src/sentry/lang/java/view_hierarchies.py

@@ -7,7 +7,7 @@
     get_attachments_for_event,
     store_attachments_for_event,
 )
-from sentry.ingest.consumer.processors import CACHE_TIMEOUT
+from sentry.ingest.consumer import CACHE_TIMEOUT
 from sentry.models.project import Project
 
 

src/sentry/lang/native/flamegraphs.py

@@ -0,0 +1,159 @@
+from collections.abc import Generator
+from typing import Any
+
+import rapidjson
+
+from sentry import features
+from sentry.attachments import (
+    CachedAttachment,
+    MissingAttachmentChunks,
+    get_attachments_for_event,
+    store_attachments_for_event,
+)
+from sentry.ingest.consumer import CACHE_TIMEOUT
+from sentry.models.project import Project
+
+# See: https://develop.sentry.dev/sdk/telemetry/attachments/#flame-graph-validation-and-limits
+MAX_FLAMEGRAPH_SIZE = 10 * 1024 * 1024
+MAX_SYMBOLICATION_FRAMES = 2500
+MAX_SAMPLE_COUNT = 9007199254740991
+
+
+def _is_valid_flamegraph(payload: Any) -> bool:
+    if not isinstance(payload, dict) or payload.get("version") != "1":
+        return False
+    if not isinstance(payload.get("platform"), str):
+        return False
+    frames = payload.get("frames")
+    trees = payload.get("trees")
+    if not isinstance(frames, list) or not frames or not isinstance(trees, list) or not trees:
+        return False
+    for frame in frames:
+        if not isinstance(frame, dict) or not any(
+            isinstance(frame.get(field), str) and frame[field]
+            for field in ("filename", "function", "instruction_addr")
+        ):
+            return False
+        if "instruction_addr" in frame and not isinstance(frame["instruction_addr"], str):
+            return False
+        if "addr_mode" in frame and not isinstance(frame["addr_mode"], str):
+            return False
+        if "platform" in frame and not isinstance(frame["platform"], str):
+            return False
+        if "data" in frame and not isinstance(frame["data"], dict):
+            return False
+    debug_meta = payload.get("debug_meta", {})
+    if not isinstance(debug_meta, dict):
+        return False
+    images = debug_meta.get("images", [])
+    if not isinstance(images, list) or not all(isinstance(image, dict) for image in images):
+        return False
+    for image in images:
+        if any(
+            image.get(field) is not None and not isinstance(image[field], str)
+            for field in ("debug_id", "id", "uuid")
+        ):
+            return False
+    for tree in trees:
+        if (
+            not isinstance(tree, dict)
+            or not isinstance(tree.get("roots"), list)
+            or not tree["roots"]
+        ):
+            return False
+        nodes = list(tree["roots"])
+        while nodes:
+            node = nodes.pop()
+            if not isinstance(node, dict):
+                return False
+            frame_id = node.get("frame_id")
+            count = node.get("sample_count")
+            if type(frame_id) is not int or not 0 <= frame_id < len(frames):
+                return False
+            if type(count) is not int or not 0 < count <= MAX_SAMPLE_COUNT:
+                return False
+            children = node.get("children", [])
+            if not isinstance(children, list) or not all(
+                isinstance(child, dict) for child in children
+            ):
+                return False
+            child_counts = [child.get("sample_count") for child in children]
+            if any(type(value) is not int or value <= 0 for value in child_counts):
+                return False
+            if sum(child_counts) > count:
+                return False
+            nodes.extend(children)
+    return True
+
+
+def expand_inline_frames(
+    trees: list[dict[str, Any]], chains: dict[int, list[int]]
+) -> list[dict[str, Any]]:
+    output = [{**tree, "roots": []} for tree in trees]
+    nodes = [
+        (root, new_tree["roots"])
+        for tree, new_tree in zip(trees, output)
+        for root in reversed(tree["roots"])
+    ]
+    while nodes:
+        original, siblings = nodes.pop()
+        node = dict(original)
+        siblings.append(node)
+        children = node.pop("children", None)
+        current = node
+        for frame_id in chains.get(node["frame_id"], [])[1:]:
+            child = {"frame_id": frame_id, "sample_count": node["sample_count"]}
+            current["children"] = [child]
+            current = child
+        if children is not None:
+            current["children"] = []
+            nodes.extend((child, current["children"]) for child in reversed(children))
+    return output
+
+
+class Flamegraphs:
+    def __init__(self, project: Project, data: Any):
+        self._project = project
+        self._event = data
+        self._attachments = list(get_attachments_for_event(data))
+
+    def get_flamegraphs(self) -> Generator[tuple[int, dict[str, Any]]]:
+        if not any(attachment.type == "event.flamegraph" for attachment in self._attachments):
+            return
+        if not features.has("organizations:flamegraph-attachments", self._project.organization):
+            return
+        for index, attachment in enumerate(self._attachments):
+            if attachment.type != "event.flamegraph":
+                continue
+            try:
+                raw = attachment.load_data(self._project)
+            except (MissingAttachmentChunks, FileNotFoundError):
+                continue
+            if len(raw) > MAX_FLAMEGRAPH_SIZE:
+                continue
+            try:
+                payload = rapidjson.loads(raw, allow_nan=False)
+            except (rapidjson.JSONDecodeError, UnicodeDecodeError):
+                continue
+            if not isinstance(payload, dict):
+                continue
+            state = payload.get("_symbolication")
+            if isinstance(state, dict) and state.get("status") == "completed":
+                continue
+            if _is_valid_flamegraph(payload):
+                yield index, payload
+
+    def save(self, attachment_index: int, payload: dict[str, Any]) -> bool:
+        serialized = rapidjson.dumps(payload, ensure_ascii=False, allow_nan=False).encode()
+        if len(serialized) > MAX_FLAMEGRAPH_SIZE:
+            return False
+        attachment = self._attachments[attachment_index]
+        metadata = attachment.meta()
+        metadata.update(data=serialized, chunks=None, size=len(serialized))
+        # Checkpoint only this attachment, without reading or rewriting unrelated payloads.
+        checkpoint = dict(self._event)
+        store_attachments_for_event(
+            self._project, checkpoint, [CachedAttachment(**metadata)], timeout=CACHE_TIMEOUT
+        )
+        self._event["_attachments"][attachment_index] = checkpoint["_attachments"][0]
+        return True

src/sentry/lang/native/processing.py

@@ -6,11 +6,17 @@
 from typing import Any
 
 import sentry_sdk
+from requests.exceptions import RequestException
 from symbolic.debuginfo import normalize_debug_id
 from symbolic.exceptions import ParseDebugIdError
 
 from sentry import options
 from sentry.lang.native.error import SymbolicationFailed, write_error
+from sentry.lang.native.flamegraphs import (
+    MAX_SYMBOLICATION_FRAMES,
+    Flamegraphs,
+    expand_inline_frames,
+)
 from sentry.lang.native.symbolicator import FrameOrder, Symbolicator, SymbolicatorFunction
 from sentry.lang.native.utils import (
     get_event_attachment,
@@ -20,6 +26,7 @@
     is_applecrashreport_event,
     is_minidump_event,
     is_native_event,
+    is_native_image,
     is_native_platform,
     native_images_from_data,
     signal_from_data,
@@ -447,7 +454,102 @@ def get_frames_for_symbolication(
     return rv
 
 
+def _get_flamegraph_stacktraces(
+    payload: dict[str, Any],
+) -> tuple[list[dict[str, Any]], list[int], list[dict[str, Any]]]:
+    frame_ids = []
+    stacktraces = []
+    original_modules = payload.get("debug_meta", {}).get("images", [])
+    modules: list[dict[str, Any]] = []
+    module_indices: dict[int, int] = {}
+    for index, image in enumerate(original_modules):
+        if is_native_image(image):
+            module_indices[index] = len(modules)
+            modules.append(image)
+    if not modules:
+        return [], [], []
+    for frame_id, frame in enumerate(payload["frames"][:MAX_SYMBOLICATION_FRAMES]):
+        frames = get_frames_for_symbolication([frame], payload, original_modules, adjustment="none")
+        if not frames:
+            continue
+        # Resolve against the captured image table before remapping to the filtered request.
+        addr_mode = frames[0].get("addr_mode")
+        if frame.get("addr_mode", "").startswith("rel:") and addr_mode is None:
+            continue
+        if addr_mode is not None:
+            module_index = module_indices.get(int(addr_mode[4:]))
+            if module_index is None:
+                continue
+            frames[0]["addr_mode"] = f"rel:{module_index}"
+        frame_ids.append(frame_id)
+        stacktraces.append({"frames": frames})
+    return modules, frame_ids, stacktraces
+
+
+def _merge_flamegraph_frames(
+    payload: dict[str, Any], frame_ids: list[int], stacktraces: list[dict[str, Any]]
+) -> dict[str, Any]:
+    output = {**payload, "frames": list(payload["frames"])}
+    chains = {}
+    for frame_id, stacktrace in zip(frame_ids, stacktraces):
+        original = payload["frames"][frame_id]
+        resolved = stacktrace.get("frames", [])
+        if not resolved or any(
+            frame.get("status") not in (None, "symbolicated") for frame in resolved
+        ):
+            continue
+        chain = []
+        for position, frame in enumerate(resolved):
+            merged = dict(original) if position == 0 else {}
+            if "platform" in original:
+                merged["platform"] = original["platform"]
+            if "data" in merged:
+                merged["data"] = dict(merged["data"])
+            _merge_frame(merged, frame, original.get("platform", payload["platform"]))
+            # Symbolicator may adjust addresses or normalize relative addressing modes.
+            # The diagnostic must retain its captured symbolication inputs.
+            merged["instruction_addr"] = original["instruction_addr"]
+            if "addr_mode" in original:
+                merged["addr_mode"] = original["addr_mode"]
+            else:
+                merged.pop("addr_mode", None)
+            if position == 0:
+                output["frames"][frame_id] = merged
+                chain.append(frame_id)
+            else:
+                chain.append(len(output["frames"]))
+                output["frames"].append(merged)
+        chains[frame_id] = chain
+    output["trees"] = expand_inline_frames(payload["trees"], chains)
+    output["_symbolication"] = {"status": "completed"}
+    return output
+
+
 def process_native_stacktraces(symbolicator: Symbolicator, data: Any) -> Any:
+    flamegraphs = Flamegraphs(symbolicator.project, data)
+    attachments_changed = False
+    for attachment_index, payload in flamegraphs.get_flamegraphs():
+        modules, frame_ids, stacktraces = _get_flamegraph_stacktraces(payload)
+        if not stacktraces:
+            continue
+        try:
+            response = symbolicator.process_payload(
+                platform=payload["platform"],
+                stacktraces=stacktraces,
+                modules=modules,
+                frame_order=FrameOrder.caller_first,
+                apply_source_context=False,
+            )
+        except (OSError, RequestException) as error:
+            sentry_sdk.capture_exception(error)
+            continue
+        if not response or response.get("status") != "completed":
+            continue
+        resolved_stacktraces = response.get("stacktraces", [])
+        if len(resolved_stacktraces) != len(frame_ids):
+            continue
+        output = _merge_flamegraph_frames(payload, frame_ids, resolved_stacktraces)
+        attachments_changed |= flamegraphs.save(attachment_index, output)
     stacktrace_infos = [
         stacktrace
         for stacktrace in find_stacktraces_in_data(data)
@@ -470,7 +572,7 @@ def process_native_stacktraces(symbolicator: Symbolicator, data: Any) -> Any:
     ]
 
     if not any(stacktrace["frames"] for stacktrace in stacktraces):
-        return
+        return data if attachments_changed else None
 
     signal = signal_from_data(data)
 
@@ -639,7 +741,9 @@ def get_native_symbolication_functions(
         functions.append(SymbolicatorFunction.native)
         return functions  # early return to prevent duplicates
 
-    if has_native_stacktraces(stacktraces):
+    if has_native_stacktraces(stacktraces) or any(
+        attachment.get("type") == "event.flamegraph" for attachment in data.get("_attachments", [])
+    ):
         # Run native symbolication first, to prevent sending minidump / ACR events to
         # symbolicator twice (after symbolication a minidump event will have native stacktraces).
         functions.insert(0, SymbolicatorFunction.native)

src/sentry/reprocessing2.py

@@ -236,7 +236,7 @@ def pull_event_data(project_id: int, event_id: str) -> ReprocessableEvent:
 
 
 def reprocess_event(project_id: int, event_id: str, start_time: float) -> None:
-    from sentry.ingest.consumer.processors import CACHE_TIMEOUT
+    from sentry.ingest.consumer import CACHE_TIMEOUT
     from sentry.tasks.store import preprocess_event_from_reprocessing
 
     reprocessable_event = pull_event_data(project_id, event_id)

tests/sentry/api/serializers/test_organization.py

@@ -88,6 +88,11 @@ def test_simple(self) -> None:
             "uptime",
         ]
 
+    @with_feature("organizations:flamegraph-attachments")
+    def test_flamegraph_feature_is_exposed(self) -> None:
+        result = serialize(self.organization, self.user)
+        assert "flamegraph-attachments" in result["features"]
+
     @mock.patch("sentry.features.batch_has")
     def test_organization_batch_has(self, mock_batch: mock.MagicMock) -> None:
         user = self.create_user()