chore(replay): add a log sampling filter and sample replay ingest logs

src/sentry/logging/handlers.py

@@ -1,4 +1,5 @@
 import logging
+import random
 import re
 
 from django.utils.timezone import now
@@ -149,3 +150,22 @@ def emit(self, record, logger=None):
         key = metrics_badchars_re.sub("", key)
         key = ".".join(key.split(".")[:3])
         metrics.incr(key, skip_internal=False)
+
+
+class SamplingFilter(logging.Filter):
+    """
+    A logging filter to sample messages at different rates.
+
+    prob_mapping -- a mapping of messages to probabilities. The msg must be an exact match.
+      Messages not in the mapping aren't filtered.
+    """
+
+    def __init__(self, prob_mapping: dict[str, float]):
+        super().__init__()
+        self.prob_mapping = prob_mapping
+
+    def filter(self, record: logging.LogRecord) -> bool:
+        msg = record.msg
+        if msg in self.prob_mapping:
+            return random.random() < self.prob_mapping[msg]
+        return True

src/sentry/replays/consumers/recording_buffered.py

@@ -56,13 +56,18 @@
 from sentry_kafka_schemas.schema_types.ingest_replay_recordings_v1 import ReplayRecording
 
 from sentry.conf.types.kafka_definition import Topic, get_topic_codec
+from sentry.logging.handlers import SamplingFilter
 from sentry.models.project import Project
 from sentry.replays.lib.storage import (
     RecordingSegmentStorageMeta,
     make_recording_filename,
     storage_kv,
 )
-from sentry.replays.usecases.ingest import process_headers, track_initial_segment_event
+from sentry.replays.usecases.ingest import (
+    MOBILE_EVENT_SAMPLE_RATE,
+    process_headers,
+    track_initial_segment_event,
+)
 from sentry.replays.usecases.ingest.dom_index import (
     ReplayActionsEvent,
     emit_replay_actions,
@@ -72,6 +77,13 @@
 from sentry.utils import json, metrics
 
 logger = logging.getLogger(__name__)
+logger.addFilter(
+    SamplingFilter(
+        {
+            "mobile_event": MOBILE_EVENT_SAMPLE_RATE,
+        }
+    )
+)
 
 RECORDINGS_CODEC: Codec[ReplayRecording] = get_topic_codec(Topic.INGEST_REPLAYS_RECORDINGS)
 

src/sentry/replays/usecases/ingest/__init__.py

@@ -12,6 +12,7 @@
 from sentry_sdk.tracing import Span
 
 from sentry.constants import DataCategory
+from sentry.logging.handlers import SamplingFilter
 from sentry.models.project import Project
 from sentry.replays.lib.storage import (
     RecordingSegmentStorageMeta,
@@ -24,7 +25,18 @@
 from sentry.utils import json, metrics
 from sentry.utils.outcomes import Outcome, track_outcome
 
+MOBILE_EVENT_SAMPLE_RATE = 0.5
+RRWEB_EVENT_COUNT_SAMPLE_RATE = 0.5
+
 logger = logging.getLogger("sentry.replays")
+logger.addFilter(
+    SamplingFilter(
+        {
+            "mobile_event": MOBILE_EVENT_SAMPLE_RATE,
+            "rrweb_event_count": RRWEB_EVENT_COUNT_SAMPLE_RATE,
+        }
+    )
+)
 
 CACHE_TIMEOUT = 3600
 COMMIT_FREQUENCY_SEC = 1

tests/sentry/logging/test_handler.py

@@ -4,7 +4,7 @@
 
 import pytest
 
-from sentry.logging.handlers import JSONRenderer, StructLogHandler
+from sentry.logging.handlers import JSONRenderer, SamplingFilter, StructLogHandler
 
 
 @pytest.fixture
@@ -119,3 +119,18 @@ def test_logging_raiseExcpetions_enabled_generic_logging(caplog, snafu):
 def test_logging_raiseExcpetions_disabled_generic_logging(caplog, snafu):
     logger = logging.getLogger(__name__)
     logger.log(logging.INFO, snafu)
+
+
+@mock.patch("random.random", lambda: 0.5)
+def test_sampling_filter(caplog):
+    logger = logging.getLogger(__name__)
+    logger.addFilter(SamplingFilter({"msg1": 0.8, "message.2": 0.3}))
+
+    logger.info("msg1")
+    logger.info("message.2")
+    logger.info("hello")
+
+    captured_msgs = list(map(lambda r: r.msg, caplog.records))
+    assert "msg1" in captured_msgs
+    assert "message.2" not in captured_msgs
+    assert "hello" in captured_msgs