Crawler: support for object ownership (#2774)

## Changes

As part of #2761 we need to have a way for determining the user
responsible for some of our inventory types. This PR updates the crawler
framework so that:

- There is a way to identify an owner of a resource referred to by
inventory records.
- When the owner cannot be identified, a workspace or account
administrator is used instead.

### Linked issues

Progresses #2761.

### Functionality

- A component for locating an administrator user.
- Ownership information for the following inventory types:

  - [X] `ClusterInfo`
  - [x] `DirectFsAccess` (stubbed)
  - [X] `Grant`
  - [x] `JobInfo`
  - [x] `PipelineInfo`
  - [X] `PolicyInfo`
  - [x] `Table`
  - [x] `TableMigrationStatus`
  - [x] `UDF`

### Tests

- [x] added unit tests
- [x] added integration tests

Author: asnare

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

@@ -29,6 +29,7 @@
 )
 from databricks.labs.ucx.assessment.init_scripts import CheckInitScriptMixin
 from databricks.labs.ucx.framework.crawlers import CrawlerBase
+from databricks.labs.ucx.framework.owners import Ownership
 from databricks.labs.ucx.framework.utils import escape_sql_identifier
 
 logger = logging.getLogger(__name__)
@@ -43,6 +44,7 @@ class ClusterInfo:
     policy_id: str | None = None
     cluster_name: str | None = None
     creator: str | None = None
+    """User-name of the creator of the cluster, if known."""
 
 
 class CheckClusterMixin(CheckInitScriptMixin):
@@ -154,7 +156,8 @@ def _assess_clusters(self, all_clusters):
         for cluster in all_clusters:
             if cluster.cluster_source == ClusterSource.JOB:
                 continue
-            if not cluster.creator_user_name:
+            creator = cluster.creator_user_name or None
+            if not creator:
                 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"
@@ -164,7 +167,7 @@ def _assess_clusters(self, all_clusters):
                 cluster_name=cluster.cluster_name,
                 policy_id=cluster.policy_id,
                 spark_version=cluster.spark_version,
-                creator=cluster.creator_user_name,
+                creator=creator,
                 success=1,
                 failures="[]",
             )
@@ -179,6 +182,16 @@ def _try_fetch(self) -> Iterable[ClusterInfo]:
             yield ClusterInfo(*row)
 
 
+class ClusterOwnership(Ownership[ClusterInfo]):
+    """Determine ownership of clusters in the inventory.
+
+    This is the cluster creator (if known).
+    """
+
+    def _maybe_direct_owner(self, record: ClusterInfo) -> str | None:
+        return record.creator
+
+
 @dataclass
 class PolicyInfo:
     policy_id: str
@@ -188,6 +201,7 @@ class PolicyInfo:
     spark_version: str | None = None
     policy_description: str | None = None
     creator: str | None = None
+    """User-name of the creator of the cluster policy, if known."""
 
 
 class PoliciesCrawler(CrawlerBase[PolicyInfo], CheckClusterMixin):
@@ -210,7 +224,7 @@ def _assess_policies(self, all_policices) -> Iterable[PolicyInfo]:
             except KeyError:
                 spark_version = None
             policy_name = policy.name
