Update permissions crawling so that it doesn't fail if a secret scope disappears during crawling

src/databricks/labs/ucx/workspace_access/manager.py

@@ -31,6 +31,7 @@ def _crawl(self) -> Iterable[Permissions]:
         logger.debug("Crawling permissions")
         crawler_tasks = list(self._get_crawler_tasks())
         logger.info(f"Starting to crawl permissions. Total tasks: {len(crawler_tasks)}")
+        # Note: tasks can return Permissions | None, but threads.gather() filters out None results.
         items, errors = Threads.gather("crawl permissions", crawler_tasks)
         acute_errors = []
         for error in errors:

src/databricks/labs/ucx/workspace_access/secrets.py

@@ -6,6 +6,7 @@
 
 from databricks.labs.blueprint.limiter import rate_limited
 from databricks.sdk import WorkspaceClient
+from databricks.sdk.errors import NotFound
 from databricks.sdk.retries import retried
 from databricks.sdk.service import workspace
 from databricks.sdk.service.workspace import AclItem
@@ -31,14 +32,19 @@ def __init__(
         self._verify_timeout = verify_timeout
         self._include_object_permissions = include_object_permissions
 
-    def get_crawler_tasks(self):
-        def _crawler_task(scope: workspace.SecretScope):
+    def get_crawler_tasks(self) -> Iterable[Callable[[], Permissions | None]]:
+        def _crawler_task(scope: workspace.SecretScope) -> Permissions | None:
             assert scope.name is not None
-            acl_items = self._ws.secrets.list_acls(scope.name)
+            try:
+                acl_items = self._ws.secrets.list_acls(scope.name)
+                acl_items_raw = [item.as_dict() for item in acl_items]
+            except NotFound:
+                logger.warning(f"Secret scope disappeared, cannot assess: {scope.name}")
+                return None
             return Permissions(
                 object_id=scope.name,
                 object_type="secrets",
-                raw=json.dumps([item.as_dict() for item in acl_items]),
+                raw=json.dumps(acl_items_raw),
             )
 
         if self._include_object_permissions:

tests/unit/workspace_access/test_secrets.py

@@ -1,10 +1,13 @@
 import json
+import logging
+from collections.abc import Iterator
 from datetime import timedelta
 from unittest.mock import call, create_autospec
 
 import pytest
 from databricks.sdk import WorkspaceClient
 from databricks.sdk.service import workspace
+from databricks.sdk.errors import ResourceDoesNotExist
 
 from databricks.labs.ucx.workspace_access.groups import MigrationState
 from databricks.labs.ucx.workspace_access.secrets import (
@@ -42,6 +45,36 @@ def test_secret_scopes_crawler():
     assert item.raw == '[{"permission": "MANAGE", "principal": "test"}]'
 
 
+def test_secret_scopes_disappearing_during_crawl(ws, caplog) -> None:
+    """Verify that when crawling secret scopes we continue instead of failing when a secret scope disappears."""
+
+    ws.secrets.list_scopes.return_value = [
+        workspace.SecretScope(name="will_remain"),
+        workspace.SecretScope(name="will_disappear"),
+    ]
+
+    def mock_list_acls(scope: str) -> Iterator[workspace.AclItem]:
+        if scope == "will_disappear":
+            raise ResourceDoesNotExist("Simulated disappearance")
+        yield workspace.AclItem(principal="a_principal", permission=workspace.AclPermission.MANAGE)
+
+    ws.secrets.list_acls = mock_list_acls
+
+    sup = SecretScopesSupport(ws=ws)
+
+    tasks = list(sup.get_crawler_tasks())
+    with caplog.at_level(logging.WARNING):
+        task_results = [task() for task in tasks]
+
+    assert task_results == [
+        Permissions(
+            object_id="will_remain", object_type="secrets", raw='[{"permission": "MANAGE", "principal": "a_principal"}]'
+        ),
+        None,
+    ]
+    assert "Secret scope disappeared, cannot assess: will_disappear" in caplog.messages
+
+
 def test_secret_scopes_crawler_include():
     ws = create_autospec(WorkspaceClient)
     ws.secrets.list_acls.return_value = [