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

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.
@@ -1348,6 +1349,8 @@ def SOCIAL_AUTH_DEFAULT_USERNAME() -> str:
     "organizations:crons-timeline-listing-page": False,
     # 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

src/sentry/features/__init__.py

@@ -259,8 +259,10 @@
 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)
 
+
 # Project scoped features
 default_manager.add("projects:alert-filters", ProjectFeature, FeatureHandlerStrategy.INTERNAL)
 default_manager.add("projects:custom-inbound-filters", ProjectFeature, FeatureHandlerStrategy.INTERNAL)

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:

src/sentry/integrations/request_buffer.py

@@ -0,0 +1,133 @@
+import logging
+from datetime import datetime, timedelta
+
+from django.conf import settings
+
+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(self):
+        """
+        Returns the list of daily aggregate error counts.
+        """
+        return [
+            self._convert_obj_to_dict(obj) for obj in self._get_all_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
+
+        """
+        data = [
+            datetime.strptime(item.get("date"), "%Y-%m-%d").date()
+            for item in self._get()
+            if item.get("fatal_count", 0) > 0 and item.get("date")
+        ][0:IS_BROKEN_RANGE]
+
+        if len(data) > 0:
+            return True
+
+        data = [
+            datetime.strptime(item.get("date"), "%Y-%m-%d").date()
+            for item in self._get()
+            if item.get("error_count", 0) > 0
+            and item.get("success_count", 0) == 0
+            and item.get("date")
+        ][0:IS_BROKEN_RANGE]
+
+        if not len(data):
+            return False
+
+        if len(data) < IS_BROKEN_RANGE:
+            return False
+
+        date_set = {data[0] - timedelta(x) for x in range((data[0] - data[-1]).days)}
+        missing = list(date_set - set(data))
+        if len(missing):
+            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]
+        buffer_key = self.integrationkey
+        now = datetime.now().strftime("%Y-%m-%d")
+
+        pipe = self.client.pipeline()
+
+        # get first element from array
+        recent_item_array = self.client.lrange(buffer_key, 0, 1)
+        if len(recent_item_array):
+            recent_item = json.loads(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.expire(buffer_key, KEY_EXPIRY)
+        pipe.execute()
+
+    def record_error(self):
+        self.add("error")
+
+    def record_success(self):
+        self.add("success")
+
+    def record_fatal(self):
+        self.add("fatal")

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:

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

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()