-            creator_name = policy.creator_user_name
+            creator_name = policy.creator_user_name or None
 
             policy_info = PolicyInfo(
                 policy_id=policy.policy_id,
@@ -229,3 +243,13 @@ def _assess_policies(self, all_policices) -> Iterable[PolicyInfo]:
     def _try_fetch(self) -> Iterable[PolicyInfo]:
         for row in self._fetch(f"SELECT * FROM {escape_sql_identifier(self.full_name)}"):
             yield PolicyInfo(*row)
+
+
+class ClusterPolicyOwnership(Ownership[PolicyInfo]):
+    """Determine ownership of cluster policies in the inventory.
+
+    This is the creator of the cluster policy (if known).
+    """
+
+    def _maybe_direct_owner(self, record: PolicyInfo) -> str | None:
+        return record.creator

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

@@ -25,6 +25,7 @@
 from databricks.labs.ucx.assessment.clusters import CheckClusterMixin
 from databricks.labs.ucx.assessment.crawlers import spark_version_compatibility
 from databricks.labs.ucx.framework.crawlers import CrawlerBase
+from databricks.labs.ucx.framework.owners import Ownership
 from databricks.labs.ucx.framework.utils import escape_sql_identifier
 
 logger = logging.getLogger(__name__)
@@ -37,6 +38,7 @@ class JobInfo:
     failures: str
     job_name: str | None = None
     creator: str | None = None
+    """User-name of the creator of the pipeline, if known."""
 
 
 class JobsMixin:
@@ -106,7 +108,8 @@ def _prepare(all_jobs) -> tuple[dict[int, set[str]], dict[int, JobInfo]]:
             if not job.job_id:
                 continue
             job_assessment[job.job_id] = set()
-            if not job.creator_user_name:
+            creator_user_name = job.creator_user_name or None
+            if not creator_user_name:
                 logger.warning(
                     f"Job {job.job_id} have Unknown creator, it means that the original creator has been deleted "
                     f"and should be re-created"
@@ -122,7 +125,7 @@ def _prepare(all_jobs) -> tuple[dict[int, set[str]], dict[int, JobInfo]]:
             job_details[job.job_id] = JobInfo(
                 job_id=str(job.job_id),
                 job_name=job_name,
-                creator=job.creator_user_name,
+                creator=creator_user_name,
                 success=1,
                 failures="[]",
             )
@@ -140,6 +143,16 @@ def _check_jar_task(self, all_task: list[RunTask]) -> list[str]:
         return task_failures
 
 
+class JobOwnership(Ownership[JobInfo]):
+    """Determine ownership of jobs (workflows) in the inventory.
+
+    This is the job creator (if known).
+    """
+
+    def _maybe_direct_owner(self, record: JobInfo) -> str | None:
+        return record.creator
+
+
 @dataclass
 class SubmitRunInfo:
     run_ids: str  # JSON-encoded list of run ids

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

@@ -8,6 +8,7 @@
 
 from databricks.labs.ucx.assessment.clusters import CheckClusterMixin
 from databricks.labs.ucx.framework.crawlers import CrawlerBase
+from databricks.labs.ucx.framework.owners import Ownership
 from databricks.labs.ucx.framework.utils import escape_sql_identifier
 
 logger = logging.getLogger(__name__)
@@ -20,6 +21,7 @@ class PipelineInfo:
     failures: str
     pipeline_name: str | None = None
     creator_name: str | None = None
+    """User-name of the creator of the pipeline, if known."""
 
 
 class PipelinesCrawler(CrawlerBase[PipelineInfo], CheckClusterMixin):
@@ -33,15 +35,16 @@ def _crawl(self) -> Iterable[PipelineInfo]:
 
     def _assess_pipelines(self, all_pipelines) -> Iterable[PipelineInfo]:
         for pipeline in all_pipelines:
-            if not pipeline.creator_user_name:
+            creator_name = pipeline.creator_user_name or None
+            if not creator_name:
                 logger.warning(
                     f"Pipeline {pipeline.name} have Unknown creator, it means that the original creator "
                     f"has been deleted and should be re-created"
                 )
             pipeline_info = PipelineInfo(
                 pipeline_id=pipeline.pipeline_id,
                 pipeline_name=pipeline.name,
-                creator_name=pipeline.creator_user_name,
+                creator_name=creator_name,
                 success=1,
                 failures="[]",
             )
@@ -73,3 +76,13 @@ def _pipeline_clusters(self, clusters, failures):
     def _try_fetch(self) -> Iterable[PipelineInfo]:
         for row in self._fetch(f"SELECT * FROM {escape_sql_identifier(self.full_name)}"):
             yield PipelineInfo(*row)
+
+
+class PipelineOwnership(Ownership[PipelineInfo]):
+    """Determine ownership of pipelines in the inventory.
+
+    This is the pipeline creator (if known).
+    """
+
+    def _maybe_direct_owner(self, record: PipelineInfo) -> str | None:
+        return record.creator_name

src/databricks/labs/ucx/contexts/application.py

@@ -28,6 +28,7 @@
 from databricks.labs.ucx.assessment.export import AssessmentExporter
 from databricks.labs.ucx.aws.credentials import CredentialManager
 from databricks.labs.ucx.config import WorkspaceConfig
+from databricks.labs.ucx.framework.owners import AdministratorLocator
 from databricks.labs.ucx.hive_metastore import ExternalLocations, Mounts, TablesCrawler
 from databricks.labs.ucx.hive_metastore.catalog_schema import CatalogSchema
 from databricks.labs.ucx.hive_metastore.grants import (
@@ -514,6 +515,10 @@ def migration_recon(self) -> MigrationRecon:
             self.config.recon_tolerance_percent,
         )
 
+    @cached_property
+    def administrator_locator(self) -> AdministratorLocator:
+        return AdministratorLocator(self.workspace_client)
+
 
 class CliContext(GlobalContext, abc.ABC):
     @cached_property

src/databricks/labs/ucx/contexts/workflow_task.py

@@ -73,7 +73,7 @@ def pipelines_crawler(self) -> PipelinesCrawler:
 
     @cached_property
     def table_size_crawler(self) -> TableSizeCrawler:
-        return TableSizeCrawler(self.sql_backend, self.inventory_database, self.config.include_databases)
+        return TableSizeCrawler(self.tables_crawler)
 
     @cached_property
     def policies_crawler(self) -> PoliciesCrawler:

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

@@ -21,7 +21,7 @@ class DataclassInstance(Protocol):
 
 
 class CrawlerBase(ABC, Generic[Result]):
-    def __init__(self, backend: SqlBackend, catalog: str, schema: str, table: str, klass: type[Result]):
+    def __init__(self, backend: SqlBackend, catalog: str, schema: str, table: str, klass: type[Result]) -> None:
         """
         Initializes a CrawlerBase instance.