Address review on dropping source-side aggregation

Docs: the autoscaling guide still described the removed flag, simple mode as the
default, and aggregation_function as inert. Rewrite Stage 2/3 and add an upgrade
note, since aggregation_function now always applies -- a deployment that set max
or min and never enabled the flag changes behavior -- and controller-side merging
costs more than summing pre-computed scalars.

HandleMetricReport.total_requests summed the last sample of each series, but a
handle is only dropped once its report is stale, so a queue that drained on that
final sample logged nothing: exactly the case the log exists for. Sum the
per-series peak instead, and pin it in test_common.py.

aggregate_sum, aggregate_avg, timeseries_count and _aggregate_reduce lost their
last callers along with simple mode; delete them and re-point the store tests at
the stored series. Fold _calculate_total_requests_aggregate_mode into
get_total_num_requests now that there is one mode, drop the duplicate empty check
in _get_queued_requests, drop the guard in _get_metrics_report that re-tested an
already-asserted local, and route the async-inference delay through
_record_metrics_delay so the third copy of the delay-and-tag block goes away.

_collect_handle_running_requests is the sole path under the collect-on-handle
default and scanned every running replica for every handle; iterate the report of
each handle instead, which the order-independence of the merge allows.

Give test_num_replicas_auto_basic a collect-on-replica target so its looser
upscaling assertion is reachable at all, and set
RAY_SERVE_COLLECT_AUTOSCALING_METRICS_ON_HANDLE explicitly in both metric
variants rather than inheriting the default.

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

Author: johntaylor-cell

doc/source/serve/advanced-guides/advanced-autoscaling.md

@@ -125,33 +125,19 @@ Replicas and deployment handles continuously record autoscaling metrics:
 
 Periodically, replicas and handles push their metrics to the controller:
 - **Frequency**: Every 10s (configurable via `RAY_SERVE_REPLICA_AUTOSCALING_METRIC_PUSH_INTERVAL_S` and `RAY_SERVE_HANDLE_AUTOSCALING_METRIC_PUSH_INTERVAL_S`)
-- **Data sent**: Both raw timeseries data and pre-aggregated metrics
-  - **Raw timeseries**: Data points are clipped to the [`look_back_period_s`](../api/doc/ray.serve.config.AutoscalingConfig.rst) window before sending (only recent measurements within the window are sent)
-  - **Pre-aggregated metrics**: A simple average computed over the [`look_back_period_s`](../api/doc/ray.serve.config.AutoscalingConfig.rst) window at the replica/handle
-- **Controller usage**: The controller decides which data to use based on the `RAY_SERVE_AGGREGATE_METRICS_AT_CONTROLLER` setting (see Stage 3 below)
+- **Data sent**: Raw timeseries data. Data points are clipped to the [`look_back_period_s`](../api/doc/ray.serve.config.AutoscalingConfig.rst) window before sending (only recent measurements within the window are sent)
 
 #### Stage 3: Metric aggregation
 
-The controller aggregates metrics to compute total ongoing requests across all replicas. Ray Serve supports two aggregation modes (controlled by `RAY_SERVE_AGGREGATE_METRICS_AT_CONTROLLER`):
-
-**Simple mode (default - `RAY_SERVE_AGGREGATE_METRICS_AT_CONTROLLER=0`):**
-- **Input**: Pre-aggregated simple averages from replicas/handles (already clipped to [`look_back_period_s`](../api/doc/ray.serve.config.AutoscalingConfig.rst))
-- **Method**: Sums the pre-aggregated values from all sources. Each component computes a simple average (arithmetic mean) before sending.
-- **Output**: Single value representing total ongoing requests
-- **Characteristics**: Lightweight and works well for most workloads. However, because it uses simple averages rather than time-weighted averages, it can be less accurate when replicas have different metric reporting intervals or when metrics arrive at different times.
-
-**Aggregate mode (experimental - `RAY_SERVE_AGGREGATE_METRICS_AT_CONTROLLER=1`):**
+The controller aggregates the raw timeseries to compute total ongoing requests across all replicas:
 - **Input**: Raw timeseries data from replicas/handles (already clipped to [`look_back_period_s`](../api/doc/ray.serve.config.AutoscalingConfig.rst))
 - **Method**: Time-weighted aggregation using the [`aggregation_function`](../api/doc/ray.serve.config.AutoscalingConfig.rst) (mean, max, or min). Uses an instantaneous merge approach that treats metrics as right-continuous step functions.
 - **Output**: Single value representing total ongoing requests
-- **Characteristics**: Provides more mathematically accurate aggregation, especially when replicas report metrics at different intervals or you need precise time-weighted averages. The trade-off is increased controller overhead.
-
-:::{note}
-The [`aggregation_function`](../api/doc/ray.serve.config.AutoscalingConfig.rst) parameter only applies in aggregate mode. In simple mode, the aggregation is always a sum of the pre-computed simple averages.
-:::
 
 :::{note}
