Removed thread pool for any IAM Group removals and additions (#394)

Resolves #384
Closes #383

Author: nfx

src/databricks/labs/ucx/framework/parallel.py

@@ -48,24 +48,22 @@ def _run(self) -> (list[Result], list[Exception]):
             if result is None:
                 continue
             collected.append(result)
-        self._on_finish(given_cnt, len(errors))
+        self._on_finish(given_cnt, len(collected), len(errors))
 
         return collected, errors
 
-    def _on_finish(self, given_cnt, failed_cnt):
+    def _on_finish(self, given_cnt: int, collected_cnt: int, failed_cnt: int):
         since = dt.datetime.now() - self._started
-        failed_pct = 0
-        if failed_cnt > 0:
-            failed_pct = failed_cnt / given_cnt * 100
-        stats = f"{failed_pct:.0f}% ({failed_cnt}/{given_cnt}). Took {since}"
+        success_pct = collected_cnt / given_cnt * 100
+        stats = f"{success_pct:.0f}% results available ({collected_cnt}/{given_cnt}). Took {since}"
         if failed_cnt == given_cnt:
             logger.critical(f"All '{self._name}' tasks failed!!!")
-        elif failed_pct >= self._task_fail_error_pct:
+        elif failed_cnt > 0 and success_pct <= self._task_fail_error_pct:
             logger.error(f"More than half '{self._name}' tasks failed: {stats}")
         elif failed_cnt > 0:
             logger.warning(f"Some '{self._name}' tasks failed: {stats}")
         else:
-            logger.info(f"Finished '{self._name}' tasks: non-empty {stats}")
+            logger.info(f"Finished '{self._name}' tasks: {stats}")
 
     def _execute(self):
         with ThreadPoolExecutor(self._num_threads) as pool:

src/databricks/labs/ucx/framework/tasks.py

@@ -4,6 +4,7 @@
 from functools import wraps
 from pathlib import Path
 
+from databricks.labs.ucx.__about__ import __version__
 from databricks.labs.ucx.config import WorkspaceConfig
 from databricks.labs.ucx.framework.logger import _install
 
@@ -95,6 +96,8 @@ def trigger(*argv):
         msg = f'task "{task_name}" not found. Valid tasks are: {", ".join(_TASKS.keys())}'
         raise KeyError(msg)
 
+    print(f"UCX v{__version__}")
+
     current_task = _TASKS[task_name]
     print(current_task.doc)
 

src/databricks/labs/ucx/hive_metastore/grants.py

@@ -192,7 +192,7 @@ def for_schema_info(self, schema: SchemaInfo):
     def _grants(
         self,
         *,
-        catalog: str = False,
+        catalog: str | None = None,
         database: str | None = None,
         table: str | None = None,
         view: str | None = None,
@@ -257,5 +257,6 @@ def _grants(
                     anonymous_function=anonymous_function,
                 )
         except Exception as e:
+            # TODO: https://github.com/databrickslabs/ucx/issues/406
             logger.error(f"Couldn't fetch grants for object {on_type} {key}: {e}")
             return []

src/databricks/labs/ucx/mixins/fixtures.py

@@ -447,7 +447,7 @@ def create(*, name: str | None = None, **kwargs):
             name = f"sdk-{make_random(4)}"
         if "definition" not in kwargs:
             kwargs["definition"] = json.dumps(
-                {"spark_conf.spark.databricks.delta.preview.enabled": {"type": "fixed", "value": True}}
+                {"spark_conf.spark.databricks.delta.preview.enabled": {"type": "fixed", "value": "true"}}
             )
         return ws.cluster_policies.create(name, **kwargs)
 
@@ -711,7 +711,7 @@ def inventory_schema(make_schema):
 
 
 @pytest.fixture
-def make_catalog(ws, sql_backend, make_random):
+def make_catalog(ws, sql_backend, make_random) -> Callable[..., CatalogInfo]:
     def create() -> CatalogInfo:
         name = f"ucx_C{make_random(4)}".lower()
         sql_backend.execute(f"CREATE CATALOG {name}")
@@ -726,7 +726,7 @@ def create() -> CatalogInfo:
 
 
 @pytest.fixture
-def make_schema(sql_backend, make_random):
+def make_schema(sql_backend, make_random) -> Callable[..., SchemaInfo]:
     def create(*, catalog_name: str = "hive_metastore", name: str | None = None) -> SchemaInfo:
         if name is None:
             name = f"ucx_S{make_random(4)}"
@@ -742,7 +742,7 @@ def create(*, catalog_name: str = "hive_metastore", name: str | None = None) ->
 
 
 @pytest.fixture
-def make_table(sql_backend, make_schema, make_random):
+def make_table(sql_backend, make_schema, make_random) -> Callable[..., TableInfo]:
     def create(
         *,
         catalog_name="hive_metastore",

src/databricks/labs/ucx/runtime.py

@@ -232,9 +232,9 @@ def migrate_permissions(cfg: WorkspaceConfig):
         workspace_start_path=cfg.workspace_start_path,
     )
 
-    permission_manager.apply_group_permissions(group_manager.migration_groups_provider, destination="backup")
+    permission_manager.apply_group_permissions(group_manager.migration_state, destination="backup")
     group_manager.replace_workspace_groups_with_account_groups()
-    permission_manager.apply_group_permissions(group_manager.migration_groups_provider, destination="account")
+    permission_manager.apply_group_permissions(group_manager.migration_state, destination="account")
 
 
 @task("migrate-groups-cleanup", depends_on=[migrate_permissions])

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

@@ -98,6 +98,40 @@ def _crawler_task(self, object_type: str, object_id: str) -> Permissions | None:
             raw=json.dumps(permissions.as_dict()),
         )
 
+    def _load_as_request(self, object_type: str, object_id: str) -> list[iam.AccessControlRequest]:
+        loaded = self._safe_get_permissions(object_type, object_id)
+        if loaded is None:
+            return []
+        acl = []
+        for v in loaded.access_control_list:
+            for permission in v.all_permissions:
+                if permission.inherited:
+                    continue
+                acl.append(
+                    iam.AccessControlRequest(
+                        permission_level=permission.permission_level,
+                        service_principal_name=v.service_principal_name,
+                        group_name=v.group_name,
+                        user_name=v.user_name,
+                    )
+                )
+        # sort to return deterministic results
+        return sorted(acl, key=lambda v: f"{v.group_name}:{v.user_name}:{v.service_principal_name}")
+
+    def load_as_dict(self, object_type: str, object_id: str) -> dict[str, iam.PermissionLevel]:
+        result = {}
+        for acl in self._load_as_request(object_type, object_id):
+            result[self._key_for_acl_dict(acl)] = acl.permission_level
+        return result
+
+    @staticmethod
+    def _key_for_acl_dict(acl: iam.AccessControlRequest) -> str:
+        if acl.group_name is not None:
+            return acl.group_name
+        if acl.user_name is not None:
+            return acl.user_name
+        return acl.service_principal_name
+
     # TODO remove after ES-892977 is fixed
     @retried(on=[RetryableError])
     def _safe_get_permissions(self, object_type: str, object_id: str) -> iam.ObjectPermissions | None: