[metrics 1/5] Add scoped vLLM offload and latency window metrics

[SumanthRH]

What does this PR do?

Add vLLM preemption, external prefix-cache, KV reload and request-latency metrics to SkyRL's existing step collection.

TLDR: Collect the launched servers' metrics, reduce counter and histogram deltas, and use request TPOT for tpot_seconds_avg.

Why this is needed

The existing scraper does not report preemptions or external-cache/offload activity, and its TPOT field uses inter-token latency. These gaps make it difficult to compare cache configurations alongside training throughput. Unfiltered cluster samples can also include other deployments.

How it works

• Add VLLMServerActor.get_ray_worker_id() and resolve the API server actors' WorkerIds during setup. Filter samples to those workers. The lookup has a 10-second timeout; failure disables collection without stopping training. External deployments without owned actors retain unfiltered collection.
• Sum engine counters before taking deltas. Merge compatible classic-histogram buckets before interpolating P90. Failed endpoint scrapes discard partial snapshots; missing baselines and invalid latency windows are omitted. A missing bucket baseline is treated as zero only when the previous histogram count is explicitly zero.
• Log preemption counts and preemptions per million output tokens, external hit rate, KV reload bytes/s, and TTFT/request TPOT mean and P90. TPOT uses request_time_per_output_token_seconds. ITL, P50 and offload store throughput remain available as raw metrics for Grafana.

Sync keys retain vllm/train/* and vllm/eval/*; fully async keys retain vllm/*. Collection uses the existing generator.inference_engine.enable_ray_prometheus_stats flag.

Validation

Collector tests cover worker filtering, bounded identity lookup, partial HTTP failures, counter resets, latency formulas and lazy histogram buckets. API-server source and historical actor PID/worker/export records confirm that the lookup returns the WorkerId used by vLLM's Ray metrics logger.

Combined-stack CPU check at 04290d61 (#2401): 70 tests passed. --noconftest skips the repository's automatic Ray lifecycle fixture so these mock/CPU checks do not attach to the live cluster. Ray 2.58.0 matches the validation cluster.

uv run --isolated --extra skyrl-train --extra dev --with ray==2.58.0 pytest --noconftest \
  tests/train/test_grafana_annotations.py tests/train/test_metrics_lifecycle.py \
  tests/train/test_vllm_run_statistics.py tests/train/test_vllm_metrics_scraper.py \
  tests/train/test_vllm_window_statistics.py tests/train/test_vllm_engine_imbalance.py \
  tests/train/test_vllm_pd_metrics.py tests/train/test_tracking.py -q

No new GPU benchmark was run for this layer.

---

Note

Medium Risk
Changes how TPOT and percentiles are defined for logged metrics and adds setup-time Ray worker identity resolution; failures are handled by disabling collection, but mis-scoped or missing metrics could affect monitoring during training.

Overview
Expands step-level vLLM metrics logged to the trainer tracker and scopes scraping to SkyRL-launched inference servers by resolving each VLLMServerActor's Ray WorkerId at setup (10s timeout; lookup failure disables collection without aborting training).

The scraper now emits preemptions, external prefix-cache hit rate, KV offload load throughput, and TTFT/request-TPOT averages plus P90 (merged histogram bucket deltas via new vllm_window_statistics). tpot_seconds_avg switches from inter-token latency to request_time_per_output_token_seconds, so it is not comparable to older runs. Partial node scrape failures drop the whole snapshot and reset the delta baseline.

Docs and tests cover worker filtering, incomplete snapshots, and the new reduction paths.

^{Reviewed by Cursor Bugbot for commit 1e2064c. Bugbot is set up for automated code reviews on this repo. Configure here.}

[gemini-code-assist]

Code Review

This pull request enhances vLLM metrics collection by adding several new metrics (such as preemptions, external prefix cache hit rates, and P90 latency percentiles), filtering snapshots by Ray worker IDs, and introducing a new window statistics utility. The review feedback highlights a critical bug in get_ray_worker_id where calling a non-existent get_worker_id() method on the Ray runtime context will raise an AttributeError. Additionally, the feedback suggests robustness improvements to prevent a single counter reset from invalidating all latency metrics, as well as performance optimizations to avoid redundant dictionary allocations and multiple iterations over metric labels during scraping.

[gemini-code-assist]

The ray.get_runtime_context() object does not have a get_worker_id() method. Calling it will raise an AttributeError, which will silently disable vLLM metrics collection during setup. Instead, use the worker_id property and convert it to a hex string.

Suggested change

	return ray.get_runtime_context().get_worker_id()
	return ray.get_runtime_context().worker_id.hex()

[gemini-code-assist]

Returning None when any single counter resets invalidates all latency metrics (both TTFT and TPOT), even if the reset occurred in an unrelated counter (like speculative decoding or prefix cache queries). Skipping the resetting counter using continue is much more robust, as the downstream latency_metrics and histogram_quantile functions are already designed to handle missing or incomplete counters safely.

Suggested change

	if not math.isfinite(delta) or delta < 0:
	return None
	if not math.isfinite(delta) or delta < 0:
	continue

[gemini-code-assist]

Converting key[1] (which is a frozenset of label tuples) to a dictionary via dict(key[1]) for every single metric sample on every step is inefficient. We can perform a direct generator check over the frozenset tuples to avoid allocating a new dictionary for every key.

Suggested change

	if self._worker_ids is not None:
	parsed = {key: value for key, value in parsed.items() if dict(key[1]).get("WorkerId") in self._worker_ids}
	if self._worker_ids is not None:
	parsed = {
	key: value
	for key, value in parsed.items()
	if any(k == "WorkerId" and v in self._worker_ids for k, v in key[1])
	}

[gemini-code-assist]

Creating a dictionary dict(labels) and then iterating over labels again to construct owner is inefficient as it performs multiple passes over the labels. We can extract bound and construct the owner pairs in a single pass over labels.

            bound = None
            owner_pairs = []
            for k, v in labels:
                if k == "le":
                    bound = v
                else:
                    owner_pairs.append((k, v))
            if bound is None:
                continue
            owner = frozenset(owner_pairs)

[greptile-apps]

Confidence Score: 4/5

[Medium risk] Adds new vLLM metrics collection with worker filtering.

The PR should not merge until an unrelated counter reset can no longer discard valid latency windows and aggregate summaries.

Findings

1.  Unrelated resets discard latency ▶
2.  Filtered samples disappear silently ▶

Diagram

%%{init: {'theme': 'neutral'}}%%
flowchart LR
  A[Launched server actors] -->|WorkerIds| F[Snapshot filter]
  R[Ray metrics endpoints] --> F
  F --> C[Merge counters and histogram buckets]
  C --> W[Window deltas]
  W --> T[Step metrics and tracker]
  W --> S[Higher-stack run summaries]

Loading

_{Reviews (1) · Last reviewed commit: "Use metric constants and expose the Ray ..."}

[greptile-apps]

Filtered samples disappear silently If a successful scrape contains samples but none match the fixed actor membership, this filter returns no snapshot without a warning: the empty-sample warning runs before filtering. The run then loses vLLM telemetry, and operators have no diagnostic pointing to the membership mismatch.

Knowledge Base Used: Trainer execution and evaluation

[cursor]

Cursor Bugbot has reviewed your changes and found 1 potential issue.

^{❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, have a team admin enable autofix in the Cursor dashboard.}

^{Reviewed by Cursor Bugbot for commit 1e2064c. Configure here.}