feat(profiles): rollout data compression (#92133)

Author: john-z-yang

src/sentry/options/defaults.py

@@ -3113,6 +3113,13 @@
     flags=FLAG_AUTOMATOR_MODIFIABLE,
 )
 
+register(
+    "taskworker.try_compress.profile_metrics.rollout",
+    default=0.0,
+    type=Float,
+    flags=FLAG_AUTOMATOR_MODIFIABLE,
+)
+
 # Taskbroker flags
 register(
     "taskworker.try_compress.profile_metrics.level",

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

@@ -18,32 +18,39 @@ def process_message(message: Message[KafkaPayload]) -> None:
     sampled = is_sampled(message.payload.headers)
 
     if sampled or options.get("profiling.profile_metrics.unsampled_profiles.enabled"):
-        b64encoded = b64encode(message.payload.value).decode("utf-8")
-        process_profile_task.delay(payload=b64encoded, sampled=sampled, compressed_profile=False)
+        b64encoded_uncompressed = b64encode(message.payload.value).decode("utf-8")
 
-        if random.random() < options.get("taskworker.try_compress.profile_metrics"):
+        if random.random() < options.get("taskworker.try_compress.profile_metrics.rollout"):
             import time
             import zlib
 
-            metrics.distribution("profiling.profile_metrics.uncompressed_bytes", len(b64encoded))
+            metrics.distribution(
+                "profiling.profile_metrics.uncompressed_bytes", len(b64encoded_uncompressed)
+            )
 
             start_time = time.perf_counter()
+            b64encoded_compressed = b64encode(
+                zlib.compress(
+                    message.payload.value,
+                    level=options.get("taskworker.try_compress.profile_metrics.level"),
+                )
+            )
             metrics.distribution(
                 "profiling.profile_metrics.compressed_bytes",
-                len(
-                    b64encode(
-                        zlib.compress(
-                            message.payload.value,
-                            level=options.get("taskworker.try_compress.profile_metrics.level"),
-                        )
-                    )
-                ),
+                len(b64encoded_compressed),
             )
             end_time = time.perf_counter()
             metrics.distribution(
                 "profiling.profile_metrics.compression_time",
                 end_time - start_time,
             )
+            process_profile_task.delay(
+                payload=b64encoded_compressed, sampled=sampled, compressed_profile=True
+            )
+        else:
+            process_profile_task.delay(
+                payload=b64encoded_uncompressed, sampled=sampled, compressed_profile=False
+            )
 
 
 class ProcessProfileStrategyFactory(ProcessingStrategyFactory[KafkaPayload]):

tests/sentry/processing/backpressure/test_checking.py

@@ -30,7 +30,7 @@
         "backpressure.checking.interval": 5,
         "backpressure.monitoring.enabled": True,
         "backpressure.status_ttl": 60,
-        "taskworker.try_compress.profile_metrics": False,
+        "taskworker.try_compress.profile_metrics.rollout": 0,
     }
 )
 def test_backpressure_unhealthy_profiles():
@@ -54,7 +54,7 @@ def test_backpressure_unhealthy_profiles():
         "backpressure.checking.interval": 5,
         "backpressure.monitoring.enabled": False,
         "backpressure.status_ttl": 60,
-        "taskworker.try_compress.profile_metrics": False,
+        "taskworker.try_compress.profile_metrics.rollout": 0,
     }
 )
 def test_bad_config():
@@ -69,7 +69,7 @@ def test_bad_config():
         "backpressure.checking.interval": 5,
         "backpressure.monitoring.enabled": True,
         "backpressure.status_ttl": 60,
-        "taskworker.try_compress.profile_metrics": False,
+        "taskworker.try_compress.profile_metrics.rollout": 0,
     }
 )
 def test_backpressure_healthy_profiles(process_profile_task):
@@ -141,7 +141,7 @@ def test_backpressure_healthy_events(preprocess_event):
     {
         "backpressure.checking.enabled": False,
         "backpressure.checking.interval": 5,
-        "taskworker.try_compress.profile_metrics": False,
+        "taskworker.try_compress.profile_metrics.rollout": 0,
     }
 )
 def test_backpressure_not_enabled(process_profile_task):