ref(hc): Move PagerDutyService into control silo (#53084)

Notably, this migration is a net zero SQL, it only has to change the
state of these columns in django.

I'll want to add the constraint on organization_integration
incrementally as a separate PR for 2 reasons

1. It is unsafe to do so synchronously -- it will be plausible that
there are some organization_integration_ids that don't point to existing
rows.
2. Related to ^^ but since this column was a HybridCloudForeignKey, it
means the cascade was applied asynchronously for deletes. The goal will
be to roll this out, making future deletes stop cascading, then manually
clean up any objects pointing to missing OIs, *then* roll out the
migration that adds the constraint back in.

migrations_lockfile.txt

@@ -6,5 +6,5 @@ To resolve this, rebase against latest master and regenerate your migration. Thi
 will then be regenerated, and you should be able to merge without conflicts.
 
 nodestore: 0002_nodestore_no_dictfield
-sentry: 0515_slugify_invalid_monitors
+sentry: 0516_switch_pagerduty_silo
 social_auth: 0001_initial

src/sentry/api/serializers/rest_framework/notification_action.py

@@ -7,7 +7,6 @@
 from sentry.api.serializers.rest_framework.project import ProjectField
 from sentry.constants import SentryAppInstallationStatus
 from sentry.integrations.slack.utils.channel import get_channel_id, validate_channel_id
-from sentry.models.integrations.pagerduty_service import PagerDutyService
 from sentry.models.integrations.sentry_app_installation import SentryAppInstallation
 from sentry.models.notificationaction import ActionService, ActionTarget, NotificationAction
 from sentry.models.project import Project
@@ -220,14 +219,16 @@ def validate_pagerduty_service(
             return data
 
         service_id = data.get("target_identifier")
+        ois = integration_service.get_organization_integrations(
+            organization_id=self.context["organization"].id,
+            integration_id=self.integration.id,
+        )
 
         if not service_id:
             pd_service_options = [
                 f"{pds['id']} ({pds['service_name']})"
-                for pds in PagerDutyService.objects.filter(
-                    organization_id=self.context["organization"].id,
-                    integration_id=self.integration.id,
-                ).values("id", "service_name")
+                for oi in ois
+                for pds in oi.config.get("pagerduty_services", [])
             ]
 
             raise serializers.ValidationError(
@@ -236,18 +237,21 @@ def validate_pagerduty_service(
                 }
             )
 
-        pds = PagerDutyService.objects.filter(
-            organization_id=self.context["organization"].id,
-            integration_id=self.integration.id,
-        ).first()
-        if not pds or str(pds.id) != service_id:
+        try:
+            pds = next(
+                pds
+                for oi in ois
+                for pds in oi.config.get("pagerduty_services", [])
+                if service_id == str(pds["id"])
+            )
+        except StopIteration:
             raise serializers.ValidationError(
                 {
                     "target_identifier": f"Could not find associated PagerDuty service for the '{self.integration.name}' account. If it exists, ensure Sentry has access."
                 }
             )
-        data["target_display"] = pds.service_name
-        data["target_identifier"] = pds.id
+        data["target_display"] = pds["service_name"]
+        data["target_identifier"] = pds["id"]
         return data
 
     def validate(self, data: NotificationActionInputData) -> NotificationActionInputData:

src/sentry/integrations/pagerduty/actions/form.py

@@ -5,8 +5,8 @@
 from django import forms
 from django.utils.translation import gettext_lazy as _
 
-from sentry.models import PagerDutyService
 from sentry.services.hybrid_cloud.integration import integration_service
+from sentry.types.integrations import ExternalProviders
 
 
 def _validate_int_field(field: str, cleaned_data: Mapping[str, Any]) -> int | None:
@@ -43,25 +43,23 @@ def __init__(self, *args, **kwargs):
         self.fields["service"].choices = services
         self.fields["service"].widget.choices = self.fields["service"].choices
 
-    def _validate_service(self, service_id: int, integration_id: int | None) -> None:
+    def _validate_service(self, service_id: int, integration_id: int) -> None:
         params = {
             "account": dict(self.fields["account"].choices).get(integration_id),
             "service": dict(self.fields["service"].choices).get(service_id),
         }
 
-        try:
-            service = PagerDutyService.objects.get(id=service_id)
-        except PagerDutyService.DoesNotExist:
-            raise forms.ValidationError(
-                _('The service "%(service)s" does not exist in the %(account)s Pagerduty account.'),
-                code="invalid",
-                params=params,
-            )
-
-        integration = integration_service.get_integration(
-            organization_integration_id=service.organization_integration_id
+        org_integrations = integration_service.get_organization_integrations(
+            integration_id=integration_id,
+            providers=[ExternalProviders.PAGERDUTY.name],
         )
-        if not integration or integration.id != integration_id:
+
+        if not any(
+            pds
+            for oi in org_integrations
+            for pds in oi.config.get("pagerduty_services", [])
+            if pds["id"] == service_id
+        ):
             # We need to make sure that the service actually belongs to that integration,
             # meaning that it belongs under the appropriate account in PagerDuty.
             raise forms.ValidationError(

src/sentry/integrations/pagerduty/actions/notification.py

@@ -1,11 +1,10 @@
 from __future__ import annotations
 
 import logging
-from typing import Sequence
+from typing import Sequence, Tuple
 
 from sentry.integrations.pagerduty.actions import PagerDutyNotifyServiceForm
 from sentry.integrations.pagerduty.client import PagerDutyProxyClient
-from sentry.models import PagerDutyService
 from sentry.rules.actions import IntegrationEventAction
 from sentry.shared_integrations.client.proxy import infer_org_integration
 from sentry.shared_integrations.exceptions import ApiError
@@ -32,7 +31,13 @@ def __init__(self, *args, **kwargs):
         }
 
     def _get_service(self):
-        return PagerDutyService.objects.get(id=self.get_option("service"))
+        oi = self.get_organization_integration()
+        if not oi:
+            return None
+        for pds in oi.config.get("pagerduty_services", []):
+            if pds["id"] == self.get_option("service"):
+                return pds
+        return None
 
     def after(self, event, state):
         integration = self.get_integration()
@@ -41,9 +46,8 @@ def after(self, event, state):
             # integration removed but rule still exists
             return
 
-        try:
-            service = self._get_service()
-        except PagerDutyService.DoesNotExist:
+        service = self._get_service()
+        if not service:
             logger.exception("The PagerDuty does not exist anymore while integration does.")
             return
 
@@ -54,11 +58,11 @@ def send_notification(event, futures):
                 org_integration_id = org_integration.id
             else:
                 org_integration_id = infer_org_integration(
-                    integration_id=service.integration_id, ctx_logger=logger
+                    integration_id=service["integration_id"], ctx_logger=logger
                 )
             client = PagerDutyProxyClient(
                 org_integration_id=org_integration_id,
-                integration_key=service.integration_key,
+                integration_key=service["integration_key"],
             )
             try:
                 resp = client.send_trigger(event)
@@ -67,8 +71,8 @@ def send_notification(event, futures):
                     "rule.fail.pagerduty_trigger",
                     extra={
                         "error": str(e),
-                        "service_name": service.service_name,
-                        "service_id": service.id,
+                        "service_name": service["service_name"],
+                        "service_id": service["id"],
                         "project_id": event.project_id,
                         "event_id": event.event_id,
                     },
@@ -83,33 +87,31 @@ def send_notification(event, futures):
                     "status_code": resp.status_code,
                     "project_id": event.project_id,
                     "event_id": event.event_id,
-                    "service_name": service.service_name,
-                    "service_id": service.id,
+                    "service_name": service["service_name"],
+                    "service_id": service["id"],
                 },
             )
 
-        key = f"pagerduty:{integration.id}:{service.id}"
+        key = f"pagerduty:{integration.id}:{service['id']}"
         yield self.future(send_notification, key=key)
 
-    def get_services(self) -> Sequence[PagerDutyService]:
+    def get_services(self) -> Sequence[Tuple[int, str]]:
         from sentry.services.hybrid_cloud.integration import integration_service
 
         organization_integrations = integration_service.get_organization_integrations(
             providers=[self.provider], organization_id=self.project.organization_id
         )
-        integration_ids = [oi.integration_id for oi in organization_integrations]
-
         return [
-            service
-            for service in PagerDutyService.objects.filter(
-                organization_id=self.project.organization_id, integration_id__in=integration_ids
-            ).values_list("id", "service_name")
+            (v["id"], v["service_name"])
+            for oi in organization_integrations
+            for v in oi.config.get("pagerduty_services", [])
         ]
 
     def render_label(self):
-        try:
-            service_name = self._get_service().service_name
-        except PagerDutyService.DoesNotExist:
+        s = self._get_service()
+        if s:
+            service_name = s["service_name"]
+        else:
             service_name = "[removed]"
 
         return self.label.format(account=self.get_integration_name(), service=service_name)

src/sentry/integrations/pagerduty/integration.py

@@ -122,6 +122,8 @@ def update_organization_config(self, data):
                         organization_integration_id=self.org_integration.id,
                         service_name=service_name,
                         integration_key=key,
+                        integration_id=self.model.id,
+                        organization_id=self.organization_id,
                     )
 
     def get_config_data(self):
