Stop computing autoscaling aggregates at the source

Removing simple mode left the producers still computing and shipping the
pre-aggregated scalars that nothing reads any more: the replica averaged its
running-requests window on every report, and the router averaged queued and
per-replica running requests.

Drop that work and the three vestigial fields. HandleMetricReport.total_requests
keeps its name and its four callers -- it is only used to report how much traffic a
dropped handle was carrying -- and is recomputed from the raw series the report
already carries.

Co-Authored-By: Claude <noreply@anthropic.com>
Signed-off-by: John Taylor <john.taylor@anyscale.com>

Author: johntaylor-cell

python/ray/serve/_private/common.py

@@ -989,14 +989,9 @@ class HandleMetricReport:
         handle_source: Describes what kind of entity holds this
             deployment handle: a Serve proxy, a Serve replica, or
             unknown.
-        aggregated_queued_requests: average number of queued requests at the
-            handle over the past look_back_period_s seconds.
         queued_requests: list of values of queued requests at the
             handle over the past look_back_period_s seconds. This is a list because
             we take multiple measurements over time.
-        aggregated_metrics: A map of metric name to the aggregated value over the past
-            look_back_period_s seconds at the handle for each replica. Replica keys
-            use ReplicaID.to_full_id_str() for efficient controller-side lookups.
         metrics: A map of metric name to the list of values running at that handle for each replica
             over the past look_back_period_s seconds. Replica keys use to_full_id_str().
             This is a list because we take multiple measurements over time.
@@ -1007,21 +1002,24 @@ class HandleMetricReport:
     handle_id: str
     actor_id: str
     handle_source: DeploymentHandleSource
-    aggregated_queued_requests: float
     queued_requests: TimeSeries
-    aggregated_metrics: Dict[
-        str, Dict[str, float]
-    ]  # replica key = ReplicaID.to_full_id_str()
     metrics: Dict[
         str, Dict[str, TimeSeries]
     ]  # replica key = ReplicaID.to_full_id_str()
     timestamp: float
 
     @property
     def total_requests(self) -> float:
