♻️ ref(integrations): Add SLO for link_all_repos Celery Task (#81029)

`link_all_repos` is only for Github at the moment, but added
observability on the task here.

also updated the `SCMIntegrationInteractionEvent` to accept org_id

Author: iamrajjoshi

src/sentry/integrations/github/tasks/link_all_repos.py

@@ -2,6 +2,11 @@
 
 from sentry.constants import ObjectStatus
 from sentry.integrations.services.integration import integration_service
+from sentry.integrations.source_code_management.metrics import (
+    LinkAllReposHaltReason,
+    SCMIntegrationInteractionEvent,
+    SCMIntegrationInteractionType,
+)
 from sentry.organizations.services.organization import organization_service
 from sentry.plugins.providers.integration_repository import (
     RepoExistsError,
@@ -35,52 +40,71 @@ def link_all_repos(
     integration_id: int,
     organization_id: int,
 ):
-    integration = integration_service.get_integration(
-        integration_id=integration_id, status=ObjectStatus.ACTIVE
-    )
-    if not integration:
-        logger.error(
-            "%s.link_all_repos.integration_missing",
-            integration_key,
-            extra={"organization_id": organization_id},
-        )
-        metrics.incr("github.link_all_repos.error", tags={"type": "missing_integration"})
-        return
-
-    rpc_org = organization_service.get(id=organization_id)
-    if rpc_org is None:
-        logger.error(
-            "%s.link_all_repos.organization_missing",
-            integration_key,
-            extra={"organization_id": organization_id},
-        )
-        metrics.incr(
-            f"{integration_key}.link_all_repos.error",
-            tags={"type": "missing_organization"},
-        )
-        return
-
-    installation = integration.get_installation(organization_id=organization_id)
 
-    client = installation.get_client()
+    with SCMIntegrationInteractionEvent(
+        interaction_type=SCMIntegrationInteractionType.LINK_ALL_REPOS,
+        provider_key=integration_key,
+    ).capture() as lifecycle:
+        lifecycle.add_extra("organization_id", organization_id)
+        integration = integration_service.get_integration(
+            integration_id=integration_id, status=ObjectStatus.ACTIVE
+        )
+        if not integration:
+            # TODO: Remove this logger in favor of context manager
+            logger.error(
+                "%s.link_all_repos.integration_missing",
+                integration_key,
+                extra={"organization_id": organization_id},
+            )
+            metrics.incr("github.link_all_repos.error", tags={"type": "missing_integration"})
+            lifecycle.record_failure(str(LinkAllReposHaltReason.MISSING_INTEGRATION))
+            return
 
-    try:
-        repositories = client.get_repositories(fetch_max_pages=True)
-    except ApiError as e:
-        if installation.is_rate_limited_error(e):
+        rpc_org = organization_service.get(id=organization_id)
+        if rpc_org is None:
+            logger.error(
+                "%s.link_all_repos.organization_missing",
+                integration_key,
+                extra={"organization_id": organization_id},
+            )
+            metrics.incr(
+                f"{integration_key}.link_all_repos.error",
+                tags={"type": "missing_organization"},
+            )
+            lifecycle.record_failure(str(LinkAllReposHaltReason.MISSING_ORGANIZATION))
             return
 
-        metrics.incr(f"{integration_key}.link_all_repos.api_error")
-        raise
+        installation = integration.get_installation(organization_id=organization_id)
 
-    integration_repo_provider = get_integration_repository_provider(integration)
+        client = installation.get_client()
 
-    for repo in repositories:
         try:
-            config = get_repo_config(repo, integration_id)
-            integration_repo_provider.create_repository(repo_config=config, organization=rpc_org)
-        except KeyError:
-            continue
-        except RepoExistsError:
-            metrics.incr("sentry.integration_repo_provider.repo_exists")
-            continue
+            repositories = client.get_repositories(fetch_max_pages=True)
+        except ApiError as e:
+            if installation.is_rate_limited_error(e):
+                lifecycle.record_halt(str(LinkAllReposHaltReason.RATE_LIMITED))
+                return
+
+            metrics.incr(f"{integration_key}.link_all_repos.api_error")
+            raise
+
+        integration_repo_provider = get_integration_repository_provider(integration)
+
+        # If we successfully create any repositories, we'll set this to True
+        success = False
+
+        for repo in repositories:
+            try:
+                config = get_repo_config(repo, integration_id)
+                integration_repo_provider.create_repository(
+                    repo_config=config, organization=rpc_org
+                )
+                success = True
+            except KeyError:
+                continue
+            except RepoExistsError:
+                metrics.incr("sentry.integration_repo_provider.repo_exists")
+                continue
+
+        if not success:
+            lifecycle.record_halt(str(LinkAllReposHaltReason.REPOSITORY_NOT_CREATED))

src/sentry/integrations/source_code_management/metrics.py

@@ -1,5 +1,5 @@
 from collections.abc import Mapping
-from enum import Enum
+from enum import Enum, StrEnum
 from typing import Any
 
 from attr import dataclass
@@ -29,6 +29,9 @@ class SCMIntegrationInteractionType(Enum):
     CREATE_COMMENT = "CREATE_COMMENT"
     UPDATE_COMMENT = "UPDATE_COMMENT"
 
+    # Tasks
+    LINK_ALL_REPOS = "LINK_ALL_REPOS"
+
     def __str__(self) -> str:
         return self.value.lower()
 
@@ -60,3 +63,12 @@ def get_extras(self) -> Mapping[str, Any]:
             "organization_id": (self.organization.id if self.organization else None),
             "org_integration_id": (self.org_integration.id if self.org_integration else None),
         }
+
+
+class LinkAllReposHaltReason(StrEnum):
+    """Common reasons why a link all repos task may halt without success/failure."""
+
+    MISSING_INTEGRATION = "missing_integration"
+    MISSING_ORGANIZATION = "missing_organization"
+    RATE_LIMITED = "rate_limited"
+    REPOSITORY_NOT_CREATED = "repository_not_created"

tests/sentry/integrations/github/tasks/test_link_all_repos.py

@@ -7,11 +7,17 @@
 from sentry.constants import ObjectStatus
 from sentry.integrations.github.integration import GitHubIntegrationProvider
 from sentry.integrations.github.tasks.link_all_repos import link_all_repos
+from sentry.integrations.source_code_management.metrics import LinkAllReposHaltReason
+from sentry.integrations.types import EventLifecycleOutcome
 from sentry.models.repository import Repository
 from sentry.shared_integrations.exceptions import ApiError
 from sentry.silo.base import SiloMode
 from sentry.testutils.cases import IntegrationTestCase
 from sentry.testutils.silo import assume_test_silo_mode, control_silo_test
+from tests.sentry.integrations.utils.test_assert_metrics import (
+    assert_failure_metric,
+    assert_halt_metric,
+)
 
 
 @control_silo_test
@@ -41,8 +47,9 @@ def _add_responses(self):
             },
         )
 