@@ -174,6 +176,8 @@ def post_install(
                     organization_integration_id=org_integration.id,
                     integration_key=service["integration_key"],
                     service_name=service["name"],
+                    organization_id=organization.id,
+                    integration_id=integration.id,
                 )
 
     def build_integration(self, state):

src/sentry/migrations/0516_switch_pagerduty_silo.py

@@ -0,0 +1,53 @@
+# Generated by Django 3.2.20 on 2023-07-18 16:07
+
+import django.db.models.deletion
+from django.db import migrations
+
+import sentry.db.models.fields.foreignkey
+import sentry.db.models.fields.hybrid_cloud_foreign_key
+from sentry.new_migrations.migrations import CheckedMigration
+
+
+class Migration(CheckedMigration):
+    # This flag is used to mark that a migration shouldn't be automatically run in production. For
+    # the most part, this should only be used for operations where it's safe to run the migration
+    # after your code has deployed. So this should not be used for most operations that alter the
+    # schema of a table.
+    # Here are some things that make sense to mark as dangerous:
+    # - Large data migrations. Typically we want these to be run manually by ops so that they can
+    #   be monitored and not block the deploy for a long period of time while they run.
+    # - Adding indexes to large tables. Since this can take a long time, we'd generally prefer to
+    #   have ops run this and not block the deploy. Note that while adding an index is a schema
+    #   change, it's completely safe to run the operation after the code has deployed.
+    is_dangerous = False
+
+    dependencies = [
+        ("sentry", "0515_slugify_invalid_monitors"),
+    ]
+
+    state_operations = [
+        migrations.RemoveField(
+            model_name="pagerdutyservice",
+            name="organization_integration_id",
+        ),
+        migrations.AddField(
+            model_name="pagerdutyservice",
+            name="organization_integration",
+            field=sentry.db.models.fields.foreignkey.FlexibleForeignKey(
+                default=-1,
+                on_delete=django.db.models.deletion.CASCADE,
+                to="sentry.organizationintegration",
+                db_constraint=False,
+            ),
+            preserve_default=False,
+        ),
+        migrations.AlterField(
+            model_name="pagerdutyservice",
+            name="organization_id",
+            field=sentry.db.models.fields.hybrid_cloud_foreign_key.HybridCloudForeignKey(
+                "sentry.Organization", db_index=True, on_delete="CASCADE"
+            ),
+        ),
+    ]
+
+    operations = [migrations.SeparateDatabaseAndState(state_operations=state_operations)]