-        """Total number of queued and running requests."""
-        return self.aggregated_queued_requests + sum(
-            self.aggregated_metrics.get(RUNNING_REQUESTS_KEY, {}).values()
+        """Most recently observed queued + running requests across this handle's
+        replicas. Diagnostic only (reported when a handle's metrics are dropped), so
+        the latest sample is enough and no windowing is applied."""
+
+        def latest(series: TimeSeries) -> float:
+            return series[-1].value if series else 0.0
+
+        return latest(self.queued_requests) + sum(
+            latest(series)
+            for series in self.metrics.get(RUNNING_REQUESTS_KEY, {}).values()
         )
 
     @property
@@ -1045,16 +1043,13 @@ class ReplicaMetricReport:
 
     Args:
         replica_id: The replica ID of the replica.
-        aggregated_metrics: A map of metric name to the aggregated value over the past
-            look_back_period_s seconds at the replica.
         metrics: A map of metric name to the list of values running at that replica
             over the past look_back_period_s seconds. This is a list because
             we take multiple measurements over time.
         timestamp: The time at which this report was created.
     """
 
     replica_id: ReplicaID
-    aggregated_metrics: Dict[str, float]
     metrics: Dict[str, TimeSeries]
     timestamp: float
 

python/ray/serve/_private/replica.py

@@ -962,21 +962,12 @@ def _push_autoscaling_metrics(self) -> None:
         look_back_period = self._autoscaling_config.look_back_period_s
         self._metrics_store.prune_keys_and_compact_data(time.time() - look_back_period)
 
-        new_aggregated_metrics = {}
         # The store keys are `Hashable`; this replica only ever records `str` keys.
         new_metrics = cast(Dict[str, TimeSeries], {**self._metrics_store.data})
 
-        if self.should_collect_ongoing_requests():
-            # Keep the legacy window_avg ongoing requests in the merged metrics dict
-            window_avg = (
-                self._metrics_store.aggregate_avg([RUNNING_REQUESTS_KEY])[0] or 0.0
-            )
-            new_aggregated_metrics.update({RUNNING_REQUESTS_KEY: window_avg})
-
         replica_metric_report = ReplicaMetricReport(
             replica_id=self._replica_id,
             timestamp=time.time(),
-            aggregated_metrics=new_aggregated_metrics,
             metrics=new_metrics,
         )
         with self._metrics_push_lock:

python/ray/serve/_private/router.py

@@ -491,54 +491,26 @@ def _add_autoscaling_metrics_point(self):
     def _get_metrics_report(self) -> HandleMetricReport:
         timestamp = time.time()
         running_requests = dict()
-        avg_running_requests = dict()
         autoscaling_config = self.autoscaling_config
         assert autoscaling_config is not None
         look_back_period = autoscaling_config.look_back_period_s
         self.metrics_store.prune_keys_and_compact_data(time.time() - look_back_period)
-        avg_queued_requests = self.metrics_store.aggregate_avg([QUEUED_REQUESTS_KEY])[0]
-        if avg_queued_requests is None:
-            # If the queued requests timeseries is empty, we set the
-            # average to the current number of queued requests.
-            avg_queued_requests = self.num_queued_requests
-        # If the queued requests timeseries is empty, we set the number of data points to 1.
-        # This is to avoid division by zero.
-        num_data_points = self.metrics_store.timeseries_count(QUEUED_REQUESTS_KEY) or 1
         queued_requests = self.metrics_store.data.get(
             QUEUED_REQUESTS_KEY, [TimeStampedValue(timestamp, self.num_queued_requests)]
         )
         if RAY_SERVE_COLLECT_AUTOSCALING_METRICS_ON_HANDLE and self.autoscaling_config:
             for replica_id, num_requests in self.num_requests_sent_to_replicas.items():
-                # Calculate avg running requests.
-                # NOTE (abrar): The number of data points from queued requests is often higher than
-                # those from running requests. This is because replica metrics are only collected
-                # once a replica is up, whereas queued request metrics are collected continuously
-                # as long as the handle is alive. To approximate the true average of ongoing requests,
-                # we should normalize by using the same number of data points for both queued and
-                # running request time series.
-                running_requests_sum = self.metrics_store.aggregate_sum([replica_id])[0]
-                if running_requests_sum is None:
-                    # If the running requests timeseries is empty, we set the sum
-                    # to the current number of requests.
-                    running_requests_sum = num_requests
-                replica_str = replica_id.to_full_id_str()
-                avg_running_requests[replica_str] = (
-                    running_requests_sum / num_data_points
-                )
-                # Get running requests data
-                running_requests[replica_str] = self.metrics_store.data.get(
+                running_requests[
+                    replica_id.to_full_id_str()
+                ] = self.metrics_store.data.get(
                     replica_id, [TimeStampedValue(timestamp, num_requests)]
                 )
         handle_metric_report = HandleMetricReport(
             deployment_id=self._deployment_id,
             handle_id=self._handle_id,
             actor_id=self._self_actor_id,
             handle_source=self._handle_source,
-            aggregated_queued_requests=avg_queued_requests,
             queued_requests=queued_requests,
-            aggregated_metrics={
-                RUNNING_REQUESTS_KEY: avg_running_requests,
-            },
             metrics={
                 RUNNING_REQUESTS_KEY: running_requests,
             },

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

@@ -2725,13 +2725,6 @@ def test_autoscaling_with_mixed_deployment_types(
                 actor_id="actor_id",
                 handle_source=DeploymentHandleSource.UNKNOWN,
                 queued_requests=[TimeStampedValue(timestamp_offset, 0)],
-                aggregated_queued_requests=0,
-                aggregated_metrics={
-                    RUNNING_REQUESTS_KEY: {
-                        r1.to_full_id_str(): 3,
-                        r2.to_full_id_str(): 3,
-                    }
-                },
                 metrics={
                     RUNNING_REQUESTS_KEY: {
                         r1.to_full_id_str(): [TimeStampedValue(timestamp_offset, 3)],
@@ -2745,7 +2738,6 @@ def test_autoscaling_with_mixed_deployment_types(
             for i in [1, 2]:
                 replica_report = ReplicaMetricReport(
                     replica_id=ReplicaID(unique_id=f"replica_{i}", deployment_id=d1_id),
-                    aggregated_metrics={RUNNING_REQUESTS_KEY: 3},
                     metrics={
                         RUNNING_REQUESTS_KEY: [TimeStampedValue(timestamp_offset, 3)]
                     },
@@ -2872,7 +2864,6 @@ def test_autoscale_multiple_apps_independent(
         for replica_id in app1_d1_replicas + app1_d2_replicas:
             replica_report = ReplicaMetricReport(
                 replica_id=replica_id,
-                aggregated_metrics={RUNNING_REQUESTS_KEY: 3},
                 metrics={RUNNING_REQUESTS_KEY: [TimeStampedValue(timestamp_offset, 3)]},
                 timestamp=time.time(),
             )
@@ -2882,7 +2873,6 @@ def test_autoscale_multiple_apps_independent(
         for replica_id in app2_d1_replicas + app2_d2_replicas:
             replica_report = ReplicaMetricReport(
                 replica_id=replica_id,
-                aggregated_metrics={RUNNING_REQUESTS_KEY: 0},
                 metrics={RUNNING_REQUESTS_KEY: [TimeStampedValue(timestamp_offset, 0)]},
                 timestamp=time.time(),
             )
@@ -2946,7 +2936,6 @@ def test_autoscale_with_partial_deployment_details(
         for i in [1, 2]:
             replica_report = ReplicaMetricReport(
                 replica_id=ReplicaID(unique_id=f"d1_replica_{i}", deployment_id=d1_id),
-                aggregated_metrics={RUNNING_REQUESTS_KEY: 3},
                 metrics={RUNNING_REQUESTS_KEY: [TimeStampedValue(timestamp_offset, 3)]},
                 timestamp=time.time(),
             )
@@ -3027,7 +3016,6 @@ def test_autoscale_single_deployment_in_app(self, mocked_application_state_manag
         for i in [1, 2]:
             replica_report = ReplicaMetricReport(
                 replica_id=ReplicaID(unique_id=f"replica_{i}", deployment_id=d1_id),
-                aggregated_metrics={RUNNING_REQUESTS_KEY: 4},
                 metrics={RUNNING_REQUESTS_KEY: [TimeStampedValue(timestamp_offset, 4)]},
                 timestamp=time.time(),
             )
@@ -3156,7 +3144,6 @@ def test_autoscale_many_deployments_in_app(self, mocked_application_state_manage
             for replica in replicas:
                 replica_report = ReplicaMetricReport(
                     replica_id=replica,
-                    aggregated_metrics={RUNNING_REQUESTS_KEY: load},
                     metrics={
                         RUNNING_REQUESTS_KEY: [TimeStampedValue(timestamp_offset, load)]
                     },
@@ -3231,7 +3218,6 @@ def test_autoscale_with_min_equals_max_replicas(
         for i in range(3):
             replica_report = ReplicaMetricReport(
                 replica_id=ReplicaID(unique_id=f"replica_{i}", deployment_id=d1_id),
-                aggregated_metrics={RUNNING_REQUESTS_KEY: 10},
                 metrics={
                     RUNNING_REQUESTS_KEY: [TimeStampedValue(timestamp_offset, 10)]
                 },
@@ -3368,13 +3354,6 @@ def _record_handle_metrics(
             actor_id="actor_id",
             handle_source=DeploymentHandleSource.UNKNOWN,
             queued_requests=[TimeStampedValue(timestamp_offset, 0)],
-            aggregated_queued_requests=0,
-            aggregated_metrics={
-                RUNNING_REQUESTS_KEY: {
-                    d1_r1.to_full_id_str(): d1_load,
-                    d1_r2.to_full_id_str(): d1_load,
-                }
-            },
             metrics={
                 RUNNING_REQUESTS_KEY: {
                     d1_r1.to_full_id_str(): [
@@ -3398,13 +3377,6 @@ def _record_handle_metrics(
             actor_id="actor_id",
             handle_source=DeploymentHandleSource.UNKNOWN,
             queued_requests=[TimeStampedValue(timestamp_offset, 0)],
-            aggregated_queued_requests=0,
-            aggregated_metrics={
-                RUNNING_REQUESTS_KEY: {
-                    d2_r3.to_full_id_str(): d2_load,
-                    d2_r4.to_full_id_str(): d2_load,
-                }
-            },
             metrics={
                 RUNNING_REQUESTS_KEY: {
                     d2_r3.to_full_id_str(): [
@@ -3427,7 +3399,6 @@ def _record_replica_metrics(
         for i in [1, 2]:
             replica_report = ReplicaMetricReport(
                 replica_id=ReplicaID(unique_id=f"replica_{i}", deployment_id=d1_id),
-                aggregated_metrics={RUNNING_REQUESTS_KEY: d1_load},
                 metrics={
                     RUNNING_REQUESTS_KEY: [TimeStampedValue(timestamp_offset, d1_load)]
                 },
@@ -3439,7 +3410,6 @@ def _record_replica_metrics(
         for i in [3, 4]:
             replica_report = ReplicaMetricReport(
                 replica_id=ReplicaID(unique_id=f"replica_{i}", deployment_id=d2_id),
-                aggregated_metrics={RUNNING_REQUESTS_KEY: d2_load},
                 metrics={
                     RUNNING_REQUESTS_KEY: [TimeStampedValue(timestamp_offset, d2_load)]
                 },