Added Dashboard widget to show the list of cluster policies along with DBR version (#1013)

## Changes
Adding new table called policies and loading cluster policies along with
dbr version in the same


### Linked issues
#830 

Resolves #..

### Functionality 

- [ ] added relevant user documentation
- [ ] added new CLI command
- [ ] modified existing command: `databricks labs ucx ...`
- [ ] added a new workflow
- [ ] modified existing workflow: `...`
- [x] added a new table
- [x] modified existing table: `...`

### Tests
<img width="749" alt="image"
src="https://github.com/databrickslabs/ucx/assets/127273819/117b8a78-9c9e-46dc-a838-7329c68c0bcb">




- [x] manually tested
- [x] added unit tests
- [x] added integration tests
- [ ] verified on staging environment (screenshot attached)

Author: prajin-29

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

@@ -37,6 +37,8 @@ class ClusterInfo:
     cluster_id: str
     success: int
     failures: str
+    spark_version: str | None = None
+    policy_id: str | None = None
     cluster_name: str | None = None
     creator: str | None = None
 
@@ -156,6 +158,8 @@ def _assess_clusters(self, all_clusters):
             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=cluster.creator_user_name,
                 success=1,
                 failures="[]",
@@ -172,3 +176,58 @@ def snapshot(self) -> Iterable[ClusterInfo]:
     def _try_fetch(self) -> Iterable[ClusterInfo]:
         for row in self._fetch(f"SELECT * FROM {self._schema}.{self._table}"):
             yield ClusterInfo(*row)
+
+
+@dataclass
+class PolicyInfo:
+    policy_id: str
+    policy_name: str
+    success: int
+    failures: str
+    spark_version: str | None = None
+    policy_description: str | None = None
+    creator: str | None = None
+
+
+class PoliciesCrawler(CrawlerBase[PolicyInfo], CheckClusterMixin):
+    def __init__(self, ws: WorkspaceClient, sbe: SqlBackend, schema):
+        super().__init__(sbe, "hive_metastore", schema, "policies", PolicyInfo)
+        self._ws = ws
+
+    def _crawl(self) -> Iterable[PolicyInfo]:
+        all_policices = list(self._ws.cluster_policies.list())
+        return list(self._assess_policies(all_policices))
+
+    def _assess_policies(self, all_policices):
+        for policy in all_policices:
+            failures: list[str] = []
+            if policy.policy_id is None:
+                continue
+            failures.extend(self._check_cluster_policy(policy.policy_id, "policy"))
+            try:
+                spark_version = json.dumps(json.loads(policy.definition)["spark_version"])
+            except KeyError:
+                spark_version = None
+            policy_name = policy.name
+            creator_name = policy.creator_user_name
+
+            policy_info = PolicyInfo(
+                policy_id=policy.policy_id,
+                policy_description=policy.description,
+                policy_name=policy_name,
+                spark_version=spark_version,
+                success=1,
+                failures="[]",
+                creator=creator_name,
+            )
+            if len(failures) > 0:
+                policy_info.success = 0
+                policy_info.failures = json.dumps(failures)
+            yield policy_info
+
+    def snapshot(self) -> Iterable[PolicyInfo]:
+        return self._snapshot(self._try_fetch, self._crawl)
+
+    def _try_fetch(self) -> Iterable[PolicyInfo]:
+        for row in self._fetch(f"SELECT * FROM {self._schema}.{self._table}"):
+            yield PolicyInfo(*row)

src/databricks/labs/ucx/install.py

@@ -56,7 +56,7 @@
 
 from databricks.labs.ucx.__about__ import __version__
 from databricks.labs.ucx.assessment.azure import AzureServicePrincipalInfo
-from databricks.labs.ucx.assessment.clusters import ClusterInfo
+from databricks.labs.ucx.assessment.clusters import ClusterInfo, PolicyInfo
 from databricks.labs.ucx.assessment.init_scripts import GlobalInitScriptInfo
 from databricks.labs.ucx.assessment.jobs import JobInfo, SubmitRunInfo
 from databricks.labs.ucx.assessment.pipelines import PipelineInfo
@@ -160,6 +160,7 @@ def deploy_schema(sql_backend: SqlBackend, inventory_schema: str):
             functools.partial(table, "workspace_objects", WorkspaceObjectInfo),
             functools.partial(table, "permissions", Permissions),
             functools.partial(table, "submit_runs", SubmitRunInfo),
+            functools.partial(table, "policies", PolicyInfo),
             functools.partial(table, "migration_status", MigrationStatus),
         ],
     )

