[Serve] set last scale up/down time on autoscaling context (#59057)

fixes #59001

---------

Signed-off-by: abrar <abrar@anyscale.com>

Author: abrarsheikh

python/ray/serve/_private/autoscaling_state.py

@@ -56,6 +56,9 @@ def __init__(self, deployment_id: DeploymentID):
         self._running_replicas: List[ReplicaID] = []
         self._target_capacity: Optional[float] = None
         self._target_capacity_direction: Optional[TargetCapacityDirection] = None
+        # Track timestamps of last scale up and scale down events
+        self._last_scale_up_time: Optional[float] = None
+        self._last_scale_down_time: Optional[float] = None
 
     def register(self, info: DeploymentInfo, curr_target_num_replicas: int) -> int:
         """Registers an autoscaling deployment's info.
@@ -121,6 +124,14 @@ def update_running_replica_ids(self, running_replicas: List[ReplicaID]):
         """Update cached set of running replica IDs for this deployment."""
         self._running_replicas = running_replicas
 
+    def record_scale_up(self):
+        """Record a scale up event by updating the timestamp."""
+        self._last_scale_up_time = time.time()
+
+    def record_scale_down(self):
+        """Record a scale down event by updating the timestamp."""
+        self._last_scale_down_time = time.time()
+
     def is_within_bounds(self, num_replicas_running_at_target_version: int):
         """Whether or not this deployment is within the autoscaling bounds.
 
@@ -253,8 +264,8 @@ def get_autoscaling_context(self, curr_target_num_replicas) -> AutoscalingContex
             total_queued_requests=self._get_queued_requests,
             aggregated_metrics=self._get_aggregated_custom_metrics,
             raw_metrics=self._get_raw_custom_metrics,
-            last_scale_up_time=None,
-            last_scale_down_time=None,
+            last_scale_up_time=self._last_scale_up_time,
+            last_scale_down_time=self._last_scale_down_time,
         )
 
     def _collect_replica_running_requests(self) -> List[TimeSeries]:
@@ -792,6 +803,16 @@ def update_running_replica_ids(
             running_replicas
         )
 
+    def record_scale_up(self, deployment_id: DeploymentID):
+        """Record a scale up event for a deployment."""
+        if deployment_id in self._deployment_autoscaling_states:
+            self._deployment_autoscaling_states[deployment_id].record_scale_up()
+
+    def record_scale_down(self, deployment_id: DeploymentID):
+        """Record a scale down event for a deployment."""
+        if deployment_id in self._deployment_autoscaling_states:
+            self._deployment_autoscaling_states[deployment_id].record_scale_down()
+
     def on_replica_stopped(self, replica_id: ReplicaID):
         dep_id = replica_id.deployment_id
         if dep_id in self._deployment_autoscaling_states:
@@ -943,6 +964,26 @@ def update_running_replica_ids(
         if app_state:
             app_state.update_running_replica_ids(deployment_id, running_replicas)
 
+    def record_scale_up(self, deployment_id: DeploymentID):
+        """Record a scale up event for a deployment.
+
+        Args:
+            deployment_id: The ID of the deployment being scaled up.
+        """
+        app_state = self._app_autoscaling_states.get(deployment_id.app_name)
+        if app_state:
+            app_state.record_scale_up(deployment_id)
+
+    def record_scale_down(self, deployment_id: DeploymentID):
+        """Record a scale down event for a deployment.
+
+        Args:
+            deployment_id: The ID of the deployment being scaled down.
+        """
+        app_state = self._app_autoscaling_states.get(deployment_id.app_name)
+        if app_state:
+            app_state.record_scale_down(deployment_id)
+
     def on_replica_stopped(self, replica_id: ReplicaID):
         app_state = self._app_autoscaling_states.get(replica_id.deployment_id.app_name)
         if app_state:

python/ray/serve/_private/deployment_state.py

@@ -2361,6 +2361,7 @@ def autoscale(self, decision_num_replicas: int) -> bool:
                 trigger=DeploymentStatusInternalTrigger.AUTOSCALE_UP,
                 message=f"Upscaling from {old_num} to {new_num} replicas.",
             )
