[iris] Add heartbeat and slice-lifecycle debug logging (#4796)

Per-task slow_log timers on submit_task (500ms) and synchronous kill
(2000ms) inside handle_heartbeat identify which task stalls a heartbeat.
The worker service heartbeat entrypoint gets an outer slow_log (1000ms)
and a DEBUG payload-size line to correlate with controller-side sync
timing. Slice ready/failed transitions log registered worker counts and
ids to expose partial bootstrap on large slices.

Complements #4792 and #4793.

Author: rjpower

lib/iris/src/iris/cluster/controller/autoscaler/scaling_group.py

@@ -466,6 +466,13 @@ def mark_slice_ready(self, slice_id: str, worker_ids: list[str], timestamp: Time
                 state.last_active = timestamp
         if state is not None:
             self._db_upsert_slice(slice_id, state)
+            logger.info(
+                "slice ready group=%s slice=%s n_workers=%d worker_ids=%s",
+                self._config.name,
+                slice_id,
+                len(worker_ids),
+                worker_ids,
+            )
 
     def mark_slice_failed(self, slice_id: str, error_message: str = "") -> None:
         """Mark a slice as FAILED. Called when bootstrap fails."""
@@ -474,8 +481,19 @@ def mark_slice_failed(self, slice_id: str, error_message: str = "") -> None:
             if state is not None:
                 state.lifecycle = SliceLifecycleState.FAILED
                 state.error_message = error_message
+                registered = list(state.worker_ids)
+            else:
+                registered = []
         if state is not None:
             self._db_upsert_slice(slice_id, state)
+            logger.warning(
+                "slice failed group=%s slice=%s n_registered=%d registered=%s error=%s",
+                self._config.name,
+                slice_id,
+                len(registered),
+                registered,
+                error_message,
+            )
 
     def reconcile(self) -> None:
         """Discover and adopt existing slices from the cloud.

lib/iris/src/iris/cluster/worker/service.py

@@ -18,6 +18,7 @@
 from iris.rpc import job_pb2
 from iris.rpc import worker_pb2
 from iris.rpc.errors import rpc_error_handler
+from rigging.log_setup import slow_log
 from rigging.timing import Timer
 
 logger = logging.getLogger(__name__)
@@ -100,7 +101,7 @@ def heartbeat(
 
         Processes tasks_to_run and tasks_to_kill, then returns current state.
         """
-        with rpc_error_handler("heartbeat"):
+        with rpc_error_handler("heartbeat"), slow_log(logger, "heartbeat rpc", threshold_ms=1000):
             # Chaos injection for testing heartbeat failures and delays
             if rule := chaos("worker.heartbeat"):
                 if rule.delay_seconds > 0:
@@ -111,6 +112,13 @@ def heartbeat(
                 if not rule.delay_seconds:
                     raise RuntimeError("chaos: worker.heartbeat")
 
+            logger.debug(
+                "heartbeat rpc received n_run=%d n_kill=%d n_expected=%d req_bytes=%d",
+                len(request.tasks_to_run),
+                len(request.tasks_to_kill),
+                len(request.expected_tasks),
+                request.ByteSize(),
+            )
             # Delegate to worker for reconciliation
             return self._provider.handle_heartbeat(request)
 

lib/iris/src/iris/cluster/worker/worker.py

@@ -765,7 +765,8 @@ def handle_heartbeat(self, request: job_pb2.HeartbeatRequest) -> job_pb2.Heartbe
             with slow_log(logger, "heartbeat submit_tasks", threshold_ms=200):
                 for run_req in request.tasks_to_run:
                     try:
-                        self.submit_task(run_req)
+                        with slow_log(logger, f"heartbeat submit_task[{run_req.task_id}]", threshold_ms=500):
+                            self.submit_task(run_req)
                         logger.info("Heartbeat: submitted task %s", run_req.task_id)
                     except Exception as e:
                         logger.warning("Heartbeat: failed to submit task %s: %s", run_req.task_id, e)
@@ -778,7 +779,8 @@ def handle_heartbeat(self, request: job_pb2.HeartbeatRequest) -> job_pb2.Heartbe
                     try:
                         current = self._get_current_attempt(task_id)
                         if current:
-                            self._kill_task_attempt(task_id, current.attempt_id, async_kill=False)
+                            with slow_log(logger, f"heartbeat kill_task[{task_id}]", threshold_ms=2000):
+                                self._kill_task_attempt(task_id, current.attempt_id, async_kill=False)
                             logger.info("Heartbeat: killed task %s", task_id)
                     except Exception as e:
                         logger.warning("Heartbeat: failed to kill task %s: %s", task_id, e)