feat(taskworker) Collect task parameter distributions (#91985)

Start collecting distributions on task payloads. We want to validate
assumptions about the distribution of task parameter sizes in
production.

Include this in the other parameter verification sample rate logic we're
running on tasks.

Author: markstory

src/sentry/celery.py

@@ -11,7 +11,7 @@
 from django.db import models
 from django.utils.safestring import SafeString
 
-from sentry.utils import metrics
+from sentry.utils import json, metrics
 
 logger = logging.getLogger("celery.pickle")
 
@@ -114,6 +114,19 @@ def apply_async(self, *args, **kwargs):
         )
         should_sample = random.random() <= settings.CELERY_PICKLE_ERROR_REPORT_SAMPLE_RATE
         if should_complain or should_sample:
+            try:
+                param_size = json.dumps({"args": args, "kwargs": kwargs})
+                metrics.distribution(
+                    "celery.task.parameter_bytes",
+                    len(param_size.encode("utf8")),
+                    tags={"taskname": self.name},
+                    sample_rate=1.0,
+                )
+            except Exception as e:
+                logger.warning(
+                    "task.payload.measure.failure", extra={"error": str(e), "task": self.name}
+                )
+
             try:
                 good_use_of_pickle_or_bad_use_of_pickle(self, args, kwargs)
             except TypeError:

tests/sentry/tasks/test_base.py

@@ -7,6 +7,7 @@
 from sentry.tasks.base import instrumented_task, retry
 from sentry.taskworker.config import TaskworkerConfig
 from sentry.taskworker.namespaces import test_tasks
+from sentry.testutils.helpers.options import override_options
 
 
 @instrumented_task(
@@ -112,6 +113,28 @@ def test_exclude_exception_retry(capture_exception):
     assert capture_exception.call_count == 0
 
 
+@override_settings(
+    CELERY_COMPLAIN_ABOUT_BAD_USE_OF_PICKLE=True,
+    CELERY_PICKLE_ERROR_REPORT_SAMPLE_RATE=1.0,
+)
+@override_options(
+    {
+        "taskworker.test.rollout": {"*": 0.0},
+        "taskworker.route.overrides": {},
+    }
+)
+@patch("sentry.tasks.base.metrics.distribution")
+def test_capture_payload_metrics(mock_distribution):
+    region_task.apply_async(args=("bruh",))
+
+    mock_distribution.assert_called_once_with(
+        "celery.task.parameter_bytes",
+        71,
+        tags={"taskname": "test.tasks.test_base.region_task"},
+        sample_rate=1.0,
+    )
+
+
 @patch("sentry.taskworker.retry.current_task")
 @patch("sentry.tasks.base.capture_exception")
 def test_retry_on(capture_exception, current_task):