Merge branch 'main' into more-workflow-history-snapshots

Author: asnare

src/databricks/labs/ucx/source_code/jobs.py

@@ -406,17 +406,19 @@ def __init__(
 
     def refresh_report(self, sql_backend: SqlBackend, inventory_database: str) -> None:
         tasks = []
-        all_jobs = list(self._ws.jobs.list())
-        logger.info(f"Preparing {len(all_jobs)} linting tasks...")
-        for i, job in enumerate(all_jobs):
-            if self._debug_listing_upper_limit is not None and i >= self._debug_listing_upper_limit:
-                logger.warning(f"Debug listing limit reached: {self._debug_listing_upper_limit}")
-                break
+        items_listed = 0
+        for job in self._ws.jobs.list():
             if self._include_job_ids and job.job_id not in self._include_job_ids:
-                logger.info(f"Skipping job {job.job_id}...")
+                logger.info(f"Skipping job_id={job.job_id}")
                 continue
+            if self._debug_listing_upper_limit is not None and items_listed >= self._debug_listing_upper_limit:
+                logger.warning(f"Debug listing limit reached: {self._debug_listing_upper_limit}")
+                break
+            if job.settings is not None and job.settings.name is not None:
+                logger.info(f"Found job_id={job.job_id}: {job.settings.name}")
             tasks.append(functools.partial(self.lint_job, job.job_id))
-        logger.info(f"Running {tasks} linting tasks in parallel...")
+            items_listed += 1
+        logger.info(f"Running {len(tasks)} linting tasks in parallel...")
         job_results, errors = Threads.gather('linting workflows', tasks)
         job_problems: list[JobProblem] = []
         job_dfsas: list[DirectFsAccess] = []

src/databricks/labs/ucx/source_code/queries.py

@@ -106,26 +106,20 @@ def _dump_used_tables(
         self._used_tables_crawler.dump_all(processed_tables)
 
     def _lint_dashboards(self, context: _ReportingContext) -> None:
-        dashboard_ids = self._dashboard_ids_in_scope()
-        logger.info(f"Running {len(dashboard_ids)} linting tasks...")
-        for i, dashboard_id in enumerate(dashboard_ids):
-            if self._debug_listing_upper_limit is not None and i >= self._debug_listing_upper_limit:
-                logger.warning(f"Debug listing limit reached: {self._debug_listing_upper_limit}")
-                break
+        for dashboard_id in self._dashboard_ids_in_scope():
             dashboard = self._ws.dashboards.get(dashboard_id=dashboard_id)
+            logger.info(f"Linting dashboard_id={dashboard_id}: {dashboard.name}")
             problems, dfsas, tables = self._lint_and_collect_from_dashboard(dashboard, context.linted_queries)
             context.all_problems.extend(problems)
             context.all_dfsas.extend(dfsas)
             context.all_tables.extend(tables)
 
     def _lint_queries(self, context: _ReportingContext) -> None:
-        for i, query in enumerate(self._queries_in_scope()):
-            if self._debug_listing_upper_limit is not None and i >= self._debug_listing_upper_limit:
-                logger.warning(f"Debug listing limit reached: {self._debug_listing_upper_limit}")
-                break
+        for query in self._queries_in_scope():
             assert query.id is not None
             if query.id in context.linted_queries:
                 continue
+            logger.info(f"Linting query_id={query.id}: {query.name}")
             context.linted_queries.add(query.id)
             problems = self.lint_query(query)
             context.all_problems.extend(problems)
@@ -137,14 +131,32 @@ def _lint_queries(self, context: _ReportingContext) -> None:
     def _dashboard_ids_in_scope(self) -> list[str]:
         if self._include_dashboard_ids is not None:  # an empty list is accepted
             return self._include_dashboard_ids
-        all_dashboards = self._ws.dashboards.list()
-        return [dashboard.id for dashboard in all_dashboards if dashboard.id]
+        items_listed = 0
+        dashboard_ids = []
+        # redash APIs are very slow to paginate, especially for large number of dashboards, so we limit the listing
+        # to a small number of items in debug mode for the assessment workflow just to complete.
+        for dashboard in self._ws.dashboards.list():
+            if self._debug_listing_upper_limit is not None and items_listed >= self._debug_listing_upper_limit:
+                logger.warning(f"Debug listing limit reached: {self._debug_listing_upper_limit}")
+                break
+            if dashboard.id is None:
+                continue
+            dashboard_ids.append(dashboard.id)
+            items_listed += 1
+        return dashboard_ids
 
     def _queries_in_scope(self) -> list[LegacyQuery]:
         if self._include_dashboard_ids is not None:  # an empty list is accepted
             return []
-        all_queries = self._ws.queries_legacy.list()
-        return [query for query in all_queries if query.id]
+        items_listed = 0
+        legacy_queries = []
+        for query in self._ws.queries_legacy.list():
+            if self._debug_listing_upper_limit is not None and items_listed >= self._debug_listing_upper_limit:
+                logger.warning(f"Debug listing limit reached: {self._debug_listing_upper_limit}")
+                break
+            legacy_queries.append(query)
+            items_listed += 1
+        return legacy_queries
 
     def _lint_and_collect_from_dashboard(
         self, dashboard: Dashboard, linted_queries: set[str]

tests/integration/install/test_installation.py

@@ -18,7 +18,6 @@
     AlreadyExists,
     InvalidParameterValue,
     NotFound,
-    ResourceConflict,
 )
 from databricks.sdk.retries import retried
 from databricks.sdk.service import compute
@@ -91,9 +90,10 @@ def factory(
         pending.remove()
 
 
-@retried(on=[NotFound, ResourceConflict], timeout=timedelta(minutes=10))
 def test_experimental_permissions_migration_for_group_with_same_name(
-    installation_ctx, make_cluster_policy, make_cluster_policy_permissions
+    installation_ctx,
+    make_cluster_policy,
+    make_cluster_policy_permissions,
 ):
     ws_group, acc_group = installation_ctx.make_ucx_group()
     migrated_group = MigratedGroup.partial_info(ws_group, acc_group)
@@ -112,7 +112,7 @@ def test_experimental_permissions_migration_for_group_with_same_name(
 
     installation_ctx.workspace_installation.run()
 
-    installation_ctx.deployed_workflows.run_workflow("migrate-groups-experimental")
+    installation_ctx.deployed_workflows.run_workflow("migrate-groups")
 
     object_permissions = installation_ctx.generic_permissions_support.load_as_dict(
         "cluster-policies", cluster_policy.policy_id