src/databricks/labs/ucx/mixins/fixtures.py

@@ -647,7 +647,9 @@ def create(*, name: str | None = None, **kwargs):
             name = f"sdk-{make_random(4)}"
         if "definition" not in kwargs:
             kwargs["definition"] = json.dumps(
-                {"spark_conf.spark.databricks.delta.preview.enabled": {"type": "fixed", "value": "true"}}
+                {
+                    "spark_conf.spark.databricks.delta.preview.enabled": {"type": "fixed", "value": "true"},
+                }
             )
         cluster_policy = ws.cluster_policies.create(name, **kwargs)
         logger.info(

src/databricks/labs/ucx/queries/assessment/main/08_0_cluster_policies.sql

@@ -0,0 +1,9 @@
+-- viz type=table, name=Cluster Policies, columns=policy_name,cluster_dbr_version,policy_spark_version
+-- widget title=Cluster Policies, row=8, col=0, size_x=3, size_y=8
+SELECT
+    distinct policy_name,
+    cluster.spark_version as cluster_dbr_version,
+    policy.spark_version as policy_spark_version
+FROM $inventory.clusters as cluster
+JOIN $inventory.policies as policy
+ON cluster.policy_id=policy.policy_id

src/databricks/labs/ucx/runtime.py

@@ -6,7 +6,7 @@
 from databricks.sdk import WorkspaceClient
 
 from databricks.labs.ucx.assessment.azure import AzureServicePrincipalCrawler
-from databricks.labs.ucx.assessment.clusters import ClustersCrawler
+from databricks.labs.ucx.assessment.clusters import ClustersCrawler, PoliciesCrawler
 from databricks.labs.ucx.assessment.init_scripts import GlobalInitScriptCrawler
 from databricks.labs.ucx.assessment.jobs import JobsCrawler, SubmitRunsCrawler
 from databricks.labs.ucx.assessment.pipelines import PipelinesCrawler
@@ -154,6 +154,19 @@ def assess_incompatible_submit_runs(cfg: WorkspaceConfig, ws: WorkspaceClient, s
     crawler.snapshot()
 
 
+@task("assessment")
+def crawl_cluster_policies(cfg: WorkspaceConfig, ws: WorkspaceClient, sql_backend: SqlBackend):
+    """This module scans through all the Cluster Policies and get the necessary information
+
+    It looks for:
+      - Clusters Policies with Databricks Runtime (DBR) version earlier than 11.3
+
+      Subsequently, a list of all the policies with matching configurations are stored in the
+    `$inventory.policies` table."""
+    crawler = PoliciesCrawler(ws, sql_backend, cfg.inventory_database)
+    crawler.snapshot()
+
+
 @task("assessment", cloud="azure")
 def assess_azure_service_principals(cfg: WorkspaceConfig, ws: WorkspaceClient, sql_backend: SqlBackend):
     """This module scans through all the clusters configurations, cluster policies, job cluster configurations,
@@ -237,6 +250,7 @@ def crawl_groups(cfg: WorkspaceConfig, ws: WorkspaceClient, sql_backend: SqlBack
         assess_jobs,
         assess_incompatible_submit_runs,
         assess_clusters,
+        crawl_cluster_policies,
         assess_azure_service_principals,
         assess_pipelines,
         assess_global_init_scripts,

src/databricks/labs/ucx/upgrades/v0.16.0_changing_cluster_table_schema.py

@@ -0,0 +1,17 @@
+# pylint: disable=invalid-name
+
+import logging
+
+from databricks.labs.blueprint.installation import Installation
+from databricks.sdk import WorkspaceClient
+
+from databricks.labs.ucx.config import WorkspaceConfig
+
+logger = logging.getLogger(__name__)
+
+
+def upgrade(installation: Installation, ws: WorkspaceClient):
+    config = installation.load(WorkspaceConfig)
+    warehouse_id = str(config.warehouse_id)
+    sql = f"ALTER TABLE hive_metastore.{config.inventory_database}.clusters ADD COLUMNS(policy_id string,spark_version string)"
+    ws.statement_execution.execute_statement(sql, warehouse_id=warehouse_id)

tests/integration/assessment/test_clusters.py

@@ -1,10 +1,11 @@
+import json
 from datetime import timedelta
 
 from databricks.sdk.errors import NotFound
 from databricks.sdk.retries import retried
 from databricks.sdk.service.compute import DataSecurityMode
 
-from databricks.labs.ucx.assessment.clusters import ClustersCrawler
+from databricks.labs.ucx.assessment.clusters import ClustersCrawler, PoliciesCrawler
 
 from .test_assessment import _SPARK_CONF
 
@@ -36,3 +37,31 @@ def test_cluster_crawler_no_isolation(ws, make_cluster, inventory_schema, sql_ba
 
     assert len(results) == 1
     assert results[0].failures == '["No isolation shared clusters not supported in UC"]'
+
+
+@retried(on=[NotFound], timeout=timedelta(minutes=6))
+def test_policy_crawler(ws, make_cluster_policy, inventory_schema, sql_backend):
+    created_policy = make_cluster_policy(
+        name="test_policy_check",
+        definition=json.dumps({"spark_version": {'type': 'fixed', 'value': '14.3.x-scala2.12'}}),
+    )
+    policy_definition = {
+        "spark_version": {'type': 'fixed', 'value': '14.3.x-scala2.12'},
+        "spark_conf.fs.azure.account.auth.type": {"type": "fixed", "value": "OAuth", "hidden": True},
+    }
+    created_policy_2 = make_cluster_policy(name="test_policy_check2", definition=json.dumps(policy_definition))
+    policy_crawler = PoliciesCrawler(ws=ws, sbe=sql_backend, schema=inventory_schema)
+    policies = policy_crawler.snapshot()
+    results = []
+    for policy in policies:
+        if policy.policy_id in (created_policy.policy_id, created_policy_2.policy_id):
+            results.append(policy)
+
+    assert results[0].policy_name == "test_policy_check"
+    assert results[0].success == 1
+    assert results[0].failures == "[]"
+    assert results[0].spark_version == json.dumps({'type': 'fixed', 'value': '14.3.x-scala2.12'})
+
+    assert results[1].policy_name == "test_policy_check2"
+    assert results[1].success == 0
+    assert results[1].failures == '["Uses azure service principal credentials config in policy."]'

tests/unit/assessment/__init__.py

@@ -40,6 +40,7 @@ def _load_fixture(filename: str):
     BaseRun: '../assessment/jobruns',
     ClusterDetails: '../assessment/clusters',
     PipelineStateInfo: '../assessment/pipelines',
+    Policy: '../assessment/policies',
 }
 
 
@@ -77,11 +78,13 @@ def workspace_client_mock(
     pipeline_ids: list[str] | None = None,
     job_ids: list[str] | None = None,
     jobruns_ids: list[str] | None = None,
+    policy_ids: list[str] | None = None,
     warehouse_config="single-config.json",
     secret_exists=True,
 ):
     ws = create_autospec(WorkspaceClient)
     ws.clusters.list.return_value = _id_list(ClusterDetails, cluster_ids)
+    ws.cluster_policies.list.return_value = _id_list(Policy, policy_ids)
     ws.cluster_policies.get = _cluster_policy
     ws.pipelines.list_pipelines.return_value = _id_list(PipelineStateInfo, pipeline_ids)
     ws.pipelines.get = _pipeline

tests/unit/assessment/policies/single-user-with-spn-no-sparkversion.json

@@ -0,0 +1,7 @@
+{
+  "policy_id": "single-user-with-spn-no-sparkversion",
+  "name": "test_policy",
+  "definition": "{\"azure_attributes.availability\": {\"type\": \"fixed\", \"value\": \"ON_DEMAND_AZURE\", \"hidden\": true}}",
+  "policy_family_definition_overrides":{
+  }
+}

tests/unit/assessment/policies/single-user-with-spn-policyid.json

@@ -0,0 +1,24 @@
+{
+  "policy_id": "single-user-with-spn-policyid",
+  "definition": "{\"spark_version\": {\"type\": \"unlimited\",\"defaultValue\": \"auto:latest-ml\"},\"spark_conf.fs.azure.account.auth.type\": {\"type\": \"fixed\", \"value\": \"OAuth\", \"hidden\": true}}",
+  "policy_family_definition_overrides": {
+    "spark_conf.fs.azure.account.auth.type": {
+      "type": "fixed",
+      "value": "OAuth",
+      "hidden": true
+    },
+    "not.a.type": {
+      "type": "fixed",
+      "value": "not.a.matching.type",
+      "hidden": true
+    },
+    "not.a.matching.type": {
+      "type": "fixed",
+      "value": "https://login.microsoftonline.com/1234ededed/oauth2/token",
+      "hidden": true
+    }
+  },
+  "name": "test_policy",
+  "description": "test",
+  "creator_user_name": "test_creator"
+}