Scope crawled pipelines in PipelineCrawler (#3513)

## Changes
Scope crawled pipelines in PipelineCrawler using the new
include_pipeline_ids argument

### Linked issues
Introduces #3493 

### Tests
- [x] added unit tests

Author: pritishpai

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

@@ -29,31 +29,41 @@ class PipelineInfo:
 
 
 class PipelinesCrawler(CrawlerBase[PipelineInfo], CheckClusterMixin):
-    def __init__(self, ws: WorkspaceClient, sql_backend: SqlBackend, schema):
+    def __init__(
+        self, ws: WorkspaceClient, sql_backend: SqlBackend, schema, include_pipeline_ids: list[str] | None = None
+    ):
         super().__init__(sql_backend, "hive_metastore", schema, "pipelines", PipelineInfo)
         self._ws = ws
+        self._include_pipeline_ids = include_pipeline_ids
 
     def _crawl(self) -> Iterable[PipelineInfo]:
-        all_pipelines = list(self._ws.pipelines.list_pipelines())
-        for pipeline in all_pipelines:
+
+        pipeline_ids = []
+        if self._include_pipeline_ids is not None:
+            pipeline_ids = self._include_pipeline_ids
+        else:
+            pipeline_ids = [p.pipeline_id for p in self._ws.pipelines.list_pipelines() if p.pipeline_id]
+
+        for pipeline_id in pipeline_ids:
+            try:
+                pipeline = self._ws.pipelines.get(pipeline_id)
+                assert pipeline.pipeline_id is not None
+                assert pipeline.spec is not None
+            except NotFound:
+                logger.warning(f"Pipeline not found: {pipeline_id}")
+                continue
+
             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"
                 )
-            try:
-                assert pipeline.pipeline_id is not None
-                pipeline_response = self._ws.pipelines.get(pipeline.pipeline_id)
-            except NotFound:
-                logger.warning(f"Pipeline disappeared, cannot assess: {pipeline.name} (id={pipeline.pipeline_id})")
-                continue
-            assert pipeline_response.spec is not None
-            pipeline_config = pipeline_response.spec.configuration
+            pipeline_config = pipeline.spec.configuration
             failures = []
             if pipeline_config:
                 failures.extend(self._check_spark_conf(pipeline_config, "pipeline"))
-            clusters = pipeline_response.spec.clusters
+            clusters = pipeline.spec.clusters
             if clusters:
                 self._pipeline_clusters(clusters, failures)
             failures_as_json = json.dumps(failures)

tests/unit/assessment/test_pipelines.py

@@ -53,6 +53,15 @@ def test_pipeline_list_with_no_config():
     assert len(crawler) == 0
 
 
+def test_include_pipeline_ids():
+    ws = mock_workspace_client(pipeline_ids=['empty-spec', 'spec-with-spn'])
+    crawler = PipelinesCrawler(ws, MockBackend(), "ucx", include_pipeline_ids=['empty-spec'])
+    result_set = list(crawler.snapshot())
+
+    assert len(result_set) == 1
+    assert result_set[0].pipeline_id == 'empty-spec'
+
+
 def test_pipeline_disappears_during_crawl(ws, mock_backend, caplog) -> None:
     """Check that crawling doesn't fail if a pipeline is deleted after we list the pipelines but before we assess it."""
     ws.pipelines.list_pipelines.return_value = (
@@ -63,7 +72,7 @@ def test_pipeline_disappears_during_crawl(ws, mock_backend, caplog) -> None:
     def mock_get(pipeline_id: str) -> GetPipelineResponse:
         if pipeline_id == "2":
             raise ResourceDoesNotExist("Simulated disappearance")
-        return GetPipelineResponse(pipeline_id=pipeline_id, spec=PipelineSpec(id=pipeline_id))
+        return GetPipelineResponse(pipeline_id=pipeline_id, spec=PipelineSpec(id=pipeline_id), name="will_remain")
 
     ws.pipelines.get = mock_get
 
@@ -73,7 +82,7 @@ def mock_get(pipeline_id: str) -> GetPipelineResponse:
     assert results == [
         PipelineInfo(pipeline_id="1", pipeline_name="will_remain", creator_name=None, success=1, failures="[]")
     ]
-    assert "Pipeline disappeared, cannot assess: will_disappear (id=2)" in caplog.messages
+    assert "Pipeline not found: 2" in caplog.messages
 
 
 def test_pipeline_crawler_creator():
@@ -83,7 +92,12 @@ def test_pipeline_crawler_creator():
         PipelineStateInfo(pipeline_id="2", creator_user_name=""),
         PipelineStateInfo(pipeline_id="3", creator_user_name="bob"),
     )
-    ws.pipelines.get = create_autospec(GetPipelineResponse)  # pylint: disable=mock-no-usage
+    ws.pipelines.get = create_autospec(GetPipelineResponse)
+    ws.pipelines.get.side_effect = [
+        GetPipelineResponse(pipeline_id="1", spec=PipelineSpec(id="1"), creator_user_name=None),
+        GetPipelineResponse(pipeline_id="2", spec=PipelineSpec(id="2"), creator_user_name=""),
+        GetPipelineResponse(pipeline_id="3", spec=PipelineSpec(id="3"), creator_user_name="bob"),
+    ]
     result = PipelinesCrawler(ws, MockBackend(), "ucx").snapshot(force_refresh=True)
 
     expected_creators = [None, None, "bob"]