ref(reprocessing): optionally write unprocessed payload to nodestore instead of redis

src/sentry/event_manager.py

@@ -125,7 +125,7 @@
 from sentry.receivers.features import record_event_processed
 from sentry.receivers.onboarding import record_release_received
 from sentry.releases.auto_creation import should_auto_create_releases
-from sentry.reprocessing2 import is_reprocessed_event
+from sentry.reprocessing2 import delete_unprocessed_event, is_reprocessed_event
 from sentry.seer.signed_seer_api import SeerViewerContext, make_signed_seer_api_request
 from sentry.services.eventstore.processing import event_processing_store
 from sentry.signals import (
@@ -555,6 +555,9 @@ def save_error_events(
             raise
 
         if not group_info:
+            # Returning here skips the nodestore write below, so the event body never
+            # lands and its unprocessed copy can no longer be reached by reprocessing.
+            delete_unprocessed_event(job["event"].project_id, job["event"].event_id)
             return job["event"]
 
         # store a reference to the group id to guarantee validation of isolation
@@ -1087,7 +1090,8 @@ def _nodestore_save_many(jobs: Sequence[Job], app_feature: str) -> None:
         subkeys = {}
 
         event = job["event"]
-        # We only care about `unprocessed` for error events
+        # We only care about `unprocessed` for error events. Events whose unprocessed
+        # copy went straight to nodestore have nothing here and need no subkey.
         if event.get_event_type() not in ("transaction", "generic") and job["groups"]:
             unprocessed = event_processing_store.get(
                 cache_key_for_event({"project": event.project_id, "event_id": event.event_id}),

src/sentry/options/defaults.py

@@ -1083,6 +1083,16 @@
 # Killswitch to stop storing any reprocessing payloads.
 register("store.reprocessing-force-disable", default=False, flags=FLAG_AUTOMATOR_MODIFIABLE)
 
+# Rollout for writing the unprocessed copy of an event straight to nodestore during
+# preprocessing, instead of parking it in the processing store for event manager to
+# promote to a subkey at save time.
+register(
+    "store.reprocessing-nodestore-backup.rollout",
+    type=Float,
+    default=0.0,
+    flags=FLAG_MODIFIABLE_RATE | FLAG_AUTOMATOR_MODIFIABLE,
+)
+
 register(
     "store.ingest-events-raw-task.inline-save-event",
     type=Bool,

src/sentry/reprocessing2.py

@@ -162,16 +162,33 @@ def __init__(self, reason: CannotReprocessReason):
         Exception.__init__(self, reason)
 
 
+def unprocessed_node_id(project_id: int, event_id: str) -> str:
+    """
+    Nodestore id holding the unprocessed copy of an event.
+    """
+    return Event.generate_node_id(project_id, event_id) + ":u"
+
+
 def backup_unprocessed_event(data: Mapping[str, Any]) -> None:
     """
-    Backup unprocessed event payload into redis. Only call if event should be
-    able to be reprocessed.
+    Backup unprocessed event payload. Only call if event should be able to be
+    reprocessed.
     """
 
     if options.get("store.reprocessing-force-disable"):
         return
 
-    event_processing_store.store(dict(data), unprocessed=True)
+    if in_random_rollout("store.reprocessing-nodestore-backup.rollout"):
+        nodestore.backend.set(unprocessed_node_id(data["project"], data["event_id"]), dict(data))
+    else:
+        event_processing_store.store(dict(data), unprocessed=True)
+
+
+def delete_unprocessed_event(project_id: int, event_id: str) -> None:
+    """
+    Drop the unprocessed copy of an event that will never be saved.
+    """
+    nodestore.backend.delete(unprocessed_node_id(project_id, event_id))
 
 
 @dataclass
@@ -191,8 +208,14 @@ def pull_event_data(project_id: int, event_id: str) -> ReprocessableEvent:
         raise CannotReprocess("event.not_found")
 
     with start_span(op="reprocess_events.nodestore.get", name="reprocess_events.nodestore.get"):
+        # Events that went through the processing store carry their unprocessed copy as a
+        # subkey of the node.
         node_id = Event.generate_node_id(project_id, event_id)
         data = nodestore.backend.get(node_id, subkey="unprocessed")
+        if data is None:
+            # If the data isn't there as a subkey, check whether it was saved to nodestore
+            # as its own node.
+            data = nodestore.backend.get(unprocessed_node_id(project_id, event_id))
 
     # Check data after checking presence of event to avoid too many instances.
     if data is None:

src/sentry/tasks/store.py

@@ -579,6 +579,10 @@ def _do_save_event(
             metrics.incr(
                 "events.failed", tags={"reason": "cache", "stage": "post"}, skip_internal=False
             )
+            # Same as the discard case below: the event will never reach nodestore, so
+            # its unprocessed copy is unreachable. This task does not retry.
+            if event_id and project_id:
+                reprocessing2.delete_unprocessed_event(project_id, event_id)
             return
 
         all_attachments = []
@@ -634,6 +638,12 @@ def _do_save_event(
             if cache_key:
                 processing_store.delete_by_key(cache_key)
 
+            # Clean up the unprocessed copy from nodestore as it won't go through
+            # reprocessing. No-op if the unprocessed copy was stored in rc-processing as
+            # it will expire on its own.
+            if event_id:
+                reprocessing2.delete_unprocessed_event(project_id, event_id)
+
             # Mark all the attachments as `rate_limited`, so they are being properly cleaned up in the `finally` block:
             for attachment in all_attachments:
                 attachment.rate_limited = True