feat(occurrences on eap): Implement EAP query for eventstore get_events (#112423)

Implements an EAP query (double read) for the eventstore's
`__get_events` function (used by both `get_events` &
`get_unfetched_events`) in
`src/sentry/services/eventstore/snuba/backend.py`.

Rather than write logic to convert an `eventstore.Filter` value to the
equivalent format needed for an EAP query, I'm adding logic here to
create `eap_conditions` at the individual callsites.

Author: shashjar

src/sentry/api/serializers/models/userreport.py

@@ -8,6 +8,7 @@
 from sentry.models.group import Group
 from sentry.models.project import Project
 from sentry.models.userreport import UserReport
+from sentry.search.eap.occurrences.query_utils import build_event_id_in_filter
 from sentry.services import eventstore
 from sentry.services.eventstore.models import Event
 from sentry.snuba.dataset import Dataset
@@ -46,12 +47,14 @@ def get_attrs(self, item_list, user, **kwargs):
         project = Project.objects.get(id=item_list[0].project_id)
         retention = quotas.backend.get_event_retention(organization=project.organization)
 
+        event_ids = [item.event_id for item in item_list]
         events = eventstore.backend.get_events(
             filter=eventstore.Filter(
-                event_ids=[item.event_id for item in item_list],
+                event_ids=event_ids,
                 project_ids=[project.id],
                 start=timezone.now() - timedelta(days=retention) if retention else None,
             ),
+            eap_conditions=build_event_id_in_filter(event_ids),
             referrer="UserReportSerializer.get_attrs",
             dataset=Dataset.Events,
             tenant_ids={"organization_id": project.organization_id},

src/sentry/deletions/tasks/nodestore.py

@@ -4,6 +4,7 @@
 from typing import Any
 
 import sentry_sdk
+from sentry_protos.snuba.v1.trace_item_filter_pb2 import TraceItemFilter
 from snuba_sdk import DeleteQuery, Request
 from taskbroker_client.retry import Retry
 
@@ -13,6 +14,11 @@
 from sentry.exceptions import DeleteAborted
 from sentry.models.eventattachment import EventAttachment
 from sentry.models.userreport import UserReport
+from sentry.search.eap.occurrences.query_utils import (
+    build_group_id_in_filter,
+    build_keyset_pagination_filter,
+)
+from sentry.search.eap.rpc_utils import and_trace_item_filters
 from sentry.services import eventstore
 from sentry.services.eventstore.models import Event
 from sentry.silo.base import SiloMode
@@ -155,20 +161,29 @@ def fetch_events_from_eventstore(
 ) -> list[Event]:
     logger.info("Fetching %s events for deletion.", limit)
     conditions = []
+    eap_conditions: TraceItemFilter | None = build_group_id_in_filter(group_ids)
     if last_event_id and last_event_timestamp:
         conditions.extend(
             [
                 ["timestamp", "<=", last_event_timestamp],
                 [["timestamp", "<", last_event_timestamp], ["event_id", "<", last_event_id]],
             ]
         )
+        eap_conditions = and_trace_item_filters(
+            eap_conditions,
+            build_keyset_pagination_filter(
+                timestamp_value=last_event_timestamp,
+                event_id=last_event_id,
+            ),
+        )
 
     events = eventstore.backend.get_unfetched_events(
         filter=eventstore.Filter(
             conditions=conditions,
             project_ids=[project_id],
             group_ids=group_ids,
         ),
+        eap_conditions=eap_conditions,
         limit=limit,
         referrer=referrer,
         orderby=["-timestamp", "-event_id"],

src/sentry/feedback/tasks/update_user_reports.py

@@ -9,6 +9,7 @@
 from sentry.feedback.usecases.ingest.shim_to_feedback import shim_to_feedback
 from sentry.models.project import Project
 from sentry.models.userreport import UserReport
+from sentry.search.eap.occurrences.query_utils import build_event_id_in_filter
 from sentry.services import eventstore
 from sentry.silo.base import SiloMode
 from sentry.snuba.referrer import Referrer
@@ -90,7 +91,9 @@ def update_user_reports(
             )
             try:
                 events_chunk = eventstore.backend.get_events(
-                    filter=snuba_filter, referrer=Referrer.TASKS_UPDATE_USER_REPORTS.value
+                    filter=snuba_filter,
+                    eap_conditions=build_event_id_in_filter(event_id_chunk),
+                    referrer=Referrer.TASKS_UPDATE_USER_REPORTS.value,
                 )
                 events.extend(events_chunk)
             except Exception:

src/sentry/issues/endpoints/organization_eventid.py

@@ -3,6 +3,8 @@
 from drf_spectacular.utils import extend_schema
 from rest_framework.request import Request
 from rest_framework.response import Response
+from sentry_protos.snuba.v1.trace_item_attribute_pb2 import AttributeKey, AttributeValue
+from sentry_protos.snuba.v1.trace_item_filter_pb2 import ComparisonFilter, TraceItemFilter
 
 from sentry.api.api_owners import ApiOwner
 from sentry.api.api_publish_status import ApiPublishStatus
@@ -18,6 +20,8 @@
 from sentry.models.organization import Organization
 from sentry.models.project import Project
 from sentry.ratelimits.config import RateLimitConfig
+from sentry.search.eap.occurrences.query_utils import build_event_id_in_filter
+from sentry.search.eap.rpc_utils import and_trace_item_filters
 from sentry.services import eventstore
 from sentry.types.ratelimit import RateLimit, RateLimitCategory
 from sentry.utils.validators import INVALID_ID_DETAILS, is_event_id
@@ -76,8 +80,21 @@ def get(self, request: Request, organization: Organization, event_id: str) -> Re
                 project_ids=list(project_slugs_by_id.keys()),
                 event_ids=[event_id],
             )
+            eap_conditions = and_trace_item_filters(
+                build_event_id_in_filter([event_id]),
+                TraceItemFilter(
+                    comparison_filter=ComparisonFilter(
+                        key=AttributeKey(name="type", type=AttributeKey.TYPE_STRING),
+                        op=ComparisonFilter.OP_NOT_EQUALS,
+                        value=AttributeValue(val_str="transaction"),
+                    )
+                ),
+            )
             event = eventstore.backend.get_events(
-                filter=snuba_filter, limit=1, tenant_ids={"organization_id": organization.id}
+                filter=snuba_filter,
+                eap_conditions=eap_conditions,
+                limit=1,
+                tenant_ids={"organization_id": organization.id},
             )[0]
         except IndexError:
             raise ResourceDoesNotExist()

src/sentry/issues/endpoints/project_events.py

@@ -4,6 +4,7 @@
 from rest_framework.exceptions import ParseError
 from rest_framework.request import Request
 from rest_framework.response import Response
+from sentry_protos.snuba.v1.trace_item_filter_pb2 import TraceItemFilter
 
 from sentry.api.api_owners import ApiOwner
 from sentry.api.api_publish_status import ApiPublishStatus
@@ -95,6 +96,7 @@ def get(self, request: Request, project: Project) -> Response:
         data_fn = partial(
             eventstore.backend.get_events,
             filter=event_filter,
+            eap_conditions=TraceItemFilter(),  # TODO: not currently taking the query into account
             referrer="api.project-events",
             tenant_ids={"organization_id": project.organization_id},
         )

src/sentry/models/group.py

@@ -18,6 +18,8 @@
 from django.utils import timezone
 from django.utils.http import urlencode
 from django.utils.translation import gettext_lazy as _
+from sentry_protos.snuba.v1.trace_item_attribute_pb2 import AttributeKey
+from sentry_protos.snuba.v1.trace_item_filter_pb2 import ExistsFilter, TraceItemFilter
 from snuba_sdk import Column, Condition, Op
 
 from sentry import eventstore, eventtypes, options, tagstore
@@ -44,6 +46,8 @@
 from sentry.models.commit import Commit
 from sentry.models.grouphistory import record_group_history, record_group_history_from_activity_type
 from sentry.models.organization import Organization
+from sentry.search.eap.occurrences.query_utils import build_event_id_in_filter
+from sentry.search.eap.rpc_utils import and_trace_item_filters
 from sentry.services.eventstore.models import GroupEvent
 from sentry.snuba.dataset import Dataset
 from sentry.snuba.referrer import Referrer
@@ -435,6 +439,14 @@ def filter_by_event_id(self, project_ids, event_id, tenant_ids=None):
                 project_ids=project_ids,
                 conditions=[["group_id", "IS NOT NULL", None]],
             ),
+            eap_conditions=and_trace_item_filters(
+                build_event_id_in_filter([event_id]),
+                TraceItemFilter(
+                    exists_filter=ExistsFilter(
+                        key=AttributeKey(name="group_id", type=AttributeKey.TYPE_INT)
+                    )
+                ),
+            ),
             limit=max(len(project_ids), 100),
             referrer="Group.filter_by_event_id",
             tenant_ids=tenant_ids,

src/sentry/search/eap/occurrences/query_utils.py

@@ -2,9 +2,18 @@
 from datetime import datetime
 from typing import Any
 
+from sentry_protos.snuba.v1.trace_item_attribute_pb2 import (
+    AttributeKey,
+    AttributeValue,
+    IntArray,
+    StrArray,
+)
+from sentry_protos.snuba.v1.trace_item_filter_pb2 import ComparisonFilter, TraceItemFilter
+
 from sentry.models.environment import Environment
 from sentry.models.organization import Organization
 from sentry.models.project import Project
+from sentry.search.eap.rpc_utils import and_trace_item_filters, or_trace_item_filters
 from sentry.search.events.types import SnubaParams
 
 
@@ -74,3 +83,56 @@ def _to_count_map(rows: Sequence[Mapping[str, Any]]) -> dict[Hashable, int]:
         return False
 
     return all(exp_count <= control_map[key] for key, exp_count in experimental_map.items())
+
+
+def build_group_id_in_filter(group_ids: Sequence[int]) -> TraceItemFilter:
+    return TraceItemFilter(
+        comparison_filter=ComparisonFilter(
+            key=AttributeKey(name="group_id", type=AttributeKey.TYPE_INT),
+            op=ComparisonFilter.OP_IN,
+            value=AttributeValue(val_int_array=IntArray(values=list(group_ids))),
+        )
+    )
+
+
+def build_event_id_in_filter(event_ids: Sequence[str]) -> TraceItemFilter:
+    return TraceItemFilter(
+        comparison_filter=ComparisonFilter(
+            key=AttributeKey(name="sentry.item_id", type=AttributeKey.TYPE_STRING),
+            op=ComparisonFilter.OP_IN,
+            value=AttributeValue(val_str_array=StrArray(values=list(event_ids))),
+        )
+    )
+
+
+def build_keyset_pagination_filter(
+    timestamp_value: str,
+    event_id: str,
+) -> TraceItemFilter | None:
+    ts_epoch = datetime.fromisoformat(timestamp_value).timestamp()
+    timestamp_key = AttributeKey(name="sentry.timestamp", type=AttributeKey.TYPE_DOUBLE)
+    event_id_key = AttributeKey(name="sentry.item_id", type=AttributeKey.TYPE_STRING)
+
+    ts_lte = TraceItemFilter(
+        comparison_filter=ComparisonFilter(
+            key=timestamp_key,
+            op=ComparisonFilter.OP_LESS_THAN_OR_EQUALS,
+            value=AttributeValue(val_double=ts_epoch),
+        )
+    )
+    ts_lt = TraceItemFilter(
+        comparison_filter=ComparisonFilter(
+            key=timestamp_key,
+            op=ComparisonFilter.OP_LESS_THAN,
+            value=AttributeValue(val_double=ts_epoch),
+        )
+    )
+    eid_lt = TraceItemFilter(
+        comparison_filter=ComparisonFilter(
+            key=event_id_key,
+            op=ComparisonFilter.OP_LESS_THAN,
+            value=AttributeValue(val_str=event_id),
+        )
+    )
+
+    return and_trace_item_filters(ts_lte, or_trace_item_filters(ts_lt, eid_lt))

src/sentry/seer/explorer/tools.py

@@ -32,6 +32,7 @@
 from sentry.replays.post_process import process_raw_response
 from sentry.replays.query import query_replay_id_by_prefix, query_replay_instance
 from sentry.search.eap.constants import BOOLEAN, DOUBLE, INT, STRING
+from sentry.search.eap.occurrences.query_utils import build_event_id_in_filter
 from sentry.search.eap.resolver import SearchResolver
 from sentry.search.eap.types import SearchResolverConfig
 from sentry.search.events.constants import ISSUE_ID_ALIAS
@@ -1255,6 +1256,7 @@ def get_event_details(
                         organization_id=organization_id,
                         project_ids=project_ids,
                     ),
+                    eap_conditions=build_event_id_in_filter([event_id]),
                     limit=1,
                     tenant_ids={"organization_id": organization_id},
                     dataset=dataset,
@@ -1355,6 +1357,7 @@ def get_issue_and_event_details_v2(
                         organization_id=organization_id,
                         project_ids=project_ids,
                     ),
+                    eap_conditions=build_event_id_in_filter([event_id]),
                     limit=1,
                     tenant_ids={"organization_id": organization_id},
                     dataset=dataset,

src/sentry/services/eventstore/base.py

@@ -7,12 +7,14 @@
 from typing import Any, Literal, Self, overload
 
 import sentry_sdk
+from sentry_protos.snuba.v1.trace_item_filter_pb2 import TraceItemFilter
 from snuba_sdk import Condition
 
 from sentry import nodestore
 from sentry.services.eventstore.models import Event, GroupEvent
 from sentry.snuba.dataset import Dataset
 from sentry.snuba.events import Columns
+from sentry.snuba.referrer import Referrer
 from sentry.utils.services import Service
 
 
@@ -166,10 +168,11 @@ class EventStorage(Service):
     def get_events(
         self,
         filter: Filter,
+        eap_conditions: TraceItemFilter | None = None,
         orderby: Sequence[str] | None = None,
         limit: int = 100,
         offset: int = 0,
-        referrer: str = "eventstore.get_events",
+        referrer: str = Referrer.EVENTSTORE_GET_EVENTS.value,
         dataset: Dataset = Dataset.Events,
         tenant_ids: Mapping[str, Any] | None = None,
     ) -> list[Event]:
@@ -180,11 +183,12 @@ def get_events(
         transaction events. Returns an empty list if no events match the filter.
 
         Arguments:
-        snuba_filter (Filter): Filter
+        filter (Filter): Snuba query filter
+        eap_conditions (TraceItemFilter | None): EAP query conditions
         orderby (Sequence[str]): List of fields to order by - default ['-time', '-event_id']
         limit (int): Query limit - default 100
         offset (int): Query offset - default 0
-        referrer (string): Referrer - default "eventstore.get_events"
+        referrer (string): Referrer
         """
         raise NotImplementedError
 
@@ -208,28 +212,30 @@ def get_events_snql(
     def get_unfetched_events(
         self,
         filter: Filter,
+        eap_conditions: TraceItemFilter | None = None,
         orderby: Sequence[str] | None = None,
         limit: int = 100,
         offset: int = 0,
-        referrer: str = "eventstore.get_unfetched_events",
+        referrer: str = Referrer.EVENTSTORE_GET_UNFETCHED_EVENTS.value,
         dataset: Dataset = Dataset.Events,
         tenant_ids: Mapping[str, Any] | None = None,
     ) -> list[Event]:
         """
-        Same as get_events but returns events without their node datas loaded.
-        Only the event ID, projectID, groupID and timestamp field will be present without
-        an additional fetch to nodestore.
+        Same as get_events but returns events without their node data loaded.
+        Only the event ID, project ID, group ID, and timestamp fields will be present
+        without an additional fetch to nodestore.
 
         Used for fetching large volumes of events that do not need data loaded
         from nodestore. Currently this is just used for event data deletions where
         we just need the event IDs in order to process the deletions.
 
         Arguments:
-        snuba_filter (Filter): Filter
+        filter (Filter): Snuba query filter
+        eap_conditions (TraceItemFilter | None): EAP query conditions
         orderby (Sequence[str]): List of fields to order by - default ['-time', '-event_id']
         limit (int): Query limit - default 100
         offset (int): Query offset - default 0
-        referrer (string): Referrer - default "eventstore.get_unfetched_events"
+        referrer (string): Referrer
         """
         raise NotImplementedError