src/sentry/models/integrations/pagerduty_service.py

@@ -1,21 +1,24 @@
 from django.db import models
+from django.db.models import CASCADE
 from django.utils import timezone
 
-from sentry.db.models import BoundedBigIntegerField, DefaultFieldsModel, region_silo_only_model
+from sentry.db.models import BoundedBigIntegerField, DefaultFieldsModel, FlexibleForeignKey
+from sentry.db.models.base import ModelSiloLimit
 from sentry.db.models.fields.hybrid_cloud_foreign_key import HybridCloudForeignKey
-from sentry.models.integrations.organization_integrity_backfill_mixin import (
-    OrganizationIntegrityBackfillMixin,
-)
+from sentry.silo import SiloMode
 
 
-@region_silo_only_model
-class PagerDutyService(OrganizationIntegrityBackfillMixin, DefaultFieldsModel):
+# Temporary -- this will become a control silo model again after a future getsentry merge.
+@ModelSiloLimit(SiloMode.CONTROL, SiloMode.REGION)
+class PagerDutyService(DefaultFieldsModel):
     __include_in_export__ = False
 
-    organization_integration_id = HybridCloudForeignKey(
-        "sentry.OrganizationIntegration", on_delete="CASCADE"
+    organization_integration = FlexibleForeignKey(
+        "sentry.OrganizationIntegration", on_delete=CASCADE, db_constraint=False
     )
-    organization_id = BoundedBigIntegerField(db_index=True)
+
+    organization_id = HybridCloudForeignKey("sentry.Organization", on_delete="cascade")
+
     # From a region point of view, you really only have per organization scoping.
     integration_id = BoundedBigIntegerField(db_index=False)
     integration_key = models.CharField(max_length=255)

src/sentry/services/hybrid_cloud/integration/impl.py

@@ -187,7 +187,7 @@ def get_organization_integrations(
         if not oi_kwargs:
             return []
 
-        ois = OrganizationIntegration.objects.filter(**oi_kwargs)
+        ois = OrganizationIntegration.objects.filter(**oi_kwargs).select_related("integration")
 
         if limit is not None:
             ois = ois[:limit]
@@ -233,7 +233,7 @@ def get_organization_contexts(
         )
         return (
             serialize_integration(integration),
-            [serialize_organization_integration(oi) for oi in organization_integrations],
+            organization_integrations,
         )
 
     def update_integrations(
@@ -325,7 +325,7 @@ def update_organization_integration(
             grace_period_end=grace_period_end,
             set_grace_period_end_null=set_grace_period_end_null,
         )
-        return serialize_organization_integration(ois[0]) if len(ois) > 0 else None
+        return ois[0] if len(ois) > 0 else None
 
     def send_incident_alert_notification(
         self,

src/sentry/services/hybrid_cloud/integration/serial.py

@@ -1,5 +1,8 @@
-from sentry.models import Integration, OrganizationIntegration
+from typing import Any, Dict
+
+from sentry.models import Integration, OrganizationIntegration, PagerDutyService
 from sentry.services.hybrid_cloud.integration import RpcIntegration, RpcOrganizationIntegration
+from sentry.types.integrations import ExternalProviders
 
 
 def serialize_integration(integration: Integration) -> RpcIntegration:
@@ -14,12 +17,24 @@ def serialize_integration(integration: Integration) -> RpcIntegration:
 
 
 def serialize_organization_integration(oi: OrganizationIntegration) -> RpcOrganizationIntegration:
+    config: Dict[str, Any] = dict(**oi.config)
+    if oi.integration.provider == ExternalProviders.PAGERDUTY.name:
+        config["pagerduty_services"] = [
+            dict(
+                integration_id=pds.integration_id,
+                integration_key=pds.integration_key,
+                service_name=pds.service_name,
+                id=pds.id,
+            )
+            for pds in PagerDutyService.objects.filter(organization_integration_id=oi.id)
+        ]
+
     return RpcOrganizationIntegration(
         id=oi.id,
         default_auth_id=oi.default_auth_id,
         organization_id=oi.organization_id,
         integration_id=oi.integration_id,
-        config=oi.config,
+        config=config,
         status=oi.status,
         grace_period_end=oi.grace_period_end,
     )