Integrate AsyncHttpTransport as a new experimental option

Add a new experimental option that uses AsyncHttpTransport and integrates it with synchronous client interfaces.

GH-4601

Author: srothh

sentry_sdk/client.py

@@ -3,6 +3,7 @@
 import uuid
 import random
 import socket
+import asyncio
 from collections.abc import Mapping
 from datetime import datetime, timezone
 from importlib import import_module
@@ -25,7 +26,7 @@
 )
 from sentry_sdk.serializer import serialize
 from sentry_sdk.tracing import trace
-from sentry_sdk.transport import HttpTransportCore, make_transport
+from sentry_sdk.transport import HttpTransportCore, make_transport, AsyncHttpTransport
 from sentry_sdk.consts import (
     SPANDATA,
     DEFAULT_MAX_VALUE_LENGTH,
@@ -914,36 +915,50 @@ def get_integration(
 
         return self.integrations.get(integration_name)
 
+    def _close_components(self) -> None:
+        """Kill all client components in the correct order."""
+        self.session_flusher.kill()
+        if self.log_batcher is not None:
+            self.log_batcher.kill()
+        if self.monitor:
+            self.monitor.kill()
+        if self.transport is not None:
+            self.transport.kill()
+            self.transport = None
+
     def close(
         self,
         timeout: Optional[float] = None,
         callback: Optional[Callable[[int, float], None]] = None,
-    ) -> None:
+    ) -> Optional[asyncio.Task[None]]:
         """
         Close the client and shut down the transport. Arguments have the same
-        semantics as :py:meth:`Client.flush`.
+        semantics as :py:meth:`Client.flush`. When using the async transport, close needs to be awaited to block.
         """
         if self.transport is not None:
-            self.flush(timeout=timeout, callback=callback)
+            if isinstance(self.transport, AsyncHttpTransport):
 
-            self.session_flusher.kill()
+                def _on_flush_done(_: asyncio.Task[None]) -> None:
+                    self._close_components()
 
-            if self.log_batcher is not None:
-                self.log_batcher.kill()
-
-            if self.monitor:
-                self.monitor.kill()
-
-            self.transport.kill()
-            self.transport = None
+                flush_task = self.transport.loop.create_task(
+                    self._flush_async(timeout, callback)
+                )
+                # Enforce flush before shutdown
+                flush_task.add_done_callback(_on_flush_done)
+                return flush_task
+            else:
+                self.flush(timeout=timeout, callback=callback)
+                self._close_components()
+                return None
 
     def flush(
         self,
         timeout: Optional[float] = None,
         callback: Optional[Callable[[int, float], None]] = None,
-    ) -> None:
+    ) -> Optional[asyncio.Task[None]]:
         """
-        Wait for the current events to be sent.
+        Wait for the current events to be sent. When using the async transport, flush needs to be awaited to block.
 
         :param timeout: Wait for at most `timeout` seconds. If no `timeout` is provided, the `shutdown_timeout` option value is used.
 
@@ -952,12 +967,28 @@ def flush(
         if self.transport is not None:
             if timeout is None:
                 timeout = self.options["shutdown_timeout"]
+
             self.session_flusher.flush()
 
             if self.log_batcher is not None:
                 self.log_batcher.flush()
 
-            self.transport.flush(timeout=timeout, callback=callback)
+            if isinstance(self.transport, AsyncHttpTransport):
+                return self.transport.loop.create_task(
+                    self._flush_async(timeout, callback)
+                )
+            else:
+                self.transport.flush(timeout=timeout, callback=callback)
+
+            return None
+
+    async def _flush_async(
+        self, timeout: float, callback: Optional[Callable[[int, float], None]]
+    ) -> None:
+        self.session_flusher.flush()
+        if self.log_batcher is not None:
+            self.log_batcher.flush()
+        await self.transport.flush_async(timeout=timeout, callback=callback)  # type: ignore
 
     def __enter__(self) -> _Client:
         return self

sentry_sdk/consts.py

@@ -75,6 +75,7 @@ class CompressionAlgo(Enum):
             "transport_compression_algo": Optional[CompressionAlgo],
             "transport_num_pools": Optional[int],
             "transport_http2": Optional[bool],
+            "transport_async": Optional[bool],
             "enable_logs": Optional[bool],
             "before_send_log": Optional[Callable[[Log, Hint], Optional[Log]]],
         },

sentry_sdk/transport.py

@@ -647,18 +647,18 @@ async def send_envelope_wrapper() -> None:
     def capture_envelope(self: Self, envelope: Envelope) -> None:
         # Synchronous entry point
         try:
-            asyncio.get_running_loop()
             # We are on the main thread running the event loop
+            asyncio.get_running_loop()
             task = asyncio.create_task(self._capture_envelope(envelope))
             self.background_tasks.add(task)
             task.add_done_callback(self.background_tasks.discard)
         except RuntimeError:
             # We are in a background thread, not running an event loop,
             # have to launch the task on the loop in a threadsafe way.
-            if self._loop and self._loop.is_running():
+            if self.loop and self.loop.is_running():
                 asyncio.run_coroutine_threadsafe(
                     self._capture_envelope(envelope),
-                    self._loop,
+                    self.loop,
                 )
             else:
                 # The event loop is no longer running
@@ -680,7 +680,7 @@ async def flush_async(
 
     def _get_pool_options(self: Self) -> Dict[str, Any]:
         options: Dict[str, Any] = {
-            "http2": False,  # no HTTP2 for now
+            "http2": False,  # no HTTP2 for now, should probably just work with this setting
             "retries": 3,
         }
 
@@ -1022,8 +1022,7 @@ def make_transport(options: Dict[str, Any]) -> Optional[Transport]:
     if use_async_transport and asyncio.get_running_loop() is not None:
         transport_cls: Type[Transport] = AsyncHttpTransport
     else:
-        use_http2 = use_http2_transport
-        transport_cls = Http2Transport if use_http2 else HttpTransport
+        transport_cls = Http2Transport if use_http2_transport else HttpTransport
 
     if isinstance(ref_transport, Transport):
         return ref_transport