diff --git a/dapr/aio/clients/health.py b/dapr/aio/clients/health.py index 9ab66ebba..9bd2afbc8 100644 --- a/dapr/aio/clients/health.py +++ b/dapr/aio/clients/health.py @@ -18,6 +18,7 @@ import aiohttp +from dapr.clients.health import _attempt_timeout from dapr.clients.http.conf import DAPR_API_TOKEN_HEADER, DAPR_USER_AGENT, USER_AGENT_HEADER from dapr.clients.http.helpers import get_api_url from dapr.conf import settings @@ -32,22 +33,31 @@ async def wait_for_sidecar(): headers[DAPR_API_TOKEN_HEADER] = settings.DAPR_API_TOKEN timeout = float(settings.DAPR_HEALTH_TIMEOUT) - start = time.time() + start = time.monotonic() ssl_context = DaprHealth.get_ssl_context() connector = aiohttp.TCPConnector(ssl=ssl_context) async with aiohttp.ClientSession(connector=connector) as session: while True: try: - async with session.get(health_url, headers=headers) as response: + # Cap each probe at the remaining budget instead of aiohttp's + # 300 s default, so a sidecar that never answers cannot block + # past DAPR_HEALTH_TIMEOUT. + probe_timeout = aiohttp.ClientTimeout(total=_attempt_timeout(start + timeout)) + async with session.get( + health_url, headers=headers, timeout=probe_timeout + ) as response: if 200 <= response.status < 300: break + except asyncio.TimeoutError: + # aiohttp's timeout error has an empty message; say what happened. + print(f'Health check on {health_url} failed: timed out') except aiohttp.ClientError as e: print(f'Health check on {health_url} failed: {e}') except Exception as e: print(f'Unexpected error during health check: {e}') - remaining = (start + timeout) - time.time() + remaining = (start + timeout) - time.monotonic() if remaining <= 0: raise TimeoutError(f'Dapr health check timed out, after {timeout}.') await asyncio.sleep(min(1, remaining)) diff --git a/dapr/clients/health.py b/dapr/clients/health.py index 8e1002292..3ce4a9f82 100644 --- a/dapr/clients/health.py +++ b/dapr/clients/health.py @@ -22,6 +22,16 @@ from dapr.clients.http.helpers import get_api_url from dapr.conf import settings +# Shortest time one health probe may wait for the sidecar. Each probe is otherwise capped +# at what is left of DAPR_HEALTH_TIMEOUT; the floor lets the last probe still get a reply. +_MIN_ATTEMPT_TIMEOUT_SECONDS = 0.5 + + +def _attempt_timeout(deadline: float) -> float: + """Seconds one health probe may take: what is left until ``deadline`` (a + ``time.monotonic()`` value), but never less than _MIN_ATTEMPT_TIMEOUT_SECONDS.""" + return max(deadline - time.monotonic(), _MIN_ATTEMPT_TIMEOUT_SECONDS) + class DaprHealth: @staticmethod @@ -41,19 +51,32 @@ def wait_for_sidecar(): headers[DAPR_API_TOKEN_HEADER] = settings.DAPR_API_TOKEN timeout = float(settings.DAPR_HEALTH_TIMEOUT) - start = time.time() + start = time.monotonic() while True: try: req = urllib.request.Request(health_url, headers=headers) - with urllib.request.urlopen(req, context=DaprHealth.get_ssl_context()) as response: + # Without a timeout, a sidecar that accepts the connection but never + # answers would block here forever, past DAPR_HEALTH_TIMEOUT. urllib + # applies this per socket operation (connect, each read), not to the + # whole request, so a probe can overrun the budget somewhat, but it + # can no longer block forever. + with urllib.request.urlopen( + req, + timeout=_attempt_timeout(start + timeout), + context=DaprHealth.get_ssl_context(), + ) as response: if 200 <= response.status < 300: break except urllib.error.URLError as e: print(f'Health check on {health_url} failed: {e.reason}') + except TimeoutError: + # urllib wraps only connect/send errors in URLError; a read timeout in + # getresponse() (a sidecar that accepts but never answers) is bare. + print(f'Health check on {health_url} failed: timed out') except Exception as e: print(f'Unexpected error during health check: {e}') - remaining = (start + timeout) - time.time() + remaining = (start + timeout) - time.monotonic() if remaining <= 0: raise TimeoutError(f'Dapr health check timed out, after {timeout}.') time.sleep(min(1, remaining)) diff --git a/tests/clients/fake_servers.py b/tests/clients/fake_servers.py new file mode 100644 index 000000000..32f44b435 --- /dev/null +++ b/tests/clients/fake_servers.py @@ -0,0 +1,27 @@ +# -*- coding: utf-8 -*- + +""" +Copyright 2025 The Dapr Authors +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + http://www.apache.org/licenses/LICENSE-2.0 +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +""" + +import socket +import unittest + + +def listen_without_replying(test: unittest.TestCase) -> int: + """Open a local port that accepts connections (the OS completes the handshake from + the listen backlog) but never sends a reply, like a hung sidecar. Returns the port.""" + listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + test.addCleanup(listener.close) + listener.bind(('127.0.0.1', 0)) + listener.listen(16) + return listener.getsockname()[1] diff --git a/tests/clients/test_healthcheck.py b/tests/clients/test_healthcheck.py index c5b49aee9..535c45e32 100644 --- a/tests/clients/test_healthcheck.py +++ b/tests/clients/test_healthcheck.py @@ -13,6 +13,7 @@ limitations under the License. """ +import threading import time import unittest from unittest.mock import MagicMock, patch @@ -21,6 +22,8 @@ from dapr.conf import settings from dapr.version import __version__ +from .fake_servers import listen_without_replying + class DaprHealthCheckTests(unittest.TestCase): @patch.object(settings, 'DAPR_HTTP_ENDPOINT', 'http://domain.com:3500') @@ -75,3 +78,39 @@ def test_wait_for_sidecar_timeout(self, mock_urlopen): self.assertGreaterEqual(time.time() - start, 2.5) self.assertGreater(mock_urlopen.call_count, 1) + + +class DaprHealthCheckUnresponsiveSidecarTests(unittest.TestCase): + # How long the test waits before calling wait_for_sidecar() hung. Well above + # DAPR_HEALTH_TIMEOUT so a slow CI machine does not fail it. + HANG_LIMIT_SECONDS = 15 + + @patch.object(settings, 'DAPR_HEALTH_TIMEOUT', '1') + def test_wait_for_sidecar_times_out_when_sidecar_never_replies(self): + port = listen_without_replying(self) + outcome = {} + + def wait(): + try: + DaprHealth.wait_for_sidecar() + except Exception as error: + outcome['error'] = error + + with ( + patch.object(settings, 'DAPR_HTTP_ENDPOINT', f'http://127.0.0.1:{port}'), + patch('builtins.print') as mock_print, + ): + # Run in a daemon thread so a regression fails this test instead of hanging + # the whole session. + waiter = threading.Thread(target=wait, daemon=True) + start = time.time() + waiter.start() + waiter.join(self.HANG_LIMIT_SECONDS) + + self.assertFalse(waiter.is_alive(), 'wait_for_sidecar() blocked past DAPR_HEALTH_TIMEOUT') + self.assertIsInstance(outcome.get('error'), TimeoutError) + self.assertIn('Dapr health check timed out', str(outcome['error'])) + self.assertLess(time.time() - start, self.HANG_LIMIT_SECONDS) + logged = [str(call.args[0]) for call in mock_print.call_args_list if call.args] + self.assertTrue(any(line.endswith('failed: timed out') for line in logged), logged) + self.assertFalse(any(line.startswith('Unexpected error') for line in logged), logged) diff --git a/tests/clients/test_healthcheck_async.py b/tests/clients/test_healthcheck_async.py index 5f497bb8d..4f18889e3 100644 --- a/tests/clients/test_healthcheck_async.py +++ b/tests/clients/test_healthcheck_async.py @@ -22,6 +22,8 @@ from dapr.conf import settings from dapr.version import __version__ +from .fake_servers import listen_without_replying + class DaprHealthCheckAsyncTests(unittest.IsolatedAsyncioTestCase): @patch.object(settings, 'DAPR_HTTP_ENDPOINT', 'http://domain.com:3500') @@ -192,6 +194,27 @@ async def test_multiple_health_checks_concurrent(self, mock_get): # Verify multiple calls were made self.assertGreaterEqual(mock_get.call_count, 3) + @patch.object(settings, 'DAPR_HEALTH_TIMEOUT', '1') + async def test_wait_for_sidecar_times_out_when_sidecar_never_replies(self): + port = listen_without_replying(self) + hang_limit_seconds = 15 # well above DAPR_HEALTH_TIMEOUT, for slow CI machines + + with ( + patch.object(settings, 'DAPR_HTTP_ENDPOINT', f'http://127.0.0.1:{port}'), + patch('builtins.print') as mock_print, + ): + task = asyncio.ensure_future(DaprHealth.wait_for_sidecar()) + done, _ = await asyncio.wait({task}, timeout=hang_limit_seconds) + if not done: + task.cancel() + self.fail('wait_for_sidecar() blocked past DAPR_HEALTH_TIMEOUT') + + error = task.exception() + self.assertIsInstance(error, TimeoutError) + self.assertIn('Dapr health check timed out', str(error)) + logged = [str(call.args[0]) for call in mock_print.call_args_list if call.args] + self.assertTrue(any(line.endswith('failed: timed out') for line in logged), logged) + if __name__ == '__main__': unittest.main()