feat(integration-slack): store request error counts and disable on broken (#53621)

Continuing #52994 after revert

> 
> Continuing #51126
> Milestone 1 of [Notify on Disabled Integration
project](https://www.notion.so/sentry/Tech-Spec-Notify-on-Disabled-Integration-Spec-e7ea0f86ccd6419cb3e564067cf4a2ef?pvs=4)
> 
> Implementing IntegrationRequestBuffer using Redis for logging errors
and detecting broken integrations to be disabled. Request buffer stores
daily aggregate error, fatal and success counts for an integration for
30 days. If an integration consistently fails for 7 days or has a fatal
response, then the integration is broken. Later users will be alerted of
broken integrations with an email or in-app.
> 
> Disabling implemented under feature flag "slack-disable-on-broken".
> Conditions for logging Implemented in BaseApiClient and SlackClient.

---------

Co-authored-by: getsantry[bot] <66042841+getsantry[bot]@users.noreply.github.com>
Co-authored-by: Colleen O'Rourke <[email protected]>

Author: 3 people

Diff for: src/sentry/conf/server.py

@@ -112,6 +112,7 @@ def env(
 SENTRY_TRANSACTION_NAMES_REDIS_CLUSTER = "default"
 SENTRY_WEBHOOK_LOG_REDIS_CLUSTER = "default"
 SENTRY_ARTIFACT_BUNDLES_INDEXING_REDIS_CLUSTER = "default"
+SENTRY_INTEGRATION_ERROR_LOG_REDIS_CLUSTER = "default"
 SENTRY_DEBUG_FILES_REDIS_CLUSTER = "default"
 
 # Hosts that are allowed to use system token authentication.
@@ -1340,6 +1341,8 @@ def SOCIAL_AUTH_DEFAULT_USERNAME() -> str:
     "organizations:create": True,
     # Enable usage of customer domains on the frontend
     "organizations:customer-domains": False,
+    # Allow disabling integrations when broken is detected
+    "organizations:slack-disable-on-broken": False,
     # Enable the 'discover' interface.
     "organizations:discover": False,
     # Enables events endpoint rate limit

Diff for: src/sentry/features/__init__.py

@@ -259,9 +259,11 @@
 default_manager.add("organizations:pr-comment-bot", OrganizationFeature, FeatureHandlerStrategy.INTERNAL)
 default_manager.add("organizations:ds-org-recalibration", OrganizationFeature, FeatureHandlerStrategy.INTERNAL)
 default_manager.add("organizations:slack-use-new-lookup", OrganizationFeature, FeatureHandlerStrategy.REMOTE)
+default_manager.add("organizations:slack-disable-on-broken", OrganizationFeature, FeatureHandlerStrategy.REMOTE)
 default_manager.add("organizations:sourcemaps-bundle-flat-file-indexing", OrganizationFeature, FeatureHandlerStrategy.REMOTE)
 default_manager.add("organizations:recap-server", OrganizationFeature, FeatureHandlerStrategy.INTERNAL)
 
+
 # Project scoped features
 default_manager.add("projects:alert-filters", ProjectFeature, FeatureHandlerStrategy.INTERNAL)
 default_manager.add("projects:custom-inbound-filters", ProjectFeature, FeatureHandlerStrategy.INTERNAL)

Diff for: src/sentry/integrations/discord/client.py

@@ -37,7 +37,7 @@ def __init__(
             org_integration_id = infer_org_integration(
                 integration_id=integration_id, ctx_logger=logger
             )
-        super().__init__(org_integration_id, verify_ssl, logging_context)
+        super().__init__(integration_id, org_integration_id, verify_ssl, logging_context)
 
     @control_silo_function
     def authorize_request(self, prepared_request: PreparedRequest) -> PreparedRequest:

Diff for: src/sentry/integrations/request_buffer.py

@@ -0,0 +1,147 @@
+import logging
+from datetime import datetime
+
+from django.conf import settings
+from redis.exceptions import WatchError
+
+from sentry.utils import json, redis
+
+BUFFER_SIZE = 30  # 30 days
+KEY_EXPIRY = 60 * 60 * 24 * 30  # 30 days
+
+IS_BROKEN_RANGE = 7  # 7 days
+
+
+class IntegrationRequestBuffer:
+    """
+    Create a data structure to store daily successful and failed request counts for each installed integration in Redis
+    This should store the aggregate counts of each type for last 30 days for each integration
+    """
+
+    def __init__(self, key):
+        self.integrationkey = key
+        logger = logging.getLogger(__name__)
+
+        try:
+            cluster_id = settings.SENTRY_INTEGRATION_ERROR_LOG_REDIS_CLUSTER
+            self.client = redis.redis_clusters.get(cluster_id)
+        except KeyError as e:
+            logger.info("no_redis_cluster", extra={"error": e, "cluster_id": cluster_id})
+
+    def _convert_obj_to_dict(self, redis_object):
+        """
+        Convert the request string stored in Redis to a python dict
+        """
+
+        return json.loads(redis_object)
+
+    def _get_all_from_buffer(self, buffer_key):
+        """
+        Get the list at the buffer key.
+        """
+
+        return self.client.lrange(buffer_key, 0, BUFFER_SIZE - 1)
+
+    def _get_broken_range_from_buffer(self, buffer_key):
+        """
+        Get the list at the buffer key in the broken range.
+        """
+
+        return self.client.lrange(buffer_key, 0, IS_BROKEN_RANGE - 1)
+
+    def _get(self):
+        """
+        Returns the list of daily aggregate error counts.
+        """
+        return [
+            self._convert_obj_to_dict(obj)
+            for obj in self._get_broken_range_from_buffer(self.integrationkey)
+        ]
+
+    def is_integration_broken(self):
+        """
+        Integration is broken if we have 7 consecutive days of errors and no successes OR have a fatal error
+
+        """
+        items = self._get()
+
+        data = [
+            datetime.strptime(item.get("date"), "%Y-%m-%d").date()
+            for item in items
+            if item.get("fatal_count", 0) > 0 and item.get("date")
+        ]
+
+        if len(data) > 0:
+            return True
+
+        data = [
+            datetime.strptime(item.get("date"), "%Y-%m-%d").date()
+            for item in items
+            if item.get("error_count", 0) > 0
+            and item.get("success_count", 0) == 0
+            and item.get("date")
+        ]
+
+        if not len(data):
+            return False
+
+        if len(data) < IS_BROKEN_RANGE:
+            return False
+
+        return True
+
+    def add(self, count: str):
+        VALID_KEYS = ["success", "error", "fatal"]
+        if count not in VALID_KEYS:
+            raise Exception("Requires a valid key param.")
+
+        other_count1, other_count2 = list(set(VALID_KEYS).difference([count]))[0:2]
+        now = datetime.now().strftime("%Y-%m-%d")
+
+        buffer_key = self.integrationkey
+        pipe = self.client.pipeline()
+
+        while True:
+            try:
+                pipe.watch(buffer_key)
+                recent_item_array = pipe.lrange(buffer_key, 0, 1)  # get first element from array
+                pipe.multi()
+                if len(recent_item_array):
+                    recent_item = self._convert_obj_to_dict(recent_item_array[0])
+                    if recent_item.get("date") == now:
+                        recent_item[f"{count}_count"] += 1
+                        pipe.lset(buffer_key, 0, json.dumps(recent_item))
+                    else:
+                        data = {
+                            "date": now,
+                            f"{count}_count": 1,
+                            f"{other_count1}_count": 0,
+                            f"{other_count2}_count": 0,
+                        }
+                        pipe.lpush(buffer_key, json.dumps(data))
+
+                else:
+                    data = {
+                        "date": now,
+                        f"{count}_count": 1,
+                        f"{other_count1}_count": 0,
+                        f"{other_count2}_count": 0,
+                    }
+                    pipe.lpush(buffer_key, json.dumps(data))
+                pipe.ltrim(buffer_key, 0, BUFFER_SIZE - 1)
+                pipe.expire(buffer_key, KEY_EXPIRY)
+                pipe.execute()
+                break
+            except WatchError:
+                continue
+            finally:
+                pipe.reset()
+
+    def record_error(self):
+        self.add("error")
+
+    def record_success(self):
+        self.add("success")
+
+    def record_fatal(self):
+        self.add("fatal")

Diff for: src/sentry/integrations/slack/client.py

@@ -38,7 +38,12 @@ def __init__(
                 integration_id=self.integration_id, ctx_logger=logger
             )
 
-        super().__init__(org_integration_id, verify_ssl, logging_context)
+        super().__init__(
+            org_integration_id=org_integration_id,
+            verify_ssl=verify_ssl,
+            integration_id=integration_id,
+            logging_context=logging_context,
+        )
 
     @control_silo_function
     def authorize_request(self, prepared_request: PreparedRequest) -> PreparedRequest:
@@ -57,13 +62,19 @@ def authorize_request(self, prepared_request: PreparedRequest) -> PreparedReques
         if not integration:
             logger.info("no_integration", extra={"path_url": prepared_request.path_url})
             return prepared_request
-
         token = (
             integration.metadata.get("user_access_token") or integration.metadata["access_token"]
         )
         prepared_request.headers["Authorization"] = f"Bearer {token}"
         return prepared_request
 
+    def is_response_fatal(self, response: Response) -> bool:
+        resp_json = response.json()
+        if not resp_json.get("ok"):
+            if "account_inactive" == resp_json.get("error", ""):
+                return True
+            return False
+
     def track_response_data(
         self,
         code: Union[str, int],
@@ -74,7 +85,6 @@ def track_response_data(
         # if no span was passed, create a dummy to which to add data to avoid having to wrap every
         # span call in `if span`
         span = span or Span()
-
         try:
             span.set_http_status(int(code))
         except ValueError:

Diff for: src/sentry/integrations/slack/integration.py

@@ -77,7 +77,7 @@ class SlackIntegration(SlackNotifyBasicMixin, IntegrationInstallation):
     def get_client(self) -> SlackClient:
         if not self.org_integration:
             raise IntegrationError("Organization Integration does not exist")
-        return SlackClient(org_integration_id=self.org_integration.id)
+        return SlackClient(org_integration_id=self.org_integration.id, integration_id=self.model.id)
 
     def get_config_data(self) -> Mapping[str, str]:
         metadata_ = self.model.metadata

Diff for: src/sentry/models/integrations/integration.py

@@ -68,7 +68,7 @@ def get_provider(self) -> IntegrationProvider:
 
     def delete(self, *args, **kwds):
         with outbox_context(
-            transaction.atomic(router.db_for_write(OrganizationIntegration)), flush=False
+            transaction.atomic(using=router.db_for_write(OrganizationIntegration)), flush=False
         ):
             for organization_integration in self.organizationintegration_set.all():
                 organization_integration.delete()
@@ -107,7 +107,7 @@ def add_organization(self, organization: RpcOrganization, user=None, default_aut
         from sentry.models import OrganizationIntegration
 
         try:
-            with transaction.atomic(router.db_for_write(OrganizationIntegration)):
+            with transaction.atomic(using=router.db_for_write(OrganizationIntegration)):
                 org_integration, created = OrganizationIntegration.objects.get_or_create(
                     organization_id=organization.id,
                     integration_id=self.id,
@@ -134,3 +134,11 @@ def add_organization(self, organization: RpcOrganization, user=None, default_aut
                 },
             )
             return False
+
+    def disable(self):
+        """
+        Disable this integration
+        """
+
+        self.update(status=ObjectStatus.DISABLED)
+        self.save()