Log warnings when mounts are discovered on incorrect cluster type (#2929)

closes #2498

Author: FastLee

src/databricks/labs/ucx/cli.py

@@ -646,6 +646,17 @@ def migrate_tables(
         workspace_contexts = _get_workspace_contexts(w, a, run_as_collection)
     for workspace_context in workspace_contexts:
         deployed_workflows = workspace_context.deployed_workflows
+
+        try:
+            workspace_context.verify_progress_tracking.verify()
+        except RuntimeWarning:
+            logger.warning(
+                "We couldn't detect a successful run of the assessment workflow."
+                "The assessment workflow is a prerequisite for the migrate-tables workflow."
+                "It can be run by using the `ensure-assessment-run` command."
+            )
+            return
+
         deployed_workflows.run_workflow("migrate-tables")
 
         tables = list(workspace_context.tables_crawler.snapshot())

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

@@ -320,8 +320,15 @@ def _deduplicate_mounts(mounts: list) -> list:
 
     def _crawl(self) -> Iterable[Mount]:
         mounts = []
-        for mount_point, source, _ in self._dbutils.fs.mounts():
-            mounts.append(Mount(mount_point, source))
+        try:
+            for mount_point, source, _ in self._dbutils.fs.mounts():
+                mounts.append(Mount(mount_point, source))
+        except Exception as error:  # pylint: disable=broad-except
+            if "com.databricks.backend.daemon.dbutils.DBUtilsCore.mounts() is not whitelisted" in str(error):
+                logger.warning(
+                    "dbutils.fs.mounts() is not whitelisted. Skipping mount point discovery."
+                    "Please make sure you run the assessment workflow."
+                )
         return self._deduplicate_mounts(mounts)
 
     def _try_fetch(self) -> Iterable[Mount]:

tests/unit/hive_metastore/test_locations.py

@@ -139,6 +139,20 @@ def test_mounts_inventory_should_contain_deduped_mounts_without_variable_volume_
     assert expected == backend.rows_written_for("hive_metastore.test.mounts", "overwrite")
 
 
+def test_mount_inventory_warning_on_incompatible_compute(caplog):
+    client = create_autospec(WorkspaceClient)
+    client.dbutils.fs.mounts.side_effect = Exception(
+        "Blah Blah com.databricks.backend.daemon.dbutils.DBUtilsCore.mounts() is not whitelisted"
+    )
+    backend = MockBackend()
+    instance = Mounts(backend, client, "test")
+
+    instance.snapshot()
+
+    expected_warning = "dbutils.fs.mounts() is not whitelisted"
+    assert expected_warning in caplog.text
+
+
 def test_external_locations():
     row_factory = type("Row", (Row,), {"__columns__": ["location", "storage_properties"]})
     sql_backend = MockBackend(

tests/unit/test_cli.py

@@ -68,6 +68,7 @@
 from databricks.labs.ucx.hive_metastore import TablesCrawler, ExternalLocations
 from databricks.labs.ucx.hive_metastore.locations import ExternalLocation
 from databricks.labs.ucx.hive_metastore.tables import Table
+from databricks.labs.ucx.progress.install import VerifyProgressTracking
 from databricks.labs.ucx.source_code.linters.files import LocalFileMigrator
 
 
@@ -929,6 +930,7 @@ def test_migrate_tables_calls_migrate_table_job_run_now(
     )
     for workspace_client in workspace_clients:
         workspace_client.jobs.wait_get_run_job_terminated_or_skipped.return_value = run
+        workspace_client.jobs.list_runs.return_value = [Run(state=RunState(result_state=RunResultState.SUCCESS))]
 
     migrate_tables(workspace_clients[0], MockPrompts({}), run_as_collection=run_as_collection, a=acc_client)
 
@@ -937,6 +939,30 @@ def test_migrate_tables_calls_migrate_table_job_run_now(
         workspace_client.jobs.wait_get_run_job_terminated_or_skipped.assert_called_once()
 
 
+@pytest.mark.parametrize("run_as_collection", [False, True])
+def test_migrate_tables_errors_out_before_assessment(
+    run_as_collection,
+    workspace_clients,
+    acc_client,
+) -> None:
+    if not run_as_collection:
+        workspace_clients = [workspace_clients[0]]
+    run = Run(
+        state=RunState(result_state=RunResultState.SUCCESS),
+        start_time=0,
+        end_time=1000,
+        run_duration=1000,
+    )
+    for workspace_client in workspace_clients:
+        workspace_client.jobs.wait_get_run_job_terminated_or_skipped.return_value = run
+        workspace_client.jobs.list_runs.return_value = [Run(state=RunState(result_state=RunResultState.FAILED))]
+
+    migrate_tables(workspace_clients[0], MockPrompts({}), run_as_collection=run_as_collection, a=acc_client)
+
+    for workspace_client in workspace_clients:
+        workspace_client.jobs.run_now.assert_not_called()
+
+
 def test_migrate_tables_calls_external_hiveserde_tables_job_run_now(ws) -> None:
     # TODO: Test for running on a collection when context injection for multiple workspaces is supported.
     tables_crawler = create_autospec(TablesCrawler)
@@ -948,7 +974,9 @@ def test_migrate_tables_calls_external_hiveserde_tables_job_run_now(ws) -> None:
         table_format="HIVE",
     )
     tables_crawler.snapshot.return_value = [table]
-    ctx = WorkspaceContext(ws).replace(tables_crawler=tables_crawler)
+    verify_progress_tracking = create_autospec(VerifyProgressTracking)
+    verify_progress_tracking.verify.return_value = None
+    ctx = WorkspaceContext(ws).replace(tables_crawler=tables_crawler, verify_progress_tracking=verify_progress_tracking)
     ws.jobs.wait_get_run_job_terminated_or_skipped.return_value = Run(
         state=RunState(result_state=RunResultState.SUCCESS),
         start_time=0,
@@ -979,7 +1007,10 @@ def test_migrate_tables_calls_external_tables_ctas_job_run_now(ws) -> None:
         table_format="EXTERNAL",
     )
     tables_crawler.snapshot.return_value = [table]
-    ctx = WorkspaceContext(ws).replace(tables_crawler=tables_crawler)
+    verify_progress_tracking = create_autospec(VerifyProgressTracking)
+    verify_progress_tracking.verify.return_value = None
+    ctx = WorkspaceContext(ws).replace(tables_crawler=tables_crawler, verify_progress_tracking=verify_progress_tracking)
+
     ws.jobs.wait_get_run_job_terminated_or_skipped.return_value = Run(
         state=RunState(result_state=RunResultState.SUCCESS),
         start_time=0,