+            self._autoscaling_state_manager.record_scale_up(self._id)
         elif new_num < old_num:
             logger.info(
                 f"Downscaling {self._id} from {old_num} to {new_num} replicas. "
@@ -2370,6 +2371,8 @@ def autoscale(self, decision_num_replicas: int) -> bool:
                 trigger=DeploymentStatusInternalTrigger.AUTOSCALE_DOWN,
                 message=f"Downscaling from {old_num} to {new_num} replicas.",
             )
+            self._autoscaling_state_manager.record_scale_down(self._id)
+
         return True
 
     def delete(self) -> bool:

python/ray/serve/tests/unit/test_deployment_state.py

@@ -3785,6 +3785,176 @@ def test_handle_metrics_on_dead_serve_actor(self, mock_deployment_state_manager)
         dsm.update()
         check_counts(ds1, total=2, by_state=[(ReplicaState.STOPPING, 2, None)])
 
+    def test_autoscaling_timestamps(self, mock_deployment_state_manager):
+        """Test that last_scale_up_time and last_scale_down_time are properly tracked.
+
+        This test verifies that:
+        1. Timestamps are None initially
+        2. last_scale_up_time is set after a scale-up event
+        3. last_scale_down_time is set after a scale-down event
+        4. Timestamps are available in AutoscalingContext
+        """
+        # Create deployment state manager
+        create_dsm, timer, _, asm = mock_deployment_state_manager
+        dsm: DeploymentStateManager = create_dsm()
+        asm: AutoscalingStateManager = asm
+
+        # Deploy deployment with autoscaling
+        info, _ = deployment_info(
+            autoscaling_config={
+                "target_ongoing_requests": 1,
+                "min_replicas": 1,
+                "max_replicas": 5,
+                "initial_replicas": 2,
+                "upscale_delay_s": 0,
+                "downscale_delay_s": 0,
+                "metrics_interval_s": 100,
+            }
+        )
+        dsm.deploy(TEST_DEPLOYMENT_ID, info)
+        ds: DeploymentState = dsm._deployment_states[TEST_DEPLOYMENT_ID]
+
+        # Make replicas ready
+        dsm.update()
+        for replica in ds._replicas.get():
+            replica._actor.set_ready()
+        dsm.update()
+
+        # Get autoscaling state
+        app_state = asm._app_autoscaling_states[TEST_DEPLOYMENT_ID.app_name]
+        dep_autoscaling_state = app_state._deployment_autoscaling_states[
+            TEST_DEPLOYMENT_ID
+        ]
+
+        # Initially, timestamps should be None
+        ctx = dep_autoscaling_state.get_autoscaling_context(2)
+        assert ctx.last_scale_up_time is None
+        assert ctx.last_scale_down_time is None
+
+        # Trigger scale-up by setting high request metrics
+        replicas = ds._replicas.get()
+        req_per_replica = 5  # High load to trigger scale-up
+
+        if RAY_SERVE_COLLECT_AUTOSCALING_METRICS_ON_HANDLE:
+            handle_metric_report = HandleMetricReport(
+                deployment_id=TEST_DEPLOYMENT_ID,
+                handle_id="test_handle",
+                actor_id="test_actor",
+                handle_source=DeploymentHandleSource.UNKNOWN,
+                queued_requests=[TimeStampedValue(timer.time() - 0.1, 0)],
+                aggregated_queued_requests=0,
+                aggregated_metrics={
+                    RUNNING_REQUESTS_KEY: {
+                        replica._actor.replica_id: req_per_replica
+                        for replica in replicas
+                    }
+                },
+                metrics={
+                    RUNNING_REQUESTS_KEY: {
+                        replica._actor.replica_id: [
+                            TimeStampedValue(timer.time() - 0.1, req_per_replica)
+                        ]
+                        for replica in replicas
+                    }
+                },
+                timestamp=timer.time(),
+            )
+            asm.record_request_metrics_for_handle(handle_metric_report)
+        else:
+            for replica in replicas:
+                replica_metric_report = ReplicaMetricReport(
+                    replica_id=replica._actor.replica_id,
+                    aggregated_metrics={RUNNING_REQUESTS_KEY: req_per_replica},
+                    metrics={
+                        RUNNING_REQUESTS_KEY: [
+                            TimeStampedValue(timer.time() - 0.1, req_per_replica)
+                        ]
+                    },
+                    timestamp=timer.time(),
+                )
+                asm.record_request_metrics_for_replica(replica_metric_report)
+
+        # Record time before scale-up
+        time_before_scale_up = timer.time()
+
+        # Trigger autoscaling decision
+        self.scale(dsm, asm, [TEST_DEPLOYMENT_ID])
+
+        # After scale-up, last_scale_up_time should be set and greater than the time before
+        ctx_after_scale_up = dep_autoscaling_state.get_autoscaling_context(5)
+        assert ctx_after_scale_up.last_scale_up_time is not None
+        assert ctx_after_scale_up.last_scale_up_time >= time_before_scale_up
+        assert ctx_after_scale_up.last_scale_down_time is None
+
+        scale_up_time = ctx_after_scale_up.last_scale_up_time
+
+        # Advance timer to simulate time passing
+        timer.advance(10)
+
+        # Set replicas ready
+        dsm.update()
+        for replica in ds._replicas.get():
+            replica._actor.set_ready()
+        dsm.update()
+
+        # Now trigger scale-down by setting low request metrics
+        replicas = ds._replicas.get()
+        req_per_replica = 0  # No load to trigger scale-down
+
+        if RAY_SERVE_COLLECT_AUTOSCALING_METRICS_ON_HANDLE:
+            handle_metric_report = HandleMetricReport(
+                deployment_id=TEST_DEPLOYMENT_ID,
+                handle_id="test_handle",
+                actor_id="test_actor",
+                handle_source=DeploymentHandleSource.UNKNOWN,
+                queued_requests=[TimeStampedValue(timer.time() - 0.1, 0)],
+                aggregated_queued_requests=0,
+                aggregated_metrics={
+                    RUNNING_REQUESTS_KEY: {
+                        replica._actor.replica_id: req_per_replica
+                        for replica in replicas
+                    }
+                },
+                metrics={
+                    RUNNING_REQUESTS_KEY: {
+                        replica._actor.replica_id: [
+                            TimeStampedValue(timer.time() - 0.1, req_per_replica)
+                        ]
+                        for replica in replicas
+                    }
+                },
+                timestamp=timer.time(),
+            )
+            asm.record_request_metrics_for_handle(handle_metric_report)
+        else:
+            for replica in replicas:
+                replica_metric_report = ReplicaMetricReport(
+                    replica_id=replica._actor.replica_id,
+                    aggregated_metrics={RUNNING_REQUESTS_KEY: req_per_replica},
+                    metrics={
+                        RUNNING_REQUESTS_KEY: [
+                            TimeStampedValue(timer.time() - 0.1, req_per_replica)
+                        ]
+                    },
+                    timestamp=timer.time(),
+                )
+                asm.record_request_metrics_for_replica(replica_metric_report)
+
+        # Record time before scale-down
+        time_before_scale_down = timer.time()
+
+        # Trigger autoscaling decision for scale-down
+        self.scale(dsm, asm, [TEST_DEPLOYMENT_ID])
+
+        # After scale-down, last_scale_down_time should be set and greater than the time before
+        ctx_after_scale_down = dep_autoscaling_state.get_autoscaling_context(1)
+        assert (
+            ctx_after_scale_down.last_scale_up_time == scale_up_time
+        )  # Should remain unchanged
+        assert ctx_after_scale_down.last_scale_down_time is not None
+        assert ctx_after_scale_down.last_scale_down_time >= time_before_scale_down
+        assert ctx_after_scale_down.last_scale_down_time > scale_up_time
+
 
 class TestTargetCapacity:
     """