feat(crons): Add api for fetching checkin processing errors for an org (

#70563)

This allows us to return processing errors for organizations. Can be
filtered by project id

<!-- Describe your PR here. -->

src/sentry/api/urls.py

@@ -648,6 +648,9 @@
 
 __all__ = ("urlpatterns",)
 
+from ..monitors.endpoints.organization_monitor_processing_errors_index import (
+    OrganizationMonitorProcessingErrorsIndexEndpoint,
+)
 
 # issues endpoints are available both top level (by numerical ID) as well as coupled
 # to the organization (and queryable via short ID)
@@ -1646,6 +1649,11 @@ def create_group_urls(name_prefix: str) -> list[URLPattern | URLResolver]:
         OrganizationMonitorIndexStatsEndpoint.as_view(),
         name="sentry-api-0-organization-monitor-index-stats",
     ),
+    re_path(
+        r"^(?P<organization_slug>[^\/]+)/processing-errors/$",
+        OrganizationMonitorProcessingErrorsIndexEndpoint.as_view(),
+        name="sentry-api-0-organization-monitor-processing-errors-index",
+    ),
     re_path(
         r"^(?P<organization_slug>[^\/]+)/monitors-schedule-data/$",
         OrganizationMonitorScheduleSampleDataEndpoint.as_view(),

src/sentry/monitors/endpoints/organization_monitor_processing_errors_index.py

@@ -0,0 +1,57 @@
+from drf_spectacular.utils import extend_schema
+from rest_framework.response import Response
+
+from sentry.api.api_owners import ApiOwner
+from sentry.api.api_publish_status import ApiPublishStatus
+from sentry.api.base import region_silo_endpoint
+from sentry.api.bases import OrganizationEndpoint
+from sentry.api.paginator import SequencePaginator
+from sentry.api.serializers import serialize
+from sentry.apidocs.constants import RESPONSE_FORBIDDEN, RESPONSE_NOT_FOUND, RESPONSE_UNAUTHORIZED
+from sentry.apidocs.parameters import GlobalParams
+from sentry.apidocs.utils import inline_sentry_response_serializer
+from sentry.models.organization import Organization
+from sentry.monitors.processing_errors import (
+    CheckinProcessErrorsManager,
+    CheckinProcessingErrorData,
+)
+from sentry.utils.auth import AuthenticatedHttpRequest
+
+
+@region_silo_endpoint
+@extend_schema(tags=["Crons"])
+class OrganizationMonitorProcessingErrorsIndexEndpoint(OrganizationEndpoint):
+    publish_status = {
+        "GET": ApiPublishStatus.PRIVATE,
+    }
+    owner = ApiOwner.CRONS
+
+    @extend_schema(
+        operation_id="Retrieve checkin processing errors for an Organization",
+        parameters=[
+            GlobalParams.ORG_SLUG,
+            GlobalParams.PROJECT_ID_OR_SLUG,
+        ],
+        responses={
+            200: inline_sentry_response_serializer(
+                "CheckinProcessingError", list[CheckinProcessingErrorData]
+            ),
+            401: RESPONSE_UNAUTHORIZED,
+            403: RESPONSE_FORBIDDEN,
+            404: RESPONSE_NOT_FOUND,
+        },
+    )
+    def get(self, request: AuthenticatedHttpRequest, organization: Organization) -> Response:
+        """
+        Retrieves checkin processing errors for a monitor
+        """
+        projects = self.get_projects(request, organization)
+        paginator = SequencePaginator(
+            list(enumerate(CheckinProcessErrorsManager().get_for_projects(projects)))
+        )
+
+        return self.paginate(
+            request=request,
+            paginator=paginator,
+            on_results=lambda results: serialize(results, request.user),
+        )

src/sentry/monitors/processing_errors.py

@@ -5,6 +5,7 @@
 import uuid
 from datetime import timedelta
 from enum import Enum
+from itertools import chain
 from typing import Any, TypedDict
 
 from django.conf import settings
@@ -159,19 +160,29 @@ def build_monitor_identifier(self, monitor: Monitor) -> str:
         return f"monitor:{monitor.id}"
 
     def get_for_monitor(self, monitor: Monitor) -> list[CheckinProcessingError]:
-        return self._get_for_entity(self.build_monitor_identifier(monitor))
+        return self._get_for_entities([self.build_monitor_identifier(monitor)])
 
     def build_project_identifier(self, project_id: int) -> str:
         return f"project:{project_id}"
 
-    def get_for_project(self, project: Project) -> list[CheckinProcessingError]:
-        return self._get_for_entity(self.build_project_identifier(project.id))
+    def get_for_projects(self, projects: list[Project]) -> list[CheckinProcessingError]:
+        return self._get_for_entities(
+            [self.build_project_identifier(project.id) for project in projects]
+        )
 
-    def _get_for_entity(self, identifier: str) -> list[CheckinProcessingError]:
+    def _get_for_entities(self, identifiers: list[str]) -> list[CheckinProcessingError]:
         redis = self._get_cluster()
-        error_key = f"monitors.processing_errors.{identifier}"
-        raw_errors = redis.zrange(error_key, 0, MAX_ERRORS_PER_SET, desc=True)
-        return [CheckinProcessingError.from_dict(json.loads(raw_error)) for raw_error in raw_errors]
+        pipeline = redis.pipeline()
+        for identifier in identifiers:
+            pipeline.zrange(
+                f"monitors.processing_errors.{identifier}", 0, MAX_ERRORS_PER_SET, desc=True
+            )
+        errors = [
+            CheckinProcessingError.from_dict(json.loads(raw_error))
+            for raw_error in chain(*pipeline.execute())
+        ]
+        errors.sort(key=lambda error: error.checkin.ts.timestamp(), reverse=True)
+        return errors
 
 
 def handle_processing_errors(item: CheckinItem, error: CheckinValidationError):

tests/sentry/monitors/endpoints/test_organization_monitor_processing_errors.py

@@ -0,0 +1,62 @@
+from sentry.api.serializers import serialize
+from sentry.monitors.processing_errors import (
+    CheckinProcessErrorsManager,
+    ProcessingError,
+    ProcessingErrorType,
+)
+from sentry.monitors.testutils import build_checkin_processing_error
+from sentry.testutils.cases import APITestCase, MonitorTestCase
+from sentry.utils import json
+
+
+class OrganizationMonitorProcessingErrorsIndexEndpointTest(MonitorTestCase, APITestCase):
+    endpoint = "sentry-api-0-organization-monitor-processing-errors-index"
+
+    def setUp(self):
+        super().setUp()
+        self.login_as(user=self.user)
+
+    def test_empty(self):
+        resp = self.get_success_response(self.organization.slug)
+        assert resp.data == []
+
+    def test(self):
+        monitor = self.create_monitor()
+        project_2 = self.create_project()
+
+        manager = CheckinProcessErrorsManager()
+        monitor_error = build_checkin_processing_error(
+            [ProcessingError(ProcessingErrorType.CHECKIN_INVALID_GUID, {"guid": "bad"})],
+            message_overrides={"project_id": self.project.id},
+            payload_overrides={"monitor_slug": monitor.slug},
+        )
+
+        project_errors = [
+            build_checkin_processing_error(
+                [ProcessingError(ProcessingErrorType.ORGANIZATION_KILLSWITCH_ENABLED)],
+                message_overrides={"project_id": self.project.id},
+            ),
+            build_checkin_processing_error(
+                [ProcessingError(ProcessingErrorType.MONITOR_DISABLED, {"some": "data"})],
+                message_overrides={"project_id": self.project.id},
+            ),
+            build_checkin_processing_error(
+                [ProcessingError(ProcessingErrorType.MONITOR_DISABLED, {"some": "data"})],
+                message_overrides={"project_id": project_2.id},
+            ),
+        ]
+
+        manager.store(monitor_error, monitor)
+        for error in project_errors:
+            manager.store(error, None)
+
+        resp = self.get_success_response(
+            self.organization.slug, project=[self.project.id, project_2.id]
+        )
+        assert resp.data == json.loads(json.dumps(serialize(list(reversed(project_errors)))))
+
+        resp = self.get_success_response(self.organization.slug, project=[self.project.id])
+        assert resp.data == json.loads(json.dumps(serialize(list(reversed(project_errors[:2])))))
+
+        resp = self.get_success_response(self.organization.slug, project=[project_2.id])
+        assert resp.data == json.loads(json.dumps(serialize(list(reversed(project_errors[2:])))))

tests/sentry/monitors/test_processing_errors.py

@@ -61,7 +61,7 @@ def test_store_with_slug_not_exist(self):
         )
 
         manager.store(processing_error, None)
-        fetched_processing_error = manager.get_for_project(self.project)
+        fetched_processing_error = manager.get_for_projects([self.project])
         assert len(fetched_processing_error) == 1
         self.assert_processing_errors_equal(processing_error, fetched_processing_error[0])
 
@@ -107,7 +107,7 @@ def test_get_for_monitor_empty(self):
 
     def test_get_for_project(self):
         manager = CheckinProcessErrorsManager()
-        assert len(manager.get_for_project(self.project)) == 0
+        assert len(manager.get_for_projects([self.project])) == 0
 
 
 class HandleProcessingErrorsTest(TestCase):