[serve] Push-based replica health: replicas self-check and report over existing channels

python/ray/serve/_private/autoscaling_state.py

@@ -210,6 +210,9 @@ def get_num_replicas_upper_bound(self) -> int:
 
     def update_running_replica_ids(self, running_replicas: List[ReplicaID]):
         """Update cached set of running replica IDs for this deployment."""
+        if running_replicas is self._running_replicas:
+            # Same (cached) list object -- membership unchanged, skip rebuild.
+            return
         self._running_replicas = running_replicas
         self._cached_running_replica_strs = {
             r.to_full_id_str() for r in running_replicas
@@ -395,7 +398,9 @@ def get_autoscaling_context(
             app_name=self._deployment_id.app_name,
             current_num_replicas=len(self._running_replicas),
             target_num_replicas=curr_target_num_replicas,
-            running_replicas=self._running_replicas,
+            # list(): AutoscalingContext.running_replicas is List[ReplicaID] on a
+            # stable PublicAPI; the tuple stays internal so the identity skip works.
+            running_replicas=list(self._running_replicas),
             total_num_requests=self.get_total_num_requests,
             capacity_adjusted_min_replicas=self.get_num_replicas_lower_bound(),
             capacity_adjusted_max_replicas=self.get_num_replicas_upper_bound(),

python/ray/serve/_private/common.py

@@ -1061,6 +1061,11 @@ class ReplicaMetricReport:
     aggregated_metrics: Dict[str, float]
     metrics: Dict[str, TimeSeries]
     timestamp: float
+    # Replica-pushed self-health (None = sender does not push health; the
+    # controller then falls back to pull probes).
+    healthy: Optional[bool] = None
+    health_checked_at: Optional[float] = None
+    health_consecutive_failures: Optional[int] = None
 
 
 @dataclass

python/ray/serve/_private/controller.py

@@ -61,6 +61,7 @@
 from ray.serve._private.deployment_info import DeploymentInfo
 from ray.serve._private.deployment_state import (
     DeploymentStateManager,
+    ReplicaHealthPushRegistry,
 )
 from ray.serve._private.endpoint_state import EndpointState
 from ray.serve._private.exceptions import ExternalScalerDisabledError
@@ -254,13 +255,15 @@ async def __init__(
         ]
 
         self.autoscaling_state_manager = AutoscalingStateManager()
+        self._replica_health_push_registry = ReplicaHealthPushRegistry()
         self.deployment_state_manager = DeploymentStateManager(
             self.kv_store,
             self.long_poll_host,
             all_serve_actor_names,
             get_all_live_placement_group_names(),
             self.cluster_node_info_cache,
             self.autoscaling_state_manager,
+            health_push_registry=self._replica_health_push_registry,
         )
 
         # Manage all applications' state
@@ -391,10 +394,30 @@ def record_autoscaling_metrics_from_replica(
         )
         # Track in health metrics
         self._health_metrics_tracker.record_replica_metrics_delay(latency_ms)
+        if replica_metric_report.healthy is not None:
+            self._replica_health_push_registry.record(
+                replica_metric_report.replica_id.unique_id,
+                replica_metric_report.health_checked_at
+                or replica_metric_report.timestamp,
+                replica_metric_report.healthy,
+                replica_metric_report.health_consecutive_failures,
+            )
         self.autoscaling_state_manager.record_request_metrics_for_replica(
             replica_metric_report
         )
 
+    def record_replica_health(
+        self,
+        replica_unique_id: str,
+        checked_at: float,
+        healthy: bool,
+        consecutive_failures: Optional[int] = None,
+    ):
+        """Self-health heartbeat from replicas that do not push metric reports."""
+        self._replica_health_push_registry.record(
+            replica_unique_id, checked_at, healthy, consecutive_failures
+        )
+
     def record_autoscaling_metrics_from_handle(
         self, handle_metric_report: Union[HandleMetricReport, bytes]
     ):
@@ -533,7 +556,8 @@ def _update_proxy_nodes(self):
         Controller decides where proxy actors should run
         (head node and nodes with deployment replicas).
         """
-        new_proxy_nodes = self.deployment_state_manager.get_active_node_ids()
+        # set(): the getter returns a frozenset memo, and this is mutated below.
+        new_proxy_nodes = set(self.deployment_state_manager.get_active_node_ids())
         new_proxy_nodes = new_proxy_nodes - set(
             self.cluster_node_info_cache.get_draining_nodes()
         )
@@ -618,6 +642,14 @@ async def run_control_loop_step(
             dsm_duration = time.time() - dsm_update_start_time
             self.dsm_update_duration_gauge_s.set(dsm_duration)
             self._health_metrics_tracker.record_dsm_update_duration(dsm_duration)
+            try:
+                _gs = self._replica_health_push_registry.gap_stats()
+                self._health_metrics_tracker.push_checks_recorded = _gs["count"]
+                self._health_metrics_tracker.push_check_gap_p50_s = _gs["p50_s"]
+                self._health_metrics_tracker.push_check_gap_p99_s = _gs["p99_s"]
+                self._health_metrics_tracker.push_check_gap_max_s = _gs["max_s"]
+            except Exception:
+                pass
             if not self.done_recovering_event.is_set() and not any_recovering:
                 self.done_recovering_event.set()
                 if num_loops > 0:

python/ray/serve/_private/controller_health_metrics_tracker.py

@@ -50,6 +50,13 @@ class ControllerHealthMetricsTracker:
     num_control_loops: int = 0
     last_control_loop_time: float = 0.0
 
+    # Replica self-check intervals observed via health pushes (set by the
+    # controller loop from the push registry).
+    push_checks_recorded: int = 0
+    push_check_gap_p50_s: float = 0.0
+    push_check_gap_p99_s: float = 0.0
+    push_check_gap_max_s: float = 0.0
+
     def record_loop_duration(self, duration: float):
         self.loop_durations.append(duration)
 
@@ -145,4 +152,8 @@ def collect_metrics(self) -> ControllerHealthMetrics:
             handle_metrics_delay_ms=handle_delay_stats,
             replica_metrics_delay_ms=replica_delay_stats,
             process_memory_mb=process_memory_mb,
+            push_checks_recorded=self.push_checks_recorded,
+            push_check_gap_p50_s=self.push_check_gap_p50_s,
+            push_check_gap_p99_s=self.push_check_gap_p99_s,
+            push_check_gap_max_s=self.push_check_gap_max_s,
         )