feat(sentry apps): Add circuit breaker into webhook code (#111723)

Author: Christinarlong

.github/CODEOWNERS

@@ -448,6 +448,7 @@ tests/sentry/api/endpoints/test_organization_attribute_mappings.py          @get
 /src/sentry/sentry_apps/                                             @getsentry/product-owners-settings-integrations @getsentry/ecosystem
 /tests/sentry/sentry_apps/                                           @getsentry/product-owners-settings-integrations @getsentry/ecosystem
 /src/sentry/utils/sentry_apps/                                       @getsentry/ecosystem
+/tests/sentry/utils/sentry_apps/                                     @getsentry/ecosystem
 /src/sentry/middleware/integrations/                                 @getsentry/ecosystem
 /src/sentry/api/endpoints/project_rule*.py                           @getsentry/alerts-notifications
 /src/sentry/api/serializers/models/rule.py                           @getsentry/alerts-notifications

src/sentry/features/temporary.py

@@ -492,6 +492,8 @@ def register_temporary_features(manager: FeatureManager) -> None:
     manager.add("organizations:conduit-demo", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=True)
     # Enable hard timeout alarm for webhooks
     manager.add("organizations:sentry-app-webhook-hard-timeout", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=False)
+    # Enable circuit breaker for webhook endpoint failure detection
+    manager.add("organizations:sentry-app-webhook-circuit-breaker", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=False)
 
     # Enables organization access to the new notification platform
     manager.add("organizations:notification-platform.internal-testing", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=True)

src/sentry/options/defaults.py

@@ -2597,6 +2597,29 @@
     flags=FLAG_AUTOMATOR_MODIFIABLE,
 )
 
