fix(ssrs): streamline test connection, stream reports, and retry transient failures

- Test connection's GetDashboards step no longer paginates every report; it
  now issues a single $top=1 probe to /Reports, matching the existing
  CheckAccess probe against /Folders.
- SsrsClient.get_reports is now a generator that yields each page as it is
  fetched, so ingestion memory stays flat regardless of report count.
- Session retries transient failures (connect errors, read timeouts, 5xx)
  up to 2 times with exponential backoff to survive flaky SSRS endpoints.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: ulixius9

ingestion/src/metadata/ingestion/source/dashboard/ssrs/client.py

@@ -12,10 +12,12 @@
 SSRS REST client
 """
 import traceback
-from typing import List, Optional, Union
+from typing import Iterator, List, Optional, Union
 
 import requests
+from requests.adapters import HTTPAdapter
 from requests_ntlm import HttpNtlmAuth
+from urllib3.util.retry import Retry
 
 from metadata.generated.schema.entity.services.connections.dashboard.ssrsConnection import (
     SsrsConnection,
@@ -35,6 +37,9 @@
 API_VERSION = "api/v2.0"
 DEFAULT_TIMEOUT = 30
 PAGE_SIZE = 100
+MAX_RETRIES = 2
+BACKOFF_FACTOR = 1
+RETRY_STATUS_CODES = (500, 502, 503, 504)
 
 
 class SsrsClient:
@@ -53,6 +58,19 @@ def __init__(
         self.session.headers.update({"Accept": "application/json"})
         if verify_ssl is not None:
             self.session.verify = verify_ssl
+        retry = Retry(
+            total=MAX_RETRIES,
+            connect=MAX_RETRIES,
+            read=MAX_RETRIES,
+            status=MAX_RETRIES,
+            backoff_factor=BACKOFF_FACTOR,
+            status_forcelist=RETRY_STATUS_CODES,
+            allowed_methods=frozenset(["GET"]),
+            raise_on_status=False,
+        )
+        adapter = HTTPAdapter(max_retries=retry)
+        self.session.mount("http://", adapter)
+        self.session.mount("https://", adapter)
 
     def close(self) -> None:
         if self.session:
@@ -72,6 +90,14 @@ def test_access(self) -> None:
                 f"Failed to connect to SSRS: {exc}"
             ) from exc
 
+    def test_get_reports(self) -> None:
+        try:
+            self._get("/Reports", params={"$top": "1"})
+        except Exception as exc:
+            raise SourceConnectionException(
+                f"Failed to fetch SSRS reports: {exc}"
+            ) from exc
+
     def get_folders(self) -> List[SsrsFolder]:
         try:
             results: List[SsrsFolder] = []
@@ -91,21 +117,18 @@ def get_folders(self) -> List[SsrsFolder]:
             logger.warning("Failed to fetch SSRS folders: %s", exc)
         return []
 
-    def get_reports(self) -> List[SsrsReport]:
+    def get_reports(self) -> Iterator[SsrsReport]:
         try:
-            results: List[SsrsReport] = []
             skip = 0
             while True:
                 data = self._get(
                     "/Reports", params={"$top": str(PAGE_SIZE), "$skip": str(skip)}
                 )
                 response = SsrsReportListResponse(**data)
-                results.extend(response.value)
+                yield from response.value
                 if len(response.value) < PAGE_SIZE:
                     break
                 skip += PAGE_SIZE
-            return results
         except Exception as exc:
             logger.debug(traceback.format_exc())
             logger.warning("Failed to fetch SSRS reports: %s", exc)
-        return []

ingestion/src/metadata/ingestion/source/dashboard/ssrs/connection.py

@@ -46,7 +46,7 @@ def test_connection(
 ) -> TestConnectionResult:
     test_fn = {
         "CheckAccess": client.test_access,
-        "GetDashboards": client.get_reports,
+        "GetDashboards": client.test_get_reports,
     }
 
     return test_connection_steps(

ingestion/src/metadata/ingestion/source/dashboard/ssrs/metadata.py

@@ -12,7 +12,7 @@
 SSRS source module
 """
 import traceback
