ref(replay): move summary log parsing to a seer rpc (#99547)

Author: aliu39

src/sentry/features/temporary.py

@@ -348,6 +348,8 @@ def register_temporary_features(manager: FeatureManager) -> None:
     manager.add("organizations:release-comparison-performance", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=True)
     # Enable replay AI summaries
     manager.add("organizations:replay-ai-summaries", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=True)
+    # Enable replay summary log parsing via Seer RPC
+    manager.add("organizations:replay-ai-summaries-rpc", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE)
     # Enable replay list selection
     manager.add("organizations:replay-list-select", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=True)
     # Enable version 2 of release serializer

src/sentry/replays/blueprints/api.md

@@ -581,13 +581,7 @@ A POST request is issued with no body. The URL and authorization context is used
 
 - Response 204
 
-## Replay Summarize [/projects/<organization_id_or_slug>/<project_id_or_slug>/replays/<replay_id>/summarize/]
-
-- Parameters
-  - start (optional, string) - ISO 8601 format (`YYYY-MM-DDTHH:mm:ss.sssZ`).
-  - end (optional, string) - ISO 8601 format. Required if `start` is set.
-
-`start` and `end` default to the last 90 days. If the replay is not found in the specified time range, this endpoint will 404.
+## Replay Summary [/projects/<organization_id_or_slug>/<project_id_or_slug>/replays/<replay_id>/summarize/]
 
 ### Fetch Replay Summary Task State [GET]
 

src/sentry/replays/endpoints/project_replay_summary.py

@@ -12,6 +12,7 @@
 from sentry.api.api_publish_status import ApiPublishStatus
 from sentry.api.base import region_silo_endpoint
 from sentry.api.bases.project import ProjectEndpoint, ProjectPermission
+from sentry.api.utils import default_start_end_dates
 from sentry.models.project import Project
 from sentry.replays.lib.seer_api import seer_summarization_connection_pool
 from sentry.replays.lib.storage import storage
@@ -178,6 +179,37 @@ def post(self, request: Request, project: Project, replay_id: str) -> Response:
                 )
                 num_segments = MAX_SEGMENTS_TO_SUMMARIZE
 
