feat(hybridcloud): async apigateway (#111307)

This changes the `apigateway` proxy to be async, with the idea to serve
the relevant deployment of control silo in ASGI rather than WSGI.

The rationale here is to avoid situations in which we exhaust the
server's threadpool by just waiting for apigateway requests to complete,
as we saw in INCs 2054/2056.

**Note:** the APIGateway changes are gated into a separated Python
module, the async flow is enabled through `SENTRY_ASYNC_APIGW`
environment variable. This allows us to control the rollout of the
change in prod. Tests and local devserver are instead always using the
new code.

Detailed changes:
- [x] Make APIGateway proxy `async`, switching inner client impl from
`requests` to `httpx`
- [x] Change APIGateway middleware to work both in ASGI and WSGI
contexts (with the latter using `async_to_sync`)
- [x] Update relevant tests interacting with APIGateway
- [x] Fix proxy acceptance test
- <s> Fix ORM calls in the custom SDK integration</s>
- [x] Bypass ORM calls in SDK custom logging integration
- [x] Restore/adapt circuit brakers

Author: gi0baro

.github/CODEOWNERS

@@ -35,6 +35,7 @@
 /src/sentry/projectoptions/                             @getsentry/app-backend
 /src/sentry/receivers/                                  @getsentry/app-backend
 /src/sentry/ratelimits/                                 @getsentry/app-backend
+/src/sentry/asgi.py                                     @getsentry/app-backend
 
 /tests/js/sentry-test/                                  @getsentry/app-frontend
 /static/app/utils/                                      @getsentry/app-frontend

pyproject.toml

@@ -37,14 +37,15 @@ dependencies = [
   "google-cloud-storage-transfer>=1.17.0",
   "google-crc32c>=1.6.0",
   "googleapis-common-protos>=1.63.2",
-  "granian[pname,reload]>=2.7",
+  "granian[pname,reload,uvloop]>=2.7",
   "grpc-google-iam-v1>=0.13.1",
   # Note, grpcio>1.30.0 requires setting GRPC_POLL_STRATEGY=epoll1
   # See https://github.com/grpc/grpc/issues/23796 and
   # https://github.com/grpc/grpc/blob/v1.35.x/doc/core/grpc-polling-engines.md#polling-engine-implementations-in-grpc
   "grpcio>=1.67.0",
   # not directly used, but provides a speedup for redis
   "hiredis>=2.3.2",
+  "httpx>=0.28.1",
   "jsonschema>=4.20.0",
   "lxml>=5.3.0",
   "maxminddb>=2.3.0",

src/sentry/asgi.py

@@ -0,0 +1,18 @@
+import os.path
+import sys
+
+# Add the project to the python path
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), os.pardir))
+
+# Configure the application only if it seemingly isn't already configured
+from django.conf import settings
+
+if not settings.configured:
+    from sentry.runner import configure
+
+    configure()
+
+from django.core.handlers.asgi import ASGIHandler
+
+# Run ASGI handler for the application
+application = ASGIHandler()

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_APIGW_ASYNC", "").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",
@@ -3181,6 +3187,9 @@ def custom_parameter_sort(parameter: dict) -> tuple[str, int]:
 }
 # Used in tests to skip forwarding relay paths to a region silo that does not exist.
 APIGATEWAY_PROXY_SKIP_RELAY = False
+APIGATEWAY_PROXY_MAX_CONCURRENCY = int(os.environ.get("SENTRY_APIGW_PROXY_MAX_CONCURRENCY", 100))
+APIGATEWAY_PROXY_MAX_FAILURES = int(os.environ.get("SENTRY_APIGW_PROXY_MAX_FAILURES", 100))
+APIGATEWAY_PROXY_FAILURE_WINDOW = int(os.environ.get("SENTRY_APIGW_PROXY_FAILURE_WINDOW", 60))
 
 # Shared resource ids for accounting
 EVENT_PROCESSING_STORE = "rc_processing_redis"

src/sentry/hybridcloud/apigateway_async/__init__.py

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

src/sentry/hybridcloud/apigateway_async/apigateway.py

