fix(taskworker) Remove pickle usage from sentryapp hooks (#91435)

Redo of #91372 which was reverted because of rate limiting. This time
around, use an option to control task variants, and only generate a
payload when a hook is going to be sent. For projects with multiple
service hooks, this should result in fewer snuba calls as events are
only serialized once.

Author: markstory

src/sentry/options/defaults.py

@@ -3343,3 +3343,5 @@
 # Orgs for which compression should be disabled in the chunk upload endpoint.
 # This is intended to circumvent sporadic 503 errors reported by some customers.
 register("chunk-upload.no-compression", default=[], flags=FLAG_AUTOMATOR_MODIFIABLE)
+
+register("process_service_hook.payload.rollout", default=0.0, flags=FLAG_AUTOMATOR_MODIFIABLE)

src/sentry/sentry_apps/tasks/service_hooks.py

@@ -1,8 +1,12 @@
 import logging
 from time import time
+from typing import Any
 
+from sentry import nodestore
 from sentry.api.serializers import serialize
+from sentry.eventstore.models import Event, GroupEvent
 from sentry.http import safe_urlopen
+from sentry.models.group import Group
 from sentry.sentry_apps.models.servicehook import ServiceHook
 from sentry.silo.base import SiloMode
 from sentry.tasks.base import instrumented_task, retry
@@ -46,14 +50,36 @@ def get_payload_v0(event):
     ),
 )
 @retry
-def process_service_hook(servicehook_id, event, **kwargs):
+def process_service_hook(
+    servicehook_id: int,
+    event: Any | None = None,
+    project_id: int | None = None,
+    group_id: int | None = None,
+    event_id: str | None = None,
+    **kwargs,
+):
     try:
         servicehook = ServiceHook.objects.get(id=servicehook_id)
     except ServiceHook.DoesNotExist:
         return
 
+    if not project_id and event:
+        project_id = event.project_id
+
+    if project_id and group_id and event_id:
+        node_id = Event.generate_node_id(project_id, event_id)
+        group = Group.objects.get_from_cache(id=group_id)
+        nodedata = nodestore.backend.get(node_id)
+        event = GroupEvent(
+            project_id=project_id,
+            event_id=event_id,
+            group=group,
+            data=nodedata,
+        )
+
+    assert event, "Event must exist at this point"
     if servicehook.version == 0:
-        payload = get_payload_v0(event)
+        payload = json.dumps(get_payload_v0(event))
     else:
         raise NotImplementedError
 
@@ -65,10 +91,8 @@ def process_service_hook(servicehook_id, event, **kwargs):
         "Content-Type": "application/json",
         "X-ServiceHook-Timestamp": str(int(time())),
         "X-ServiceHook-GUID": servicehook.guid,
-        "X-ServiceHook-Signature": servicehook.build_signature(json.dumps(payload)),
+        "X-ServiceHook-Signature": servicehook.build_signature(payload),
     }
 
-    safe_urlopen(
-        url=servicehook.url, data=json.dumps(payload), headers=headers, timeout=5, verify_ssl=False
-    )
-    logger.info("service_hook.success", extra={"project_id": event.project_id})
+    safe_urlopen(url=servicehook.url, data=payload, headers=headers, timeout=5, verify_ssl=False)
+    logger.info("service_hook.success", extra={"project_id": project_id})

src/sentry/tasks/post_process.py

@@ -18,6 +18,7 @@
 from sentry.issues.grouptype import GroupCategory
 from sentry.issues.issue_occurrence import IssueOccurrence
 from sentry.killswitches import killswitch_matches_context
+from sentry.options.rollout import in_random_rollout
 from sentry.replays.lib.event_linking import transform_event_for_linking_payload
 from sentry.replays.lib.kafka import initialize_replays_publisher
 from sentry.sentry_metrics.client import generic_metrics_backend
@@ -72,7 +73,7 @@ class PostProcessJob(TypedDict, total=False):
     has_escalated: bool
 
 
-def _get_service_hooks(project_id):
+def _get_service_hooks(project_id: int) -> list[tuple[int, list[str]]]:
     from sentry.sentry_apps.models.servicehook import ServiceHook
 
     cache_key = f"servicehooks:1:{project_id}"
@@ -1193,7 +1194,16 @@ def process_service_hooks(job: PostProcessJob) -> None:
         if has_alert:
             allowed_events.add("event.alert")
 
