llm-monitor/tests/test_rates.py

"""Derived throughput: Δcounter/Δt and the lifetime-average fallback.

The deployed vLLM exports no rate gauge of any kind (verified: zero series with a
throughput/rate name), so prompt tok/s can only be *computed*. These tests pin the
precedence — native gauge > measured Δcounter > estimate > n/a — and the counter-reset
handling, because a naive delta shows up as a giant negative spike after an engine
restart.
"""

from __future__ import annotations

import pytest

from app import RateEstimator, parse_gufo_metrics, parse_vllm_metrics

# gufo's generation-speed gauge is ONE request's last decode pass; the counter is every
# slot's tokens. 30 tok/s on the gauge with 60 tokens landing per second = two slots.
GUFO_SCRAPE = """
llamacpp:requests_processing 2
llamacpp:tokens_predicted_total 1000
llamacpp:predicted_tokens_seconds 30
"""

COUNTERS = """
vllm:num_requests_running{engine="0"} 2
vllm:prompt_tokens_total{engine="0"} 1000
vllm:generation_tokens_total{engine="0"} 400
vllm:inter_token_latency_seconds_sum 0.04
vllm:inter_token_latency_seconds_count 2
"""


def test_first_scrape_has_nothing_to_diff_against() -> None:
    rates = RateEstimator()
    assert rates.rate("k", 100.0, now=10.0) == (None, None)


def test_second_scrape_measures_the_delta() -> None:
    rates = RateEstimator()
    rates.rate("k", 100.0, now=10.0)
    rate, window = rates.rate("k", 130.0, now=11.0)
    assert rate == pytest.approx(30.0)
    assert window == pytest.approx(1.0)


def test_counter_reset_yields_none_not_a_negative_spike() -> None:
    rates = RateEstimator()
    rates.rate("k", 5000.0, now=10.0)
    assert rates.rate("k", 12.0, now=11.0) == (None, None)  # engine restarted
    # and it recovers on the following scrape
    assert rates.rate("k", 22.0, now=12.0)[0] == pytest.approx(10.0)


def test_absent_counter_and_zero_window_are_none() -> None:
    rates = RateEstimator()
    assert rates.rate("k", None, now=1.0) == (None, None)
    rates.rate("k", 1.0, now=5.0)
    assert rates.rate("k", 2.0, now=5.0) == (None, None)  # dt == 0
    assert rates.rate("other", 1.0, now=6.0) == (None, None)  # unknown key


def test_rate_smooths_over_the_span_once_enough_samples_exist() -> None:
    # a bare 1s delta jitters (prompts arrive in bursts); the span averages it out
    rates = RateEstimator(span=10.0)
    rates.rate("k", 0.0, now=0.0)    # outside the span by t=12
    rates.rate("k", 100.0, now=5.0)
    rate, window = rates.rate("k", 400.0, now=12.0)
    assert window == pytest.approx(7.0)  # measured from the oldest in-span sample
    assert rate == pytest.approx(300.0 / 7.0)


def test_only_two_in_window_samples_still_produce_a_shorter_rate() -> None:
    rates = RateEstimator(span=10.0)
    rates.rate("k", 0.0, now=0.0)
    rates.rate("k", 100.0, now=20.0)  # first sample is long gone
    rate, window = rates.rate("k", 150.0, now=21.0)
    assert window == pytest.approx(1.0)
    assert rate == pytest.approx(50.0)


def test_history_is_bounded() -> None:
    rates = RateEstimator(span=1.0, max_samples=8)
    for i in range(50):
        rates.rate("k", float(i), now=float(i))
    assert len(rates._samples["k"]) <= 8


def test_keys_are_independent() -> None:
    rates = RateEstimator()
    rates.rate("a", 0.0, now=0.0)
    rates.rate("b", 0.0, now=0.0)
    assert rates.rate("a", 10.0, now=1.0)[0] == pytest.approx(10.0)
    assert rates.rate("b", 4.0, now=1.0)[0] == pytest.approx(4.0)


# --------------------------------------------------------------------------- #
# gufo aggregate decode rate: the counter adds slots, the gauge cannot
# --------------------------------------------------------------------------- #
def test_gufo_decode_rate_adds_every_slot_generating_at_once() -> None:
    """The whole point: two slots at 30 tok/s must read 60, not 30.

    gufo's `predicted_tokens_seconds` is a single unlabeled gauge holding one request's
    last decode pass, so it can never sum concurrency. `tokens_predicted_total` advances
    once per token from every slot, so Δ/Δt is the server-aggregate.
    """
    rates = RateEstimator()
    parse_gufo_metrics(GUFO_SCRAPE, rates, now=100.0)
    m = parse_gufo_metrics(GUFO_SCRAPE.replace("1000", "1060"), rates, now=101.0)
    d = m["decode_tokens_per_second"]
    assert d["value"] == pytest.approx(60.0)
    assert d["derived"] is True
    assert d["window"] == pytest.approx(1.0)
    assert d["source"] == "computed:Δllamacpp:tokens_predicted_total/Δt"
    # the single-request gauge stays exposed, untouched, as the native reference
    assert m["decode_tokens_per_second_native"]["value"] == pytest.approx(30.0)


