feat(transport): Experimental sdk outcomes support

mitsuhiko · mitsuhiko · commit edb55facae9f · 2021-09-02T13:38:40.000+02:00
diff --git a/sentry_sdk/client.py b/sentry_sdk/client.py
@@ -200,6 +200,7 @@ def _prepare_event(
                 new_event = before_send(event, hint or {})
             if new_event is None:
                 logger.info("before send dropped event (%s)", event)
+                self.tansport.record_lost_event("before_send")
             event = new_event  # type: ignore
 
         return event
diff --git a/sentry_sdk/envelope.py b/sentry_sdk/envelope.py
@@ -236,6 +236,8 @@ def data_category(self):
             return "transaction"
         elif ty == "event":
             return "error"
+        elif ty == "sdk_outcomes":
+            return "sdk_outcomes"
         else:
             return "default"
 
diff --git a/sentry_sdk/transport.py b/sentry_sdk/transport.py
@@ -4,12 +4,13 @@
 import urllib3  # type: ignore
 import certifi
 import gzip
+import time
 
 from datetime import datetime, timedelta
 
 from sentry_sdk.utils import Dsn, logger, capture_internal_exceptions, json_dumps
 from sentry_sdk.worker import BackgroundWorker
-from sentry_sdk.envelope import Envelope
+from sentry_sdk.envelope import Envelope, Item, PayloadRef
 
 from sentry_sdk._types import MYPY
 
@@ -131,6 +132,8 @@ def __init__(
         self._auth = self.parsed_dsn.to_auth("sentry.python/%s" % VERSION)
         self._disabled_until = {}  # type: Dict[DataCategory, datetime]
         self._retry = urllib3.util.Retry()
+        self._event_loss_counters = {}
+        self._last_event_loss_sent = None
 
         self._pool = self._make_pool(
             self.parsed_dsn,
@@ -143,6 +146,13 @@ def __init__(
 
         self.hub_cls = Hub
 
+    def record_lost_event(self, reason):
+        """This increments a counter for event loss by reason.  The counters are flushed
+        periodically automatically (typically once a minute).
+        """
+        # This is not locked because we are okay with small mismeasuring.
+        self._event_loss_counters[reason] = self._event_loss_counters.get(reason, 0) + 1
+
     def _update_rate_limits(self, response):
         # type: (urllib3.HTTPResponse) -> None
 
@@ -207,7 +217,24 @@ def _send_request(
 
     def on_dropped_event(self, reason):
         # type: (str) -> None
-        pass
+        self.record_lost_event(reason)
+
+    def _flush_stats(self, force=False):
+        if not (force or self._last_event_loss_sent is None or \
+            self._last_event_loss_sent < time.time() - 60):
+            return
+        outcomes = self._event_loss_counters
+        self._event_loss_counters = {}
+
+        if outcomes:
+            self.capture_envelope(Envelope(
+                items=[Item(PayloadRef(json={
+                    "timestamp": time.time(),
+                    "outcomes": outcomes,
+                }), type="sdk_outomes")],
+            ))
+
+        self._last_event_loss_sent = time.time()
 
     def _check_disabled(self, category):
         # type: (str) -> bool
@@ -254,9 +281,15 @@ def _send_envelope(
         # type: (...) -> None
 
         # remove all items from the envelope which are over quota
-        envelope.items[:] = [
-            x for x in envelope.items if not self._check_disabled(x.data_category)
-        ]
+        new_items = []
+        for item in envelope.items:
+            if self._check_disabled(item.data_category):
+                if item.data_category in ("transaction", "error", "default"):
+                    self.on_dropped_event("self_rate_limits")
+            else:
+                new_items.append(item)
+
+        envelope.items[:] = new_items
         if not envelope.items:
             return None
 
@@ -341,6 +374,8 @@ def send_event_wrapper():
         if not self._worker.submit(send_event_wrapper):
             self.on_dropped_event("full_queue")
 
+        self._flush_stats()
+
     def capture_envelope(
         self, envelope  # type: Envelope
     ):
@@ -356,13 +391,16 @@ def send_envelope_wrapper():
         if not self._worker.submit(send_envelope_wrapper):
             self.on_dropped_event("full_queue")
 
+        self._flush_stats()
+
     def flush(
         self,
         timeout,  # type: float
         callback=None,  # type: Optional[Any]
     ):
         # type: (...) -> None
         logger.debug("Flushing HTTP transport")
+        self._flush_stats(force=True)
         if timeout > 0:
             self._worker.flush(timeout, callback)
 
