fix(spark): [O11YINFRA-46] retry connection error to account for autodiscovery race condition

Author: jhoenzsch

spark/CHANGELOG.md

@@ -2,6 +2,12 @@
 
 <!-- towncrier release notes start -->
 
+## Unreleased
+
+***Fixed***:
+
+* Reduce false-positive Spark connection errors by debouncing only `pod_phase:succeeded` and alerting immediately for other phases ([#21922](https://github.com/DataDog/integrations-core/pull/21922))
+
 ## 7.2.1 / 2025-10-31
 
 ***Fixed***:

spark/datadog_checks/spark/spark.py

@@ -83,6 +83,7 @@ def __init__(self, name, init_config, instances):
             raise ConfigurationError('The cluster_name must be specified in the instance configuration')
 
         self.master_address = self._get_master_address()
+        self._connection_errors = 0
 
     def check(self, _):
         tags = list(self.tags)
@@ -192,6 +193,8 @@ def _get_running_apps(self):
     def _collect_version(self, base_url, tags):
         try:
             version_json = self._rest_request_to_json(base_url, SPARK_VERSION_PATH, SPARK_SERVICE_CHECK, tags)
+            if version_json is None:
+                return False
             version = version_json['spark']
         except Exception as e:
             self.log.debug("Failed to collect version information: %s", e)
@@ -210,6 +213,9 @@ def _driver_init(self, tags):
             self.master_address, SPARK_APPS_PATH, SPARK_DRIVER_SERVICE_CHECK, tags
         )
 
+        if metrics_json is None:
+            return running_apps
+
         for app_json in metrics_json:
             app_id = app_json.get('id')
             app_name = app_json.get('name')
@@ -231,6 +237,9 @@ def _standalone_init(self, pre_20_mode, tags):
             self.master_address, SPARK_MASTER_STATE_PATH, SPARK_STANDALONE_SERVICE_CHECK, tags
         )
 
+        if metrics_json is None:
+            return {}
+
         running_apps = {}
         version_set = False
 
@@ -251,10 +260,11 @@ def _standalone_init(self, pre_20_mode, tags):
                             applist = self._rest_request_to_json(
                                 app_url, SPARK_APPS_PATH, SPARK_STANDALONE_SERVICE_CHECK, tags
                             )
-                            for appl in applist:
-                                aid = appl.get('id')
-                                aname = appl.get('name')
-                                running_apps[aid] = (aname, app_url)
+                            if applist:
+                                for appl in applist:
+                                    aid = appl.get('id')
+                                    aname = appl.get('name')
+                                    running_apps[aid] = (aname, app_url)
                         else:
                             running_apps[app_id] = (app_name, app_url)
                 except Exception:
@@ -279,6 +289,9 @@ def _mesos_init(self, tags):
 
         metrics_json = self._rest_request_to_json(self.master_address, MESOS_MASTER_APP_PATH, MESOS_SERVICE_CHECK, tags)
 
+        if metrics_json is None:
+            return running_apps
+
         if metrics_json.get('frameworks'):
             for app_json in metrics_json.get('frameworks'):
                 app_id = app_json.get('id')
@@ -330,6 +343,9 @@ def _get_standalone_app_url(self, app_id, tags):
             self.master_address, SPARK_MASTER_APP_PATH, SPARK_STANDALONE_SERVICE_CHECK, tags, appId=app_id
         )
 
+        if app_page is None:
+            return None
+
         dom = BeautifulSoup(app_page.text, 'html.parser')
         app_detail_ui_links = dom.find_all('a', string='Application Detail UI')
 
@@ -352,6 +368,9 @@ def _yarn_get_running_spark_apps(self, tags):
             applicationTypes=YARN_APPLICATION_TYPES,
         )
 
+        if metrics_json is None:
+            return {}
+
         running_apps = {}
 
         if metrics_json.get('apps'):
@@ -379,6 +398,8 @@ def _get_spark_app_ids(self, running_apps, tags):
                 if not version_set:
                     version_set = self._collect_version(tracking_url, tags)
                 response = self._rest_request_to_json(tracking_url, SPARK_APPS_PATH, SPARK_SERVICE_CHECK, tags)
+                if response is None:
+                    continue
             except Exception as e:
                 self.log.warning("Exception happened when fetching app ids for %s: %s", tracking_url, e)
                 continue
@@ -405,6 +426,8 @@ def _describe_app(self, property, running_apps, addl_tags):
                 response = self._rest_request(
                     base_url, SPARK_APPS_PATH, SPARK_SERVICE_CHECK, addl_tags, app_id, property
                 )
+                if response is None:
+                    continue
             except HTTPError:
                 self.log.debug("Got an error collecting %s", property, exc_info=True)
                 continue
@@ -512,6 +535,8 @@ def _spark_structured_streams_metrics(self, running_apps, addl_tags):
                 response = self._rest_request_to_json(
                     base_url, self.metricsservlet_path, SPARK_SERVICE_CHECK, addl_tags
                 )