def test_gufo_first_scrape_falls_back_to_the_gauge_not_na() -> None:
    # nothing to diff against yet, but the scrape can still answer
    m = parse_gufo_metrics(GUFO_SCRAPE, RateEstimator(), now=10.0)
    assert m["decode_tokens_per_second"] == {
        "value": 30.0,
        "source": "llamacpp:predicted_tokens_seconds",
        "derived": False,
    }


def test_gufo_idle_reads_zero_not_the_stale_gauge() -> None:
    # an engine that generated nothing in the window is at 0 tok/s; the gauge would
    # happily keep showing the last session's speed forever
    rates = RateEstimator()
    parse_gufo_metrics(GUFO_SCRAPE, rates, now=0.0)
    m = parse_gufo_metrics(GUFO_SCRAPE, rates, now=1.0)
    assert m["decode_tokens_per_second"]["value"] == 0.0
    assert m["decode_tokens_per_second_native"]["value"] == pytest.approx(30.0)


def test_gufo_counter_reset_falls_back_instead_of_going_negative() -> None:
    rates = RateEstimator()
    parse_gufo_metrics(GUFO_SCRAPE, rates, now=0.0)
    # engine restarted: the counter dropped. Never -995 tok/s.
    m = parse_gufo_metrics(GUFO_SCRAPE.replace("1000", "5"), rates, now=1.0)
    assert m["decode_tokens_per_second"]["value"] == pytest.approx(30.0)
    assert m["decode_tokens_per_second"]["derived"] is False
    # and it recovers on the following scrape
    m = parse_gufo_metrics(GUFO_SCRAPE.replace("1000", "25"), rates, now=2.0)
    assert m["decode_tokens_per_second"]["value"] == pytest.approx(20.0)


def test_gufo_alias_rename_starts_a_fresh_window_instead_of_a_fake_rate() -> None:
    # the estimator keys off the RESOLVED series name, so a renamed counter cannot diff
    # against a different series' history (and a flapping alias cannot invent throughput)
    rates = RateEstimator()
    parse_gufo_metrics("llamacpp:tokens_generated_total 1000\n", rates, now=0.0)
    m = parse_gufo_metrics("llamacpp:tokens_predicted_total 1000\n", rates, now=1.0)
    assert m["decode_tokens_per_second"]["value"] is None


def test_gufo_prompt_rate_is_measured_and_adds_slots_too() -> None:
    """Prefill gets the same treatment as decode, so a cross-engine prompt sum is additive.

    gufo's `prompt_tokens_seconds` is one request's last prefill pass; the counter is every
    slot's newly-computed prompt tokens.
    """
    text = "llamacpp:prompt_tokens_total 1000\nllamacpp:prompt_tokens_seconds 700\n"
    rates = RateEstimator()
    parse_gufo_metrics(text, rates, now=0.0)
    m = parse_gufo_metrics(text.replace("1000", "3400"), rates, now=2.0)
    p = m["prompt_tokens_per_second"]
    assert p["value"] == pytest.approx(1200.0)
    assert p["source"] == "computed:Δllamacpp:prompt_tokens_total/Δt"
    assert p["derived"] is True and p["window"] == pytest.approx(2.0)
    # the one-request gauge stays exposed next to it
    assert m["prompt_tokens_per_second_native"]["value"] == pytest.approx(700.0)


def test_gufo_prompt_first_scrape_falls_back_to_the_gauge() -> None:
    text = "llamacpp:prompt_tokens_total 1000\nllamacpp:prompt_tokens_seconds 700\n"
    m = parse_gufo_metrics(text, RateEstimator(), now=5.0)
    assert m["prompt_tokens_per_second"] == {
        "value": 700.0,
        "source": "llamacpp:prompt_tokens_seconds",
        "derived": False,
    }


def test_gufo_prompt_and_decode_keep_separate_windows() -> None:
    # one estimator, but the two counters must not read each other's history
    rates = RateEstimator()
    base = "llamacpp:prompt_tokens_total 1000\nllamacpp:tokens_predicted_total 500\n"
    parse_gufo_metrics(base, rates, now=0.0)
    m = parse_gufo_metrics(base.replace("1000", "3000").replace("500", "550"), rates, now=2.0)
    assert m["prompt_tokens_per_second"]["value"] == pytest.approx(1000.0)
    assert m["decode_tokens_per_second"]["value"] == pytest.approx(25.0)


def test_gufo_prompt_idle_reads_zero_not_the_stale_gauge() -> None:
    text = "llamacpp:prompt_tokens_total 1000\nllamacpp:prompt_tokens_seconds 700\n"
    rates = RateEstimator()
    parse_gufo_metrics(text, rates, now=0.0)
    m = parse_gufo_metrics(text, rates, now=1.0)
    assert m["prompt_tokens_per_second"]["value"] == 0.0
    assert m["prompt_tokens_per_second_native"]["value"] == pytest.approx(700.0)


