Changed _check_cluster_failures to be more resilient. (#845)

Author: FastLee

src/databricks/labs/ucx/assessment/clusters.py

@@ -4,7 +4,6 @@
 
 from databricks.sdk import WorkspaceClient
 from databricks.sdk.errors import NotFound
-from databricks.sdk.service import compute
 from databricks.sdk.service.compute import ClusterDetails, ClusterSource, Policy
 
 from databricks.labs.ucx.assessment.crawlers import (
@@ -68,21 +67,12 @@ def _check_init_scripts(self, cluster, failures):
                 continue
             failures.append(f"{_AZURE_SP_CONF_FAILURE_MSG} cluster.")
 
-    def _check_cluster_failures(self, cluster: ClusterDetails | compute.ClusterSpec):
+    def _check_cluster_failures(self, cluster: ClusterDetails):
         failures = []
-        if isinstance(cluster, ClusterDetails) and not cluster.creator_user_name:
-            logger.warning(
-                f"Cluster {cluster.cluster_id} have Unknown creator, it means that the original creator "
-                f"has been deleted and should be re-created"
-            )
-        cluster_id = cluster.cluster_id if isinstance(cluster, ClusterDetails) and cluster.cluster_id else ""
-        creator_user_name = (
-            cluster.creator_user_name if isinstance(cluster, ClusterDetails) and cluster.creator_user_name else None
-        )
         cluster_info = ClusterInfo(
-            cluster_id=cluster_id,
+            cluster_id=cluster.cluster_id if cluster.cluster_id else "",
             cluster_name=cluster.cluster_name,
-            creator=creator_user_name,
+            creator=cluster.creator_user_name,
             success=1,
             failures="[]",
         )
@@ -115,6 +105,11 @@ def _assess_clusters(self, all_clusters):
         for cluster in all_clusters:
             if cluster.cluster_source == ClusterSource.JOB:
                 continue
+            if not cluster.creator_user_name:
+                logger.warning(
+                    f"Cluster {cluster.cluster_id} have Unknown creator, it means that the original creator "
+                    f"has been deleted and should be re-created"
+                )
             yield self._check_cluster_failures(cluster)
 
     def snapshot(self) -> Iterable[ClusterInfo]:

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

@@ -3,6 +3,7 @@
 from dataclasses import dataclass
 
 from databricks.sdk import WorkspaceClient
+from databricks.sdk.service.compute import ClusterDetails
 from databricks.sdk.service.jobs import BaseJob
 
 from databricks.labs.ucx.assessment.clusters import ClustersMixin
@@ -84,7 +85,8 @@ def _assess_jobs(self, all_jobs: list[BaseJob], all_clusters_by_id) -> Iterable[
             job_id = job.job_id
             if not job_id:
                 continue
-            cluster_failures = self._check_cluster_failures(cluster_config)
+            cluster_details = ClusterDetails.from_dict(cluster_config.as_dict())
+            cluster_failures = self._check_cluster_failures(cluster_details)
             for failure in json.loads(cluster_failures.failures):
                 job_assessment[job_id].add(failure)