feat(taskworker): Compress send_resource_change_webhook task parameters (#95337)

In our US region, the `send_resource_change_webhook` is sending large
task payloads to kafka. As we observed, this has negative effects on the
taskbroker system. Now that taskworker supports compression, this PR is
responsible for compressing the task parameters of this task before
producing to kafka.

Rollout:
When this PR is merged and deployed, it should be a no-op to start.
Since compression is gated behind a sentry option, the only difference
is that we should see more options checks in postgres. When ready, we
can incrementally sample tasks to be compressed in smaller environments
first before reaching US region.

Author: enochtangg

src/sentry/sentry_apps/tasks/sentry_apps.py

@@ -46,6 +46,7 @@
 from sentry.silo.base import SiloMode
 from sentry.tasks.base import instrumented_task, retry
 from sentry.taskworker.config import TaskworkerConfig
+from sentry.taskworker.constants import CompressionType
 from sentry.taskworker.namespaces import sentryapp_control_tasks, sentryapp_tasks
 from sentry.taskworker.retry import Retry, retry_task
 from sentry.types.rules import RuleFuture
@@ -654,6 +655,7 @@ def get_webhook_data(
             times=3,
             delay=60 * 5,
         ),
+        compression_type=CompressionType.ZSTD,
     ),
     **TASK_OPTIONS,
 )

src/sentry/tasks/base.py

@@ -208,6 +208,7 @@ def _wrapped(*args, **kwargs):
                 processing_deadline_duration=taskworker_config.processing_deadline_duration,
                 at_most_once=taskworker_config.at_most_once,
                 wait_for_delivery=taskworker_config.wait_for_delivery,
+                compression_type=taskworker_config.compression_type,
             )(func)
 
             task = override_task(task, taskworker_task, taskworker_config, name)

src/sentry/taskworker/config.py

@@ -1,5 +1,6 @@
 import datetime
 
+from sentry.taskworker.constants import CompressionType
 from sentry.taskworker.registry import TaskNamespace
 from sentry.taskworker.retry import Retry
 
@@ -18,10 +19,12 @@ def __init__(
         processing_deadline_duration: int | datetime.timedelta | None = None,
         at_most_once: bool = False,
         wait_for_delivery: bool = False,
+        compression_type: CompressionType = CompressionType.PLAINTEXT,
     ):
         self.namespace = namespace
         self.retry = retry
         self.expires = expires
         self.processing_deadline_duration = processing_deadline_duration
         self.at_most_once = at_most_once
         self.wait_for_delivery = wait_for_delivery
+        self.compression_type = compression_type