-The long-term plan is to deprecate simple mode in favor of aggregate mode. Aggregate mode provides more accurate metrics aggregation and will become the default in a future release. Consider testing aggregate mode(`RAY_SERVE_AGGREGATE_METRICS_AT_CONTROLLER=1`) in your deployments to prepare for this transition.
+Earlier releases also supported a "simple mode" that summed averages pre-computed at each replica and handle, selected by `RAY_SERVE_AGGREGATE_METRICS_AT_CONTROLLER`. That flag and that mode are removed: the controller now always aggregates raw timeseries. Two consequences if you are upgrading from a release that had the flag, where it defaulted to simple mode:
+- [`aggregation_function`](../api/doc/ray.serve.config.AutoscalingConfig.rst) now always applies. Under simple mode it was ignored, so a deployment that set `max` or `min` and never enabled the flag scaled on a mean and now scales on a peak or a trough.
+- Aggregating timeseries costs the controller more than summing pre-computed averages, and the cost grows with the number of replicas. Watch controller CPU on deployments with very high replica counts.
 :::
 
 #### Stage 4: Policy execution
@@ -201,10 +187,6 @@ Several environment variables control autoscaling behavior at a lower level. The
 
 * **`RAY_SERVE_MIN_HANDLE_METRICS_TIMEOUT_S`** (default: 10.0s): Minimum timeout for handle metrics collection. The system uses the maximum of this value and `2 * `[`metrics_interval_s`](../api/doc/ray.serve.config.AutoscalingConfig.rst) to determine when to drop stale handle metrics.
 
-#### Advanced feature flags
-
-* **`RAY_SERVE_AGGREGATE_METRICS_AT_CONTROLLER`** (default: false): Enables an experimental metrics aggregation mode where the controller aggregates raw timeseries data instead of using pre-aggregated metrics. This mode provides more accurate time-weighted averages but may increase controller overhead. See Stage 3 in "How autoscaling metrics work" for details.
-
 
 ## Model composition example
 

python/ray/serve/_private/autoscaling_state.py

@@ -462,10 +462,11 @@ def _collect_handle_running_requests(self) -> List[TimeSeries]:
 
         for handle_metric in self._handle_requests.values():
             running_reqs = handle_metric.metrics.get(RUNNING_REQUESTS_KEY, {})
-            for replica_str in self._cached_running_replica_strs:
-                if replica_str not in running_reqs:
-                    continue
-                timeseries_list.append(running_reqs[replica_str])
+            # Iterate the handle's own replicas, not every running replica: a handle
+            # usually routes to a subset, and the merge is order-independent.
+            for replica_str, timeseries in running_reqs.items():
+                if replica_str in self._cached_running_replica_strs:
+                    timeseries_list.append(timeseries)
 
         return timeseries_list
 