@@ -0,0 +1,108 @@
+from __future__ import annotations
+
+import logging
+from collections.abc import Callable
+from typing import Any
+
+from django.conf import settings
+from django.http.response import HttpResponseBase
+from rest_framework.request import Request
+
+from sentry.silo.base import SiloLimit, SiloMode
+from sentry.types.cell import get_cell_by_name
+from sentry.utils import metrics
+
+from .proxy import (
+    proxy_cell_request,
+    proxy_error_embed_request,
+    proxy_request,
+)
+
+logger = logging.getLogger(__name__)
+
+
+def _get_view_silo_mode(view_func: Callable[..., HttpResponseBase]) -> frozenset[SiloMode] | None:
+    view_class = getattr(view_func, "view_class", None)
+    if not view_class:
+        return None
+    if not hasattr(view_class, "silo_limit"):
+        return None
+    endpoint_silo_limit: SiloLimit = view_class.silo_limit
+    return endpoint_silo_limit.modes
+
+
+async 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.
+    returns None if proxying is not required, or a response if the proxy was successful.
+    """
+    current_silo_mode = SiloMode.get_current_mode()
+    if current_silo_mode != SiloMode.CONTROL:
+        return None
+
+    silo_modes = _get_view_silo_mode(view_func)
+    if not silo_modes or current_silo_mode in silo_modes:
+        return None
+
+    url_name = "unknown"
+    if request.resolver_match:
+        url_name = request.resolver_match.url_name or url_name
+
+    if "organization_slug" in view_kwargs or "organization_id_or_slug" in view_kwargs:
+        org_id_or_slug = str(
+            view_kwargs.get("organization_slug") or view_kwargs.get("organization_id_or_slug", "")
+        )
+
+        metrics.incr(
+            "apigateway.proxy_request",
+            tags={
+                "url_name": url_name,
+                "kind": "orgslug",
+            },
+        )
+        return await 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.
+        dsn = request.GET["dsn"]
+        metrics.incr(
+            "apigateway.proxy_request",
+            tags={
+                "url_name": url_name,
+                "kind": "error-embed",
+            },
+        )
+        return await proxy_error_embed_request(request, dsn, url_name)
+
+    if (
+        request.resolver_match
+        and request.resolver_match.url_name in settings.REGION_PINNED_URL_NAMES
+    ):
+        cell = get_cell_by_name(settings.SENTRY_MONOLITH_REGION)
+        metrics.incr(
+            "apigateway.proxy_request",
+            tags={
+                "url_name": url_name,
+                "kind": "regionpin",
+            },
+        )
+
+        return await 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
+        # URL handling and that needs to be fixed.
+        metrics.incr(
+            "apigateway.proxy_request",
+            tags={
+                "kind": "noop",
+                "url_name": url_name,
+            },
+        )
+        logger.info("apigateway.unknown_url", extra={"url": request.path})
+
+    return None

src/sentry/hybridcloud/apigateway_async/circuitbreaker.py

@@ -0,0 +1,104 @@
+from __future__ import annotations
+
+import asyncio
+import time
+from collections import defaultdict
+from typing import Any
+
+from django.conf import settings
+
+
+class CircuitBreaker:
+    __slots__ = [
+        "concurrency",
+        "counter_window",
+        "failures",
+        "semaphore",
+        "_clock",
+        "_counters",
+        "_counter_idx",
+    ]
+
+    def __init__(self, concurrency: int, failures: tuple[int, int]) -> None:
+        self.concurrency = concurrency
+        self.counter_window = failures[0]
+        self.failures = failures[1]
+        self.semaphore = asyncio.Semaphore(self.concurrency)
+        self._clock = 0
+        self._counters = [0, 0]
+        self._counter_idx = 0
+
+    def _counter_flip(self, clock: int) -> None:
+        self._clock = clock
+        prev = self._counter_idx
+        self._counter_idx = 1 - prev
+        self._counters[prev] = 0
+
+    def _maybe_counter_flip(self) -> None:
+        now = int(time.monotonic())
+        delta = now - self._clock
+        if delta > 0:
+            if delta // self.counter_window:
+                self._counter_flip(now)
+
+    def counter_incr(self) -> None:
+        self._maybe_counter_flip()
+        self._counters[self._counter_idx] += 1
+
+    def window_overflow(self) -> bool:
+        self._maybe_counter_flip()
+        return self._counters[self._counter_idx] > self.failures
+
+    def overflow(self) -> bool:
+        return self.semaphore.locked()
+
+    def ctx(self) -> CircuitBreakerCtx:
+        return CircuitBreakerCtx(self)
+
+
+class CircuitBreakerOverflow(Exception): ...
+
+
+class CircuitBreakerWindowOverflow(Exception): ...
+
+
+class CircuitBreakerCtx:
+    __slots__ = ["cb"]
+
+    def __init__(self, cb: CircuitBreaker):
+        self.cb = cb
+
+    def incr_failures(self) -> None:
+        self.cb.counter_incr()
+
+    async def __aenter__(self) -> CircuitBreakerCtx:
+        if self.cb.overflow():
+            raise CircuitBreakerOverflow
+        await self.cb.semaphore.acquire()
+        if self.cb.window_overflow():
+            self.cb.semaphore.release()
+            raise CircuitBreakerWindowOverflow
+        return self
+
+    async def __aexit__(self, exc_type: Any, exc_value: Any, exc_tb: Any) -> None:
+        self.cb.semaphore.release()
+
+
+class CircuitBreakerManager:
+    __slots__ = ["objs"]
+
+    def __init__(
+        self,
+        max_concurrency: int | None = None,
+        failures: int | None = None,
+        failure_window: int | None = None,
+    ):
+        concurrency = max_concurrency or settings.APIGATEWAY_PROXY_MAX_CONCURRENCY
+        failures = failures or settings.APIGATEWAY_PROXY_MAX_FAILURES
+        failure_window = failure_window or settings.APIGATEWAY_PROXY_FAILURE_WINDOW
+        self.objs: dict[str, CircuitBreaker] = defaultdict(
+            lambda: CircuitBreaker(concurrency, (failure_window, failures))
+        )
+
+    def get(self, key: str) -> CircuitBreakerCtx:
+        return self.objs[key].ctx()

src/sentry/hybridcloud/apigateway_async/middleware.py

@@ -0,0 +1,78 @@
+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
+
+from . import proxy_request_if_needed
+
+
+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)
+        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)
+        if proxy_response is not None:
+            return proxy_response
+        return None