[serve] Transform replica level metrics to AutoScalingContext constructor args (ray-project#57202)

## Changes

1. Wire up the AutoScalingContext constructor args to make metrics
readable in the custom AutoScalingPolicy function.
2. dropped `requests_per_replica` since its expensive to compute
3. renamed `queued_requests` to `total_queued_requests` for consistency
with `total_num_requests`
4. added `total_running_requests`
5. added tests assert new fields are populated correctly
6. run custom metrics tests with
`RAY_SERVE_AGGREGATE_METRICS_AT_CONTROLLER` = 0 and 1
7. updated docs

---------

Signed-off-by: abrar <abrar@anyscale.com>
Signed-off-by: Arthur Leung <arcyleung@gmail.com>
Signed-off-by: Arthur Leung <arcyleung+github@gmail.com>
Co-authored-by: abrar <abrar@anyscale.com>
Co-authored-by: Arthur Leung <arcyleung@gmail.com>

Author: 3 people

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

@@ -668,5 +668,3 @@ In your policy, access custom metrics via:
 * **`ctx.raw_metrics[metric_name]`** — A mapping of replica IDs to lists of raw metric values.
   The number of data points stored for each replica depends on the [`look_back_period_s`](../api/doc/ray.serve.config.AutoscalingConfig.look_back_period_s.rst) (the sliding window size) and [`metrics_interval_s`](../api/doc/ray.serve.config.AutoscalingConfig.metrics_interval_s.rst) (the metric recording interval).
 * **`ctx.aggregated_metrics[metric_name]`** — A time-weighted average computed from the raw metric values for each replica.
-
-> Today, aggregation is a time-weighted average. In future releases, additional aggregation options may be supported.

doc/source/serve/doc_code/autoscaling_policy.py

@@ -29,8 +29,10 @@ def custom_metrics_autoscaling_policy(
 ) -> tuple[int, Dict[str, Any]]:
     cpu_usage_metric = ctx.aggregated_metrics.get("cpu_usage", {})
     memory_usage_metric = ctx.aggregated_metrics.get("memory_usage", {})
-    max_cpu_usage = max(cpu_usage_metric.values())
-    max_memory_usage = max(memory_usage_metric.values())
+    max_cpu_usage = list(cpu_usage_metric.values())[-1] if cpu_usage_metric else 0
+    max_memory_usage = (
+        list(memory_usage_metric.values())[-1] if memory_usage_metric else 0
+    )
 
     if max_cpu_usage > 80 or max_memory_usage > 85:
         return min(ctx.capacity_adjusted_max_replicas, ctx.current_num_replicas + 1), {}

doc/source/serve/doc_code/custom_metrics_autoscaling.py

@@ -9,6 +9,7 @@
     autoscaling_config={
         "min_replicas": 1,
         "max_replicas": 5,
+        "metrics_interval_s": 0.1,
         "policy": {
             "policy_function": "autoscaling_policy:custom_metrics_autoscaling_policy"
         },
@@ -21,7 +22,7 @@ def __init__(self):
         self.memory_usage = 60.0
 
     def __call__(self) -> str:
-        time.sleep(0.1)
+        time.sleep(0.5)
         self.cpu_usage = min(100, self.cpu_usage + 5)
         self.memory_usage = min(100, self.memory_usage + 3)
         return "Hello, world!"
@@ -39,10 +40,10 @@ def record_autoscaling_stats(self) -> Dict[str, float]:
 app = CustomMetricsDeployment.bind()
 # __serve_example_end__
 
-# TODO: uncomment after autoscaling context is populated with all metrics
-# if __name__ == "__main__":
-#     import requests  # noqa
+if __name__ == "__main__":
+    import requests  # noqa
 
-#     serve.run(app)
-#     resp = requests.get("http://localhost:8000/")
-#     assert resp.text == "Hello, world!"
+    serve.run(app)
+    for _ in range(10):
+        resp = requests.get("http://localhost:8000/")
+        assert resp.text == "Hello, world!"

doc/source/serve/doc_code/scheduled_batch_processing.py

@@ -12,6 +12,7 @@
         policy=AutoscalingPolicy(
             policy_function="autoscaling_policy:scheduled_batch_processing_policy"
         ),
+        metrics_interval_s=0.1,
     ),
     max_ongoing_requests=3,
 )