@@ -543,11 +544,17 @@ def _merge_and_aggregate_timeseries(
 
         return 0.0
 
-    def _calculate_total_requests_aggregate_mode(self) -> float:
-        """Calculate total requests using aggregate metrics mode with timeseries data.
+    def get_total_num_requests(self) -> float:
+        """Get average total number of requests aggregated over the past
+        `look_back_period_s` number of seconds.
 
-        This method works with raw timeseries metrics data and performs aggregation
-        at the controller level.
+        Works with raw timeseries metrics data and performs aggregation at the
+        controller level. If there are 0 running replicas, then returns the total
+        number of requests queued at handles.
+
+        This code assumes that the metrics are either emmited on handles
+        or on replicas, but not both. Its the responsibility of the writer
+        to ensure enclusivity of the metrics.
 
         Processing Steps:
             1. Collect raw timeseries data (eg: running request) from replicas (if available)
@@ -556,12 +563,6 @@ def _calculate_total_requests_aggregate_mode(self) -> float:
             4. Merge timeseries using instantaneous approach for mathematically correct totals
             5. Calculate time-weighted average running requests from the merged timeseries
 
-        Key Differences from Simple Mode:
-            - Uses raw timeseries data instead of pre-aggregated metrics
-            - Performs instantaneous merging for exact gauge semantics
-            - Aggregates at the controller level rather than using pre-computed averages
-            - Uses time-weighted averaging over the look_back_period_s interval for accurate calculations
-
         Metrics Collection:
             Running requests are collected with either replica-level or handle-level metrics.
 
@@ -637,19 +638,6 @@ def _calculate_total_requests_aggregate_mode(self) -> float:
 
         return ongoing_requests
 
-    def get_total_num_requests(self) -> float:
-        """Get average total number of requests aggregated over the past
-        `look_back_period_s` number of seconds.
-
-        If there are 0 running replicas, then returns the total number
-        of requests queued at handles
-
-        This code assumes that the metrics are either emmited on handles
-        or on replicas, but not both. Its the responsibility of the writer
-        to ensure enclusivity of the metrics.
-        """
-        return self._calculate_total_requests_aggregate_mode()
-
     def get_replica_metrics(self) -> Dict[str, List[TimeSeries]]:
         """Get the raw replica metrics dict."""
         metric_values: Dict[str, List[TimeSeries]] = defaultdict(list)
@@ -666,11 +654,9 @@ def _get_queued_requests(self) -> float:
         Returns:
             Sum of queued requests at all handles, aggregated from handle timeseries.
         """
-        queued_timeseries = self._collect_handle_queued_requests()
-        if not queued_timeseries:
-            return 0.0
-
-        return self._merge_and_aggregate_timeseries(queued_timeseries)
+        return self._merge_and_aggregate_timeseries(
+            self._collect_handle_queued_requests()
+        )
 
     def _get_aggregated_custom_metrics(self) -> Dict[str, Dict[ReplicaID, float]]:
         """Aggregate custom metrics from replica metric reports.

python/ray/serve/_private/common.py

@@ -1010,17 +1010,12 @@ class HandleMetricReport:
 
     @property
     def total_requests(self) -> float:
-        """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()
-        )
+        """Upper bound on queued + running requests over this handle's reported window.
+        Diagnostic only (logged when a handle's metrics are dropped); peaks are summed
+        so a handle that was busy earlier in the window is not reported as idle."""
+        running = self.metrics.get(RUNNING_REQUESTS_KEY, {}).values()
+        series = [self.queued_requests, *running]
+        return sum(max(point.value for point in s) for s in series if s)
 
     @property
     def is_serve_component_source(self) -> bool:

python/ray/serve/_private/constants.py

@@ -1118,7 +1118,6 @@
 if RAY_SERVE_INGRESS_REQUEST_ROUTER_METRICS_ENABLED:
     RAY_SERVE_HAPROXY_METRICS_ENABLED = True
 
-
 # Feature flag to include high-cardinality source tags on Serve controller metrics.
 # Disable this to keep deployment/application tags while dropping source identifiers
 # like replica IDs from controller-emitted metrics.

python/ray/serve/_private/controller.py

@@ -390,20 +390,21 @@ def _record_metrics_delay(
         self,
         timestamp: float,
         deployment_id: DeploymentID,
-        histogram: metrics.Histogram,
-        record_delay: Callable[[float], None],
+        report_delay: Callable[..., None],
+        record_delay: Optional[Callable[[float], None]] = None,
     ) -> None:
-        """Report ingest delay. A histogram lets Prometheus aggregate reports from all
-        sources of a deployment, so the per-source tag is omitted to bound cardinality."""
+        """Report ingest delay. Prometheus aggregates reports from all sources of a
+        deployment, so the per-source tag is omitted to bound cardinality."""
         delay_ms = (time.time() - timestamp) * 1000
-        histogram.observe(
+        report_delay(
             delay_ms,
             tags={
                 "deployment": deployment_id.name,
                 "application": deployment_id.app_name,
             },
         )
-        record_delay(delay_ms)
+        if record_delay is not None:
+            record_delay(delay_ms)
 
     def record_autoscaling_metrics_from_replica(
         self, replica_metric_report: Union[ReplicaMetricReport, bytes]
@@ -415,7 +416,7 @@ def record_autoscaling_metrics_from_replica(
         self._record_metrics_delay(
             replica_metric_report.timestamp,
             replica_metric_report.replica_id.deployment_id,
-            self.replica_metrics_delay_histogram,
+            self.replica_metrics_delay_histogram.observe,
             self._health_metrics_tracker.record_replica_metrics_delay,
         )
         self.autoscaling_state_manager.record_request_metrics_for_replica(
@@ -432,7 +433,7 @@ def record_autoscaling_metrics_from_handle(
         self._record_metrics_delay(
             handle_metric_report.timestamp,
             handle_metric_report.deployment_id,
-            self.handle_metrics_delay_histogram,
+            self.handle_metrics_delay_histogram.observe,
             self._health_metrics_tracker.record_handle_metrics_delay,
         )
         self.autoscaling_state_manager.record_request_metrics_for_handle(
@@ -443,15 +444,10 @@ def record_autoscaling_metrics_from_async_inference_task_queue(
         self, report: AsyncInferenceTaskQueueMetricReport
     ):
         """Record async inference task queue metrics pushed from QueueMonitor."""
-        latency = time.time() - report.timestamp_s
-        latency_ms = latency * 1000
-        # Record the metrics delay for observability
-        self.async_inference_task_queue_metrics_delay_gauge.set(
-            latency_ms,
-            tags={
-                "deployment": report.deployment_id.name,
-                "application": report.deployment_id.app_name,
-            },
+        self._record_metrics_delay(
+            report.timestamp_s,
+            report.deployment_id,
+            self.async_inference_task_queue_metrics_delay_gauge.set,
         )
         self.autoscaling_state_manager.record_async_inference_task_queue_metrics(report)
 

python/ray/serve/_private/metrics_utils.py

@@ -1,10 +1,8 @@
 import asyncio
 import bisect
 import logging
-import statistics
 from collections import defaultdict
 from dataclasses import dataclass
-from itertools import chain
 from typing import (
     Any,
     Awaitable,
@@ -13,10 +11,8 @@
     DefaultDict,
     Dict,
     Hashable,
-    Iterable,
     List,
     Optional,
-    Tuple,
     Union,
 )
 
@@ -194,66 +190,6 @@ def _get_datapoints(
         )
         return datapoints[idx:]
 
-    def _aggregate_reduce(
-        self,
-        keys: Iterable[Hashable],
-        aggregate_fn: Callable[[Iterable[float]], float],
-    ) -> Tuple[Optional[float], int]:
-        """Reduce the entire set of timeseries values across the specified keys.
-
-        Args:
-            keys: Iterable of keys to aggregate across.
-            aggregate_fn: Function to apply across all float values, e.g., sum, max.
-
-        Returns:
-            A tuple of (float, int) where the first element is the aggregated value
-            and the second element is the number of valid keys used.
-            Returns (None, 0) if no valid keys have data.
-
-        Example:
-        Suppose the store contains:
-        >>> store = InMemoryMetricsStore()
-        >>> store.data.update({
-        ...     "a": [TimeStampedValue(0, 1.0), TimeStampedValue(1, 2.0)],
-        ...     "b": [],
-        ...     "c": [TimeStampedValue(0, 10.0)],
-        ... })
-
-        Using sum across keys:
-
-        >>> store._aggregate_reduce(keys=["a", "b", "c"], aggregate_fn=sum)
-        (13.0, 2)
-
-        Here:
-        - The aggregated value is 1.0 + 2.0 + 10.0 = 13.0
-        - Only keys "a" and "c" contribute values, so report_count = 2
-        """
-        valid_key_count = 0
-
-        def _values_generator():
-            """Generator that yields values from valid keys without storing them all in memory."""
-            nonlocal valid_key_count
-            for key in keys:
-                series = self.data.get(key, [])
-                if not series:
-                    continue
-
-                valid_key_count += 1
-                for timestamp_value in series:
-                    yield timestamp_value.value
-
-        # Create the generator and check if it has any values
-        values_gen = _values_generator()
-        try:
-            first_value = next(values_gen)
-        except StopIteration:
-            # No valid data found
-            return None, 0
-
-        # Apply aggregation to the generator (memory efficient)
-        aggregated_result = aggregate_fn(chain([first_value], values_gen))
-        return aggregated_result, valid_key_count
-
     def get_latest(
         self,
         key: Hashable,
@@ -263,47 +199,6 @@ def get_latest(
             return None
         return self.data[key][-1].value
 
-    def aggregate_sum(
-        self,
-        keys: Iterable[Hashable],
-    ) -> Tuple[Optional[float], int]:
-        """Sum the entire set of timeseries values across the specified keys.
-        Args:
-            keys: Iterable of keys to aggregate across.
-        Returns:
-            A tuple of (float, int) where the first element is the sum across
-            all values found at `keys`, and the second is the number of valid
-            keys used to compute the sum.
-            Returns (None, 0) if no valid keys have data.
-        """
-        return self._aggregate_reduce(keys, sum)
-
-    def aggregate_avg(
-        self,
-        keys: Iterable[Hashable],
-    ) -> Tuple[Optional[float], int]:
-        """Average the entire set of timeseries values across the specified keys.
-
-        Args:
-            keys: Iterable of keys to aggregate across.
-        Returns:
-            A tuple of (float, int) where the first element is the mean across
-            all values found at `keys`, and the second is the number of valid
-            keys used to compute the mean.
-            Returns (None, 0) if no valid keys have data.
-        """
-        return self._aggregate_reduce(keys, statistics.mean)
-
-    def timeseries_count(
-        self,
-        key: Hashable,
-    ) -> int:
-        """Count the number of values across all timeseries values at the specified keys."""
-        series = self.data.get(key, [])
-        if not series:
-            return 0
-        return len(series)
-
 
 def time_weighted_average(
     step_series: TimeSeries,

python/ray/serve/_private/router.py

@@ -498,7 +498,7 @@ def _get_metrics_report(self) -> HandleMetricReport:
         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:
+        if RAY_SERVE_COLLECT_AUTOSCALING_METRICS_ON_HANDLE:
             for replica_id, num_requests in self.num_requests_sent_to_replicas.items():
                 running_requests[
                     replica_id.to_full_id_str()