+    @patch("sentry.integrations.utils.metrics.EventLifecycle.record_event")
     @patch("sentry.integrations.github.tasks.link_all_repos.metrics")
-    def test_link_all_repos_inactive_integration(self, mock_metrics, _):
+    def test_link_all_repos_inactive_integration(self, mock_metrics, mock_record, _):
         self.integration.update(status=ObjectStatus.DISABLED)
 
         link_all_repos(
@@ -55,8 +62,14 @@ def test_link_all_repos_inactive_integration(self, mock_metrics, _):
             "github.link_all_repos.error", tags={"type": "missing_integration"}
         )
 
+        start, halt = mock_record.mock_calls
+        assert start.args[0] == EventLifecycleOutcome.STARTED
+        assert halt.args[0] == EventLifecycleOutcome.FAILURE
+        assert_failure_metric(mock_record, LinkAllReposHaltReason.MISSING_INTEGRATION.value)
+
     @responses.activate
-    def test_link_all_repos(self, _):
+    @patch("sentry.integrations.utils.metrics.EventLifecycle.record_event")
+    def test_link_all_repos(self, mock_record, _):
         self._add_responses()
 
         link_all_repos(
@@ -76,8 +89,13 @@ def test_link_all_repos(self, _):
         assert repos[0].name == "getsentry/sentry"
         assert repos[1].name == "getsentry/snuba"
 
+        start, halt = mock_record.mock_calls
+        assert start.args[0] == EventLifecycleOutcome.STARTED
+        assert halt.args[0] == EventLifecycleOutcome.SUCCESS
+
     @responses.activate
-    def test_link_all_repos_api_response_keyerror(self, _):
+    @patch("sentry.integrations.utils.metrics.EventLifecycle.record_event")
+    def test_link_all_repos_api_response_keyerror(self, mock_record, _):
 
         responses.add(
             responses.GET,
@@ -112,8 +130,48 @@ def test_link_all_repos_api_response_keyerror(self, _):
 
         assert repos[0].name == "getsentry/snuba"
 
+        start, halt = mock_record.mock_calls
+        assert start.args[0] == EventLifecycleOutcome.STARTED
+        assert halt.args[0] == EventLifecycleOutcome.SUCCESS
+
+    @responses.activate
+    @patch("sentry.integrations.utils.metrics.EventLifecycle.record_event")
+    def test_link_all_repos_api_response_keyerror_single_repo(self, mock_record, _):
+
+        responses.add(
+            responses.GET,
+            self.base_url + "/installation/repositories?per_page=100",
+            status=200,
+            json={
+                "total_count": 2,
+                "repositories": [
+                    {
+                        "full_name": "getsentry/sentry",
+                    },
+                ],
+            },
+        )
+
+        link_all_repos(
+            integration_key=self.key,
+            integration_id=self.integration.id,
+            organization_id=self.organization.id,
+        )
+
+        with assume_test_silo_mode(SiloMode.REGION):
+            repos = Repository.objects.all()
+        assert len(repos) == 0
+
+        start, halt = mock_record.mock_calls
+
+        start, halt = mock_record.mock_calls
+        assert start.args[0] == EventLifecycleOutcome.STARTED
+        assert halt.args[0] == EventLifecycleOutcome.HALTED
+        assert_halt_metric(mock_record, LinkAllReposHaltReason.REPOSITORY_NOT_CREATED.value)
+
+    @patch("sentry.integrations.utils.metrics.EventLifecycle.record_event")
     @patch("sentry.integrations.github.tasks.link_all_repos.metrics")
-    def test_link_all_repos_missing_integration(self, mock_metrics, _):
+    def test_link_all_repos_missing_integration(self, mock_metrics, mock_record, _):
         link_all_repos(
             integration_key=self.key,
             integration_id=0,
@@ -123,8 +181,14 @@ def test_link_all_repos_missing_integration(self, mock_metrics, _):
             "github.link_all_repos.error", tags={"type": "missing_integration"}
         )
 
+        start, halt = mock_record.mock_calls
+        assert start.args[0] == EventLifecycleOutcome.STARTED
+        assert halt.args[0] == EventLifecycleOutcome.FAILURE
+        assert_failure_metric(mock_record, LinkAllReposHaltReason.MISSING_INTEGRATION.value)
+
+    @patch("sentry.integrations.utils.metrics.EventLifecycle.record_event")
     @patch("sentry.integrations.github.tasks.link_all_repos.metrics")
-    def test_link_all_repos_missing_organization(self, mock_metrics, _):
+    def test_link_all_repos_missing_organization(self, mock_metrics, mock_record, _):
         link_all_repos(
             integration_key=self.key,
             integration_id=self.integration.id,
@@ -134,9 +198,15 @@ def test_link_all_repos_missing_organization(self, mock_metrics, _):
             "github.link_all_repos.error", tags={"type": "missing_organization"}
         )
 
+        start, halt = mock_record.mock_calls
+        assert start.args[0] == EventLifecycleOutcome.STARTED
+        assert halt.args[0] == EventLifecycleOutcome.FAILURE
+        assert_failure_metric(mock_record, LinkAllReposHaltReason.MISSING_ORGANIZATION.value)
+
+    @patch("sentry.integrations.utils.metrics.EventLifecycle.record_event")
     @patch("sentry.integrations.github.tasks.link_all_repos.metrics")
     @responses.activate
-    def test_link_all_repos_api_error(self, mock_metrics, _):
+    def test_link_all_repos_api_error(self, mock_metrics, mock_record, _):
 
         responses.add(
             responses.GET,
@@ -150,11 +220,16 @@ def test_link_all_repos_api_error(self, mock_metrics, _):
                 integration_id=self.integration.id,
                 organization_id=self.organization.id,
             )
-            mock_metrics.incr.assert_called_with("github.link_all_repos.api_error")
+        mock_metrics.incr.assert_called_with("github.link_all_repos.api_error")
 
+        start, halt = mock_record.mock_calls
+        assert start.args[0] == EventLifecycleOutcome.STARTED
+        assert halt.args[0] == EventLifecycleOutcome.FAILURE
+
+    @patch("sentry.integrations.utils.metrics.EventLifecycle.record_event")
     @patch("sentry.integrations.github.integration.metrics")
     @responses.activate
-    def test_link_all_repos_api_error_rate_limited(self, mock_metrics, _):
+    def test_link_all_repos_api_error_rate_limited(self, mock_metrics, mock_record, _):
 
         responses.add(
             responses.GET,
@@ -173,10 +248,16 @@ def test_link_all_repos_api_error_rate_limited(self, mock_metrics, _):
         )
         mock_metrics.incr.assert_called_with("github.link_all_repos.rate_limited_error")
 
+        start, halt = mock_record.mock_calls
+        assert start.args[0] == EventLifecycleOutcome.STARTED
+        assert halt.args[0] == EventLifecycleOutcome.HALTED
+        assert_halt_metric(mock_record, LinkAllReposHaltReason.RATE_LIMITED.value)
+
+    @patch("sentry.integrations.utils.metrics.EventLifecycle.record_event")
     @patch("sentry.models.Repository.objects.create")
     @patch("sentry.integrations.github.tasks.link_all_repos.metrics")
     @responses.activate
-    def test_link_all_repos_repo_creation_error(self, mock_metrics, mock_repo, _):
+    def test_link_all_repos_repo_creation_error(self, mock_metrics, mock_repo, mock_record, _):
         mock_repo.side_effect = IntegrityError
 
         self._add_responses()
@@ -189,11 +270,17 @@ def test_link_all_repos_repo_creation_error(self, mock_metrics, mock_repo, _):
 
         mock_metrics.incr.assert_called_with("sentry.integration_repo_provider.repo_exists")
 
+        start, halt = mock_record.mock_calls
+        assert start.args[0] == EventLifecycleOutcome.STARTED
+        assert halt.args[0] == EventLifecycleOutcome.HALTED
+        assert_halt_metric(mock_record, LinkAllReposHaltReason.REPOSITORY_NOT_CREATED.value)
+
+    @patch("sentry.integrations.utils.metrics.EventLifecycle.record_event")
     @patch("sentry.integrations.services.repository.repository_service.create_repository")
     @patch("sentry.plugins.providers.IntegrationRepositoryProvider.on_delete_repository")
     @responses.activate
     def test_link_all_repos_repo_creation_exception(
-        self, mock_delete_repo, mock_create_repository, _
+        self, mock_delete_repo, mock_create_repository, mock_record, _
     ):
         mock_create_repository.return_value = None
         mock_delete_repo.side_effect = Exception
@@ -206,3 +293,7 @@ def test_link_all_repos_repo_creation_exception(
                 integration_id=self.integration.id,
                 organization_id=self.organization.id,
             )
+
+        start, halt = mock_record.mock_calls
+        assert start.args[0] == EventLifecycleOutcome.STARTED
+        assert halt.args[0] == EventLifecycleOutcome.FAILURE