fix(sentryapps) Stop passing event to celery task v2 (#79042)

Another approach to https://github.com/getsentry/sentry/pull/79013.
eventstore was rate-limiting requests once the previous changes were
enabled. Fix up the option gate logic and use nodestore to read events.
Using nodestore requires adding group_id to the task parameters.

src/sentry/options/defaults.py

@@ -2801,3 +2801,10 @@ register(
     default=False,
     flags=FLAG_AUTOMATOR_MODIFIABLE,
 )
+
+register(
+    "sentryapps.process-resource-change.use-eventid",
+    type=Bool,
+    default=False,
+    flags=FLAG_AUTOMATOR_MODIFIABLE,
+)

src/sentry/sentry_apps/tasks/sentry_apps.py

@@ -9,7 +9,7 @@ from celery import Task, current_task
 from django.urls import reverse
 from requests.exceptions import RequestException
 
-from sentry import analytics
+from sentry import analytics, nodestore
 from sentry.api.serializers import serialize
 from sentry.constants import SentryAppInstallationStatus
 from sentry.db.models.base import Model
@@ -181,18 +181,33 @@ def _process_resource_change(
     # The class is serialized as a string when enqueueing the class.
     model: type[Event] | type[Model] = TYPES[sender]
     instance: Event | Model | None = None
-    # The Event model has different hooks for the different event types. The sender
-    # determines which type eg. Error and therefore the 'name' eg. error
-    if issubclass(model, Event):
-        if not kwargs.get("instance"):
-            extra = {"sender": sender, "action": action, "event_id": instance_id}
-            logger.info("process_resource_change.event_missing_event", extra=extra)
-            return
+
+    project_id: int | None = kwargs.get("project_id", None)
+    group_id: int | None = kwargs.get("group_id", None)
+    if sender == "Error" and not kwargs.get("instance", None) and project_id and group_id:
+        # Read event from nodestore as we're trying to move away from passing events in task
+        # messages.
+        node_id = Event.generate_node_id(project_id, str(instance_id))
+        nodedata = nodestore.backend.get(node_id)
+        node_event = Event(
+            project_id=project_id, group_id=group_id, event_id=str(instance_id), data=nodedata
+        )
+
+        kwargs["instance"] = node_event
         name = sender.lower()
     else:
-        # Some resources are named differently than their model. eg. Group vs Issue.
-        # Looks up the human name for the model. Defaults to the model name.
-        name = RESOURCE_RENAMES.get(model.__name__, model.__name__.lower())
+        # The Event model has different hooks for the different event types. The sender
+        # determines which type eg. Error and therefore the 'name' eg. error
+        if issubclass(model, Event):
+            if not kwargs.get("instance"):
+                extra = {"sender": sender, "action": action, "event_id": instance_id}
+                logger.info("process_resource_change.event_missing_event", extra=extra)
+                return
+            name = sender.lower()
+        else:
+            # Some resources are named differently than their model. eg. Group vs Issue.
+            # Looks up the human name for the model. Defaults to the model name.
+            name = RESOURCE_RENAMES.get(model.__name__, model.__name__.lower())
 
     # By default, use Celery's `current_task` but allow a value to be passed for the
     # bound Task.

src/sentry/tasks/post_process.py

@@ -13,7 +13,7 @@ from django.db.models.signals import post_save
 from django.utils import timezone
 from google.api_core.exceptions import ServiceUnavailable
 
-from sentry import features, projectoptions
+from sentry import features, options, projectoptions
 from sentry.exceptions import PluginError
 from sentry.issues.grouptype import GroupCategory
 from sentry.issues.issue_occurrence import IssueOccurrence
@@ -1164,9 +1164,21 @@ def process_resource_change_bounds(job: PostProcessJob) -> None:
     event, is_new = job["event"], job["group_state"]["is_new"]
 
     if event.get_event_type() == "error" and _should_send_error_created_hooks(event.project):
-        process_resource_change_bound.delay(
-            action="created", sender="Error", instance_id=event.event_id, instance=event
-        )
+        if options.get("sentryapps.process-resource-change.use-eventid"):
+            process_resource_change_bound.delay(
+                action="created",
+                sender="Error",
+                instance_id=event.event_id,
+                project_id=event.project_id,
+                group_id=event.group_id,
+            )
+        else:
+            process_resource_change_bound.delay(
+                action="created",
+                sender="Error",
+                instance_id=event.event_id,
+                instance=event,
+            )
     if is_new:
         process_resource_change_bound.delay(
             action="created", sender="Group", instance_id=event.group_id

tests/sentry/sentry_apps/tasks/test_sentry_apps.py

@@ -10,7 +10,7 @@ from django.test import override_settings
 from django.urls import reverse
 from requests.exceptions import Timeout
 
-from sentry import audit_log
+from sentry import audit_log, nodestore
 from sentry.api.serializers import serialize
 from sentry.constants import SentryAppStatus
 from sentry.integrations.models.utils import get_redis_key
@@ -36,6 +36,7 @@ from sentry.testutils.cases import TestCase
 from sentry.testutils.helpers import with_feature
 from sentry.testutils.helpers.datetime import before_now, freeze_time
 from sentry.testutils.helpers.eventprocessing import write_event_to_cache
+from sentry.testutils.helpers.options import override_options
 from sentry.testutils.outbox import outbox_runner
 from sentry.testutils.silo import assume_test_silo_mode_of, control_silo_test
 from sentry.testutils.skips import requires_snuba
@@ -304,7 +305,10 @@ class TestProcessResourceChange(TestCase):
             assert_no_errors=False,
         )
 
-        with self.tasks():
+        with self.tasks(), patch(
+            "sentry.sentry_apps.tasks.sentry_apps.nodestore.backend.get",
+            wraps=nodestore.backend.get,
+        ) as nodestore_get:
             post_process_group(
                 is_new=False,
                 is_regression=False,
@@ -313,6 +317,59 @@ class TestProcessResourceChange(TestCase):
                 group_id=event.group_id,
                 project_id=self.project.id,
             )
+            assert not nodestore_get.called
+
+        ((args, kwargs),) = safe_urlopen.call_args_list
+        data = json.loads(kwargs["data"])
+
+        assert data["action"] == "created"
+        assert data["installation"]["uuid"] == install.uuid
+        assert data["data"]["error"]["event_id"] == event.event_id
+        assert data["data"]["error"]["issue_id"] == str(event.group_id)
+        assert kwargs["headers"].keys() >= {
+            "Content-Type",
+            "Request-ID",
+            "Sentry-Hook-Resource",
+            "Sentry-Hook-Timestamp",
+            "Sentry-Hook-Signature",
+        }
+
+    @with_feature("organizations:integrations-event-hooks")
+    @override_options({"sentryapps.process-resource-change.use-eventid": True})
+    def test_error_created_sends_webhook_no_event_payload_eventid_only(self, safe_urlopen):
+        sentry_app = self.create_sentry_app(
+            organization=self.project.organization, events=["error.created"]
+        )
+        install = self.create_sentry_app_installation(
+            organization=self.project.organization, slug=sentry_app.slug
+        )
+
+        one_min_ago = before_now(minutes=1).timestamp()
+        event = self.store_event(
+            data={
+                "message": "Foo bar",
+                "exception": {"type": "Foo", "value": "oh no"},
+                "level": "error",
+                "timestamp": one_min_ago,
+            },
+            project_id=self.project.id,
+            assert_no_errors=False,
+        )
+
+        with self.tasks(), patch(
+            "sentry.sentry_apps.tasks.sentry_apps.nodestore.backend.get",
+            wraps=nodestore.backend.get,
+        ) as nodestore_get:
+            post_process_group(
+                is_new=False,
+                is_regression=False,
+                is_new_group_environment=False,
+                cache_key=write_event_to_cache(event),
+                group_id=event.group_id,
+                project_id=self.project.id,
+            )
+
+            assert nodestore_get.called
 
         ((args, kwargs),) = safe_urlopen.call_args_list
         data = json.loads(kwargs["data"])