From e4b37f299a8647025360902f2ae8bbdad47d01ad Mon Sep 17 00:00:00 2001 From: Casper Nielsen Date: Tue, 13 Oct 2026 12:56:38 +0000 Subject: [PATCH] fix(clients): give each sidecar health check attempt a timeout (#1241) * fix(health): bound each sidecar health probe by DAPR_HEALTH_TIMEOUT DaprHealth.wait_for_sidecar() called urllib.request.urlopen with no timeout. DAPR_HEALTH_TIMEOUT only limited the retry loop, so a sidecar (or any process on that port) that accepted the connection but never replied blocked the call, and every client constructor that waits for the sidecar, forever. Each probe now gets a timeout equal to what is left of DAPR_HEALTH_TIMEOUT, with a 0.5 s floor so the last probe can still get a reply. The async health check had the same shape, bounded only by aiohttp's 300 s default, and now uses the same per-probe timeout. Retry, logging and the final TimeoutError are unchanged. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Casper Nielsen * fix(health): log async probe timeouts clearly and note sync timeout scope aiohttp's timeout error has an empty message, so an async probe timeout was logged as 'Unexpected error during health check:' with nothing after it. Catch asyncio.TimeoutError and log 'timed out', like the sync client. Also note in the sync client that urllib's timeout applies per socket operation, not to the whole request. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Casper Nielsen * fix(health): log sync read timeouts as timed out, not unexpected urllib wraps only connect/send errors in URLError, so a read timeout in getresponse() reached the generic branch. Also measure the health budget with time.monotonic() in both clients so a clock jump cannot stretch it. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Casper Nielsen * refactor(tests): move listen_without_replying into a shared fake_servers module Co-Authored-By: Claude Opus 5.5 Signed-off-by: Casper Nielsen --------- Signed-off-by: Casper Nielsen Co-authored-by: Claude Opus 5.5 (cherry picked from commit b14b2838aa706b935423290966e87414ecf04006) Signed-off-by: dapr-bot --- dapr/aio/clients/health.py | 16 ++++++++-- dapr/clients/health.py | 29 ++++++++++++++++-- tests/clients/fake_servers.py | 27 +++++++++++++++++ tests/clients/test_healthcheck.py | 39 +++++++++++++++++++++++++ tests/clients/test_healthcheck_async.py | 23 +++++++++++++++ 5 files changed, 128 insertions(+), 6 deletions(-) create mode 100644 tests/clients/fake_servers.py 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()