tele(taskbroker): check zlib compression viability

Author: john-z-yang

src/sentry/options/defaults.py

@@ -3107,6 +3107,12 @@
 )
 
 # Taskbroker flags
+register(
+    "taskworker.try_compress.profile_metrics",
+    default=0.0,
+    type=Float,
+    flags=FLAG_AUTOMATOR_MODIFIABLE,
+)
 
 register(
     "taskworker.route.overrides",

src/sentry/profiles/consumers/process/factory.py

@@ -1,3 +1,4 @@
+import random
 from base64 import b64encode
 from collections.abc import Iterable, Mapping
 
@@ -10,15 +11,31 @@
 from sentry import options
 from sentry.processing.backpressure.arroyo import HealthChecker, create_backpressure_step
 from sentry.profiles.task import process_profile_task
+from sentry.utils import metrics
 
 
 def process_message(message: Message[KafkaPayload]) -> None:
     sampled = is_sampled(message.payload.headers)
 
     if sampled or options.get("profiling.profile_metrics.unsampled_profiles.enabled"):
-        process_profile_task.delay(
-            payload=b64encode(message.payload.value).decode("utf-8"), sampled=sampled
-        )
+        b64encoded = b64encode(message.payload.value).decode("utf-8")
+        process_profile_task.delay(payload=b64encoded, sampled=sampled)
+        metrics.distribution("profiling.profile_metrics.uncompressed_bytes", len(b64encoded))
+
+        if random.random() < options.get("taskworker.try_compress.profile_metrics"):
+            import time
+            import zlib
+
+            start_time = time.perf_counter()
+            metrics.distribution(
+                "profiling.profile_metrics.compressed_bytes",
+                len(b64encode(zlib.compress(message.payload.value))),
+            )
+            end_time = time.perf_counter()
+            metrics.distribution(
+                "profiling.profile_metrics.compression_time",
+                end_time - start_time,
+            )
 
 
 class ProcessProfileStrategyFactory(ProcessingStrategyFactory[KafkaPayload]):

tests/sentry/processing/backpressure/test_checking.py

@@ -67,6 +67,7 @@ def test_bad_config():
         "backpressure.checking.interval": 5,
         "backpressure.monitoring.enabled": True,
         "backpressure.status_ttl": 60,
+        "taskworker.try_compress.profile_metrics": False,
     }
 )
 def test_backpressure_healthy_profiles(process_profile_task):