+                if response is None:
+                    continue
                 self.log.debug('Structured streaming metrics: %s', response)
                 response = {
                     metric_name: v['value']
@@ -611,6 +636,10 @@ def _rest_request(self, url, object_path, service_name, tags, *args, **kwargs):
             self.log.debug('Spark check URL: %s', url)
             response = self.http.get(url, cookies=self.proxy_redirect_cookies)
             response.raise_for_status()
+
+            # Reset connection errors on success
+            self._connection_errors = 0
+
             content = response.text
             proxy_redirect_url = self._parse_proxy_redirect_url(content)
             if proxy_redirect_url:
@@ -633,6 +662,21 @@ def _rest_request(self, url, object_path, service_name, tags, *args, **kwargs):
             raise
 
         except (HTTPError, InvalidURL, ConnectionError) as e:
+            debounce_allowed = (
+                isinstance(e, ConnectionError)
+                and ("Connection refused" in str(e) or "No route to host" in str(e))
+                and self._should_debounce_connection_error(tags)
+            )
+            if debounce_allowed:
+                self._connection_errors += 1
+                if self._connection_errors <= 1:
+                    self.log.warning(
+                        "Connection failed. Retrying next run to confirm if it's a transient error or shutdown. "
+                        "Error: %s",
+                        e,
+                    )
+                    return None
+
             self.service_check(
                 service_name,
                 AgentCheck.CRITICAL,
@@ -654,6 +698,9 @@ def _rest_request_to_json(self, address, object_path, service_name, tags, *args,
         """
         response = self._rest_request(address, object_path, service_name, tags, *args, **kwargs)
 
+        if response is None:
+            return None
+
         try:
             response_json = response.json()
 
@@ -668,6 +715,14 @@ def _rest_request_to_json(self, address, object_path, service_name, tags, *args,
 
         return response_json
 
+    def _should_debounce_connection_error(self, tags):
+        """Allow one debounce unless the pod phase tag says failed/unknown."""
+        for tag in tags or []:
+            if tag.startswith('pod_phase:'):
+                phase = tag.split(':', 1)[1].strip().lower()
+                return phase not in ('failed', 'unknown')
+        return True
+
     @classmethod
     def _join_url_dir(cls, url, *args):
         """

spark/tests/test_spark.py

@@ -12,7 +12,7 @@
 import mock
 import pytest
 import urllib3
-from requests import RequestException
+from requests import RequestException, ConnectionError
 
 from datadog_checks.dev.http import MockResponse
 from datadog_checks.dev.utils import get_metadata_metrics
@@ -1465,3 +1465,54 @@ def test_integration_driver_2(aggregator, dd_run_check):
     )
     aggregator.assert_all_metrics_covered()
     aggregator.assert_metrics_using_metadata(get_metadata_metrics())
+
+
+@pytest.mark.unit
+def test_debounce_connection_failure(aggregator, dd_run_check, caplog):
+    # Mock connection failure
+    def connection_failure_mock(session, url, *args, **kwargs):
+        raise ConnectionError("Connection refused")
+
+    instance = DRIVER_CONFIG.copy()
+    instance['tags'] = list(instance.get('tags', [])) + ['pod_phase:Running']
+
+    with mock.patch('requests.Session.get', side_effect=connection_failure_mock):
+        c = SparkCheck('spark', {}, [instance])
+
+        # First run: expect warning, no CRITICAL check
+        with caplog.at_level(logging.WARNING):
+            dd_run_check(c)
+
+        assert "Connection failed. Retrying next run" in caplog.text
+
+        # Verify no CRITICAL check sent for spark.driver.can_connect
+        service_checks = aggregator.service_checks(SPARK_DRIVER_SERVICE_CHECK)
+        assert len(service_checks) == 0
+
+        # Second run: expect CRITICAL
+        # The check raises the exception, so we expect it here
+        with pytest.raises(ConnectionError):
+            dd_run_check(c)
+
+        service_checks = aggregator.service_checks(SPARK_DRIVER_SERVICE_CHECK)
+        assert len(service_checks) == 1
+        assert service_checks[0].status == SparkCheck.CRITICAL
+
+
+@pytest.mark.unit
+def test_debounce_connection_failure_terminal_phase(aggregator, dd_run_check, caplog):
+    def connection_failure_mock(session, url, *args, **kwargs):
+        raise ConnectionError("Connection refused")
+
+    instance = DRIVER_CONFIG.copy()
+    instance['tags'] = list(instance.get('tags', [])) + ['pod_phase:Succeeded']
+
+    with mock.patch('requests.Session.get', side_effect=connection_failure_mock):
+        c = SparkCheck('spark', {}, [instance])
+
+        with pytest.raises(ConnectionError):
+            dd_run_check(c)
+
+    service_checks = aggregator.service_checks(SPARK_DRIVER_SERVICE_CHECK)
+    assert len(service_checks) == 1
+    assert service_checks[0].status == SparkCheck.CRITICAL