-        if allowed_events:
+        if in_random_rollout("process_service_hook.payload.rollout"):
+            for servicehook_id, events in _get_service_hooks(project_id=event.project_id):
+                if any(e in allowed_events for e in events):
+                    process_service_hook.delay(
+                        servicehook_id=servicehook_id,
+                        project_id=event.project_id,
+                        group_id=event.group_id,
+                        event_id=event.event_id,
+                    )
+        else:
             for servicehook_id, events in _get_service_hooks(project_id=event.project_id):
                 if any(e in allowed_events for e in events):
                     process_service_hook.delay(servicehook_id=servicehook_id, event=event)

tests/sentry/sentry_apps/tasks/test_servicehooks.py

@@ -59,6 +59,35 @@ def test_event_created_sends_service_hook(self, safe_urlopen):
             "X-ServiceHook-Signature",
         }
 
+    @patch("sentry.sentry_apps.tasks.service_hooks.safe_urlopen")
+    @responses.activate
+    def test_event_created_sends_service_hook_with_event_id(self, safe_urlopen):
+        self.hook.update(events=["event.created", "event.alert"])
+
+        event = self.store_event(
+            data={"timestamp": before_now(minutes=1).isoformat()}, project_id=self.project.id
+        )
+        assert event.group
+
+        process_service_hook(
+            self.hook.id,
+            project_id=event.project_id,
+            group_id=event.group.id,
+            event_id=event.event_id,
+        )
+
+        ((_, kwargs),) = safe_urlopen.call_args_list
+        data = json.loads(kwargs["data"])
+
+        assert kwargs["url"] == self.hook.url
+        assert data == json.loads(json.dumps(get_payload_v0(event)))
+        assert kwargs["headers"].keys() <= {
+            "Content-Type",
+            "X-ServiceHook-Timestamp",
+            "X-ServiceHook-GUID",
+            "X-ServiceHook-Signature",
+        }
+
     @responses.activate
     def test_v0_payload(self):
         responses.add(responses.POST, "https://example.com/sentry/webhook")

tests/sentry/tasks/test_post_process.py

@@ -499,6 +499,32 @@ def test_group_last_seen_buffer(self, mock_processor):
 
 
 class ServiceHooksTestMixin(BasePostProgressGroupMixin):
+    @override_options({"process_service_hook.payload.rollout": 1.0})
+    @patch("sentry.sentry_apps.tasks.service_hooks.process_service_hook")
+    def test_service_hook_fires_on_new_event_payload_param(self, mock_process_service_hook):
+        event = self.create_event(data={}, project_id=self.project.id)
+        hook = self.create_service_hook(
+            project=self.project,
+            organization=self.project.organization,
+            actor=self.user,
+            events=["event.created"],
+        )
+
+        with self.feature("projects:servicehooks"):
+            self.call_post_process_group(
+                is_new=False,
+                is_regression=False,
+                is_new_group_environment=False,
+                event=event,
+            )
+
+        mock_process_service_hook.delay.assert_called_once_with(
+            servicehook_id=hook.id,
+            project_id=self.project.id,
+            group_id=event.group_id,
+            event_id=event.event_id,
+        )
+
     @patch("sentry.sentry_apps.tasks.service_hooks.process_service_hook")
     def test_service_hook_fires_on_new_event(self, mock_process_service_hook):
         event = self.create_event(data={}, project_id=self.project.id)
@@ -518,7 +544,8 @@ def test_service_hook_fires_on_new_event(self, mock_process_service_hook):
             )
 
         mock_process_service_hook.delay.assert_called_once_with(
-            servicehook_id=hook.id, event=EventMatcher(event)
+            servicehook_id=hook.id,
+            event=EventMatcher(event),
         )
 
     @patch("sentry.sentry_apps.tasks.service_hooks.process_service_hook")
@@ -547,7 +574,8 @@ def test_service_hook_fires_on_alert(self, mock_processor, mock_process_service_
             )
 
         mock_process_service_hook.delay.assert_called_once_with(
-            servicehook_id=hook.id, event=EventMatcher(event)
+            servicehook_id=hook.id,
+            event=EventMatcher(event),
         )
 
     @patch("sentry.sentry_apps.tasks.service_hooks.process_service_hook")