From 624022757f317ec6c1fec796c6a6588127eeb80f Mon Sep 17 00:00:00 2001 From: Xuanqi He Date: Tue, 4 Aug 2026 17:32:26 -0400 Subject: [PATCH] [clustermgtd] Fix spurious errors logged on transient IMDS unavailability. The retry introduced to mitigate the impact of transient IMDS unavailability during the compute fleet status retrieval was not effective, for two reasons: * its window (3 attempts, 1s apart) was shorter than the IMDS outages observed in the fleet, which are in the order of a few seconds, so the retries were exhausted before IMDS became available again; * every failed attempt was logged as an error by _run_command, so even the attempts recovered by a successful retry left errors in the log, describing failures that no longer had any impact on the cluster. Waits are now 1s, 2s and 4s, plus a jitter of up to 1s each to avoid retrying in lockstep with the other daemons polling IMDS, and errors on the single attempts are no longer logged: the failure is reported once by get_status, only after the retries are exhausted, so that genuinely prolonged IMDS outages are still visible. --- src/slurm_plugin/clustermgtd.py | 15 ++++++++++-- tests/slurm_plugin/test_clustermgtd.py | 33 ++++++++++++++++++++++++-- 2 files changed, 44 insertions(+), 4 deletions(-) diff --git a/src/slurm_plugin/clustermgtd.py b/src/slurm_plugin/clustermgtd.py index 5ee5a751..4908491d 100644 --- a/src/slurm_plugin/clustermgtd.py +++ b/src/slurm_plugin/clustermgtd.py @@ -98,10 +98,21 @@ class ComputeFleetStatusManager: COMPUTE_FLEET_STATUS_ATTRIBUTE = "status" COMPUTE_FLEET_LAST_UPDATED_TIME_ATTRIBUTE = "lastStatusUpdatedTime" + # Retry is sized to outlast the transient IMDS unavailability windows observed in the fleet, which are in the + # order of a few seconds: waits are 1s, 2s and 4s, plus a jitter of up to 1s each to avoid retrying in lockstep + # with the other daemons polling IMDS. The overall wait is kept well below the clustermgtd loop time, because + # the fleet status is retrieved synchronously at the beginning of every loop. @staticmethod - @retry(stop_max_attempt_number=3, wait_fixed=seconds(1)) + @retry( + wait_exponential_multiplier=500, + wait_exponential_max=seconds(4), + wait_jitter_max=seconds(1), + stop_max_attempt_number=4, + ) def _get_fleet_status(): - compute_fleet_raw_data = check_command_output("get-compute-fleet-status.sh") + # Failures are logged by get_status only once the retries are exhausted, so that the ones recovered by the + # retries do not pollute the log with errors that have no impact on the cluster. + compute_fleet_raw_data = check_command_output("get-compute-fleet-status.sh", log_error=False) log.debug("Retrieved compute fleet data: %s", compute_fleet_raw_data) return ComputeFleetStatus( json.loads(compute_fleet_raw_data).get(ComputeFleetStatusManager.COMPUTE_FLEET_STATUS_ATTRIBUTE) diff --git a/tests/slurm_plugin/test_clustermgtd.py b/tests/slurm_plugin/test_clustermgtd.py index 6f63dd74..0cec13bf 100644 --- a/tests/slurm_plugin/test_clustermgtd.py +++ b/tests/slurm_plugin/test_clustermgtd.py @@ -12,6 +12,7 @@ import logging import os +import subprocess from datetime import datetime, timezone from types import SimpleNamespace from unittest.mock import ANY, call @@ -2356,9 +2357,37 @@ def test_get_status(self, mocker, get_item_response, fallback, expected_status): status = compute_fleet_status_manager.get_status(fallback) assert_that(status).is_equal_to(expected_status) if get_item_response is Exception or get_item_response == "": - assert_that(check_command_output_mocked.call_count).is_equal_to(3) + assert_that(check_command_output_mocked.call_count).is_equal_to(4) else: - check_command_output_mocked.assert_called_once_with("get-compute-fleet-status.sh") + check_command_output_mocked.assert_called_once_with("get-compute-fleet-status.sh", log_error=False) + + @pytest.mark.parametrize( + "failed_attempts, expected_status, expected_errors", + [ + (0, ComputeFleetStatus.RUNNING, 0), + (2, ComputeFleetStatus.RUNNING, 0), + (4, ComputeFleetStatus.STOPPED, 1), + ], + ids=["no_failure", "transient_failure_recovered", "failure_not_recovered"], + ) + def test_get_status_logs_errors_only_once_retries_are_exhausted( + self, mocker, caplog, failed_attempts, expected_status, expected_errors + ): + caplog.set_level(logging.ERROR) + # subprocess.run is patched instead of check_command_output, so that the error logging performed by + # _run_command on command failure is exercised as well. + command_results = [ + subprocess.CalledProcessError(1, "get-compute-fleet-status.sh", output="ERROR") + for _ in range(failed_attempts) + ] + [SimpleNamespace(stdout='{"status": "RUNNING"}')] + subprocess_run_mocked = mocker.patch("subprocess.run", autospec=True, side_effect=command_results) + mocker.patch("retrying.time.sleep") + + status = ComputeFleetStatusManager().get_status(fallback=ComputeFleetStatus.STOPPED) + + assert_that(status).is_equal_to(expected_status) + assert_that(subprocess_run_mocked.call_count).is_equal_to(min(failed_attempts + 1, 4)) + assert_that(caplog.records).is_length(expected_errors) @pytest.mark.parametrize( "desired_status, update_item_response",