-from typing import Any, Dict, Iterable, List, Optional
+from typing import Any, Dict, Iterable, Optional
 
 from metadata.generated.schema.api.data.createChart import CreateChartRequest
 from metadata.generated.schema.api.data.createDashboard import CreateDashboardRequest
@@ -81,9 +81,10 @@ def prepare(self):
         self.folder_path_map = {folder.path: folder.name for folder in folders}
         return super().prepare()
 
-    def get_dashboards_list(self) -> Optional[List[SsrsReport]]:
-        reports = self.client.get_reports()
-        return [r for r in reports if not r.hidden]
+    def get_dashboards_list(self) -> Iterable[SsrsReport]:
+        for report in self.client.get_reports():
+            if not report.hidden:
+                yield report
 
     def get_dashboard_name(self, dashboard: SsrsReport) -> str:
         return dashboard.name

ingestion/tests/integration/connections/test_ssrs_connection.py

@@ -38,6 +38,29 @@ def log_message(self, format, *args):
         pass
 
 
+class _FlakyHandler(BaseHTTPRequestHandler):
+    failures_remaining = 2
+    request_count = 0
+
+    def do_GET(self):
+        type(self).request_count += 1
+        if type(self).failures_remaining > 0:
+            type(self).failures_remaining -= 1
+            self.send_response(503)
+            self.send_header("Content-Length", "0")
+            self.end_headers()
+            return
+        body = json.dumps({"value": []}).encode()
+        self.send_response(200)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Content-Length", str(len(body)))
+        self.end_headers()
+        self.wfile.write(body)
+
+    def log_message(self, format, *args):
+        pass
+
+
 @pytest.fixture(scope="module")
 def ssrs_mock_url():
     server = HTTPServer(("127.0.0.1", 0), _MockHandler)
@@ -48,6 +71,18 @@ def ssrs_mock_url():
     server.shutdown()
 
 
+@pytest.fixture()
+def ssrs_flaky_url():
+    _FlakyHandler.failures_remaining = 2
+    _FlakyHandler.request_count = 0
+    server = HTTPServer(("127.0.0.1", 0), _FlakyHandler)
+    port = server.server_address[1]
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    yield f"http://127.0.0.1:{port}/reports"
+    server.shutdown()
+
+
 @pytest.mark.integration
 class TestSsrsConnection:
     def test_get_connection(self, ssrs_mock_url):
@@ -64,10 +99,34 @@ def test_get_connection_test_access(self, ssrs_mock_url):
         client = get_connection(connection)
         client.test_access()
 
+    def test_get_connection_test_get_reports(self, ssrs_mock_url):
+        connection = SsrsConnection(
+            hostPort=ssrs_mock_url, username="test_user", password="test_pass"
+        )
+        client = get_connection(connection)
+        client.test_get_reports()
+
     def test_connection_bad_host(self):
         connection = SsrsConnection(
             hostPort="http://localhost:1", username="test_user", password="test_pass"
         )
         client = get_connection(connection)
         with pytest.raises(SourceConnectionException):
             client.test_access()
+
+    def test_connection_bad_host_get_reports(self):
+        connection = SsrsConnection(
+            hostPort="http://localhost:1", username="test_user", password="test_pass"
+        )
+        client = get_connection(connection)
+        with pytest.raises(SourceConnectionException):
+            client.test_get_reports()
+
+    def test_get_reports_retries_transient_failures(self, ssrs_flaky_url):
+        connection = SsrsConnection(
+            hostPort=ssrs_flaky_url, username="test_user", password="test_pass"
+        )
+        client = get_connection(connection)
+        reports = list(client.get_reports())
+        assert reports == []
+        assert _FlakyHandler.request_count == 3

ingestion/tests/integration/ssrs/test_metadata.py

@@ -26,7 +26,7 @@ def test_client_get_reports(self, ssrs_service):
             hostPort=ssrs_service, username="test_user", password="test_pass"
         )
         client = SsrsClient(connection)
