Repository navigation
[serve] Columnar zero-copy autoscaling-metrics ingest (plumbing) - #66070
Conversation
There was a problem hiding this comment.
Code Review
This pull request introduces an array-native, columnar format for autoscaling metrics in Ray Serve, optimizing the decision-time merge and aggregation paths using NumPy. It adds a new autoscaling_metrics_merge module, integrates columnar and mixed-mode aggregations into AutoscalingStateManager, and adds performance tracking for metrics ingestion. The review feedback focuses on correcting a misleading comment about dropping columnar replica entries, replacing platform-specific "<i8" dtype literals with standard np.int64, and adhering to PEP 8 naming conventions by removing leading underscores from actively used loop variables.
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes using default effort and found 1 potential issue.
Reviewed by Cursor Bugbot for commit c59920d. Configure here.
afb922a to
3fdb82f
Compare
There was a problem hiding this comment.
Code Review
This pull request introduces a columnar wire format for autoscaling metrics in Ray Serve to improve performance by enabling efficient array-based processing. Key changes include adding numpy-based aggregation logic, updating the autoscaling state manager to handle columnar metrics with memoization, and implementing a unified staleness gate across wire formats. The controller is updated to wire-detect the format and track decoding metrics separately from decompression. I have kept the review comment regarding a potential type mismatch in numpy array indexing, as it identifies a critical issue that could lead to runtime errors.
4e8d94d to
2dea8d2
Compare
2dea8d2 to
75fcabe
Compare
abrarsheikh
left a comment
There was a problem hiding this comment.
High-level comments
- code is currently hard to read and requires refactoring. I think it requires some more revisions.
- offs / csr / run_offsets, which are one concept under three names — pick source_offsets and use it everywhere.
- ts now means two types in this file: a TimeSeries object list, a timestamps ndarray in _running_csr method and one other place, in the autoscaling state file.
Refactored, clarified variable names |
The controller wire-detects the SCR1 magic on each handle report and decodes columnar frames straight into flat float64 arrays, keyed per handle. Aggregation routes three ways: all-columnar, all-object, or a single merged pass when both stores hold sources. The mixed path is a steady state rather than a rollout artifact. Replica reports are never columnar, and direct-ingress deployments report from replicas even when handle collection is on, so both stores are live at once. It merges every source in one pass because summing two separate aggregations is correct for MEAN but wrong for MAX and MIN. Decode time is recorded apart from decompress, since keeping the two wire formats comparable is the point of that metric. Scoped to correctness. The tick optimisations for wide handles, a vectorised gather and a memo on the running-set generation, ride with the producer switch, since nothing exercises them until frames actually flow. Nothing emits columnar frames yet, so this is inert until that switch lands. Co-Authored-By: Claude <noreply@anthropic.com> Signed-off-by: john.taylor <john.taylor@anyscale.com>
…types Introduces MetricSamples (timestamps/values/source_offsets) and ColumnarHandleReport so the columnar and object paths share one vocabulary instead of passing bare tuples around. - Three-way aggregation dispatch (columnar / object / mixed) collapses to two: no columnar sources takes the object kernel, anything else takes the array merge, which already absorbs object sources. _columnar_aggregate_total_requests is redundant with it and is deleted. - The two near-identical stale-handle drop loops become one over both stores. - Self-free helpers move to module scope; _single_source_block, _series_to_arrays, _series_blocks and _columnar_peak_requests fold into MetricSamples. - *_blocks -> *_samples throughout, and `ts` no longer means both "timestamp" and "timeseries" in the same file. No behaviour change: the 34 columnar tests and the 894-test unit sweep over every suite touching autoscaling_state are unchanged. Co-Authored-By: Claude <noreply@anthropic.com> Signed-off-by: john.taylor <john.taylor@anyscale.com>
Follow-up to the dispatcher refactor, all comments and annotations. - from_timeseries said only what the signature already said. It now states what it returns, and its rationale no longer rests on a mixed fleet, since the dispatcher comment says no fleet runs both formats until routers emit SCR1. - _array_aggregate_total_requests claimed mixing arises from "a thin driver handle alongside wide proxy handles". Both are handles and every HandleMetricReport encodes columnar, so that pair cannot mix. Corrected to the account get_total_num_requests already gives: object replica reports alongside columnar handle reports. - _running_samples had a mangled reflow from the blocks-to-samples rename sweep. - Two params the typing sweep missed, both in code this refactor added: _handle_running_columnar_samples(running: Set[str]) and _log_dropped_handle's report, which is duck-typed across both report types and was the only unannotated param in its own signature. No behaviour change: the 894-test unit sweep over every suite touching autoscaling_state is unchanged. Co-Authored-By: Claude <noreply@anthropic.com> Signed-off-by: john.taylor <john.taylor@anyscale.com>
Sweep for other instances of the untyped-tuple pattern, not just the one reviewed. The replica running-series memo was an anonymous 2-tuple read as cached[0]/cached[1]; annotating the dict typed it without fixing what the positional reads cost a reader. It is now a _ReplicaRunningMemo NamedTuple. Renamed _replica_running_arrays to _replica_running_memo in the same pass: it holds MetricSamples, so "arrays" was left over from the pre-refactor naming. Co-Authored-By: Claude <noreply@anthropic.com> Signed-off-by: john.taylor <john.taylor@anyscale.com>
82cfdd3 to
8bef45e
Compare
Three dicts keyed by handle ID encoded an invariant that no single writer owned: a handle must be in exactly one of the object or columnar store, and the timestamp gate must track exactly what the two hold. Correctness depended on each record path remembering to evict the other format and on a separate repair pass over the gate. _HandleMetricStore now owns all three, with accept() as the only writer, so the invariant holds by construction. drop_stale_handle_metrics loses its repair loop: forget() drops a report and its gate entry together. The columnar path keeps a cheap is_fresher() pre-check so a late frame skips _running_samples, the expensive part of ingest, rather than decoding only to have accept discard the result. test_handle_store_invariant_survives_interleaved_writes asserts both halves of the invariant after every write in a mixed sequence, and after a drop. It fails if accept stops evicting the other store, and independently if forget stops clearing the gate. Co-Authored-By: Claude <noreply@anthropic.com> Signed-off-by: john.taylor <john.taylor@anyscale.com>
## Fixes Fixes #66271 ## Problem `drop_stale_handle_metrics` is documented as dropping handles the controller has not **received** updates from within `max(2 * metrics_interval_s, RAY_SERVE_MIN_HANDLE_METRICS_TIMEOUT_S)`, but it compared controller `time.time()` to the producer's `report.timestamp`. That mixes wall clocks from different machines, so skew can drop fresh metrics early (producer behind) or retain phantoms past the timeout (producer ahead). ## What changes - `_HandleMetricStore` records a controller-local **monotonic** receive time in `_received_at` when `accept()` succeeds. - `drop_stale_handle_metrics` ages handles as `time.monotonic() - received_at` for the timeout path. - `forget()` clears `_received_at` with the rest of the handle state. - Rejected (stale) accepts do **not** refresh receive time. ## What is preserved (intentional) - Producer timestamps remain the sole key for `is_fresher` / cross-format ordering (#66070). - Timeseries sample timestamps and aggregation look-back stay producer-based. - Ingest-delay metrics (`_record_metrics_delay`) stay producer-based. - Dead Serve-actor dropping is unchanged. - No wire / public API change; receive time is never put on the wire. This is scoped only to the drop freshness mismatch with the documented "received" semantics. Aggregation `last_window_s` skew and delay-metric docs are left alone on purpose. ## Tests - New unit coverage for producer clock behind/ahead, rejected accept not refreshing receive time, fresher accept refreshing it, dead-actor path, and columnar parity. - Existing columnar drop / fresher / cross-format tests updated so timeout cases patch `time.monotonic` alongside `time.time`. - `mock_deployment_state_manager` also patches `time.monotonic` so handle-timeout deployment-state tests keep working. ## Related Credit to @suppagoddo for the issue writeup and receive-time approach. Coordinates with #66070 (ordering gate) and does not wait on #66240. cc @ray-project/ray-serve @edoakes @zcin @johntaylor-cell @suppagoddo --------- Signed-off-by: Dundy Pasupuleti <dundysm@gmail.com> Co-authored-by: Abrar Sheikh <abrar@anyscale.com>
… cut per-tick work (#66240) Last of the six Handle Metric Report optimization PRs. Metric Reports carry metrics that are used by the Serve Controller to auto-scale deployments. PRs 1-5 are merged and every one of them is inert: nothing on master produces a columnar report. This PR is the switch that enables new format, and provides a revert lever. ## The series | Step | PR | Status | |------|-----|--------| | 1 · codec | [#64281](#64281) | merged 2026-09-10 — the `SCR1` wire format alone; nothing called it, so the merge was a no-op | | 2 · kernels | [#66104](#66104) | merged 2026-09-12 — `autoscaling_metrics_merge`, flat-array forms of the timeseries kernels. Imports only `typing` and `numpy` | | 3 · extraction | [#66105](#66105) | merged 2026-09-12 — pure move of the object aggregation out of `get_total_num_requests`, byte-identical body | | 4 · instrumentation | [#66106](#66106) | merged 2026-09-14 — ingest and decompress duration, report counters, windowed `ingest_cpu_fraction` on the controller health schema | | 5 · ingest | [#66070](#66070) | merged 2026-09-17 — array store, mixed aggregation path, controller decode and its own decode-duration metric | | 6 · producer flip + tick opts | this PR | the switch | ## What changes By default, routers now emit handle reports with a columnar format instead of cloudpickled python objects. Each report identifies its format. The FF RAY_SERVE_COLUMNAR_AUTOSCALING_METRICS can be used to revert to python objects. Replica reports remain object based until they become aggregated by the HAProxy Manager (future work). The change in router.py enables the new format. The rest of this PR manages down scaling, when replicas are reaped, we do not want to keep using their metrics in the autoscaling decisions. This "masking" decision is cached (memoised) to prevent any perf degradation. The mask cache lives in `_HandleMetricStore` rather than on the report, which is frozen. That also means the handle metric store's existing `accept`/`forget` invalidate it, so a dropped handle cannot leave a stale mask behind. ## Testing `test_columnar` 44 passed, `test_router` back to its baseline after porting the push assertion to the columnar frame, and the unit suites touching `autoscaling_state` are unchanged: `test_deployment_state` 258, `test_application_state` 122, `test_autoscaling_policy` 47, `test_proxy_state` 38, `test_controller` 71. Each optimization has a test that fails when it is reverted, checked by mutation: - dropping the cached mask in `accept` -> `test_mask_memo_invalidated_by_a_new_report` - bumping the running-set generation -> `test_masked_running_samples_follow_the_running_set` - filtering data-free rows -> `test_empty_running_rows_keep_the_whole_frame_fast_path` The masking test regains its `stop_at` axis. Only when the stopped replica is last are the survivors a contiguous prefix and the gather an identity permutation, so a first or middle stop is what actually exercises the index arithmetic. <img width="2025" height="810" alt="perf_A5_ingest" src="https://github.com/user-attachments/assets/8fb04c75-c889-4b62-ab8b-1d8dedbac45f" /> Most recent results just comparing columnar vs pickled: <img width="2000" height="800" alt="perf_66240_columnar_aggregation" src="https://github.com/user-attachments/assets/501dc381-59aa-41af-883d-f89b03b62ec5" /> --------- Signed-off-by: john.taylor <john.taylor@anyscale.com> Co-authored-by: Claude <noreply@anthropic.com>
…project#66307) ## Fixes Fixes ray-project#66271 ## Problem `drop_stale_handle_metrics` is documented as dropping handles the controller has not **received** updates from within `max(2 * metrics_interval_s, RAY_SERVE_MIN_HANDLE_METRICS_TIMEOUT_S)`, but it compared controller `time.time()` to the producer's `report.timestamp`. That mixes wall clocks from different machines, so skew can drop fresh metrics early (producer behind) or retain phantoms past the timeout (producer ahead). ## What changes - `_HandleMetricStore` records a controller-local **monotonic** receive time in `_received_at` when `accept()` succeeds. - `drop_stale_handle_metrics` ages handles as `time.monotonic() - received_at` for the timeout path. - `forget()` clears `_received_at` with the rest of the handle state. - Rejected (stale) accepts do **not** refresh receive time. ## What is preserved (intentional) - Producer timestamps remain the sole key for `is_fresher` / cross-format ordering (ray-project#66070). - Timeseries sample timestamps and aggregation look-back stay producer-based. - Ingest-delay metrics (`_record_metrics_delay`) stay producer-based. - Dead Serve-actor dropping is unchanged. - No wire / public API change; receive time is never put on the wire. This is scoped only to the drop freshness mismatch with the documented "received" semantics. Aggregation `last_window_s` skew and delay-metric docs are left alone on purpose. ## Tests - New unit coverage for producer clock behind/ahead, rejected accept not refreshing receive time, fresher accept refreshing it, dead-actor path, and columnar parity. - Existing columnar drop / fresher / cross-format tests updated so timeout cases patch `time.monotonic` alongside `time.time`. - `mock_deployment_state_manager` also patches `time.monotonic` so handle-timeout deployment-state tests keep working. ## Related Credit to @suppagoddo for the issue writeup and receive-time approach. Coordinates with ray-project#66070 (ordering gate) and does not wait on ray-project#66240. cc @ray-project/ray-serve @edoakes @zcin @johntaylor-cell @suppagoddo --------- Signed-off-by: Dundy Pasupuleti <dundysm@gmail.com> Co-authored-by: Abrar Sheikh <abrar@anyscale.com>
…-project#66070) ## Why are these changes needed? Follow-up to ray-project#64281 (codec), ray-project#66104 (kernels), ray-project#66105 (extraction) and ray-project#66106 (ingest metrics). This is the consumer side: the controller decodes columnar handle reports and aggregates them as arrays. The `ServeController` aggregates autoscaling metrics in a single-threaded control loop. Each handle report covers every replica that handle routes to, so at high fan-in the controller spends much of each loop cloudpickle-decoding reports, rebuilding one Python object per timestamped point. This decodes the `SCR1` format into flat float64 arrays instead, and merges them in a single vectorized pass. ### What changed The controller wire-detects the frame magic per report, so it reads whatever each sender actually emitted rather than relying on config agreement. Columnar frames decode straight into a per-handle array store; everything else takes the existing path unchanged. `get_total_num_requests` routes three ways: all-columnar, all-object, or one merged pass when both stores hold sources. The mixed case is a steady state, not a rollout artifact — replica reports are never columnar, and direct-ingress deployments report from replicas even when handle collection is on, so both stores are live simultaneously. It has to merge every source in one pass because summing two separate aggregations is correct for MEAN but wrong for MAX and MIN. Decode duration is recorded separately from decompress duration, since keeping the two wire formats comparable is the point of that metric. Scoped to correctness. The tick optimizations for wide handles — a vectorized gather and a memo on the running-set generation — ride with the producer switch, because nothing exercises them until frames actually flow. The masking here is the straightforward per-replica form. ### This commit is inert Nothing produces columnar frames yet. The producer switch is six lines in `router.py` and lands in the final PR, so merging this cannot change behaviour: the decode branch is never entered. That also makes the ordering safe by construction — a router cannot emit `SCR1` to a controller that does not yet understand it. ### Test Plan `test_columnar.py` runs 34 cases covering the array store, the three aggregation paths, controller dispatch, mixed-format fleets, cross-format staleness, stale-replica masking and pruning, and drop logging. Also ran every suite touching the changed modules: `test_autoscaling_policy` (47), `test_deployment_state` (256), `test_controller` (71), `test_application_state` (122), `test_schema` (116). --------- Signed-off-by: john.taylor <john.taylor@anyscale.com> Co-authored-by: Claude <noreply@anthropic.com>
…project#66307) ## Fixes Fixes ray-project#66271 ## Problem `drop_stale_handle_metrics` is documented as dropping handles the controller has not **received** updates from within `max(2 * metrics_interval_s, RAY_SERVE_MIN_HANDLE_METRICS_TIMEOUT_S)`, but it compared controller `time.time()` to the producer's `report.timestamp`. That mixes wall clocks from different machines, so skew can drop fresh metrics early (producer behind) or retain phantoms past the timeout (producer ahead). ## What changes - `_HandleMetricStore` records a controller-local **monotonic** receive time in `_received_at` when `accept()` succeeds. - `drop_stale_handle_metrics` ages handles as `time.monotonic() - received_at` for the timeout path. - `forget()` clears `_received_at` with the rest of the handle state. - Rejected (stale) accepts do **not** refresh receive time. ## What is preserved (intentional) - Producer timestamps remain the sole key for `is_fresher` / cross-format ordering (ray-project#66070). - Timeseries sample timestamps and aggregation look-back stay producer-based. - Ingest-delay metrics (`_record_metrics_delay`) stay producer-based. - Dead Serve-actor dropping is unchanged. - No wire / public API change; receive time is never put on the wire. This is scoped only to the drop freshness mismatch with the documented "received" semantics. Aggregation `last_window_s` skew and delay-metric docs are left alone on purpose. ## Tests - New unit coverage for producer clock behind/ahead, rejected accept not refreshing receive time, fresher accept refreshing it, dead-actor path, and columnar parity. - Existing columnar drop / fresher / cross-format tests updated so timeout cases patch `time.monotonic` alongside `time.time`. - `mock_deployment_state_manager` also patches `time.monotonic` so handle-timeout deployment-state tests keep working. ## Related Credit to @suppagoddo for the issue writeup and receive-time approach. Coordinates with ray-project#66070 (ordering gate) and does not wait on ray-project#66240. cc @ray-project/ray-serve @edoakes @zcin @johntaylor-cell @suppagoddo --------- Signed-off-by: Dundy Pasupuleti <dundysm@gmail.com> Co-authored-by: Abrar Sheikh <abrar@anyscale.com>
… cut per-tick work (ray-project#66240) Last of the six Handle Metric Report optimization PRs. Metric Reports carry metrics that are used by the Serve Controller to auto-scale deployments. PRs 1-5 are merged and every one of them is inert: nothing on master produces a columnar report. This PR is the switch that enables new format, and provides a revert lever. ## The series | Step | PR | Status | |------|-----|--------| | 1 · codec | [ray-project#64281](ray-project#64281) | merged 2026-09-10 — the `SCR1` wire format alone; nothing called it, so the merge was a no-op | | 2 · kernels | [ray-project#66104](ray-project#66104) | merged 2026-09-12 — `autoscaling_metrics_merge`, flat-array forms of the timeseries kernels. Imports only `typing` and `numpy` | | 3 · extraction | [ray-project#66105](ray-project#66105) | merged 2026-09-12 — pure move of the object aggregation out of `get_total_num_requests`, byte-identical body | | 4 · instrumentation | [ray-project#66106](ray-project#66106) | merged 2026-09-14 — ingest and decompress duration, report counters, windowed `ingest_cpu_fraction` on the controller health schema | | 5 · ingest | [ray-project#66070](ray-project#66070) | merged 2026-09-17 — array store, mixed aggregation path, controller decode and its own decode-duration metric | | 6 · producer flip + tick opts | this PR | the switch | ## What changes By default, routers now emit handle reports with a columnar format instead of cloudpickled python objects. Each report identifies its format. The FF RAY_SERVE_COLUMNAR_AUTOSCALING_METRICS can be used to revert to python objects. Replica reports remain object based until they become aggregated by the HAProxy Manager (future work). The change in router.py enables the new format. The rest of this PR manages down scaling, when replicas are reaped, we do not want to keep using their metrics in the autoscaling decisions. This "masking" decision is cached (memoised) to prevent any perf degradation. The mask cache lives in `_HandleMetricStore` rather than on the report, which is frozen. That also means the handle metric store's existing `accept`/`forget` invalidate it, so a dropped handle cannot leave a stale mask behind. ## Testing `test_columnar` 44 passed, `test_router` back to its baseline after porting the push assertion to the columnar frame, and the unit suites touching `autoscaling_state` are unchanged: `test_deployment_state` 258, `test_application_state` 122, `test_autoscaling_policy` 47, `test_proxy_state` 38, `test_controller` 71. Each optimization has a test that fails when it is reverted, checked by mutation: - dropping the cached mask in `accept` -> `test_mask_memo_invalidated_by_a_new_report` - bumping the running-set generation -> `test_masked_running_samples_follow_the_running_set` - filtering data-free rows -> `test_empty_running_rows_keep_the_whole_frame_fast_path` The masking test regains its `stop_at` axis. Only when the stopped replica is last are the survivors a contiguous prefix and the gather an identity permutation, so a first or middle stop is what actually exercises the index arithmetic. <img width="2025" height="810" alt="perf_A5_ingest" src="https://github.com/user-attachments/assets/8fb04c75-c889-4b62-ab8b-1d8dedbac45f" /> Most recent results just comparing columnar vs pickled: <img width="2000" height="800" alt="perf_66240_columnar_aggregation" src="https://github.com/user-attachments/assets/501dc381-59aa-41af-883d-f89b03b62ec5" /> --------- Signed-off-by: john.taylor <john.taylor@anyscale.com> Co-authored-by: Claude <noreply@anthropic.com>

Why are these changes needed?
Follow-up to #64281 (codec), #66104 (kernels), #66105 (extraction) and #66106 (ingest metrics). This is the consumer side: the controller decodes columnar handle reports and aggregates them as arrays.
The
ServeControlleraggregates autoscaling metrics in a single-threaded control loop. Each handle report covers every replica that handle routes to, so at high fan-in the controller spends much of each loop cloudpickle-decoding reports, rebuilding one Python object per timestamped point. This decodes theSCR1format into flat float64 arrays instead, and merges them in a single vectorized pass.What changed
The controller wire-detects the frame magic per report, so it reads whatever each sender actually emitted rather than relying on config agreement. Columnar frames decode straight into a per-handle array store; everything else takes the existing path unchanged.
get_total_num_requestsroutes three ways: all-columnar, all-object, or one merged pass when both stores hold sources. The mixed case is a steady state, not a rollout artifact — replica reports are never columnar, and direct-ingress deployments report from replicas even when handle collection is on, so both stores are live simultaneously. It has to merge every source in one pass because summing two separate aggregations is correct for MEAN but wrong for MAX and MIN.Decode duration is recorded separately from decompress duration, since keeping the two wire formats comparable is the point of that metric.
Scoped to correctness. The tick optimizations for wide handles — a vectorized gather and a memo on the running-set generation — ride with the producer switch, because nothing exercises them until frames actually flow. The masking here is the straightforward per-replica form.
This commit is inert
Nothing produces columnar frames yet. The producer switch is six lines in
router.pyand lands in the final PR, so merging this cannot change behaviour: the decode branch is never entered. That also makes the ordering safe by construction — a router cannot emitSCR1to a controller that does not yet understand it.Test Plan
test_columnar.pyruns 34 cases covering the array store, the three aggregation paths, controller dispatch, mixed-format fleets, cross-format staleness, stale-replica masking and pruning, and drop logging.Also ran every suite touching the changed modules:
test_autoscaling_policy(47),test_deployment_state(256),test_controller(71),test_application_state(122),test_schema(116).