diff --git a/distributed/http/scheduler/info.py b/distributed/http/scheduler/info.py index 78ad7f89e89..94cfacd4300 100644 --- a/distributed/http/scheduler/info.py +++ b/distributed/http/scheduler/info.py @@ -22,10 +22,44 @@ if TYPE_CHECKING: from distributed import Scheduler + from distributed.scheduler import WorkerState + +#: Row colors in the worker table for workers that have not sent a heartbeat for a +#: while, as (fraction of the worker TTL, color) in decreasing order of fraction. +WORKER_TTL_COLORS = [(0.5, "#ffcdd2"), (0.25, "#ffe0b2")] +#: Threshold in seconds used when ``distributed.scheduler.worker-ttl`` is disabled +WORKER_NO_TTL_THRESHOLD = 60 + + +def worker_row_color(scheduler: Scheduler, ws: WorkerState) -> str | None: + """Background color for a worker in the worker table, or None. + + The thresholds scale with the timeout after which the scheduler removes a + worker that stopped sending heartbeats (``distributed.scheduler.worker-ttl``), + so that the table flags a worker before the scheduler gives up on it. + """ + from distributed.scheduler import heartbeat_interval + + since = time() - ws.last_seen + if not scheduler.worker_ttl: + return WORKER_TTL_COLORS[0][1] if since > WORKER_NO_TTL_THRESHOLD else None + # Same effective TTL as Scheduler.check_worker_ttl + ttl = max(scheduler.worker_ttl, 10 * heartbeat_interval(len(scheduler.workers))) + for fraction, color in WORKER_TTL_COLORS: + if since > fraction * ttl: + return color + return None + ns = { func.__name__: func - for func in [format_bytes, format_time, datetime.fromtimestamp, time] + for func in [ + format_bytes, + format_time, + datetime.fromtimestamp, + time, + worker_row_color, + ] } rel_path_statics = {"rel_path_statics": "../../.."} diff --git a/distributed/http/scheduler/tests/test_scheduler_http.py b/distributed/http/scheduler/tests/test_scheduler_http.py index b7aac70121a..02fdf693d3c 100644 --- a/distributed/http/scheduler/tests/test_scheduler_http.py +++ b/distributed/http/scheduler/tests/test_scheduler_http.py @@ -15,6 +15,7 @@ from distributed import Event, Lock, Scheduler from distributed.client import Client, wait from distributed.core import Status +from distributed.metrics import time from distributed.utils import is_valid_xml, url_escape from distributed.utils_test import ( async_poll_for, @@ -67,6 +68,48 @@ async def test_connect(c, s, a, b): assert not re.search("href=./", body) # no absolute links +async def _worker_table_colors(s): + http_client = AsyncHTTPClient() + response = await http_client.fetch( + f"http://localhost:{s.http_server.port}/info/main/workers.html" + ) + body = response.body.decode() + colors = {} + for row in re.findall(r"