feat(issue-priority): Add a circuit breaker on consecutive failures (#67913)

If Seer fails for multiple requests in succession, we want a way to
start skipping Seer until the service recovers. This circuit breaker
keeps track of consecutive failures on the service and will start
throttling requests if the limit is hit. In order to detect if the
service is recovered, the circuit breaker will allow a configurable
number of requests through per minute. If that request succeeds, the
cache key will be cleared and we will resume making requests to Seer.

---------

Co-authored-by: getsantry[bot] <66042841+getsantry[bot]@users.noreply.github.com>

Author: snigdhas

src/sentry/event_manager.py

@@ -116,6 +116,11 @@
 from sentry.utils import json, metrics
 from sentry.utils.cache import cache_key_for_event
 from sentry.utils.canonical import CanonicalKeyDict
+from sentry.utils.circuit_breaker import (
+    ERROR_COUNT_CACHE_KEY,
+    CircuitBreakerPassthrough,
+    circuit_breaker_activated,
+)
 from sentry.utils.dates import to_datetime
 from sentry.utils.event import has_event_minified_stack_trace, has_stacktrace, is_handled
 from sentry.utils.eventuser import EventUser
@@ -141,6 +146,10 @@
 
 HIGH_SEVERITY_THRESHOLD = 0.1
 
+SEER_GLOBAL_RATE_LIMIT_DEFAULT = 20  # 20 requests per second
+SEER_PROJECT_RATE_LIMIT_DEFAULT = 5  # 5 requests per second
+SEER_ERROR_COUNT_KEY = ERROR_COUNT_CACHE_KEY("sentry.seer.severity-failures")
+
 
 @dataclass
 class GroupInfo:
@@ -1829,9 +1838,16 @@ def _create_group(project: Project, event: Event, **group_creation_kwargs: Any)
     group_data.setdefault("metadata", {}).update(sdk_metadata_from_event(event))
 
     # add severity to metadata for alert filtering
-    group_type = group_creation_kwargs.get("type", None)
-    severity = _get_severity_metadata_for_group(event, project.id, group_type)
-    group_data["metadata"].update(severity)
+    try:
+        group_type = group_creation_kwargs.get("type", None)
+        severity = _get_severity_metadata_for_group(event, project.id, group_type)
+        group_data["metadata"].update(severity)
+    except Exception as e:
+        logger.exception(
+            "Failed to get severity metadata for group",
+            repr(e),
+            extra={"event_id": event.event_id},
+        )
 
     if features.has("projects:issue-priority", project, actor=None):
         # the kwargs only include priority for non-error issue platform events, which takes precedence.
@@ -2163,7 +2179,10 @@ def _get_severity_metadata_for_group(
     from sentry.receivers.rules import PLATFORMS_WITH_PRIORITY_ALERTS
 
     if killswitch_matches_context("issues.skip-seer-requests", {"project_id": event.project_id}):
-        logger.warning("get_severity_metadata_for_group.seer_killswitch_enabled")
+        logger.warning(
+            "get_severity_metadata_for_group.seer_killswitch_enabled",
+            extra={"event_id": event.event_id, "project_id": project_id},
+        )
         metrics.incr("issues.severity.seer_killswitch_enabled")
         return {}
 
@@ -2180,25 +2199,32 @@ def _get_severity_metadata_for_group(
     if not is_supported_platform or not is_error_group:
         return {}
 
+    passthrough_data = options.get(
+        "issues.severity.seer-circuit-breaker-passthrough-limit",
+        CircuitBreakerPassthrough(limit=1, window=10),
+    )
+    if circuit_breaker_activated("sentry.seer.severity", passthrough_data=passthrough_data):
+        logger.warning(
+            "get_severity_metadata_for_group.circuit_breaker_activated",
+            extra={"event_id": event.event_id, "project_id": project_id},
+        )
+        return {}
+
     from sentry import ratelimits as ratelimiter
 
-    limit = options.get("issues.severity.seer-global-rate-limit", 20)
+    limit = options.get("issues.severity.seer-global-rate-limit", SEER_GLOBAL_RATE_LIMIT_DEFAULT)
     if ratelimiter.backend.is_limited(
-        "seer:severity-calculation:global-limit",
-        limit=limit,
-        window=1,  # starting this out 20 requests per second
+        "seer:severity-calculation:global-limit", limit=limit, window=1
     ):
         logger.warning(
             "get_severity_metadata_for_group.rate_limited_globally",
             extra={"event_id": event.event_id, "project_id": project_id},
         )
         metrics.incr("issues.severity.rate_limited_globally")
 
-    limit = options.get("issues.severity.seer-project-rate-limit", 5)
+    limit = options.get("issues.severity.seer-project-rate-limit", SEER_PROJECT_RATE_LIMIT_DEFAULT)
     if ratelimiter.backend.is_limited(
-        f"seer:severity-calculation:{project_id}",
-        limit=limit,
-        window=1,  # starting this out 5 requests per second
+        f"seer:severity-calculation:{project_id}", limit=limit, window=1
     ):
         logger.warning(
             "get_severity_metadata_for_group.rate_limited_for_project",
@@ -2215,7 +2241,7 @@ def _get_severity_metadata_for_group(
         }
     except Exception as e:
         logger.warning("Failed to calculate severity score for group", repr(e))
-
+        update_severity_error_count()
         return {}
 
 
@@ -2259,6 +2285,19 @@ def _get_priority_for_group(severity: Mapping[str, Any], kwargs: Mapping[str, An
         return PriorityLevel.MEDIUM
 
 
+def update_severity_error_count(reset=False) -> None:
+    timeout = 60 * 60  # 1 hour
+    if reset:
+        cache.set(SEER_ERROR_COUNT_KEY, 0, timeout=timeout)
+        return
+
+    try:
+        cache.incr(SEER_ERROR_COUNT_KEY)
+        cache.touch(SEER_ERROR_COUNT_KEY, timeout=timeout)
+    except ValueError:
+        cache.set(SEER_ERROR_COUNT_KEY, 1, timeout=timeout)
+
+
 def _get_severity_score(event: Event) -> tuple[float, str]:
     # Short circuit the severity value if we know the event is fatal or info/debug
     level = str(event.data.get("level", "error"))
@@ -2344,20 +2383,23 @@ def _get_severity_score(event: Event) -> tuple[float, str]:
                 extra=logger_data,
             )
             reason = "microservice_max_retry"
+            update_severity_error_count()
         except Exception as e:
             logger.warning(
                 "Unable to get severity score from microservice. Got: %s.",
                 repr(e),
                 extra=logger_data,
             )
             reason = "microservice_error"
+            update_severity_error_count()
         else:
             logger.info(
                 "Got severity score of %s for event %s",
                 severity,
                 event.data["event_id"],
                 extra=logger_data,
             )
+            update_severity_error_count(reset=True)
 
     return severity, reason
 

src/sentry/options/defaults.py

@@ -776,6 +776,13 @@
     flags=FLAG_ALLOW_EMPTY | FLAG_AUTOMATOR_MODIFIABLE,
 )
 
+register(
+    "issues.severity.seer-circuit-breaker-passthrough-limit",
+    type=Dict,
+    default={"limit": 1, "window": 10},
+    flags=FLAG_ALLOW_EMPTY | FLAG_AUTOMATOR_MODIFIABLE,
+)
+
 register(
     "issues.severity.seer-timout",
     type=Float,

src/sentry/utils/circuit_breaker.py

@@ -0,0 +1,44 @@
+from typing import TypedDict
+
+from django.core.cache import cache
+
+from sentry import ratelimits as ratelimiter
+from sentry.utils import metrics
+
+DEFAULT_ERROR_LIMIT = 30
+ERROR_COUNT_CACHE_KEY = lambda key: f"circuit_breaker:{key}-error-count"
+PASSTHROUGH_RATELIMIT_KEY = lambda key: f"circuit_breaker:{key}-passthrough"
+
+
+class CircuitBreakerPassthrough(TypedDict, total=True):
+    limit: int
+    window: int
+
+
+def circuit_breaker_activated(
+    key: str,
+    error_limit: int = DEFAULT_ERROR_LIMIT,
+    passthrough_data: CircuitBreakerPassthrough | None = None,
+) -> bool:
+    """
+    Activates the circuit breaker if the error count for a cache key exceeds the error limit.
+
+    The circuit breaker can allow a certain number of requests to pass through per minute, defined by
+    the passthrough limit if provided.
+    """
+    failure_count = cache.get_or_set(ERROR_COUNT_CACHE_KEY(key), default=0, timeout=60 * 60) or 0
+    if failure_count < error_limit:
+        return False  # not blocked
+
+    # Limit has been exceeded, check if we should allow any requests to pass through
+    if passthrough_data:
+        if not ratelimiter.backend.is_limited(
+            PASSTHROUGH_RATELIMIT_KEY(key),
+            limit=passthrough_data["limit"],
+            window=passthrough_data["window"],
+        ):
+            metrics.incr(f"circuit_breaker.{key}.bypassed")
+            return False  # not blocked
+
+    metrics.incr(f"circuit_breaker.{key}.throttled")
+    return True  # blocked

tests/sentry/event_manager/test_severity.py

@@ -4,12 +4,14 @@
 from typing import Any
 from unittest.mock import MagicMock, patch
 
+from django.core.cache import cache
 from urllib3 import HTTPResponse
 from urllib3.exceptions import MaxRetryError
 
 from sentry import options
 from sentry.event_manager import (
     NON_TITLE_EVENT_TITLES,
+    SEER_ERROR_COUNT_KEY,
     EventManager,
     _get_severity_score,
     severity_connection_pool,
@@ -86,6 +88,7 @@ def test_error_event_simple(
         )
         assert severity == 0.1231
         assert reason == "ml"
+        assert cache.get(SEER_ERROR_COUNT_KEY) == 0
 
     @patch(
         "sentry.event_manager.severity_connection_pool.urlopen",
@@ -133,6 +136,7 @@ def test_message_event_simple(
             )
             assert severity == 0.1231
             assert reason == "ml"
+            assert cache.get(SEER_ERROR_COUNT_KEY) == 0
 
     @patch(
         "sentry.event_manager.severity_connection_pool.urlopen",
@@ -270,6 +274,7 @@ def test_max_retry_exception(
         )
         assert severity == 1.0
         assert reason == "microservice_max_retry"
+        assert cache.get(SEER_ERROR_COUNT_KEY) == 1
 
     @patch(
         "sentry.event_manager.severity_connection_pool.urlopen",
@@ -314,6 +319,7 @@ def test_other_exception(
         )
         assert severity == 1.0
         assert reason == "microservice_error"
+        assert cache.get(SEER_ERROR_COUNT_KEY) == 1
 
 
 @apply_feature_flag_on_cls("projects:first-event-severity-calculation")
@@ -426,4 +432,5 @@ def test_killswitch_on(self, mock_get_severity_score: MagicMock):
 
         assert event.group
         assert "severity" not in event.group.get_event_metadata()
+        assert cache.get(SEER_ERROR_COUNT_KEY) is None
         assert mock_get_severity_score.call_count == 0

tests/sentry/utils/test_circuit_breaker.py

@@ -0,0 +1,41 @@
+import time
+from unittest.mock import patch
+
+from django.core.cache import cache
+
+from sentry.testutils.cases import TestCase
+from sentry.utils.circuit_breaker import (
+    ERROR_COUNT_CACHE_KEY,
+    CircuitBreakerPassthrough,
+    circuit_breaker_activated,
+)
+
+
+class TestCircuitBreaker(TestCase):
+    def setUp(self):
+        self.key = "test"
+        self.error_limit = 5
+        self.passthrough_data = CircuitBreakerPassthrough(limit=2, window=1)
+        cache.set(ERROR_COUNT_CACHE_KEY(self.key), self.error_limit)
+
+    def test_not_activated(self):
+        assert not circuit_breaker_activated(self.key, self.error_limit + 1)
+
+    def test_activated_at_error_limit(self):
+        assert circuit_breaker_activated(key=self.key, error_limit=self.error_limit)
+
+    @patch("sentry.utils.circuit_breaker.metrics.incr")
+    def test_passthrough(self, mock_metrics):
+        assert not circuit_breaker_activated(self.key, self.error_limit, self.passthrough_data)
+        mock_metrics.assert_called_with(f"circuit_breaker.{self.key}.bypassed")
+
+        assert not circuit_breaker_activated(self.key, self.error_limit, self.passthrough_data)
+        mock_metrics.assert_called_with(f"circuit_breaker.{self.key}.bypassed")
+
+        assert circuit_breaker_activated(self.key, self.error_limit, self.passthrough_data)
+        mock_metrics.assert_called_with(f"circuit_breaker.{self.key}.throttled")
+
+        # Wait for the passthrough window to expire and try again
+        time.sleep(1)
+        assert not circuit_breaker_activated(self.key, self.error_limit, self.passthrough_data)
+        mock_metrics.assert_called_with(f"circuit_breaker.{self.key}.bypassed")