[serve] Set status message if deployment pending for too long

python/ray/serve/deployment_state.py

@@ -71,8 +71,10 @@ class ReplicaHealthCheckResponse(Enum):
 
 
 CHECKPOINT_KEY = "serve-deployment-state-checkpoint"
-SLOW_STARTUP_WARNING_S = 30
-SLOW_STARTUP_WARNING_PERIOD_S = 30
+SLOW_STARTUP_WARNING_S = int(os.getenv("SERVE_SLOW_STARTUP_WARNING_S", 30))
+SLOW_STARTUP_WARNING_PERIOD_S = int(
+    os.getenv("SERVE_SLOW_STARTUP_WARNING_PERIOD_S", 30)
+)
 
 ALL_REPLICA_STATES = list(ReplicaState)
 USE_PLACEMENT_GROUP = os.environ.get("SERVE_USE_PLACEMENT_GROUP", "1") != "0"
@@ -1496,7 +1498,7 @@ def _check_and_update_replicas(self) -> bool:
 
             if len(pending_allocation) > 0:
                 required, available = slow_start_replicas[0][0].resource_requirements()
-                logger.warning(
+                message = (
                     f"Deployment '{self._name}' has "
                     f"{len(pending_allocation)} replicas that have taken "
                     f"more than {SLOW_STARTUP_WARNING_S}s to be scheduled. "
@@ -1506,16 +1508,36 @@ def _check_and_update_replicas(self) -> bool:
                     f"Resources required for each replica: {required}, "
                     f"resources available: {available}."
                 )
+                logger.warning(message)
                 if _SCALING_LOG_ENABLED:
                     print_verbose_scaling_log()
+                # If status is UNHEALTHY, leave the status and message as is.
+                # The issue that caused the deployment to be unhealthy should be
+                # prioritized over this resource availability issue.
+                if self._curr_status_info.status != DeploymentStatus.UNHEALTHY:
+                    self._curr_status_info = DeploymentStatusInfo(
+                        name=self._name,
+                        status=DeploymentStatus.UPDATING,
+                        message=message,
+                    )
 
             if len(pending_initialization) > 0:
-                logger.warning(
+                message = (
                     f"Deployment '{self._name}' has "
                     f"{len(pending_initialization)} replicas that have taken "
                     f"more than {SLOW_STARTUP_WARNING_S}s to initialize. This "
                     f"may be caused by a slow __init__ or reconfigure method."
                 )
+                logger.warning(message)
+                # If status is UNHEALTHY, leave the status and message as is.
+                # The issue that caused the deployment to be unhealthy should be
+                # prioritized over this resource availability issue.
+                if self._curr_status_info.status != DeploymentStatus.UNHEALTHY:
+                    self._curr_status_info = DeploymentStatusInfo(
+                        name=self._name,
+                        status=DeploymentStatus.UPDATING,
+                        message=message,
+                    )
 
             self._prev_startup_warning = time.time()
 

python/ray/serve/tests/test_standalone.py

@@ -52,6 +52,32 @@ def ray_cluster():
     cluster.shutdown()
 
 
+@pytest.fixture()
+def lower_slow_startup_threshold_and_reset():
+    original_slow_startup_warning_s = os.getenv("SERVE_SLOW_STARTUP_WARNING_S")
+    original_slow_startup_warning_period_s = os.getenv(
+        "SERVE_SLOW_STARTUP_WARNING_PERIOD_S"
+    )
+    # Lower slow startup warning threshold to 1 second to reduce test duration
+    os.environ["SERVE_SLOW_STARTUP_WARNING_S"] = "1"
+    os.environ["SERVE_SLOW_STARTUP_WARNING_PERIOD_S"] = "1"
+
+    ray.init(num_cpus=2)
+    client = serve.start(detached=True)
+
+    yield client
+
+    serve.shutdown()
+    ray.shutdown()
+
+    # Reset slow startup warning threshold to prevent state sharing across unit
+    # tests
+    os.environ["SERVE_SLOW_STARTUP_WARNING_S"] = original_slow_startup_warning_s
+    os.environ[
+        "SERVE_SLOW_STARTUP_WARNING_PERIOD_S"
+    ] = original_slow_startup_warning_period_s
+
+
 def test_shutdown(ray_shutdown):
     ray.init(num_cpus=16)
     serve.start(http_options=dict(port=8003))
@@ -701,5 +727,61 @@ def f():
     ray.shutdown()
 
 
+def test_updating_status_message(lower_slow_startup_threshold_and_reset):
+    """Check if status message says if a serve deployment has taken a long time"""
+
+    client = lower_slow_startup_threshold_and_reset
+
+    @serve.deployment(
+        num_replicas=5,
+        ray_actor_options={"num_cpus": 1},
+    )
+    def f(*args):
+        pass
+
+    f.deploy(_blocking=False)
+
+    def updating_message():
+        deployment_status = client.get_serve_status().deployment_statuses[0]
+        message_substring = "more than 1s to be scheduled."
+        return (deployment_status.status == "UPDATING") and (
+            message_substring in deployment_status.message
+        )
+
+    wait_for_condition(updating_message, timeout=20)
+
+
+def test_unhealthy_override_updating_status(lower_slow_startup_threshold_and_reset):
+    """
+    Check that if status is UNHEALTHY and there is a resource availability
+    issue, the status should not change. The issue that caused the deployment to
+    be unhealthy should be prioritized over this resource availability issue.
+    """
+
+    client = lower_slow_startup_threshold_and_reset
+
+    @serve.deployment
+    class f:
+        def __init__(self):
+            self.num = 1 / 0
+
+        def __call__(self, request):
+            pass
+
+    f.deploy(_blocking=False)
+
+    wait_for_condition(
+        lambda: client.get_serve_status().deployment_statuses[0].status == "UNHEALTHY",
+        timeout=20,
+    )
+
+    with pytest.raises(RuntimeError):
+        wait_for_condition(
+            lambda: client.get_serve_status().deployment_statuses[0].status
+            == "UPDATING",
+            timeout=10,
+        )
+
+
 if __name__ == "__main__":
     sys.exit(pytest.main(["-v", "-s", __file__]))