fix(transport): Unified hook for capturing metric about dropped events (#1100)

Author: untitaker

sentry_sdk/transport.py

@@ -150,12 +150,14 @@ def _update_rate_limits(self, response):
         # no matter of the status code to update our internal rate limits.
         header = response.headers.get("x-sentry-rate-limits")
         if header:
+            logger.warning("Rate-limited via x-sentry-rate-limits")
             self._disabled_until.update(_parse_rate_limits(header))
 
         # old sentries only communicate global rate limit hits via the
         # retry-after header on 429.  This header can also be emitted on new
         # sentries if a proxy in front wants to globally slow things down.
         elif response.status == 429:
+            logger.warning("Rate-limited via 429")
             self._disabled_until[None] = datetime.utcnow() + timedelta(
                 seconds=self._retry.get_retry_after(response) or 60
             )
@@ -173,19 +175,24 @@ def _send_request(
                 "X-Sentry-Auth": str(self._auth.to_header()),
             }
         )
-        response = self._pool.request(
-            "POST",
-            str(self._auth.get_api_url(endpoint_type)),
-            body=body,
-            headers=headers,
-        )
+        try:
+            response = self._pool.request(
+                "POST",
+                str(self._auth.get_api_url(endpoint_type)),
+                body=body,
+                headers=headers,
+            )
+        except Exception:
+            self.on_dropped_event("network")
+            raise
 
         try:
             self._update_rate_limits(response)
 
             if response.status == 429:
                 # if we hit a 429.  Something was rate limited but we already
                 # acted on this in `self._update_rate_limits`.
+                self.on_dropped_event("status_429")
                 pass
 
             elif response.status >= 300 or response.status < 200:
@@ -194,9 +201,14 @@ def _send_request(
                     response.status,
                     response.data,
                 )
+                self.on_dropped_event("status_{}".format(response.status))
         finally:
             response.close()
 
+    def on_dropped_event(self, reason):
+        # type: (str) -> None
+        pass
+
     def _check_disabled(self, category):
         # type: (str) -> bool
         def _disabled(bucket):
@@ -212,6 +224,7 @@ def _send_event(
         # type: (...) -> None
 
         if self._check_disabled("error"):
+            self.on_dropped_event("self_rate_limits")
             return None
 
         body = io.BytesIO()
@@ -325,7 +338,8 @@ def send_event_wrapper():
                 with capture_internal_exceptions():
                     self._send_event(event)
 
-        self._worker.submit(send_event_wrapper)
+        if not self._worker.submit(send_event_wrapper):
+            self.on_dropped_event("full_queue")
 
     def capture_envelope(
         self, envelope  # type: Envelope
@@ -339,7 +353,8 @@ def send_envelope_wrapper():
                 with capture_internal_exceptions():
                     self._send_envelope(envelope)
 
-        self._worker.submit(send_envelope_wrapper)
+        if not self._worker.submit(send_envelope_wrapper):
+            self.on_dropped_event("full_queue")
 
     def flush(
         self,

sentry_sdk/worker.py

@@ -109,16 +109,13 @@ def _wait_flush(self, timeout, callback):
                 logger.error("flush timed out, dropped %s events", pending)
 
     def submit(self, callback):
-        # type: (Callable[[], None]) -> None
+        # type: (Callable[[], None]) -> bool
         self._ensure_thread()
         try:
             self._queue.put_nowait(callback)
+            return True
         except Full:
-            self.on_full_queue(callback)
-
-    def on_full_queue(self, callback):
-        # type: (Optional[Any]) -> None
-        logger.error("background worker queue full, dropping event")
+            return False
 
     def _target(self):
         # type: () -> None