+            if features.has(
+                "organizations:replay-ai-summaries-rpc", project.organization, actor=request.user
+            ):
+                start, end = default_start_end_dates()
+                snuba_response = query_replay_instance(
+                    project_id=project.id,
+                    replay_id=replay_id,
+                    start=start,
+                    end=end,
+                    organization=project.organization,
+                    request_user_id=request.user.id,
+                )
+                if not snuba_response:
+                    return self.respond(
+                        {"detail": "Replay not found."},
+                        status=404,
+                    )
+
+                return self.make_seer_request(
+                    SEER_START_TASK_ENDPOINT_PATH,
+                    {
+                        "logs": [],
+                        "use_rpc": True,
+                        "num_segments": num_segments,
+                        "replay_id": replay_id,
+                        "organization_id": project.organization.id,
+                        "project_id": project.id,
+                        "temperature": temperature,
+                    },
+                )
+
             # Fetch the replay's error and trace IDs from the replay_id.
             snuba_response = query_replay_instance(
                 project_id=project.id,

src/sentry/replays/usecases/summarize.py

@@ -7,19 +7,22 @@
 import sentry_sdk
 
 from sentry import nodestore
+from sentry.api.utils import default_start_end_dates
 from sentry.constants import ObjectStatus
 from sentry.issues.grouptype import FeedbackGroup
 from sentry.models.project import Project
-from sentry.replays.query import query_trace_connected_events
+from sentry.replays.post_process import process_raw_response
+from sentry.replays.query import query_replay_instance, query_trace_connected_events
 from sentry.replays.usecases.ingest.event_parser import EventType
 from sentry.replays.usecases.ingest.event_parser import (
     get_timestamp_ms as get_replay_event_timestamp_ms,
 )
 from sentry.replays.usecases.ingest.event_parser import parse_network_content_lengths, which
+from sentry.replays.usecases.reader import fetch_segments_metadata, iter_segment_data
 from sentry.search.events.types import SnubaParams
 from sentry.services.eventstore.models import Event
 from sentry.snuba.referrer import Referrer
-from sentry.utils import json
+from sentry.utils import json, metrics
 
 logger = logging.getLogger(__name__)
 
@@ -449,3 +452,78 @@ def _parse_url(s: str, trunc_length: int) -> str:
     if len(s) > trunc_length:
         return s[:trunc_length] + " [truncated]"
     return s
+
+
+def rpc_get_replay_summary_logs(
+    project_id: int, replay_id: str, num_segments: int
+) -> dict[str, Any]:
+    """
+    RPC call for Seer. Downloads a replay's segment data, queries associated errors, and parses this into summary logs.
+    """
+
+    project = Project.objects.get(id=project_id)
+    # Last 90 days. We don't support date filters in /summarize/.
+    start, end = default_start_end_dates()
+
+    # Fetch the replay's error and trace IDs from the replay_id.
+    snuba_response = query_replay_instance(
+        project_id=project.id,
+        replay_id=replay_id,
+        start=start,
+        end=end,
+        organization=project.organization,
+        request_user_id=None,  # This is for the viewed_by_me field which is unused for summaries.
+    )
+    processed_response = process_raw_response(
+        snuba_response,
+        fields=[],  # Defaults to all fields.
+    )
+
+    # 404s should be handled in the originating Sentry endpoint.
+    # If the replay is missing here just return an empty response.
+    if not processed_response:
+        return {"logs": []}
+
+    error_ids = processed_response[0].get("error_ids", [])
+    trace_ids = processed_response[0].get("trace_ids", [])
+
+    # Fetch same-trace errors.
+    trace_connected_errors = fetch_trace_connected_errors(
+        project=project,
+        trace_ids=trace_ids,
+        start=start,
+        end=end,
+        limit=100,
+    )
+    trace_connected_error_ids = {x["id"] for x in trace_connected_errors}
+
+    # Fetch directly linked errors, if they weren't returned by the trace query.
+    direct_errors = fetch_error_details(
+        project_id=project.id,
+        error_ids=[x for x in error_ids if x not in trace_connected_error_ids],
+    )
+
+    error_events = direct_errors + trace_connected_errors
+
+    # Metric names kept for backwards compatibility.
+    metrics.distribution(
+        "replays.endpoints.project_replay_summary.direct_errors",
+        value=len(direct_errors),
+    )
+    metrics.distribution(
+        "replays.endpoints.project_replay_summary.trace_connected_errors",
+        value=len(trace_connected_errors),
+    )
+    metrics.distribution(
+        "replays.endpoints.project_replay_summary.num_trace_ids",
+        value=len(trace_ids),
+    )
+
+    # Download segment data.
+    segment_md = fetch_segments_metadata(project.id, replay_id, 0, num_segments)
+    segment_data = iter_segment_data(segment_md)
+
+    # Combine replay and error data and parse into logs.
+    logs = get_summary_logs(segment_data, error_events, project.id)
+
+    return {"logs": logs}

src/sentry/seer/endpoints/seer_rpc.py

@@ -55,6 +55,7 @@
 from sentry.integrations.types import IntegrationProviderSlug
 from sentry.models.organization import Organization, OrganizationStatus
 from sentry.models.repository import Repository
+from sentry.replays.usecases.summarize import rpc_get_replay_summary_logs
 from sentry.search.eap.resolver import SearchResolver
 from sentry.search.eap.spans.definitions import SPAN_DEFINITIONS
 from sentry.search.eap.types import SearchResolverConfig, SupportedTraceItemType
@@ -931,6 +932,9 @@ def send_seer_webhook(*, event_name: str, organization_id: int, payload: dict) -
     "get_trace_for_transaction": rpc_get_trace_for_transaction,
     "get_profiles_for_trace": rpc_get_profiles_for_trace,
     "get_issues_for_transaction": rpc_get_issues_for_transaction,
+    #
+    # Replays
+    "get_replay_summary_logs": rpc_get_replay_summary_logs,
 }