+# Circuit breaker configuration for webhook endpoint failure detection.
+# Keys match RateBasedTripStrategyConfig + CircuitBreakerConfig
+register(
+    "sentry-apps.webhook.circuit-breaker.config",
+    type=Dict,
+    default={
+        "error_limit_window": 600,  # 10 minutes
+        "broken_state_duration": 300,  # 5 minutes
+        "threshold": 0.5,  # 50% error rate
+        "floor": 500,  # 500 errors before error rate check applies
+        "metrics_key": "sentry-app.webhook",  # to avoid high cardinality slug tag
+    },
+    flags=FLAG_AUTOMATOR_MODIFIABLE,
+)
+
+# When True, the circuit breaker tracks state and emits metrics but does not block requests.
+register(
+    "sentry-apps.webhook.circuit-breaker.dry-run",
+    type=Bool,
+    default=False,
+    flags=FLAG_AUTOMATOR_MODIFIABLE,
+)
+
 # Enables statistical detectors for a project
 register(
     "statistical_detectors.enable",

src/sentry/sentry_apps/metrics.py

@@ -71,6 +71,7 @@ class SentryAppWebhookHaltReason(StrEnum):
     RESTRICTED_IP = "restricted_ip"
     CONNECTION_RESET = "connection_reset"
     HARD_TIMEOUT = "hard_timeout"
+    CIRCUIT_BROKEN = "circuit_broken"
 
 
 class SentryAppExternalRequestFailureReason(StrEnum):

src/sentry/shared_integrations/exceptions/__init__.py

@@ -224,6 +224,6 @@ def __init__(self, field_errors: Mapping[str, Any] | None = None) -> None:
 class ClientError(RequestException):
     """4xx Error Occurred"""
 
-    def __init__(self, status_code: str, url: str, response: Response | None = None) -> None:
+    def __init__(self, status_code: str | int, url: str, response: Response | None = None) -> None:
         http_error_msg = f"{status_code} Client Error: for url: {url}"
         super().__init__(http_error_msg, response=response)

src/sentry/utils/sentry_apps/circuit_breaker.py

@@ -0,0 +1,39 @@
+import logging
+from collections.abc import Generator
+from contextlib import contextmanager
+
+from sentry.utils.circuit_breaker2 import CircuitBreaker
+
+logger = logging.getLogger("sentry.sentry_apps.circuit_breaker")
+
+
+@contextmanager
+def circuit_breaker_tracking(
+    breaker: CircuitBreaker | None,
+) -> Generator[None]:
+    """Track request outcome: record_error on WebhookTimeoutError, record_success on normal exit.
+
+    Handles the None case as a no-op so callers don't need nullcontext().
+    """
+    from sentry.utils.sentry_apps.webhooks import WebhookTimeoutError
+
+    if breaker is None:
+        yield
+        return
+    try:
+        yield
+
+    # Currently we only count WebhookTimeoutError as an error in the circuit breaker as those operations are the ones that are taking too long
+    # If an app returns a say 500, in a reasonable time that's okay
+    except WebhookTimeoutError:
+        # This is gross but we don't want to propagate a redis or circuit breaker error to the webhook code
+        try:
+            breaker.record_error()
+        except Exception:
+            logger.exception("sentry_apps.circuit_breaker.record_error.failure")
+        raise
+    else:
+        try:
+            breaker.record_success()
+        except Exception:
+            logger.exception("sentry_apps.circuit_breaker.record_success.failure")

src/sentry/utils/sentry_apps/webhooks.py

@@ -4,6 +4,7 @@
 from collections.abc import Callable, Mapping
 from types import FrameType
 from typing import TYPE_CHECKING, Any, Concatenate, ParamSpec, TypeVar
+from urllib.parse import urlparse
 
 import sentry_sdk
 from requests import RequestException, Response
@@ -13,6 +14,8 @@
 from sentry import features, options
 from sentry.exceptions import RestrictedIPAddress
 from sentry.http import safe_urlopen
+from sentry.integrations.utils.metrics import EventLifecycle
+from sentry.organizations.services.organization.model import RpcUserOrganizationContext
 from sentry.organizations.services.organization.service import organization_service
 from sentry.sentry_apps.metrics import (
     SentryAppEventType,
@@ -23,7 +26,10 @@
 from sentry.sentry_apps.utils.errors import SentryAppSentryError
 from sentry.shared_integrations.exceptions import ApiHostError, ApiTimeoutError, ClientError
 from sentry.taskworker.timeout import timeout_alarm
+from sentry.utils import metrics
+from sentry.utils.circuit_breaker2 import CircuitBreaker, RateBasedTripStrategy
 from sentry.utils.sentry_apps import SentryAppWebhookRequestsBuffer
+from sentry.utils.sentry_apps.circuit_breaker import circuit_breaker_tracking
 
 if TYPE_CHECKING:
     from sentry.sentry_apps.api.serializers.app_platform_event import AppPlatformEvent
@@ -51,7 +57,6 @@ def _handle_webhook_timeout(signum: int, frame: FrameType | None) -> None:
     """Handler for when a webhook request exceeds the hard timeout deadline.
     - This is a workaround for safe_create_connection sockets hanging when the given url
     cannot be reached or resolved.
-    - TODO(christinarlong): Add sentry app disabling logic here
     """
     raise WebhookTimeoutError("Webhook request exceeded hard timeout deadline")
 
@@ -73,6 +78,79 @@ def wrapper(
     return wrapper
 
 
+def _create_circuit_breaker(
+    sentry_app: SentryApp | RpcSentryApp,
+    organization_context: RpcUserOrganizationContext | None,
+) -> CircuitBreaker | None:
+    if organization_context is None or not features.has(
+        "organizations:sentry-app-webhook-circuit-breaker",
+        organization_context.organization,
+    ):
+        return None
+    config = options.get("sentry-apps.webhook.circuit-breaker.config")
+    return CircuitBreaker(
+        key=f"sentry-app.webhook.{sentry_app.slug}",
+        config=config,
+        trip_strategy=RateBasedTripStrategy.from_config(config),
+    )
+
+
+def _circuit_breaker_allows_request(
+    circuit_breaker: CircuitBreaker | None,
+    sentry_app: SentryApp | RpcSentryApp,
+    org_id: int,
+    lifecycle: EventLifecycle,
+) -> bool:
+    if circuit_breaker is None or circuit_breaker.should_allow_request():
+        return True
+
+    dry_run = options.get("sentry-apps.webhook.circuit-breaker.dry-run")
+    if dry_run:
+        metrics.incr(
+            "sentry_app.webhook.circuit_breaker.would_block",
+            tags={"slug": sentry_app.slug},
+        )
+        logger.warning(
+            "sentry_app.webhook.circuit_breaker.would_block",
+            extra={"slug": sentry_app.slug, "org_id": org_id},
+        )
+        return True
+
+    lifecycle.record_halt(
+        halt_reason=f"send_and_save_webhook_request.{SentryAppWebhookHaltReason.CIRCUIT_BROKEN}"
+    )
+    return False
+
+
+def _send_webhook_request(
+    url: str,
+    app_platform_event: AppPlatformEvent[T],
+    organization_context: RpcUserOrganizationContext | None,
+) -> Response:
+    if organization_context is not None and features.has(
+        "organizations:sentry-app-webhook-hard-timeout",
+        organization_context.organization,
+    ):
+        # We're using a signal based timeout here because we need to interrupt the blocking
+        # socket.connect() operation. See SENTRY-5HA6 for more context. Here we're hanging at
+        # the socket.connect() call and the timeout we set in safe_urlopen is not being respected.
+        timeout_seconds = options.get("sentry-apps.webhook.hard-timeout.sec")
+        with timeout_alarm(timeout_seconds, _handle_webhook_timeout):
+            return safe_urlopen(
+                url=url,
+                data=app_platform_event.body,
+                headers=app_platform_event.headers,
+                timeout=options.get("sentry-apps.webhook.timeout.sec"),
+            )
+
+    return safe_urlopen(
+        url=url,
+        data=app_platform_event.body,
+        headers=app_platform_event.headers,
+        timeout=options.get("sentry-apps.webhook.timeout.sec"),
+    )
+
+
 @sentry_sdk.trace(name="send_and_save_webhook_request")
 @ignore_unpublished_app_errors
 def send_and_save_webhook_request(
@@ -124,28 +202,12 @@ def send_and_save_webhook_request(
                 include_projects=False,
                 include_teams=False,
             )
-            if organization_context is not None and features.has(
-                "organizations:sentry-app-webhook-hard-timeout",
-                organization_context.organization,
-            ):
-                # We're using a signal based timeout here because we need to interrupt the blocking socket.connect() opeartion.
-                # See SENTRY-5HA6 for more context. Here we're hanging at the socket.connect() call and the timeout we set
-                # in safe_urlopen is not being respected.
-                timeout_seconds = options.get("sentry-apps.webhook.hard-timeout.sec")
-                with timeout_alarm(timeout_seconds, _handle_webhook_timeout):
-                    response = safe_urlopen(
-                        url=url,
-                        data=app_platform_event.body,
-                        headers=app_platform_event.headers,
-                        timeout=options.get("sentry-apps.webhook.timeout.sec"),
-                    )
-            else:
-                response = safe_urlopen(
-                    url=url,
-                    data=app_platform_event.body,
-                    headers=app_platform_event.headers,
-                    timeout=options.get("sentry-apps.webhook.timeout.sec"),
-                )
+            circuit_breaker = _create_circuit_breaker(sentry_app, organization_context)
+            if not _circuit_breaker_allows_request(circuit_breaker, sentry_app, org_id, lifecycle):
+                return Response()
+
+            with circuit_breaker_tracking(circuit_breaker):
+                response = _send_webhook_request(url, app_platform_event, organization_context)
 
         except WebhookTimeoutError:
             lifecycle.record_halt(
@@ -186,13 +248,19 @@ def send_and_save_webhook_request(
             raise
 
         track_response_code(response.status_code, slug, event)
+
+        project_id = (
+            int(p_id)
+            if (p_id := response.headers.get("Sentry-Hook-Project")) and p_id.isdigit()
+            else None
+        )
         buffer.add_request(
             response_code=response.status_code,
             org_id=org_id,
             event=event,
             url=url,
             error_id=response.headers.get("Sentry-Hook-Error"),
-            project_id=response.headers.get("Sentry-Hook-Project"),
+            project_id=project_id,
             response=response,
             headers=app_platform_event.headers,
         )
@@ -223,13 +291,15 @@ def send_and_save_webhook_request(
             lifecycle.record_halt(
                 halt_reason=f"send_and_save_webhook_request.{SentryAppWebhookHaltReason.INTEGRATOR_ERROR}"
             )
-            raise ApiHostError.from_request(response.request)
+            raise ApiHostError(f"Unable to reach host: {urlparse(url).netloc}", url=url)
 
         elif response.status_code == status.HTTP_504_GATEWAY_TIMEOUT:
             lifecycle.record_halt(
                 halt_reason=f"send_and_save_webhook_request.{SentryAppWebhookHaltReason.INTEGRATOR_ERROR}"
             )
-            raise ApiTimeoutError.from_request(response.request)
+            raise ApiTimeoutError(
+                f"Timed out attempting to reach host: {urlparse(url).netloc}", url=url
+            )
 
         elif 400 <= response.status_code < 500:
             lifecycle.record_halt(
@@ -243,4 +313,5 @@ def send_and_save_webhook_request(
         except RequestException as e:
             lifecycle.record_halt(e)
             raise
+
         return response

tests/sentry/sentry_apps/tasks/test_sentry_apps.py

@@ -1713,7 +1713,7 @@ def test_saves_error_event_id_if_in_header(self, safe_urlopen: MagicMock) -> Non
         assert first_request["event_type"] == "issue.assigned"
         assert first_request["organization_id"] == self.install.organization_id
         assert first_request["error_id"] == "d5111da2c28645c5889d072017e3445d"
-        assert first_request["project_id"] == "1"
+        assert first_request["project_id"] == 1
 
 
 @patch("sentry.utils.sentry_apps.webhooks.safe_urlopen", return_value=MockResponseInstance)