ref(store) Update post_process to use a cache key round 2 (#20620)

This reverts commit 9018680 and
restores post_process to using a cache key instead of an AMQP payload.

I've added additional logic to ensure we don't fail when duplicate
events are processed and we don't have a cache_key, and included tests
for those scenarios. Lastly, I'm always passing cache_key to
post_process now to ensure that the processingstore is cleared when the
new paths are not run as the delete done during tasks/store has been
removed.

Author: markstory

src/sentry/eventstream/base.py

@@ -1,9 +1,12 @@
 from __future__ import absolute_import
 
+import random
 import logging
 
-from sentry.utils.services import Service
+from sentry import options
 from sentry.tasks.post_process import post_process_group
+from sentry.utils.services import Service
+from sentry.utils.cache import cache_key_for_event
 
 
 logger = logging.getLogger(__name__)
@@ -43,13 +46,30 @@ def _dispatch_post_process_group_task(
         if skip_consume:
             logger.info("post_process.skip.raw_event", extra={"event_id": event.event_id})
         else:
-            post_process_group.delay(
-                event=event,
-                is_new=is_new,
-                is_regression=is_regression,
-                is_new_group_environment=is_new_group_environment,
-                primary_hash=primary_hash,
+            random_val = random.random()
+            cache_key = cache_key_for_event(
+                {"project": event.project_id, "event_id": event.event_id}
             )
+            if options.get("postprocess.use-cache-key") > random_val:
+                post_process_group.delay(
+                    event=None,
+                    is_new=is_new,
+                    is_regression=is_regression,
+                    is_new_group_environment=is_new_group_environment,
+                    primary_hash=primary_hash,
+                    cache_key=cache_key,
+                    group_id=event.group_id,
+                )
+            else:
+                # Pass the cache key here to ensure that the processing cache is removed.
+                post_process_group.delay(
+                    event=event,
+                    is_new=is_new,
+                    is_regression=is_regression,
+                    is_new_group_environment=is_new_group_environment,
+                    primary_hash=primary_hash,
+                    cache_key=cache_key,
+                )
 
     def insert(
         self,

src/sentry/tasks/post_process.py

@@ -116,38 +116,57 @@ def handle_owner_assignment(project, group, event):
 
 
 @instrumented_task(name="sentry.tasks.post_process.post_process_group")
-def post_process_group(event, is_new, is_regression, is_new_group_environment, **kwargs):
+def post_process_group(
+    event, is_new, is_regression, is_new_group_environment, cache_key=None, group_id=None, **kwargs
+):
     """
     Fires post processing hooks for a group.
     """
-    set_current_project(event.project_id)
-
+    from sentry.eventstore.models import Event
+    from sentry.eventstore.processing import event_processing_store
     from sentry.utils import snuba
     from sentry.reprocessing2 import is_reprocessed_event
 
     with snuba.options_override({"consistent": True}):
-        if is_reprocessed_event(event.data):
+        # The event parameter will be removed after transitioning to
+        # event_processing_store is complete.
+        if cache_key and event is None:
+            data = event_processing_store.get(cache_key)
+            if not data:
+                logger.info(
+                    "post_process.skipped",
+                    extra={"cache_key": cache_key, "reason": "missing_cache"},
+                )
+                return
+            event = Event(
+                project_id=data["project"], event_id=data["event_id"], group_id=group_id, data=data
+            )
+        elif check_event_already_post_processed(event):
+            if cache_key:
+                event_processing_store.delete_by_key(cache_key)
             logger.info(
                 "post_process.skipped",
                 extra={
                     "project_id": event.project_id,
                     "event_id": event.event_id,
-                    "reason": "reprocessed",
+                    "reason": "duplicate",
                 },
             )
             return
 
-        if check_event_already_post_processed(event):
+        if is_reprocessed_event(event.data):
             logger.info(
                 "post_process.skipped",
                 extra={
                     "project_id": event.project_id,
                     "event_id": event.event_id,
-                    "reason": "duplicate",
+                    "reason": "reprocessed",
                 },
             )
             return
 
+        set_current_project(event.project_id)
+
         # NOTE: we must pass through the full Event object, and not an
         # event_id since the Event object may not actually have been stored
         # in the database due to sampling.
@@ -161,13 +180,13 @@ def post_process_group(event, is_new, is_regression, is_new_group_environment, *
         event.data = EventDict(event.data, skip_renormalization=True)
 
         if event.group_id:
-            # Re-bind Group since we're pickling the whole Event object
-            # which may contain a stale Project.
+            # Re-bind Group since we're reading the Event object
+            # from cache, which may contain a stale group and project
             event.group, _ = get_group_with_redirect(event.group_id)
             event.group_id = event.group.id
 
-        # Re-bind Project and Org since we're pickling the whole Event object
-        # which may contain stale parent models.
+        # Re-bind Project and Org since we're reading the Event object
+        # from cache which may contain stale parent models.
         event.project = Project.objects.get_from_cache(id=event.project_id)
         event.project._organization_cache = Organization.objects.get_from_cache(
             id=event.project.organization_id
@@ -232,6 +251,9 @@ def post_process_group(event, is_new, is_regression, is_new_group_environment, *
             event=event,
             primary_hash=kwargs.get("primary_hash"),
         )
+        if cache_key:
+            with metrics.timer("tasks.post_process.delete_event_cache"):
+                event_processing_store.delete_by_key(cache_key)
 
 
 def process_snoozes(group):

src/sentry/tasks/store.py

@@ -667,6 +667,7 @@ def create_failed_event(
     # modifications to take place.
     delete_raw_event(project_id, event_id)
     data = event_processing_store.get(cache_key)
+
     if data is None:
         metrics.incr("events.failed", tags={"reason": "cache", "stage": "raw"}, skip_internal=False)
         error_logger.error("process.failed_raw.empty", extra={"cache_key": cache_key})
@@ -690,7 +691,6 @@ def create_failed_event(
             type=issue["type"],
             data=issue["data"],
         )
-
     event_processing_store.delete_by_key(cache_key)
 
     return True
@@ -762,15 +762,18 @@ def _do_save_event(
                 manager.save(
                     project_id, assume_normalized=True, start_time=start_time, cache_key=cache_key
                 )
-
+                # Put the updated event back into the cache so that post_process
+                # has the most recent data.
+                data = manager.get_data()
+                if isinstance(data, CANONICAL_TYPES):
+                    data = dict(data.items())
+                with metrics.timer("tasks.store.do_save_event.write_processing_cache"):
+                    event_processing_store.store(data)
         except HashDiscarded:
             pass
 
         finally:
             if cache_key:
-                with metrics.timer("tasks.store.do_save_event.delete_cache"):
-                    event_processing_store.delete_by_key(cache_key)
-
                 with metrics.timer("tasks.store.do_save_event.delete_attachment_cache"):
                     attachment_cache.delete(cache_key)