use existing Ownership classes

Author: ericvergnaud

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

@@ -46,6 +46,18 @@ class ClusterInfo:
     creator: str | None = None
     """User-name of the creator of the cluster, if known."""
 
+    @classmethod
+    def from_cluster_details(cls, details: ClusterDetails):
+        return ClusterInfo(
+            cluster_id=details.cluster_id if details.cluster_id else "",
+            cluster_name=details.cluster_name,
+            policy_id=details.policy_id,
+            spark_version=details.spark_version,
+            creator=details.creator_user_name or None,
+            success=1,
+            failures="[]",
+        )
+
 
 class CheckClusterMixin(CheckInitScriptMixin):
     _ws: WorkspaceClient
@@ -152,7 +164,7 @@ def _crawl(self) -> Iterable[ClusterInfo]:
         all_clusters = list(self._ws.clusters.list())
         return list(self._assess_clusters(all_clusters))
 
-    def _assess_clusters(self, all_clusters):
+    def _assess_clusters(self, all_clusters: Iterable[ClusterDetails]):
         for cluster in all_clusters:
             if cluster.cluster_source == ClusterSource.JOB:
                 continue
@@ -162,15 +174,7 @@ def _assess_clusters(self, all_clusters):
                     f"Cluster {cluster.cluster_id} have Unknown creator, it means that the original creator "
                     f"has been deleted and should be re-created"
                 )
-            cluster_info = ClusterInfo(
-                cluster_id=cluster.cluster_id if cluster.cluster_id else "",
-                cluster_name=cluster.cluster_name,
-                policy_id=cluster.policy_id,
-                spark_version=cluster.spark_version,
-                creator=creator,
-                success=1,
-                failures="[]",
-            )
+            cluster_info = ClusterInfo.from_cluster_details(cluster)
             failures = self._check_cluster_failures(cluster, "cluster")
             if len(failures) > 0:
                 cluster_info.success = 0

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

@@ -20,6 +20,7 @@
     RunType,
     SparkJarTask,
     SqlTask,
+    Job,
 )
 
 from databricks.labs.ucx.assessment.clusters import CheckClusterMixin
@@ -40,6 +41,17 @@ class JobInfo:
     creator: str | None = None
     """User-name of the creator of the pipeline, if known."""
 
+    @classmethod
+    def from_job(cls, job: Job):
+        job_name = job.settings.name if job.settings and job.settings.name else "Unknown"
+        return JobInfo(
+            job_id=str(job.job_id),
+            success=1,
+            failures="[]",
+            job_name=job_name,
+            creator=job.creator_user_name or None,
+        )
+
 
 class JobsMixin:
     @classmethod
@@ -124,17 +136,7 @@ def _prepare(all_jobs) -> tuple[dict[int, set[str]], dict[int, JobInfo]]:
             job_settings = job.settings
             if not job_settings:
                 continue
-            job_name = job_settings.name
-            if not job_name:
-                job_name = "Unknown"
-
-            job_details[job.job_id] = JobInfo(
-                job_id=str(job.job_id),
-                job_name=job_name,
-                creator=creator_user_name,
-                success=1,
-                failures="[]",
-            )
+            job_details[job.job_id] = JobInfo.from_job(job)
         return job_assessment, job_details
 
     def _try_fetch(self) -> Iterable[JobInfo]:

src/databricks/labs/ucx/sequencing/sequencing.py

@@ -7,6 +7,9 @@
 from databricks.sdk import WorkspaceClient
 from databricks.sdk.service import jobs
 
+from databricks.labs.ucx.assessment.clusters import ClusterOwnership, ClusterInfo
+from databricks.labs.ucx.assessment.jobs import JobOwnership, JobInfo
+from databricks.labs.ucx.framework.owners import AdministratorLocator
 from databricks.labs.ucx.source_code.graph import DependencyGraph
 
 
@@ -93,7 +96,7 @@ def register_workflow_job(self, job: jobs.Job) -> MigrationNode:
             object_type="JOB",
             object_id=str(job.job_id),
             object_name=job_name,
-            object_owner=job.creator_user_name or "<UNKNOWN>",
+            object_owner=JobOwnership(self._admin_locator).owner_of(JobInfo.from_job(job)),
         )
         self._nodes[job_node.key] = job_node
         if job.settings and job.settings.job_clusters:
@@ -121,7 +124,7 @@ def register_cluster(self, cluster_id: str) -> MigrationNode:
             object_type="CLUSTER",
             object_id=cluster_id,
             object_name=object_name,
-            object_owner=object_owner,
+            object_owner=ClusterOwnership(self._admin_locator).owner_of(ClusterInfo.from_cluster_details(details)),
         )
         self._nodes[cluster_node.key] = cluster_node
         # TODO register warehouses and policies