def test_gufo_parse_without_rates_stays_pure() -> None:
    assert parse_gufo_metrics(GUFO_SCRAPE) == parse_gufo_metrics(GUFO_SCRAPE)
    assert parse_gufo_metrics(GUFO_SCRAPE)["decode_tokens_per_second"]["derived"] is False


def test_both_engines_share_one_estimator_without_cross_reading() -> None:
    # RATES is one object for both engines; the series names keep the histories apart
    rates = RateEstimator()
    parse_gufo_metrics(GUFO_SCRAPE, rates, now=0.0)
    parse_vllm_metrics(COUNTERS, rates, now=0.0)
    g = parse_gufo_metrics(GUFO_SCRAPE.replace("1000", "1040"), rates, now=1.0)
    v = parse_vllm_metrics(COUNTERS.replace("400", "410"), rates, now=1.0)
    assert g["decode_tokens_per_second"]["value"] == pytest.approx(40.0)
    assert v["decode_tokens_per_second"]["source"] == "computed:Δvllm:generation_tokens_total/Δt"
    assert v["decode_tokens_per_second"]["value"] == pytest.approx(10.0)


# --------------------------------------------------------------------------- #
# parser integration
# --------------------------------------------------------------------------- #
def test_prompt_and_decode_use_measured_deltas_when_rates_are_supplied() -> None:
    rates = RateEstimator()
    parse_vllm_metrics(COUNTERS, rates, now=100.0)  # seeds the previous sample
    m = parse_vllm_metrics(
        COUNTERS.replace("1000", "1500").replace("400", "450"), rates, now=101.0
    )
    p = m["prompt_tokens_per_second"]
    d = m["decode_tokens_per_second"]
    assert p["derived"] is True and p["window"] == pytest.approx(1.0)
    assert p["value"] == pytest.approx(500.0)
    assert p["source"] == "computed:Δvllm:prompt_tokens_total/Δt"
    assert d["value"] == pytest.approx(50.0)
    assert d["source"] == "computed:Δvllm:generation_tokens_total/Δt"


def test_first_scrape_falls_back_to_the_lifetime_average() -> None:
    # no previous sample to diff against, but prefill time + prompt length still answer
    text = (
        "vllm:prompt_tokens_total 1000\n"
        "vllm:request_prompt_tokens_sum 8000\nvllm:request_prompt_tokens_count 2\n"
        "vllm:request_prefill_time_seconds_sum 1.6\nvllm:request_prefill_time_seconds_count 2\n"
    )
    m = parse_vllm_metrics(text, RateEstimator(), now=10.0)
    p = m["prompt_tokens_per_second"]
    assert p["value"] == pytest.approx(5000.0)  # 4000 tok mean / 0.8 s mean
    assert p["source"] == "computed:mean(request_prompt_tokens)/mean(request_prefill_time_seconds)"
    assert m["prefill_tokens_per_second_lifetime"]["value"] == pytest.approx(5000.0)


def test_measured_delta_beats_the_lifetime_average() -> None:
    rates = RateEstimator()
    text = (
        "vllm:prompt_tokens_total 1000\n"
        "vllm:request_prompt_tokens_sum 8000\nvllm:request_prompt_tokens_count 2\n"
        "vllm:request_prefill_time_seconds_sum 1.6\nvllm:request_prefill_time_seconds_count 2\n"
    )
    parse_vllm_metrics(text, rates, now=0.0)
    m = parse_vllm_metrics(text.replace("1000", "1200"), rates, now=1.0)
    assert m["prompt_tokens_per_second"]["source"].startswith("computed:Δ")
    assert m["prompt_tokens_per_second"]["value"] == pytest.approx(200.0)


def test_native_gauge_still_wins_over_any_derivation() -> None:
    m = parse_vllm_metrics(
        "vllm:prompt_throughput 1234.5\nvllm:prompt_tokens_total 10\n",
        RateEstimator(),
        now=1.0,
    )
    assert m["prompt_tokens_per_second"] == {
        "value": 1234.5,
        "source": "vllm:prompt_throughput",
        "derived": False,
    }


def test_without_rates_the_parse_stays_pure() -> None:
    # parse(text) alone must not depend on wall-clock or hidden state
    a = parse_vllm_metrics(COUNTERS)
    b = parse_vllm_metrics(COUNTERS)
    assert a == b
    assert a["prompt_tokens_per_second"]["value"] is None  # no prefill series here
    assert a["prefill_tokens_per_second_lifetime"]["value"] is None
    # decode still answers via TPOT, which is stateless
    assert a["decode_tokens_per_second"]["source"] == "computed:running/tpot"


def test_prompt_tok_s_is_never_a_silent_zero() -> None:
    """The whole point: an engine that IS working must not render n/a because it lacks
    a gauge, and an idle engine must show 0 rather than a stale estimate."""
    rates = RateEstimator()
    text = "vllm:prompt_tokens_total 500\n"
    parse_vllm_metrics(text, rates, now=0.0)
    m = parse_vllm_metrics(text, rates, now=1.0)
    assert m["prompt_tokens_per_second"]["value"] == 0.0  # genuinely idle, not unknown