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",