chore: move async apigateway in own module

Author: gi0baro

src/sentry/conf/server.py

@@ -371,6 +371,12 @@ def env(
 # so that responses aren't modified after Content-Length is set, or have the
 # response modifying middleware reset the Content-Length header.
 # This is because CommonMiddleware Sets the Content-Length header for non-streaming responses.
+APIGW_ASYNC = os.environ.get("SENTRY_ASYNC_APIGW", "").lower() in ("1", "true", "y", "yes")
+APIGW_MIDDLEWARE = (
+    "sentry.hybridcloud.apigateway_async.middleware.ApiGatewayMiddleware"
+    if APIGW_ASYNC
+    else "sentry.hybridcloud.apigateway.middleware.ApiGatewayMiddleware"
+)
 MIDDLEWARE: tuple[str, ...] = (
     "csp.middleware.CSPMiddleware",
     "sentry.middleware.health.HealthCheck",
@@ -387,7 +393,7 @@ def env(
     "sentry.middleware.auth.AuthenticationMiddleware",
     "sentry.middleware.ai_agent.AIAgentMiddleware",
     "sentry.middleware.integrations.IntegrationControlMiddleware",
-    "sentry.hybridcloud.apigateway.middleware.ApiGatewayMiddleware",
+    APIGW_MIDDLEWARE,
     "sentry.middleware.demo_mode_guard.DemoModeGuardMiddleware",
     "sentry.middleware.customer_domain.CustomerDomainMiddleware",
     "sentry.middleware.sudo.SudoMiddleware",

src/sentry/hybridcloud/apigateway/apigateway.py

@@ -30,10 +30,8 @@ def _get_view_silo_mode(view_func: Callable[..., HttpResponseBase]) -> frozenset
     return endpoint_silo_limit.modes
 
 
-async def proxy_request_if_needed(
-    request: Request,
-    view_func: Callable[..., HttpResponseBase],
-    view_kwargs: dict[str, Any],
+def proxy_request_if_needed(
+    request: Request, view_func: Callable[..., HttpResponseBase], view_kwargs: dict[str, Any]
 ) -> HttpResponseBase | None:
     """
     Main execution flow for the API Gateway.
@@ -63,7 +61,7 @@ async def proxy_request_if_needed(
                 "kind": "orgslug",
             },
         )
-        return await proxy_request(request, org_id_or_slug, url_name)
+        return proxy_request(request, org_id_or_slug, url_name)
 
     if url_name == "sentry-error-page-embed" and "dsn" in request.GET:
         # Error embed modal is special as customers can't easily use cell URLs.
@@ -75,7 +73,7 @@ async def proxy_request_if_needed(
                 "kind": "error-embed",
             },
         )
-        return await proxy_error_embed_request(request, dsn, url_name)
+        return proxy_error_embed_request(request, dsn, url_name)
 
     if (
         request.resolver_match
@@ -90,7 +88,7 @@ async def proxy_request_if_needed(
             },
         )
 
-        return await proxy_cell_request(request, cell, url_name)
+        return proxy_cell_request(request, cell, url_name)
 
     if url_name != "unknown":
         # If we know the URL but didn't proxy it record we could be missing

src/sentry/hybridcloud/apigateway/middleware.py

@@ -1,10 +1,8 @@
 from __future__ import annotations
 
-import asyncio
 from collections.abc import Callable
 from typing import Any
 
-from asgiref.sync import async_to_sync, iscoroutinefunction, markcoroutinefunction
 from django.http.response import HttpResponseBase
 from rest_framework.request import Request
 
@@ -14,65 +12,21 @@
 class ApiGatewayMiddleware:
     """Proxy requests intended for remote silos"""
 
-    async_capable = True
-    sync_capable = True
-
     def __init__(self, get_response: Callable[[Request], HttpResponseBase]):
         self.get_response = get_response
-        if iscoroutinefunction(self.get_response):
-            markcoroutinefunction(self)
 
-    def __call__(self, request: Request) -> Any:
-        if iscoroutinefunction(self):
-            return self.__acall__(request)
+    def __call__(self, request: Request) -> HttpResponseBase:
         return self.get_response(request)
 
-    async def __acall__(self, request: Request) -> HttpResponseBase:
-        return await self.get_response(request)  # type: ignore[misc]
-
     def process_view(
         self,
         request: Request,
         view_func: Callable[..., HttpResponseBase],
         view_args: tuple[str],
         view_kwargs: dict[str, Any],
     ) -> HttpResponseBase | None:
-        return self._process_view_match(request, view_func, view_args, view_kwargs)
-
-    def _process_view_match(
-        self,
-        request: Request,
-        view_func: Callable[..., HttpResponseBase],
-        view_args: tuple[str],
-        view_kwargs: dict[str, Any],
-    ) -> Any:
-        #: we check if we're in an async or sync runtime once, then
-        #  overwrite the method with the actual impl.
-        try:
-            asyncio.get_running_loop()
-            method = self._process_view_inner
-        except RuntimeError:
-            method = self._process_view_sync  # type: ignore[assignment]
-        setattr(self, "_process_view_match", method)
-        return method(request, view_func, view_args, view_kwargs)
-
-    def _process_view_sync(
-        self,
-        request: Request,
-        view_func: Callable[..., HttpResponseBase],
-        view_args: tuple[str],
-        view_kwargs: dict[str, Any],
-    ) -> HttpResponseBase | None:
-        return async_to_sync(self._process_view_inner)(request, view_func, view_args, view_kwargs)
-
-    async def _process_view_inner(
-        self,
-        request: Request,
-        view_func: Callable[..., HttpResponseBase],
-        view_args: tuple[str],
-        view_kwargs: dict[str, Any],
-    ) -> HttpResponseBase | None:
-        proxy_response = await proxy_request_if_needed(request, view_func, view_kwargs)
+        proxy_response = proxy_request_if_needed(request, view_func, view_kwargs)
         if proxy_response is not None:
             return proxy_response
-        return None
+        else:
+            return None

src/sentry/hybridcloud/apigateway/proxy.py

@@ -4,21 +4,21 @@
 
 from __future__ import annotations
 
-import asyncio
 import logging
-from collections.abc import AsyncGenerator, AsyncIterator
+from collections.abc import Generator
 from urllib.parse import urljoin, urlparse
 from wsgiref.util import is_hop_by_hop
 
-import httpx
-from asgiref.sync import sync_to_async
 from django.conf import settings
-from django.http import HttpRequest, HttpResponse, StreamingHttpResponse
+from django.http import HttpRequest, HttpResponse, JsonResponse, StreamingHttpResponse
 from django.http.response import HttpResponseBase
+from requests import Response as ExternalResponse
+from requests import request as external_request
+from requests.exceptions import Timeout
 
 from sentry import options
 from sentry.api.exceptions import RequestTimeout
-from sentry.objectstore.endpoints.organization import get_raw_body_async
+from sentry.objectstore.endpoints.organization import ChunkedEncodingDecoder, get_raw_body
 from sentry.silo.util import (
     PROXY_APIGATEWAY_HEADER,
     PROXY_DIRECT_LOCATION_HEADER,
@@ -32,12 +32,11 @@
     get_cell_for_organization,
 )
 from sentry.utils import metrics
-from sentry.utils.http import BodyAsyncWrapper
+from sentry.utils.circuit_breaker2 import CircuitBreaker, CountBasedTripStrategy
+from sentry.utils.http import BodyWithLength
 
 logger = logging.getLogger(__name__)
 
-proxy_client = httpx.AsyncClient()
-
 # Endpoints that handle uploaded files have higher timeouts configured
 # and we need to honor those timeouts when proxying.
 # See frontend/templates/sites-enabled/sentry.io in getsentry/ops
@@ -56,57 +55,41 @@
 PROXY_CHUNK_SIZE = 512 * 1024
 
 
-async def _stream_response_and_close(response: httpx.Response) -> AsyncGenerator[bytes]:
-    """Yield chunks from an httpx response and close the connection when done."""
-    try:
-        async for chunk in response.aiter_bytes(PROXY_CHUNK_SIZE):
-            yield chunk
-    finally:
-        await response.aclose()
-
-
-def _adapt_response(response: httpx.Response, remote_url: str) -> StreamingHttpResponse:
-    """Convert an httpx Response into a Django response."""
+def _parse_response(response: ExternalResponse, remote_url: str) -> StreamingHttpResponse:
+    """
+    Convert the Responses class from requests into the drf Response
+    """
 
-    new_headers = clean_outbound_headers(response.headers)
-    content_type = new_headers.pop("Content-Type", None)
+    def stream_response() -> Generator[bytes]:
+        yield from response.iter_content(PROXY_CHUNK_SIZE)
 
     streamed_response = StreamingHttpResponse(
-        streaming_content=_stream_response_and_close(response),
+        streaming_content=stream_response(),
         status=response.status_code,
-        content_type=content_type,
+        content_type=response.headers.pop("Content-Type", None),
     )
-
-    for header, value in new_headers.items():
+    # Add Headers to response
+    for header, value in response.headers.items():
         if not is_hop_by_hop(header):
             streamed_response[header] = value
 
     streamed_response[PROXY_DIRECT_LOCATION_HEADER] = remote_url
     return streamed_response
 
 
-async def _stream_request(body: AsyncIterator[bytes]) -> AsyncGenerator[bytes]:
-    async for chunk in body:
-        yield chunk
-
-
-async def proxy_request(
-    request: HttpRequest,
-    org_id_or_slug: str,
-    url_name: str,
-) -> HttpResponseBase:
+def proxy_request(request: HttpRequest, org_id_or_slug: str, url_name: str) -> HttpResponseBase:
     """Take a django request object and proxy it to a remote location given an org_id_or_slug"""
 
     try:
-        cell = await sync_to_async(get_cell_for_organization)(org_id_or_slug)
+        cell = get_cell_for_organization(org_id_or_slug)
     except CellResolutionError as e:
         logger.info("region_resolution_error", extra={"org_slug": org_id_or_slug, "error": str(e)})
         return HttpResponse(status=404)
 
-    return await proxy_cell_request(request, cell, url_name)
+    return proxy_cell_request(request, cell, url_name)
 
 
-async def proxy_error_embed_request(
+def proxy_error_embed_request(
     request: HttpRequest, dsn: str, url_name: str
 ) -> HttpResponseBase | None:
     try:
@@ -126,65 +109,104 @@ async def proxy_error_embed_request(
         # If we don't have a o123.ingest.{cell}.{app_host} style domain
         # we forward to the monolith cell
         cell = get_cell_by_name(settings.SENTRY_MONOLITH_REGION)
-        return await proxy_cell_request(request, cell, url_name)
+        return proxy_cell_request(request, cell, url_name)
     try:
         cell_offset = len(app_segments) + 1
         cell_segment = host_segments[cell_offset * -1]
         cell = get_cell_by_name(cell_segment)
     except Exception:
         return None
 
-    return await proxy_cell_request(request, cell, url_name)
+    return proxy_cell_request(request, cell, url_name)
 
 
-async def proxy_cell_request(
-    request: HttpRequest,
-    cell: Cell,
-    url_name: str,
-) -> StreamingHttpResponse:
+def proxy_cell_request(request: HttpRequest, cell: Cell, url_name: str) -> HttpResponseBase:
     """Take a django request object and proxy it to a cell silo"""
+
+    metric_tags = {"region": cell.name, "url_name": url_name}
+    circuit_breaker: CircuitBreaker | None = None
+    # TODO(mark) remove rollout options
+    if options.get("apigateway.proxy.circuit-breaker.enabled"):
+        try:
+            circuit_breaker = CircuitBreaker(
+                key=f"apigateway.proxy.{cell.name}",
+                config=options.get("apigateway.proxy.circuit-breaker.config"),
+                trip_strategy=CountBasedTripStrategy.from_config(
+                    options.get("apigateway.proxy.circuit-breaker.config")
+                ),
+            )
+        except Exception as e:
+            logger.warning("apigateway.invalid-breaker-config", extra={"message": str(e)})
+
+    if circuit_breaker is not None:
+        if not circuit_breaker.should_allow_request():
+            metrics.incr("apigateway.proxy.circuit_breaker.rejected", tags=metric_tags)
+            if options.get("apigateway.proxy.circuit-breaker.enforce"):
+                body = {
+                    "error": "apigateway",
+                    "detail": "Downstream service temporarily unavailable",
+                }
+                return JsonResponse(body, status=503)
+
     target_url = urljoin(cell.address, request.path)
 
     content_encoding = request.headers.get("Content-Encoding")
-    content_length = request.headers.get("Content-Length")
     header_dict = clean_proxy_headers(request.headers)
     header_dict[PROXY_APIGATEWAY_HEADER] = "true"
 
+    # TODO: use requests session for connection pooling capabilities
     assert request.method is not None
     query_params = request.GET
 
-    timeout = ENDPOINT_TIMEOUT_OVERRIDE.get(url_name, settings.GATEWAY_PROXY_TIMEOUT)
-    metric_tags = {"region": cell.name, "url_name": url_name}
+    # This option has a default of None, which is cast to 0
+    timeout = options.get("apigateway.proxy.timeout")
+    if not timeout:
+        timeout = settings.GATEWAY_PROXY_TIMEOUT
+    timeout = ENDPOINT_TIMEOUT_OVERRIDE.get(url_name, timeout)
 
     # XXX: See sentry.testutils.pytest.sentry for more information
     if settings.APIGATEWAY_PROXY_SKIP_RELAY and request.path.startswith("/api/0/relays/"):
         return StreamingHttpResponse(streaming_content="relay proxy skipped", status=404)
 
+    data: bytes | Generator[bytes] | ChunkedEncodingDecoder | BodyWithLength | None = None
     if url_name == "sentry-api-0-organization-objectstore":
         if content_encoding:
             header_dict["Content-Encoding"] = content_encoding
-        data = get_raw_body_async(request)
+        data = get_raw_body(request)
     else:
-        data = BodyAsyncWrapper(request.body)
-        # With request streaming, and without `Content-Length` header,
-        # `httpx` will set chunked transfer encoding.
-        # Upstream doesn't necessarily support this,
-        # thus we re-add the header if it was present in the original request.
-        if content_length:
-            header_dict["Content-Length"] = content_length
+        data = BodyWithLength(request)
 
     try:
         with metrics.timer("apigateway.proxy_request.duration", tags=metric_tags):
-            req = proxy_client.build_request(
+            resp = external_request(
                 request.method,
-                target_url,
+                url=target_url,
                 headers=header_dict,
                 params=dict(query_params) if query_params is not None else None,
-                content=_stream_request(data) if data else None,  # type: ignore[arg-type]
+                data=data,
+                stream=True,
                 timeout=timeout,
+                # By default, external_request will resolve any redirects for any verb except for HEAD.
+                # We explicitly disable this behavior to avoid misrepresenting the original sentry.io request with the
+                # body response of the redirect.
+                allow_redirects=False,
             )
-            resp = await proxy_client.send(req, stream=True, follow_redirects=False)
-            return _adapt_response(resp, target_url)
-    except (httpx.TimeoutException, asyncio.CancelledError):
+    except Timeout:
+        metrics.incr("apigateway.proxy.request_timeout", tags=metric_tags)
+        try:
+            if circuit_breaker is not None:
+                circuit_breaker.record_error()
+        except Exception:
+            logger.exception("Failed to record circuitbreaker failure")
+
         # remote silo timeout. Use DRF timeout instead
         raise RequestTimeout()
+
+    if resp.status_code >= 500 and circuit_breaker is not None:
+        metrics.incr("apigateway.proxy.request_failed", tags=metric_tags)
+        circuit_breaker.record_error()
+
+    new_headers = clean_outbound_headers(resp.headers)
+    resp.headers.clear()
+    resp.headers.update(new_headers)
+    return _parse_response(resp, target_url)

src/sentry/hybridcloud/apigateway_async/__init__.py

@@ -0,0 +1,3 @@
+from .apigateway import proxy_request_if_needed
+
+__all__ = ("proxy_request_if_needed",)