-        reports = client.get_reports()
+        reports = list(client.get_reports())
         assert len(reports) == 4
         assert reports[0].name == "Report 1"
         assert reports[0].path == "/TestFolder/Report 1"
@@ -52,7 +52,7 @@ def test_hidden_reports_present_in_raw(self, ssrs_service):
             hostPort=ssrs_service, username="test_user", password="test_pass"
         )
         client = SsrsClient(connection)
-        reports = client.get_reports()
+        reports = list(client.get_reports())
         assert any(r.hidden for r in reports)
         visible = [r for r in reports if not r.hidden]
         assert len(visible) == 3

ingestion/tests/unit/topology/dashboard/test_ssrs.py

@@ -173,14 +173,14 @@ def test_dashboard_details(self, ssrs_source):
             assert ssrs_source.get_dashboard_details(report) == report
 
     def test_dashboards_list(self, ssrs_source):
-        ssrs_source.client.get_reports = lambda: MOCK_REPORTS
-        result = ssrs_source.get_dashboards_list()
+        ssrs_source.client.get_reports = lambda: iter(MOCK_REPORTS)
+        result = list(ssrs_source.get_dashboards_list())
         assert result == MOCK_REPORTS
         assert len(result) == 3
 
     def test_dashboards_list_filters_hidden(self, ssrs_source):
-        ssrs_source.client.get_reports = lambda: MOCK_REPORTS_WITH_HIDDEN
-        result = ssrs_source.get_dashboards_list()
+        ssrs_source.client.get_reports = lambda: iter(MOCK_REPORTS_WITH_HIDDEN)
+        result = list(ssrs_source.get_dashboards_list())
         assert len(result) == 3
         assert all(not r.hidden for r in result)
 
@@ -313,7 +313,7 @@ def test_get_reports_single_page(self):
                 ]
             }
         )
-        reports = client.get_reports()
+        reports = list(client.get_reports())
         assert len(reports) == 3
         client._get.assert_called_once()
 
@@ -343,14 +343,45 @@ def test_get_reports_multi_page(self):
         }
         client._get = MagicMock(side_effect=[page1, page2])
 
-        reports = client.get_reports()
+        reports = list(client.get_reports())
         assert len(reports) == 150
         assert client._get.call_count == 2
         _, kwargs1 = client._get.call_args_list[0]
         _, kwargs2 = client._get.call_args_list[1]
         assert kwargs1["params"]["$skip"] == "0"
         assert kwargs2["params"]["$skip"] == "100"
 
+    def test_get_reports_streams_lazily(self):
+        client = MagicMock(spec=SsrsClient)
+        client.get_reports = SsrsClient.get_reports.__get__(client)
+
+        page1 = {
+            "value": [
+                {
+                    "Id": f"r-{i}",
+                    "Name": f"Report {i}",
+                    "Path": f"/Reports/Report {i}",
+                }
+                for i in range(100)
+            ]
+        }
+        page2 = {
+            "value": [
+                {
+                    "Id": f"r-{i}",
+                    "Name": f"Report {i}",
+                    "Path": f"/Reports/Report {i}",
+                }
+                for i in range(100, 150)
+            ]
+        }
+        client._get = MagicMock(side_effect=[page1, page2])
+
+        reports_iter = client.get_reports()
+        first = next(reports_iter)
+        assert first.id == "r-0"
+        assert client._get.call_count == 1
+
     def test_get_folders_multi_page(self):
         client = MagicMock(spec=SsrsClient)
         client.get_folders = SsrsClient.get_folders.__get__(client)
@@ -377,6 +408,6 @@ def test_get_reports_empty(self):
         client = MagicMock(spec=SsrsClient)
         client.get_reports = SsrsClient.get_reports.__get__(client)
         client._get = MagicMock(return_value={"value": []})
-        reports = client.get_reports()
+        reports = list(client.get_reports())
         assert len(reports) == 0
         client._get.assert_called_once()