diff --git a/backends/exllamav3/model.py b/backends/exllamav3/model.py index a0902327..f64a3d75 100644 --- a/backends/exllamav3/model.py +++ b/backends/exllamav3/model.py @@ -37,6 +37,7 @@ from common.health import HealthManager from common.errors import ContextLengthExceededError, validate_context_requirements from common.logger import xlogger +from common.metrics import MetricsManager from common.multimodal import MultimodalEmbeddingWrapper from common.networking import DisconnectHandler from common.optional_dependencies import check_package_version @@ -351,6 +352,12 @@ async def create(cls, model_directory: pathlib.Path, hf_model: HFModel, **kwargs self.max_seq_len = max_seq_len self.cache_size = cache_size + # Size the /metrics token histograms to this model's context length. + # Their buckets are meaningless until the range they have to cover is + # known, and a ladder that overshoots it reports percentiles above any + # request the server can even accept. + MetricsManager.configure_token_buckets(max_seq_len) + # Max batch size default_mbs = 4 if self.model.caps.get("recurrent_states") else 128 self.max_batch_size = unwrap(kwargs.get("max_batch_size"), default_mbs) @@ -1065,7 +1072,8 @@ def handle_finish_chunk(self, result: dict, request_id: str, full_text: str): # Prompt prompt_tokens = result.get("prompt_tokens") cached_tokens = round(result.get("cached_tokens"), 2) - prompt_time = round(result.get("time_prefill"), 2) + raw_prompt_time = result.get("time_prefill") + prompt_time = round(raw_prompt_time, 2) prompt_ts = ( "Indeterminate" if prompt_time == 0 @@ -1111,6 +1119,20 @@ def handle_finish_chunk(self, result: dict, request_id: str, full_text: str): } ) + # Accumulate server-wide metrics for the /metrics endpoint + MetricsManager.record_generation( + prompt_tokens=prompt_tokens, + cached_tokens=cached_tokens, + gen_tokens=gen_tokens, + # Unrounded, so the aggregate rates are not skewed by the 0.01s + # display rounding on short prefills. + prompt_time=raw_prompt_time, + gen_time=gen_time, + queue_time=queue_time, + accepted_draft_tokens=accepted_draft_tokens, + rejected_draft_tokens=rejected_draft_tokens, + ) + return finish_chunk async def generate_gen( diff --git a/common/config_models.py b/common/config_models.py index fb0db10e..0528a5ee 100644 --- a/common/config_models.py +++ b/common/config_models.py @@ -81,6 +81,14 @@ class NetworkConfig(BaseConfigModel): ), ge=0, ) + enable_metrics: Optional[bool] = Field( + False, + description=( + "Enable the Prometheus-compatible /metrics endpoint (default: False).\n" + "Exposes aggregate inference stats in the text exposition format.\n" + "NOTE: This endpoint is not protected by API key authentication." + ), + ) # Converts all strings in the api_servers list to lowercase # NOTE: Expand if more models need this validator diff --git a/common/metrics.py b/common/metrics.py new file mode 100644 index 00000000..64b114a8 --- /dev/null +++ b/common/metrics.py @@ -0,0 +1,974 @@ +"""Global inference metrics for the Prometheus-compatible /metrics endpoint. + +Modeled after llama.cpp's `/metrics` exporter: process-lifetime counters are +accumulated as generations finish, while gauges (throughput, in-flight and +queued requests, KV-cache usage) are computed live at scrape time. Per-request +latency and size distributions are recorded as histograms, following the +conventional metric set for an inference server. All access happens on the +single asyncio event loop, so plain attributes are safe without locking. +""" + +import time + + +# Bucket boundaries are the ones inference-server dashboards conventionally +# expect, so they keep working after a prefix swap. Seconds-valued latency +# histograms share one coarse set; the +# time-to-first-token histogram gets a finer sub-second set since prefill is +# often fast. Per-request token counts use a 1-2-5 progression. +LATENCY_BUCKETS = [ + 0.3, + 0.5, + 0.8, + 1.0, + 1.5, + 2.0, + 2.5, + 5.0, + 10.0, + 15.0, + 20.0, + 30.0, + 40.0, + 50.0, + 60.0, + 120.0, + 240.0, + 480.0, + 960.0, + 1920.0, + 7680.0, +] +TTFT_BUCKETS = [ + 0.001, + 0.005, + 0.01, + 0.02, + 0.04, + 0.06, + 0.08, + 0.1, + 0.25, + 0.5, + 0.75, + 1.0, + 2.5, + 5.0, + 7.5, + 10.0, + 20.0, + 40.0, + 80.0, + 160.0, + 640.0, + 2560.0, +] +# Token-count buckets are built from the loaded model's context length rather +# than fixed, which is the conventional approach. A fixed ladder to 1M was the +# source of a +# real misreading: with a 256k context the top boundaries were 200k and 500k, so +# a server answering 200k-token prompts put every sample a few thousand above +# the floor of a 300k-wide bucket, and histogram_quantile -- which assumes +# samples are spread across the bucket they landed in -- reported a p99 of 493k +# against a largest-ever prompt of 206k. +# +# One boundary beyond the 1-2-5 ladder is added at max_seq_len itself. Stopping +# at the last mantissa value below the limit, as the usual construction does, +# leaves everything between there and the real limit in +Inf: at 262144 the +# ladder ends at 200000 and the top 24% of the usable range has no bucket. +DEFAULT_MAX_TOKENS = 1_000_000 + + +def build_1_2_5_buckets(max_value: int) -> list[int]: + """Increasing powers of 10 times 1, 2 and 5, up to and including max_value. + + >>> build_1_2_5_buckets(100) + [1, 2, 5, 10, 20, 50, 100] + """ + + buckets: list[int] = [] + exponent = 0 + while True: + for mantissa in (1, 2, 5): + value = mantissa * 10**exponent + if value > max_value: + if buckets and buckets[-1] < max_value: + buckets.append(max_value) + return buckets + buckets.append(value) + exponent += 1 + + +# Per-request draft acceptance is a ratio in [0, 1], so it gets its own evenly +# spaced buckets rather than the token or latency sets. +ACCEPTANCE_BUCKETS = [ + 0.05, + 0.1, + 0.2, + 0.3, + 0.4, + 0.5, + 0.6, + 0.7, + 0.8, + 0.9, + 0.95, + 1.0, +] + + +class _Histogram: + """A minimal cumulative-bucket histogram for the Prometheus text format.""" + + def __init__(self, buckets: list[float]): + # Upper bounds in ascending order. observe() tallies each value into the + # first bucket it fits; render sums them cumulatively, as Prometheus + # histogram buckets are "less than or equal" and cumulative. + self.buckets = buckets + self.counts = [0] * len(buckets) + self.sum = 0.0 + self.count = 0 + + def observe(self, value: float): + self.sum += value + self.count += 1 + for i, upper in enumerate(self.buckets): + if value <= upper: + self.counts[i] += 1 + return + # A value above the last bucket is reflected only in +Inf / _count. + + +class MetricsManagerClass: + """Tracks process-lifetime inference stats for the /metrics endpoint.""" + + def __init__(self): + self.process_start_time = time.time() + + # Counters (monotonic over the process lifetime) + # prompt_tokens_total counts only tokens actually processed, matching + # llama.cpp's counter of the same name. Tokens served from the prefix + # cache are tracked separately, so the full prompt length over the + # process lifetime is prompt_tokens_total + cached_tokens_total. + self.prompt_tokens_total = 0 + self.cached_tokens_total = 0 + self.gen_tokens_total = 0 + self.prompt_seconds_total = 0.0 + self.gen_seconds_total = 0.0 + self.requests_total = 0 + # Largest prompt and largest completion seen. The request_prompt_tokens + # and request_generation_tokens histograms describe the distribution but + # cannot report an extreme: the top sample sits above every percentile, + # and once it lands in the final bucket its magnitude is lost. These + # keep the peaks readable. Both only ever rise, so they are counters. + self.n_tokens_max = 0 + self.tokens_predicted_max = 0 + + # No prefill throughput gauge is derived from the two counters above, + # deliberately. The backend times prefill as one span per request, and + # that span cannot be split into time spent on cached versus newly + # processed tokens: a prefix cache hit takes tokens out of + # prompt_tokens_total while the lookup, page allocation and per-chunk + # overhead it still paid for stay in prompt_seconds_total. Their ratio + # therefore reads low and keeps sinking as reuse accumulates. Measured + # on a 27B model, a warm request contributing 12 tokens in 37ms drags a + # lifetime average towards ~320 T/s against a real rate of ~1646 T/s. + # + # Restricting the sample to requests that computed enough tokens for the + # ~30ms of fixed per-request overhead to vanish does fix the bias, but a + # server fronting a harness with a stable system prefix may never see + # more than one such request, and the first one carries the autotuning + # pass (measured 1.6% slow), so the estimator is pinned to its single + # worst sample. Prefill speed is a benchmark quantity; measure it with + # exllamav3's eval/perf.py or from the per-request log line. + # + # What is well defined here is the rate of work over wall clock, which + # rate(prompt_tokens_total[5m]) gives without any of this reasoning. + + # Speculative decoding counters, using the conventional spec-decode + # metric names. A "draft token" is one the drafter proposed; it is accepted + # when the target model samples the same token, otherwise it and every + # draft position after it are rejected, so accepted + rejected is the + # number of tokens drafted. Only requests served with drafting enabled + # contribute, tracked by draft_requests_total so the acceptance rate is + # not diluted by non-drafted requests. + self.draft_tokens_accepted_total = 0 + self.draft_tokens_rejected_total = 0 + self.draft_requests_total = 0 + # Decode steps over drafted requests only. exllamav3 does not count + # draft rounds, but every decode step emits exactly one token from the + # target model and the accepted drafts ride along on top of it, so + # (gen_tokens - accepted) recovers the step count exactly. + self.draft_decode_steps_total = 0 + + # Per-request distributions, as histograms. Latency is split + # into the queue / prefill / decode phases the backend already times, + # plus derived time-to-first-token (queue + prefill) and end-to-end + # totals; token counts cover the full prompt and the generation. + self.hist_queue_time = _Histogram(LATENCY_BUCKETS) + self.hist_prefill_time = _Histogram(LATENCY_BUCKETS) + self.hist_decode_time = _Histogram(LATENCY_BUCKETS) + self.hist_ttft = _Histogram(TTFT_BUCKETS) + self.hist_e2e = _Histogram(LATENCY_BUCKETS) + self.token_buckets = build_1_2_5_buckets(DEFAULT_MAX_TOKENS) + self.hist_prompt_tokens = _Histogram(self.token_buckets) + self.hist_gen_tokens = _Histogram(self.token_buckets) + self.hist_draft_acceptance = _Histogram(ACCEPTANCE_BUCKETS) + + def configure_token_buckets(self, max_seq_len: int): + """Size the token histograms for the loaded model's context length. + + Called on model load, when max_seq_len is finally known. Changing the + boundaries discards whatever those two histograms had accumulated, + since counts against a different ladder cannot be carried over, so this + is a no-op when the buckets come out unchanged. + """ + + if not max_seq_len or max_seq_len < 1: + return + buckets = build_1_2_5_buckets(max_seq_len) + if buckets == self.token_buckets: + return + self.token_buckets = buckets + self.hist_prompt_tokens = _Histogram(buckets) + self.hist_gen_tokens = _Histogram(buckets) + + def record_generation( + self, + prompt_tokens: int, + cached_tokens: float, + gen_tokens: int, + prompt_time: float, + gen_time: float, + queue_time: float = 0.0, + accepted_draft_tokens: int = None, + rejected_draft_tokens: int = None, + ): + """Accumulate stats from a single finished generation. + + `prompt_tokens` is the full prompt length and `cached_tokens` the part + of it served from the prefix cache; only the difference was processed. + `queue_time`, `prompt_time` and `gen_time` are the queue, prefill and + decode phase durations in seconds. + + `accepted_draft_tokens` / `rejected_draft_tokens` are the speculative + decoding tallies, and are None when the request ran without a drafter. + """ + + prompt_tokens = prompt_tokens or 0 + cached_tokens = cached_tokens or 0 + gen_tokens = gen_tokens or 0 + prompt_time = prompt_time or 0.0 + gen_time = gen_time or 0.0 + queue_time = queue_time or 0.0 + + self.prompt_tokens_total += prompt_tokens - cached_tokens + self.cached_tokens_total += cached_tokens + self.gen_tokens_total += gen_tokens + self.prompt_seconds_total += prompt_time + self.gen_seconds_total += gen_time + self.requests_total += 1 + self.n_tokens_max = max(self.n_tokens_max, prompt_tokens) + self.tokens_predicted_max = max(self.tokens_predicted_max, gen_tokens) + + # Time to first token is the wait in queue plus prefill; end-to-end adds + # the decode phase on top. + self.hist_queue_time.observe(queue_time) + self.hist_prefill_time.observe(prompt_time) + self.hist_decode_time.observe(gen_time) + self.hist_ttft.observe(queue_time + prompt_time) + self.hist_e2e.observe(queue_time + prompt_time + gen_time) + self.hist_prompt_tokens.observe(prompt_tokens) + self.hist_gen_tokens.observe(gen_tokens) + + # Drafting stats are absent when no drafter is configured; a request + # that ran with one but happened to draft nothing still counts, so the + # None check has to stay distinct from a zero tally. + if accepted_draft_tokens is not None: + accepted = accepted_draft_tokens or 0 + rejected = rejected_draft_tokens or 0 + drafted = accepted + rejected + + self.draft_tokens_accepted_total += accepted + self.draft_tokens_rejected_total += rejected + self.draft_requests_total += 1 + self.draft_decode_steps_total += max(gen_tokens - accepted, 0) + + if drafted > 0: + self.hist_draft_acceptance.observe(accepted / drafted) + + def _live_request_counts(self) -> tuple[int, int]: + """Read (processing, deferred) request counts from the generator. + + Returns zeros if no model is loaded or the backend does not expose + job counts. + """ + + # Imported lazily to avoid a circular import (common.model pulls in the + # backends, which import this module). + from common import model + + container = model.container + generator = getattr(container, "generator", None) if container else None + sync_generator = getattr(generator, "generator", None) if generator else None + + if sync_generator is None: + return 0, 0 + + try: + return sync_generator.num_active_jobs(), sync_generator.num_pending_jobs() + except Exception: + return 0, 0 + + def _live_kv_cache(self) -> tuple[int, int]: + """Read (used_tokens, max_tokens) of the paged KV cache from the generator. + + Usage is measured over pages currently referenced by in-flight jobs, the + instantaneous KV load. Unreferenced pages may still hold reusable prompt + prefixes but are free to be evicted, so they count as headroom rather than + usage (matching llama.cpp's kv_cache_usage_ratio). Returns zeros if no + model is loaded or the backend does not expose a page table. + """ + + # Imported lazily to avoid a circular import (common.model pulls in the + # backends, which import this module). + from common import model + + container = model.container + generator = getattr(container, "generator", None) if container else None + sync_generator = getattr(generator, "generator", None) if generator else None + pagetable = getattr(sync_generator, "pagetable", None) if sync_generator else None + + if pagetable is None: + return 0, 0 + + try: + max_pages = pagetable.max_pages + max_tokens = sync_generator.max_total_tokens + page_size = max_tokens // max_pages if max_pages else 0 + used_tokens = len(pagetable.referenced_pages) * page_size + return used_tokens, max_tokens + except Exception: + return 0, 0 + + def _live_kv_offload(self) -> dict: + """Read the state of the CPU page cache (pages evicted to system RAM). + + Returns a zero-filled dict when offloading is disabled or no model is + loaded, so the series exist unconditionally rather than appearing and + disappearing across scrapes. + + The counters here live in the generator's page cache, not in this + object, so they restart from zero whenever the generator is recreated + (model reload, or recovery from a backend error). Prometheus detects + counter resets, so rate() over them stays correct across a reload; + absolute values are only meaningful within one generator's lifetime. + """ + + from common import model + + empty = { + "tokens": 0, + "max_tokens": 0, + "usage_ratio": 0.0, + "bytes": 0, + "max_bytes": 0, + "restored_tokens": 0, + "stores": 0, + "deduped_stores": 0, + "evictions": 0, + "cold_allocs": 0, + "bytes_read": 0, + "bytes_written": 0, + } + + container = model.container + generator = getattr(container, "generator", None) if container else None + sync_generator = getattr(generator, "generator", None) if generator else None + cpu_cache = getattr(sync_generator, "cpu_page_cache", None) if sync_generator else None + + if cpu_cache is None: + return empty + + try: + counters = cpu_cache.metrics + pages = len(cpu_cache) + max_pages = cpu_cache.max_slots + # One slot holds the whole per-layer page image across the attached + # caches, so it is also the size of a single transfer in either + # direction. + slot_size = cpu_cache.slot_size + + # Page-granular counts are converted to tokens so they read on the + # same axis as kv_cache_tokens. The number of tokens per page is + # fixed by the generator's page size, which the page cache does not + # carry, so it is taken from the page table. + pagetable = getattr(sync_generator, "pagetable", None) + pt_max_pages = getattr(pagetable, "max_pages", 0) if pagetable else 0 + tokens_per_page = sync_generator.max_total_tokens // pt_max_pages if pt_max_pages else 0 + + return { + "tokens": pages * tokens_per_page, + "max_tokens": max_pages * tokens_per_page, + "usage_ratio": pages / max_pages if max_pages else 0.0, + "bytes": pages * slot_size, + "max_bytes": max_pages * slot_size, + # A restore reads back a whole page, which is one page of prompt + # that did not have to be prefilled again. These tokens are + # already counted in cached_tokens_total, which does not + # distinguish a page found in VRAM from one read back over + # PCIe; this series is that breakdown, and the ratio against + # cached_tokens_total says how much of the prefix cache is + # actually being served out of RAM. + "restored_tokens": counters["restores"] * tokens_per_page, + "stores": counters["pushes"], + "deduped_stores": counters["dedup_hits"], + "evictions": counters["evictions"], + "cold_allocs": counters["cold_allocs"], + # Every transfer moves exactly one slot, in whole, so the byte + # totals are exact rather than an estimate. + "bytes_read": counters["restores"] * slot_size, + "bytes_written": counters["pushes"] * slot_size, + } + except Exception: + return empty + + def _live_recurrent(self) -> dict: + """Read the state of the recurrent checkpoint cache on hybrid models. + + Hybrid architectures interleave full-attention layers, whose state is + the paged K/V cache, with linear-attention layers, whose state is a + single evolving tensor that cannot be indexed by position. The generator + checkpoints the latter to system RAM at page boundaries, and prompt + reuse is capped at the longest prefix that has *both* valid K/V pages + and a matching recurrent checkpoint. + + That cap is why these series matter: if the checkpoint for a prefix is + evicted, its K/V pages are unusable however well the KV cache held them, + and with offloading enabled they will have been read back over PCIe + first. recurrent_capped_tokens_total counts exactly that waste. + + Returns zeros on non-hybrid models, where there is no recurrent state. + """ + + from common import model + + empty = { + "checkpoints": 0, + "bytes": 0, + "max_bytes": 0, + "usage_ratio": 0.0, + "checkpoint_bytes": 0, + "evictions": 0, + "stranded_evictions": 0, + "live_kv_evictions": 0, + "pruned": 0, + "stranded_by_kv": 0, + "capped_tokens": 0, + } + + container = model.container + generator = getattr(container, "generator", None) if container else None + sync_generator = getattr(generator, "generator", None) if generator else None + if sync_generator is None: + return empty + + recurrent_cache = getattr(sync_generator, "recurrent_cache", None) + pagetable = getattr(sync_generator, "pagetable", None) + + out = dict(empty) + + # Read in two stages, so a cache object in an unexpected state does not + # take the page table's figures down with it. The cap is the series + # these exist for, and losing it would hide the failure it reports. + try: + if pagetable is not None: + pt_counters = pagetable.metrics + max_pages = pagetable.max_pages + tokens_per_page = sync_generator.max_total_tokens // max_pages if max_pages else 0 + out.update( + { + "capped_tokens": (pt_counters["alloc_kv_only_pages"] * tokens_per_page), + "stranded_by_kv": pt_counters["stashes_stranded"], + } + ) + except Exception: + pass + + if recurrent_cache is None: + return out + + try: + counters = recurrent_cache.metrics + checkpoints = len(recurrent_cache) + current_size = recurrent_cache.current_size + max_size = recurrent_cache.max_size + out.update( + { + "checkpoints": checkpoints, + "bytes": current_size, + "max_bytes": max_size, + "usage_ratio": current_size / max_size if max_size else 0.0, + # Published because it is the unit the budget is spent in: a + # checkpoint is indivisible, so max_bytes / checkpoint_bytes + # is how many prefixes can be resumed at all. + "checkpoint_bytes": (current_size // checkpoints if checkpoints else 0), + "evictions": counters["stash_evictions"], + "stranded_evictions": counters["stash_evictions_stranded"], + "live_kv_evictions": counters["stash_evictions_live_kv"], + "pruned": counters["stash_pruned"], + } + ) + except Exception: + # The dict above is built whole before it is applied, so a failed + # read leaves the page table's figures in place rather than a + # half-updated mix of the two. + pass + + return out + + def render_prometheus(self) -> str: + """Render all metrics in the Prometheus text exposition format.""" + + requests_processing, requests_deferred = self._live_request_counts() + kv_cache_tokens, kv_cache_max_tokens = self._live_kv_cache() + offload = self._live_kv_offload() + recurrent = self._live_recurrent() + kv_cache_usage_ratio = ( + kv_cache_tokens / kv_cache_max_tokens if kv_cache_max_tokens > 0 else 0.0 + ) + + # Prefix-cache effectiveness is exposed as the raw queries/hits token + # counters, leaving the hit ratio to be computed at query + # time with rate() so it reflects recent behavior rather than a + # lifetime average. + prefix_cache_queries = self.prompt_tokens_total + self.cached_tokens_total + prefix_cache_hits = self.cached_tokens_total + + # There is no prefill counterpart to this gauge on purpose; see the + # counter definitions. Decode time has no cached-token analogue to skew + # it, so generation tokens over decode seconds is what it claims to be. + predicted_tokens_seconds = ( + self.gen_tokens_total / self.gen_seconds_total if self.gen_seconds_total > 0 else 0.0 + ) + + # Speculative decoding effectiveness. The two raw counters are the + # primitives to rate() over; these gauges are the lifetime + # summary, cheap to read without a query language. Acceptance rate is + # per drafted token, mean accepted length is per decode step (how many + # drafts a step gets for free), and tokens per step adds the target + # model's own token, so it is the decode speedup factor over no drafter. + draft_tokens_total = self.draft_tokens_accepted_total + self.draft_tokens_rejected_total + draft_acceptance_rate = ( + self.draft_tokens_accepted_total / draft_tokens_total if draft_tokens_total > 0 else 0.0 + ) + draft_mean_accepted_len = ( + self.draft_tokens_accepted_total / self.draft_decode_steps_total + if self.draft_decode_steps_total > 0 + else 0.0 + ) + draft_tokens_per_step = ( + 1.0 + draft_mean_accepted_len if self.draft_decode_steps_total else 0.0 + ) + + # (type, name, help, value) + # Names and help text of the shared metrics are kept verbatim from + # llama.cpp's exporter so its dashboards work after a prefix swap. + # tabbyAPI-only metrics follow the ones they relate to. The one + # deliberate omission from that set is prompt_tokens_seconds; see the + # counter definitions for why it is not a quantity worth publishing. + metrics = [ + # Published so the phase-time counters below can be read against + # wall clock. Without it prompt_seconds_total and + # tokens_predicted_seconds_total give the split between prefill and + # decode but not what either costs against a real interval, and the + # standard name means the usual process dashboards pick it up. + ( + "gauge", + "process_start_time_seconds", + "Start time of the process since the Unix epoch, in seconds.", + self.process_start_time, + ), + ( + "counter", + "prompt_tokens_total", + "Number of prompt tokens processed.", + self.prompt_tokens_total, + ), + ( + "counter", + "cached_tokens_total", + "Number of prompt tokens skipped via the prefix cache.", + self.cached_tokens_total, + ), + ( + "counter", + "prefix_cache_queries", + "Prefix cache queries, in terms of number of queried tokens.", + prefix_cache_queries, + ), + ( + "counter", + "prefix_cache_hits", + "Prefix cache hits, in terms of number of cached tokens.", + prefix_cache_hits, + ), + ( + "counter", + "prompt_seconds_total", + "Prompt process time", + self.prompt_seconds_total, + ), + ( + "counter", + "tokens_predicted_total", + "Number of generation tokens processed.", + self.gen_tokens_total, + ), + ( + "counter", + "tokens_predicted_seconds_total", + "Predict process time", + self.gen_seconds_total, + ), + ( + "counter", + "n_tokens_max", + "Largest observed n_tokens.", + self.n_tokens_max, + ), + ( + "counter", + "tokens_predicted_max", + "Largest observed number of generation tokens in one request.", + self.tokens_predicted_max, + ), + ( + "counter", + "requests_total", + "Number of finished generation requests.", + self.requests_total, + ), + ( + "gauge", + "predicted_tokens_seconds", + "Average generation throughput in tokens/s.", + predicted_tokens_seconds, + ), + ( + "counter", + "spec_decode_num_draft_tokens_total", + "Number of tokens proposed by the drafter.", + draft_tokens_total, + ), + ( + "counter", + "spec_decode_num_accepted_tokens_total", + "Number of drafted tokens accepted by the target model.", + self.draft_tokens_accepted_total, + ), + ( + "counter", + "spec_decode_num_decode_steps_total", + "Number of decode steps over requests served with a drafter.", + self.draft_decode_steps_total, + ), + ( + "counter", + "spec_decode_requests_total", + "Number of finished requests served with a drafter.", + self.draft_requests_total, + ), + ( + "gauge", + "spec_decode_draft_acceptance_rate", + "Fraction of drafted tokens accepted. 1 means every draft was accepted.", + draft_acceptance_rate, + ), + ( + "gauge", + "spec_decode_mean_accepted_length", + "Average drafted tokens accepted per decode step.", + draft_mean_accepted_len, + ), + ( + "gauge", + "spec_decode_tokens_per_step", + "Average tokens emitted per decode step, including the target model's own.", + draft_tokens_per_step, + ), + ( + "gauge", + "requests_processing", + "Number of requests processing.", + requests_processing, + ), + ( + "gauge", + "requests_deferred", + "Number of requests deferred.", + requests_deferred, + ), + ( + "gauge", + "kv_cache_usage_ratio", + "KV-cache usage. 1 means 100 percent usage.", + kv_cache_usage_ratio, + ), + ( + "gauge", + "kv_cache_tokens", + "KV-cache tokens.", + kv_cache_tokens, + ), + ( + "gauge", + "kv_cache_max_tokens", + "Total KV-cache token capacity.", + kv_cache_max_tokens, + ), + # KV offload cache. Deliberately no bytes-per-second gauge: a + # lifetime average of transfer rate is as misleading here as it is + # for prefill, since it divides by wall clock that includes every + # scrape interval with no transfers at all. The bytes counters below + # give the real thing under rate(), and PCIe bandwidth is a constant + # of the machine rather than something to watch drift. + ( + "gauge", + "kv_offload_usage_ratio", + "KV offload cache usage. 1 means 100 percent usage.", + offload["usage_ratio"], + ), + ( + "gauge", + "kv_offload_tokens", + "Tokens of KV cache currently held in system RAM.", + offload["tokens"], + ), + ( + "gauge", + "kv_offload_max_tokens", + "Total KV offload cache token capacity.", + offload["max_tokens"], + ), + ( + "gauge", + "kv_offload_bytes", + "System RAM currently holding cache pages, in bytes.", + offload["bytes"], + ), + # The backend pins the whole configured capacity up front, in the + # background, so this is the RAM cost of the feature whatever the + # usage gauge above reads. + ( + "gauge", + "kv_offload_max_bytes", + "Configured size of the KV offload cache, in bytes.", + offload["max_bytes"], + ), + # Taken against cached_tokens_total this is the share of prefix + # cache hits that came back over PCIe rather than being found in + # VRAM, in the same raw-counter form as prefix_cache_queries/hits so + # rate() gives recent behavior rather than a lifetime average. + ( + "counter", + "kv_offload_restored_tokens_total", + "Prompt tokens read back from system RAM instead of being prefilled again.", + offload["restored_tokens"], + ), + # Evictions climbing towards stores means the cache is too small for + # the working set and pages are being written out only to be + # discarded before anyone reads them back. + ( + "counter", + "kv_offload_stores_total", + "Pages copied from VRAM into the KV offload cache.", + offload["stores"], + ), + ( + "counter", + "kv_offload_deduped_stores_total", + "Page stores skipped because the page was already held in system RAM.", + offload["deduped_stores"], + ), + ( + "counter", + "kv_offload_evictions_total", + "Pages dropped from the KV offload cache to make room.", + offload["evictions"], + ), + # The backend pins slabs ahead of demand on a background thread. A + # store that outruns it has to pin synchronously, at roughly + # 2.5 GB/s, on the generator's own thread. Nonzero early in a + # process is expected; nonzero later is a stall worth seeing. + ( + "counter", + "kv_offload_cold_allocs_total", + "Page stores that had to pin system memory synchronously.", + offload["cold_allocs"], + ), + ( + "counter", + "kv_offload_read_bytes_total", + "Bytes transferred from system RAM to VRAM restoring cache pages.", + offload["bytes_read"], + ), + ( + "counter", + "kv_offload_written_bytes_total", + "Bytes transferred from VRAM to system RAM evicting cache pages.", + offload["bytes_written"], + ), + # Recurrent checkpoint cache (hybrid models only). Zero everywhere on + # a pure transformer, which has no recurrent layers to checkpoint. + ( + "gauge", + "recurrent_cache_usage_ratio", + "Recurrent checkpoint cache usage. 1 means 100 percent usage.", + recurrent["usage_ratio"], + ), + ( + "gauge", + "recurrent_checkpoints", + "Recurrent state checkpoints currently held in system RAM.", + recurrent["checkpoints"], + ), + ( + "gauge", + "recurrent_cache_bytes", + "System RAM currently holding recurrent checkpoints, in bytes.", + recurrent["bytes"], + ), + ( + "gauge", + "recurrent_cache_max_bytes", + "Configured size of the recurrent checkpoint cache, in bytes.", + recurrent["max_bytes"], + ), + ( + "gauge", + "recurrent_checkpoint_bytes", + "Mean size of one recurrent checkpoint, in bytes.", + recurrent["checkpoint_bytes"], + ), + ( + "counter", + "recurrent_cache_evictions_total", + "Recurrent checkpoints dropped to make room.", + recurrent["evictions"], + ), + # The eviction breakdown is what says whether the budget is actually + # too small. A stranded checkpoint had already lost the K/V pages it + # anchors and could never have been resumed, so dropping it costs + # nothing; the same for one pruned while idle. A checkpoint dropped + # while its anchor page was still cached is the one that hurts, and + # is the direct precursor of recurrent_capped_tokens_total below. + ( + "counter", + "recurrent_cache_stranded_evictions_total", + "Recurrent checkpoints dropped that were already unresumable.", + recurrent["stranded_evictions"], + ), + ( + "counter", + "recurrent_cache_live_kv_evictions_total", + "Recurrent checkpoints dropped while their anchor KV page was still cached.", + recurrent["live_kv_evictions"], + ), + ( + "counter", + "recurrent_cache_pruned_total", + "Unresumable recurrent checkpoints reclaimed while the generator was idle.", + recurrent["pruned"], + ), + # The mirror of live_kv_evictions, and the reason the two budgets + # have to be sized against each other rather than independently: + # here the KV cache is the one under pressure, and evicting a page + # stranded the checkpoint anchored to it. + ( + "counter", + "recurrent_stranded_by_kv_total", + "Recurrent checkpoints stranded by eviction of the KV page anchoring them.", + recurrent["stranded_by_kv"], + ), + # The cost of an undersized recurrent cache, and the one series that + # ties the two caches together: these tokens had valid K/V in the + # cache and were still re-prefilled, because the recurrent state that + # goes with them was gone. With offloading on, they were also read + # back over PCIe before being discarded. + ( + "counter", + "recurrent_capped_tokens_total", + "Tokens with valid KV that were re-prefilled anyway, for lack of recurrent state.", + recurrent["capped_tokens"], + ), + ] + + # (name, help, histogram) + histograms = [ + ( + "request_queue_time_seconds", + "Histogram of time spent in the queue before prefill, in seconds.", + self.hist_queue_time, + ), + ( + "request_prefill_time_seconds", + "Histogram of prefill (prompt processing) time in seconds.", + self.hist_prefill_time, + ), + ( + "request_decode_time_seconds", + "Histogram of decode (generation) time in seconds.", + self.hist_decode_time, + ), + ( + "time_to_first_token_seconds", + "Histogram of time to first token in seconds.", + self.hist_ttft, + ), + ( + "e2e_request_latency_seconds", + "Histogram of end to end request latency in seconds.", + self.hist_e2e, + ), + ( + "request_prompt_tokens", + "Histogram of number of prompt tokens per request.", + self.hist_prompt_tokens, + ), + ( + "request_generation_tokens", + "Histogram of number of generation tokens per request.", + self.hist_gen_tokens, + ), + ( + "spec_decode_acceptance_rate", + "Histogram of per-request draft acceptance rate.", + self.hist_draft_acceptance, + ), + ] + + lines = [] + for metric_type, name, help_text, value in metrics: + full_name = f"tabbyapi:{name}" + lines.append(f"# HELP {full_name} {help_text}") + lines.append(f"# TYPE {full_name} {metric_type}") + lines.append(f"{full_name} {value}") + + for name, help_text, hist in histograms: + full_name = f"tabbyapi:{name}" + lines.append(f"# HELP {full_name} {help_text}") + lines.append(f"# TYPE {full_name} histogram") + cumulative = 0 + for upper, count in zip(hist.buckets, hist.counts, strict=True): + cumulative += count + lines.append(f'{full_name}_bucket{{le="{upper}"}} {cumulative}') + lines.append(f'{full_name}_bucket{{le="+Inf"}} {hist.count}') + lines.append(f"{full_name}_sum {hist.sum}") + lines.append(f"{full_name}_count {hist.count}") + + return "\n".join(lines) + "\n" + + +# Create an instance of the global metrics manager +MetricsManager = MetricsManagerClass() diff --git a/config_sample.yml b/config_sample.yml index 7b16b06b..8dea751e 100644 --- a/config_sample.yml +++ b/config_sample.yml @@ -36,6 +36,11 @@ network: # connections from dropping during long prefills. Set to 0 to disable. sse_ping_interval: 15 + # Enable the Prometheus-compatible /metrics endpoint (default: False). + # Exposes aggregate inference stats in the text exposition format. + # NOTE: This endpoint is not protected by API key authentication. + enable_metrics: false + # Options for logging logging: # Enable prompt logging (default: False). diff --git a/endpoints/core/router.py b/endpoints/core/router.py index ffbab464..6bf82c3c 100644 --- a/endpoints/core/router.py +++ b/endpoints/core/router.py @@ -15,6 +15,7 @@ handle_request_error, run_with_request_disconnect, ) +from common.metrics import MetricsManager from common.tabby_config import config from common.templating import PromptTemplate, get_all_templates from common.utils import unwrap @@ -69,6 +70,25 @@ async def healthcheck(response: Response) -> HealthCheckResponse: return HealthCheckResponse(status="healthy" if healthy else "unhealthy", issues=issues) +# Prometheus-compatible metrics endpoint (no auth, opt-in via config) +@router.get("/metrics") +async def metrics(): + """Exposes aggregate inference stats in the Prometheus text format.""" + + if not config.network.enable_metrics: + raise HTTPException( + 404, + "The metrics endpoint is disabled. " + "Set network.enable_metrics to true in config.yml to enable it.", + ) + + return Response( + content=MetricsManager.render_prometheus(), + media_type="text/plain; version=0.0.4", + headers={"Process-Start-Time-Unix": str(int(MetricsManager.process_start_time))}, + ) + + @router.get("/.well-known/serviceinfo") async def service_info(): return JSONResponse( diff --git a/tests/test_metrics_kv_offload.py b/tests/test_metrics_kv_offload.py new file mode 100644 index 00000000..c6abccb5 --- /dev/null +++ b/tests/test_metrics_kv_offload.py @@ -0,0 +1,162 @@ +import types +import unittest + +from common import model as model_module +from common.metrics import MetricsManagerClass + + +PAGE_SIZE = 256 +SLOT_BYTES = 16 * 1024 * 1024 + + +class FakeCPUPageCache: + """Stand-in for exllamav3's CPUPageCache, with the surface the reader uses.""" + + def __init__(self, pages=100, max_slots=400, **overrides): + self.pages = pages + self.max_slots = max_slots + self.slot_size = SLOT_BYTES + self.metrics = { + "pushes": 500, + "dedup_hits": 40, + "restores": 390, + "evictions": 100, + "cold_allocs": 3, + } + self.metrics.update(overrides) + + def __len__(self): + return self.pages + + +def install_container(cpu_cache, max_pages=1000, max_total_tokens=1000 * PAGE_SIZE): + """Point common.model.container at a stand-in generator exposing a CPU page cache.""" + + pagetable = types.SimpleNamespace(max_pages=max_pages, referenced_pages={}) + sync_generator = types.SimpleNamespace( + pagetable=pagetable, + max_total_tokens=max_total_tokens, + cpu_page_cache=cpu_cache, + ) + model_module.container = types.SimpleNamespace( + generator=types.SimpleNamespace(generator=sync_generator) + ) + + +class KVOffloadMetricsTests(unittest.TestCase): + """The /metrics view of the CPU page cache. + + The counters behind these series live in the generator's page cache rather + than in the metrics manager, so the reader has to tolerate a generator that + is missing, has offloading disabled, or reports something unexpected, + without dropping the series or raising through a scrape. + """ + + def setUp(self): + self.metrics = MetricsManagerClass() + self.original_container = model_module.container + + def tearDown(self): + model_module.container = self.original_container + + def test_page_counts_are_reported_as_tokens(self): + install_container(FakeCPUPageCache()) + + live = self.metrics._live_kv_offload() + + self.assertEqual(live["tokens"], 100 * PAGE_SIZE) + self.assertEqual(live["max_tokens"], 400 * PAGE_SIZE) + self.assertAlmostEqual(live["usage_ratio"], 0.25) + + def test_restored_tokens_measure_prefill_avoided(self): + # Every restore reads back one whole page, which is one page of prompt + # that did not have to be prefilled again. + install_container(FakeCPUPageCache()) + + live = self.metrics._live_kv_offload() + + self.assertEqual(live["restored_tokens"], 390 * PAGE_SIZE) + + def test_byte_counters_are_derived_from_whole_slot_transfers(self): + # A transfer in either direction always moves exactly one slot, so the + # byte totals follow from the transfer counts rather than being an + # estimate. + install_container(FakeCPUPageCache()) + + live = self.metrics._live_kv_offload() + + self.assertEqual(live["bytes"], 100 * SLOT_BYTES) + self.assertEqual(live["max_bytes"], 400 * SLOT_BYTES) + self.assertEqual(live["bytes_read"], 390 * SLOT_BYTES) + self.assertEqual(live["bytes_written"], 500 * SLOT_BYTES) + + def test_synchronous_pinning_is_visible(self): + # A store that outruns the background pinning thread pins on the + # generator's own thread at roughly 2.5 GB/s. That has to be observable. + install_container(FakeCPUPageCache(cold_allocs=17)) + + self.assertEqual(self.metrics._live_kv_offload()["cold_allocs"], 17) + + def test_series_are_published_when_offloading_is_disabled(self): + # Series that appear and disappear across scrapes are awkward to alert + # on, so a disabled cache reports zeros rather than nothing. + install_container(None) + + rendered = self.metrics.render_prometheus() + + for name in ( + "tabbyapi:kv_offload_usage_ratio", + "tabbyapi:kv_offload_tokens", + "tabbyapi:kv_offload_max_tokens", + "tabbyapi:kv_offload_bytes", + "tabbyapi:kv_offload_max_bytes", + "tabbyapi:kv_offload_restored_tokens_total", + "tabbyapi:kv_offload_stores_total", + "tabbyapi:kv_offload_deduped_stores_total", + "tabbyapi:kv_offload_evictions_total", + "tabbyapi:kv_offload_cold_allocs_total", + "tabbyapi:kv_offload_read_bytes_total", + "tabbyapi:kv_offload_written_bytes_total", + ): + self.assertIn(f"{name} 0", rendered) + + def test_no_model_loaded_does_not_raise(self): + model_module.container = None + + self.assertEqual(self.metrics._live_kv_offload()["tokens"], 0) + + def test_a_broken_cpu_cache_does_not_break_the_scrape(self): + install_container(types.SimpleNamespace(metrics={})) + + self.assertEqual(self.metrics._live_kv_offload()["tokens"], 0) + self.assertIn("tabbyapi:kv_offload_tokens 0", self.metrics.render_prometheus()) + + def test_no_throughput_gauge_is_published(self): + # A lifetime average of transfer rate divides by wall clock that + # includes every interval with no transfers at all. The bytes counters + # are exposed so a windowed rate can be taken at query time instead. + install_container(FakeCPUPageCache()) + + rendered = self.metrics.render_prometheus() + + for name in ( + "tabbyapi:kv_offload_bytes_seconds", + "tabbyapi:kv_offload_read_bytes_seconds", + "tabbyapi:kv_offload_written_bytes_seconds", + "tabbyapi:kv_offload_hit_ratio", + ): + self.assertNotIn(name, rendered) + + def test_rendered_values_match_the_live_read(self): + install_container(FakeCPUPageCache()) + + rendered = self.metrics.render_prometheus() + + self.assertIn(f"tabbyapi:kv_offload_tokens {100 * PAGE_SIZE}", rendered) + self.assertIn("tabbyapi:kv_offload_usage_ratio 0.25", rendered) + self.assertIn(f"tabbyapi:kv_offload_restored_tokens_total {390 * PAGE_SIZE}", rendered) + self.assertIn("tabbyapi:kv_offload_evictions_total 100", rendered) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_metrics_peak_sizes.py b/tests/test_metrics_peak_sizes.py new file mode 100644 index 00000000..8e1baa80 --- /dev/null +++ b/tests/test_metrics_peak_sizes.py @@ -0,0 +1,67 @@ +import unittest + +from common.metrics import MetricsManagerClass + + +class PeakSizeCounterTests(unittest.TestCase): + def setUp(self): + self.metrics = MetricsManagerClass() + + def record(self, prompt_tokens: int, gen_tokens: int): + self.metrics.record_generation( + prompt_tokens=prompt_tokens, + cached_tokens=0, + gen_tokens=gen_tokens, + prompt_time=1.0, + gen_time=1.0, + ) + + def test_peaks_track_the_largest_request(self): + self.record(1000, 20) + self.record(500, 8000) + self.record(9000, 15) + + self.assertEqual(self.metrics.n_tokens_max, 9000) + self.assertEqual(self.metrics.tokens_predicted_max, 8000) + + rendered = self.metrics.render_prometheus() + self.assertIn("tabbyapi:n_tokens_max 9000", rendered) + self.assertIn("tabbyapi:tokens_predicted_max 8000", rendered) + + def test_peaks_never_fall(self): + self.record(9000, 8000) + for _ in range(50): + self.record(20, 15) + + self.assertEqual(self.metrics.n_tokens_max, 9000) + self.assertEqual(self.metrics.tokens_predicted_max, 8000) + + def test_peak_survives_where_percentiles_cannot(self): + # The reason this counter exists: one large completion among many small + # ones sits above every percentile the histogram can report, so the + # distribution alone cannot show it. + self.record(100, 8000) + for _ in range(344): + self.record(100, 15) + + gen_hist = self.metrics.hist_gen_tokens + self.assertEqual(gen_hist.count, 345) + # The outlier is one sample in 345, i.e. above the 99th percentile. + self.assertLess(gen_hist.count * 0.99, 344) + self.assertEqual(self.metrics.tokens_predicted_max, 8000) + + def test_missing_counts_do_not_break_the_peaks(self): + self.metrics.record_generation( + prompt_tokens=None, + cached_tokens=None, + gen_tokens=None, + prompt_time=None, + gen_time=None, + ) + + self.assertEqual(self.metrics.n_tokens_max, 0) + self.assertEqual(self.metrics.tokens_predicted_max, 0) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_metrics_prefill_series.py b/tests/test_metrics_prefill_series.py new file mode 100644 index 00000000..2a244f09 --- /dev/null +++ b/tests/test_metrics_prefill_series.py @@ -0,0 +1,72 @@ +import unittest + +from common.metrics import MetricsManagerClass + + +class NoPrefillThroughputGaugeTests(unittest.TestCase): + """/metrics must not publish a prefill throughput gauge. + + Prefill is timed as one span per request and that span cannot be split into + time spent on cached versus newly processed tokens, so any lifetime average + of tokens over prefill seconds decays as prefix cache reuse accumulates. + Restricting the sample to requests that computed enough tokens to swamp the + fixed overhead fixes the bias but leaves too few samples to be worth + publishing on a workload with a stable system prefix. The counters are + exposed instead, so a windowed rate can be taken at query time. + """ + + def setUp(self): + self.metrics = MetricsManagerClass() + + def test_no_prefill_throughput_series_is_published(self): + self.metrics.record_generation( + prompt_tokens=4000, + cached_tokens=0, + gen_tokens=10, + prompt_time=2.0, + gen_time=1.0, + ) + + rendered = self.metrics.render_prometheus() + + for name in ( + "tabbyapi:prompt_tokens_seconds", + "tabbyapi:prompt_compute_tokens_seconds", + "tabbyapi:prompt_cold_tokens_total", + "tabbyapi:prompt_cold_seconds_total", + ): + self.assertNotIn(name, rendered) + + def test_the_counters_a_windowed_rate_needs_are_published(self): + self.metrics.record_generation( + prompt_tokens=10000, + cached_tokens=9900, + gen_tokens=10, + prompt_time=0.5, + gen_time=1.0, + ) + + rendered = self.metrics.render_prometheus() + + # Tokens actually processed, the time prefill took, and the cache hits + # that explain the difference. + self.assertIn("tabbyapi:prompt_tokens_total 100", rendered) + self.assertIn("tabbyapi:prompt_seconds_total 0.5", rendered) + self.assertIn("tabbyapi:cached_tokens_total 9900", rendered) + + def test_decode_throughput_gauge_is_kept(self): + # Decode time has no cached-token analogue, so this one measures what + # it claims to and stays. + self.metrics.record_generation( + prompt_tokens=4000, + cached_tokens=0, + gen_tokens=100, + prompt_time=2.0, + gen_time=4.0, + ) + + self.assertIn("tabbyapi:predicted_tokens_seconds 25.0", self.metrics.render_prometheus()) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_metrics_recurrent_cache.py b/tests/test_metrics_recurrent_cache.py new file mode 100644 index 00000000..41452f09 --- /dev/null +++ b/tests/test_metrics_recurrent_cache.py @@ -0,0 +1,174 @@ +import types +import unittest + +from common import model as model_module +from common.metrics import MetricsManagerClass + + +PAGE_SIZE = 256 +CHECKPOINT_BYTES = 155 * 1024**2 + + +class FakeRecurrentCache: + """Stand-in for exllamav3's RecurrentCache, with the surface the reader uses.""" + + def __init__(self, checkpoints=60, max_size=16 * 1024**3, **overrides): + self.checkpoints = checkpoints + self.current_size = checkpoints * CHECKPOINT_BYTES + self.max_size = max_size + self.metrics = { + "stash_evictions": 40, + "stash_evictions_stranded": 25, + "stash_evictions_live_kv": 12, + "stash_pruned": 8, + } + self.metrics.update(overrides) + + def __len__(self): + return self.checkpoints + + +def install_hybrid(recurrent_cache=None, capped_pages=0, stranded_by_kv=0): + pagetable = types.SimpleNamespace( + max_pages=1000, + referenced_pages={}, + metrics={ + "alloc_kv_only_pages": capped_pages, + "stashes_stranded": stranded_by_kv, + }, + ) + sync_generator = types.SimpleNamespace( + pagetable=pagetable, + max_total_tokens=1000 * PAGE_SIZE, + cpu_page_cache=None, + recurrent_cache=recurrent_cache, + ) + model_module.container = types.SimpleNamespace( + generator=types.SimpleNamespace(generator=sync_generator) + ) + + +class RecurrentCacheMetricsTests(unittest.TestCase): + """The recurrent checkpoint cache, and its coupling to the KV cache. + + On a hybrid model, prompt reuse is capped at the longest prefix that has + both valid K/V pages and a matching recurrent checkpoint. An undersized + recurrent cache therefore silently defeats the KV cache, and with offloading + enabled it wastes PCIe bandwidth doing so. + """ + + def setUp(self): + self.metrics = MetricsManagerClass() + self.original_container = model_module.container + + def tearDown(self): + model_module.container = self.original_container + + def test_checkpoint_accounting(self): + install_hybrid(FakeRecurrentCache()) + + live = self.metrics._live_recurrent() + + self.assertEqual(live["checkpoints"], 60) + self.assertEqual(live["checkpoint_bytes"], CHECKPOINT_BYTES) + self.assertEqual(live["max_bytes"], 16 * 1024**3) + self.assertAlmostEqual(live["usage_ratio"], (60 * CHECKPOINT_BYTES) / (16 * 1024**3)) + + def test_an_empty_cache_reports_no_checkpoint_size(self): + # checkpoint_bytes is a mean, so it has no value before the first store. + install_hybrid(FakeRecurrentCache(checkpoints=0)) + + live = self.metrics._live_recurrent() + + self.assertEqual(live["checkpoints"], 0) + self.assertEqual(live["checkpoint_bytes"], 0) + self.assertEqual(live["usage_ratio"], 0.0) + + def test_capped_pages_are_reported_as_tokens(self): + # The waste metric: valid KV that was re-prefilled for lack of recurrent state. + install_hybrid(FakeRecurrentCache(), capped_pages=120) + + self.assertEqual(self.metrics._live_recurrent()["capped_tokens"], 120 * PAGE_SIZE) + + def test_eviction_breakdown_separates_harmless_from_costly(self): + # A stranded checkpoint could never have been resumed, so dropping it is + # free. One dropped while its anchor page was still cached is the drop + # that turns into capped tokens later. + install_hybrid(FakeRecurrentCache(), stranded_by_kv=17) + + live = self.metrics._live_recurrent() + + self.assertEqual(live["evictions"], 40) + self.assertEqual(live["stranded_evictions"], 25) + self.assertEqual(live["live_kv_evictions"], 12) + self.assertEqual(live["pruned"], 8) + # The mirror case, counted by the page table: KV eviction stranding a + # checkpoint rather than the other way round. + self.assertEqual(live["stranded_by_kv"], 17) + + def test_capping_is_reported_without_a_recurrent_cache(self): + # The page table counts the cap even if the cache object is unavailable; + # losing the waste figure would hide the very failure it exists to show. + install_hybrid(None, capped_pages=50) + + live = self.metrics._live_recurrent() + self.assertEqual(live["capped_tokens"], 50 * PAGE_SIZE) + self.assertEqual(live["checkpoints"], 0) + + def test_non_hybrid_model_reports_zeros(self): + install_hybrid(None) + + live = self.metrics._live_recurrent() + self.assertEqual(live["checkpoints"], 0) + self.assertEqual(live["capped_tokens"], 0) + + def test_a_broken_cache_does_not_take_the_cap_with_it(self): + # The cap is counted by the page table, not the cache, and it is the + # figure these series exist for. A cache in an unexpected state must not + # cost it. + install_hybrid(None, capped_pages=50) + model_module.container.generator.generator.recurrent_cache = types.SimpleNamespace( + metrics={} + ) + + live = self.metrics._live_recurrent() + + self.assertEqual(live["capped_tokens"], 50 * PAGE_SIZE) + self.assertEqual(live["checkpoints"], 0) + + def test_series_are_published_and_survive_a_broken_cache(self): + install_hybrid(None) + model_module.container.generator.generator.recurrent_cache = types.SimpleNamespace( + metrics={} + ) + + rendered = self.metrics.render_prometheus() + for name in ( + "tabbyapi:recurrent_cache_usage_ratio", + "tabbyapi:recurrent_checkpoints", + "tabbyapi:recurrent_cache_bytes", + "tabbyapi:recurrent_cache_max_bytes", + "tabbyapi:recurrent_checkpoint_bytes", + "tabbyapi:recurrent_cache_evictions_total", + "tabbyapi:recurrent_cache_stranded_evictions_total", + "tabbyapi:recurrent_cache_live_kv_evictions_total", + "tabbyapi:recurrent_cache_pruned_total", + "tabbyapi:recurrent_stranded_by_kv_total", + "tabbyapi:recurrent_capped_tokens_total", + ): + self.assertIn(f"{name} 0", rendered) + + def test_rendered_values_match_the_live_read(self): + install_hybrid(FakeRecurrentCache(), capped_pages=120, stranded_by_kv=17) + + rendered = self.metrics.render_prometheus() + + self.assertIn("tabbyapi:recurrent_checkpoints 60", rendered) + self.assertIn(f"tabbyapi:recurrent_capped_tokens_total {120 * PAGE_SIZE}", rendered) + self.assertIn("tabbyapi:recurrent_cache_evictions_total 40", rendered) + self.assertIn("tabbyapi:recurrent_cache_live_kv_evictions_total 12", rendered) + self.assertIn("tabbyapi:recurrent_stranded_by_kv_total 17", rendered) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_metrics_time_breakdown.py b/tests/test_metrics_time_breakdown.py new file mode 100644 index 00000000..4867761e --- /dev/null +++ b/tests/test_metrics_time_breakdown.py @@ -0,0 +1,148 @@ +import importlib.util +import pathlib +import time +import types +import unittest + +from common import model as model_module +from common.metrics import MetricsManagerClass + + +def load_inspector(): + """The tool is a script rather than a package module, so it is loaded by path.""" + + path = pathlib.Path(__file__).parent.parent / "tools" / "inspect_metrics.py" + spec = importlib.util.spec_from_file_location("inspect_metrics", path) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +inspector = load_inspector() + + +def rendered(prefill: float, decode: float, requests: int = 20, uptime: float = 3600.0) -> str: + metrics = MetricsManagerClass() + metrics.process_start_time = time.time() - uptime + + pagetable = types.SimpleNamespace( + max_pages=1000, + referenced_pages={}, + metrics={"alloc_kv_only_pages": 0, "stashes_stranded": 0}, + ) + model_module.container = types.SimpleNamespace( + generator=types.SimpleNamespace( + generator=types.SimpleNamespace( + pagetable=pagetable, + max_total_tokens=256000, + cpu_page_cache=None, + recurrent_cache=None, + num_active_jobs=lambda: 0, + num_pending_jobs=lambda: 0, + ) + ) + ) + for _ in range(requests): + metrics.record_generation(20000, 2000, 300, prefill, decode, 0.05) + + scalars, hists = inspector.parse(metrics.render_prometheus()) + return inspector.render( + scalars, hists, "http://test/metrics", inspector.C(False), show_hists=False, width=200 + ) + + +def block(text: str) -> str: + """Just the time breakdown, so an assertion cannot match a number from elsewhere.""" + + lines = text.splitlines() + start = next(i for i, line in enumerate(lines) if "Where the time goes" in line) + end = next( + (i for i in range(start + 1, len(lines)) if not lines[i].startswith(" ")), len(lines) + ) + return "\n".join(lines[start:end]) + + +class TimeBreakdownTests(unittest.TestCase): + """The phase split, which is what decides whether to buy prefill or decode speed. + + A parallelism layout trades one against the other, so the number that + matters is not how fast either phase is but how much of the server's time + goes into each. + """ + + def setUp(self): + self.original_container = model_module.container + + def tearDown(self): + model_module.container = self.original_container + + def test_the_exchange_rate_is_the_ratio_of_the_shares(self): + # One percent off the dominant phase pays for (p/d) percent onto the + # other, which is the whole decision in one number and does not depend + # on any assumed speedup factor. 93.0 / 7.0 = 13.3. + text = block(rendered(prefill=8.0, decode=0.6)) + + self.assertIn("93.0%", text) + self.assertIn("1% off prefill pays for 13.3% onto decode", text) + + def test_the_rate_names_whichever_phase_dominates(self): + text = block(rendered(prefill=0.4, decode=6.0)) + + self.assertIn("1% off decode pays for", text) + self.assertIn("onto prefill", text) + + def test_an_even_split_trades_one_for_one(self): + text = block(rendered(prefill=3.0, decode=3.0)) + + self.assertIn("1% off prefill pays for 1.0% onto decode", text) + + def test_break_even_is_exact_rather_than_marginal(self): + # The rate above is a linear reading and overstates a large move, so the + # break-even for a concrete one is computed exactly: after 2x prefill on + # a 50/50 split, decode may fall to 0.5 / (1 - 0.25) = 66.7% of its speed. + text = block(rendered(prefill=3.0, decode=3.0)) + + self.assertIn("at 2x prefill", text) + self.assertIn("break-even at 66.7% of decode speed", text) + + def test_a_phase_that_costs_nothing_is_named_as_such(self): + text = block(rendered(prefill=3.0, decode=0.0)) + + self.assertIn("prefill is all of it", text) + + def test_shares_sum_to_the_whole(self): + text = block(rendered(prefill=3.0, decode=1.0)) + + self.assertIn("75.0%", text) + self.assertIn("25.0%", text) + + def test_utilization_is_against_wall_clock(self): + # 20 requests of 4s each in an hour of uptime + text = block(rendered(prefill=3.0, decode=1.0, requests=20, uptime=3600.0)) + + self.assertIn("2.2% utilization", text) + + def test_concurrency_is_named_rather_than_reported_as_over_100_percent(self): + # Summed request time can exceed the interval it ran in. That is + # concurrency, not a bug, but a utilization over 100% reads as one. + text = block(rendered(prefill=8.0, decode=8.0, requests=100, uptime=60.0)) + + self.assertIn("overlapped", text) + self.assertNotIn("utilization", text.replace("100.0% utilization", "")) + + def test_the_block_is_absent_before_any_request(self): + # Nothing to divide, and a 0/0 split would be an invented number + text = rendered(prefill=0.0, decode=0.0, requests=0) + + self.assertNotIn("Where the time goes", text) + + def test_queue_time_is_kept_out_of_the_split(self): + # A request waiting is the engine working on another one, so counting + # queue time as a phase would double-count it. + text = block(rendered(prefill=3.0, decode=1.0)) + + self.assertIn("not engine time", text) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_metrics_token_buckets.py b/tests/test_metrics_token_buckets.py new file mode 100644 index 00000000..a5625bdb --- /dev/null +++ b/tests/test_metrics_token_buckets.py @@ -0,0 +1,90 @@ +import unittest + +from common.metrics import MetricsManagerClass, build_1_2_5_buckets + + +class TokenBucketTests(unittest.TestCase): + """Token histogram boundaries, built from the model's context length. + + A fixed ladder is not merely imprecise, it produces percentiles above any + request the server can accept: histogram_quantile interpolates across the + bucket a sample landed in, so a 300k-wide bucket holding prompts clustered + at its floor reports a p99 near its ceiling. + """ + + def test_ladder_is_a_1_2_5_progression(self): + self.assertEqual(build_1_2_5_buckets(100), [1, 2, 5, 10, 20, 50, 100]) + self.assertEqual(build_1_2_5_buckets(1), [1]) + # 9 is off the ladder, so it becomes the final boundary itself + self.assertEqual(build_1_2_5_buckets(9), [1, 2, 5, 9]) + + def test_limit_is_always_a_boundary(self): + # Stopping at the last mantissa below the limit strands everything + # between there and the real limit in +Inf. At 262144 that is the top + # 24% of the usable range. + buckets = build_1_2_5_buckets(262144) + self.assertEqual(buckets[-1], 262144) + self.assertEqual(buckets[-2], 200000) + + def test_no_bucket_exceeds_the_limit(self): + for limit in (4096, 32768, 131072, 262144, 344064, 1_000_000): + self.assertLessEqual(max(build_1_2_5_buckets(limit)), limit) + self.assertEqual(sorted(set(build_1_2_5_buckets(limit))), build_1_2_5_buckets(limit)) + + def test_configure_resizes_the_histograms(self): + metrics = MetricsManagerClass() + metrics.configure_token_buckets(262144) + + self.assertEqual(metrics.token_buckets[-1], 262144) + self.assertEqual(metrics.hist_prompt_tokens.buckets, metrics.token_buckets) + self.assertEqual(metrics.hist_gen_tokens.buckets, metrics.token_buckets) + + def test_reconfiguring_to_the_same_length_keeps_the_samples(self): + # Rebuilding drops accumulated counts, so an unchanged ladder must not. + metrics = MetricsManagerClass() + metrics.configure_token_buckets(262144) + metrics.record_generation( + prompt_tokens=1000, cached_tokens=0, gen_tokens=10, prompt_time=1.0, gen_time=1.0 + ) + metrics.configure_token_buckets(262144) + + self.assertEqual(metrics.hist_prompt_tokens.count, 1) + + def test_a_different_length_resets_the_histograms(self): + # Counts against a different ladder cannot be carried over. + metrics = MetricsManagerClass() + metrics.record_generation( + prompt_tokens=1000, cached_tokens=0, gen_tokens=10, prompt_time=1.0, gen_time=1.0 + ) + metrics.configure_token_buckets(4096) + + self.assertEqual(metrics.hist_prompt_tokens.count, 0) + + def test_nonsense_lengths_are_ignored(self): + metrics = MetricsManagerClass() + before = list(metrics.token_buckets) + for bad in (0, -1, None): + metrics.configure_token_buckets(bad) + self.assertEqual(metrics.token_buckets, before) + + def test_percentiles_stay_inside_the_context_limit(self): + # The regression this exists for: every reported percentile must be a + # length the server could actually have been asked for. + metrics = MetricsManagerClass() + metrics.configure_token_buckets(262144) + for _ in range(31): + metrics.record_generation( + prompt_tokens=203000, cached_tokens=0, gen_tokens=1, prompt_time=1.0, gen_time=1.0 + ) + + rendered = metrics.render_prometheus() + boundaries = [ + float(line.split('le="')[1].split('"')[0]) + for line in rendered.splitlines() + if line.startswith("tabbyapi:request_prompt_tokens_bucket") and "+Inf" not in line + ] + self.assertLessEqual(max(boundaries), 262144) + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/inspect_metrics.py b/tools/inspect_metrics.py new file mode 100755 index 00000000..1de7d80b --- /dev/null +++ b/tools/inspect_metrics.py @@ -0,0 +1,1223 @@ +#!/usr/bin/env python3 +"""Pretty-print TabbyAPI's Prometheus /metrics endpoint. + +Stdlib only (no prometheus_client, no requests) — same constraint as the +launcher paths, so it runs on a bare community VM. + + python3 tools/inspect_metrics.py # one shot, localhost:5000 + python3 tools/inspect_metrics.py --port 8010 + python3 tools/inspect_metrics.py --watch 2 # refresh every 2s + python3 tools/inspect_metrics.py --json # machine-readable digest + +Requires network.enable_metrics to be true in config.yml. +""" + +from __future__ import annotations + +import argparse +import json +import os +import re +import shutil +import sys +import time +import urllib.error +import urllib.request + +PREFIX = "tabbyapi:" +SAMPLE_RE = re.compile(r"^(?P[a-zA-Z_:][\w:]*)(?:\{(?P[^}]*)\})?\s+(?P.+)$") + + +# ---------------------------------------------------------------- scrape ---- + + +def scrape(url: str, timeout: float) -> str: + req = urllib.request.Request(url, headers={"Accept": "text/plain"}) + with urllib.request.urlopen(req, timeout=timeout) as resp: + return resp.read().decode("utf-8", errors="replace") + + +def parse(text: str): + """-> (scalars: name->float, histograms: name->{'buckets': [(le, count)], 'sum', 'count'})""" + scalars: dict[str, float] = {} + hists: dict[str, dict] = {} + for line in text.splitlines(): + line = line.strip() + if not line or line.startswith("#"): + continue + m = SAMPLE_RE.match(line) + if not m: + continue + name = m.group("name") + if name.startswith(PREFIX): + name = name[len(PREFIX) :] + try: + value = float(m.group("value").split()[0]) + except ValueError: + continue + + if name.endswith("_bucket"): + base = name[: -len("_bucket")] + le = (m.group("labels") or "").partition("le=")[2].strip('"').strip('",') + le = float("inf") if le in ("+Inf", "Inf") else _f(le) + if le is not None: + hists.setdefault(base, {"buckets": [], "sum": None, "count": None})[ + "buckets" + ].append((le, value)) + elif name.endswith("_sum") and name[: -len("_sum")] in hists or name.endswith("_sum"): + hists.setdefault(name[: -len("_sum")], {"buckets": [], "sum": None, "count": None})[ + "sum" + ] = value + elif name.endswith("_count"): + hists.setdefault(name[: -len("_count")], {"buckets": [], "sum": None, "count": None})[ + "count" + ] = value + else: + scalars[name] = value + + for h in hists.values(): + h["buckets"].sort(key=lambda b: b[0]) + return scalars, hists + + +def _f(s: str): + try: + return float(s) + except ValueError: + return None + + +# ------------------------------------------------------------ histograms ---- + + +def quantile(hist: dict, q: float, cap: float | None = None): + """Prometheus-style linear interpolation inside the matched bucket. + + Interpolation assumes the samples are spread evenly across the bucket they + landed in, which the 1-2-5 token buckets make wildly untrue at the top end: + 31 prompts of ~200k tokens put 14 samples in the (200k, 500k] bucket, all + of them within a few thousand of its floor, and the p99 comes out at 493k. + `cap` clamps the estimate to a separately reported maximum, since no + percentile of a sample can exceed the largest value in it. + """ + buckets = hist.get("buckets") or [] + total = hist.get("count") + if total is None: + total = buckets[-1][1] if buckets else 0 + if not buckets or not total: + return None + target = q * total + prev_le, prev_count = 0.0, 0.0 + for le, count in buckets: + if count >= target: + if le == float("inf"): + return prev_le + if count == prev_count: + return le + frac = (target - prev_count) / (count - prev_count) + value = prev_le + (le - prev_le) * frac + return min(value, cap) if cap is not None else value + prev_le, prev_count = le, count + return buckets[-1][0] + + +def mean(hist: dict): + c = hist.get("count") or 0 + s = hist.get("sum") + return (s / c) if (s is not None and c) else None + + +def per_bucket(hist: dict): + """Cumulative buckets -> [(le, non-cumulative count)].""" + out, prev = [], 0.0 + for le, count in hist.get("buckets") or []: + out.append((le, count - prev)) + prev = count + return out + + +# --------------------------------------------------------------- display ---- + +BLOCKS = " ▁▂▃▄▅▆▇█" + + +def color_on() -> bool: + return sys.stdout.isatty() and not os.environ.get("NO_COLOR") + + +class C: + def __init__(self, enabled: bool): + self.e = enabled + + def _w(self, code: str, s: str) -> str: + return f"\033[{code}m{s}\033[0m" if self.e else s + + def head(self, s): + return self._w("1;36", s) + + def key(self, s): + return self._w("38;5;245", s) + + def val(self, s): + return self._w("1", s) + + def good(self, s): + return self._w("32", s) + + def warn(self, s): + return self._w("33", s) + + def bad(self, s): + return self._w("31", s) + + def dim(self, s): + return self._w("2", s) + + +def fmt_num(v, unit=""): + if v is None: + return "—" + if unit == "s": + if v < 1: + return f"{v * 1000:.0f} ms" + if v < 60: + return f"{v:.2f} s" + return f"{v / 60:.1f} min" + if unit == "tok": + if v >= 1e9: + return f"{v / 1e9:.2f}B" + if v >= 1e6: + return f"{v / 1e6:.2f}M" + if v >= 1e3: + return f"{v / 1e3:.1f}K" + return f"{v:.0f}" + if unit == "%": + return f"{v * 100:.1f}%" + if unit == "B": + for scale, suffix in ((1024**4, "TB"), (1024**3, "GB"), (1024**2, "MB"), (1024, "KB")): + if v >= scale: + return f"{v / scale:.2f} {suffix}" + return f"{v:.0f} B" + if v == int(v): + return f"{int(v)}" + return f"{v:.2f}" + + +def fmt_le(le, unit): + if le == float("inf"): + return "+Inf" + if unit == "s": + return f"{le * 1000:.0f}ms" if le < 1 else f"{le:g}s" + if unit == "tok": + return fmt_num(le, "tok") + return f"{le:g}" + + +def bar(frac: float, width: int, c: C) -> str: + frac = max(0.0, min(1.0, frac)) + filled = int(frac * width) + rem = frac * width - filled + tail = BLOCKS[int(rem * 8)] if filled < width and rem > 0 else "" + body = "█" * filled + tail + return body.ljust(width) + + +# Lookback windows for the rate table, as seconds. Six columns keep the table +# inside the width the widest existing row already occupies; more still render, +# just wider. Windows shorter than the refresh interval are dropped outright, +# since nothing could ever fill them; the rest stay blank until the session has +# run long enough to reach back that far, so the table fills in as you watch. +# +# Windows are written as durations — 30s, 5m, 1h, 24h — rather than as raw +# seconds. It matches PromQL's rate(...[5m]) spelling, which is the idiom this +# table is a stand-in for, and each token doubles as its own column header, so +# a header always reads back exactly what was asked for. A bare number is taken +# as seconds. +RATE_WINDOWS_DEFAULT = "30s,1m,5m,15m,1h,24h" +RATE_COL = 8 +DURATION_UNITS = {"s": 1, "m": 60, "h": 3600, "d": 86400, "": 1} + + +def parse_windows(spec: str) -> list: + """Parse a comma-separated duration list into (label, seconds) pairs.""" + + out = {} + for token in spec.split(","): + token = token.strip().lower() + if not token: + continue + m = re.fullmatch(r"(\d+(?:\.\d+)?)([smhd]?)", token) + if not m: + raise argparse.ArgumentTypeError( + f"bad window {token!r}: want a duration like 30s, 5m, 1h, 24h" + ) + seconds = float(m.group(1)) * DURATION_UNITS[m.group(2)] + if seconds <= 0: + raise argparse.ArgumentTypeError(f"window {token!r} must be positive") + # Keep the first spelling seen for any given length, so 60s and 1m do + # not both claim a column. + out.setdefault(seconds, token) + if not out: + raise argparse.ArgumentTypeError("no windows given") + return [(label, seconds) for seconds, label in sorted(out.items())] + + +# History retention. Full resolution for the recent past, thinned beyond that: +# a 24h window scraped every few seconds would otherwise hold tens of thousands +# of samples. Since a column divides by the true elapsed time, a baseline that +# is a little older than its nominal window stays correct — it just measures a +# slightly wider window than the header says. +HISTORY_FINE_AGE = 90.0 +HISTORY_COARSE_SPACING = 30.0 + + +def fmt_rate(v) -> str: + """Compact fixed-width rate, at most six characters wide.""" + if v is None: + return "—" + if v >= 1e6: + return f"{v / 1e6:.1f}M" + if v >= 1e3: + return f"{v / 1e3:.1f}K" + if v >= 100: + return f"{v:.0f}" + if v >= 10: + return f"{v:.1f}" + # A true zero and a rate too small to render must not look alike: one + # request in a 5m window is 0.0033/s, and printing that as "0.00" reads as + # an idle server. Bare "0" is the only cell that means nothing happened. + if v == 0: + return "0" + if v < 0.01: + return "<0.01" + return f"{v:.2f}" + + +def fmt_span(seconds: float) -> str: + """Compact duration for a column header.""" + if seconds >= 3600: + return f"{seconds / 3600:.1f}h" + if seconds >= 60: + return f"{seconds / 60:.0f}m" + return f"{seconds:.0f}s" + + +def baseline_for(history, now: float, window: float): + """The sample to difference against for a given lookback window, or None. + + Picks the newest sample at least `window` old, so the figure covers the + whole window rather than part of it. Returns None when no sample is that + old yet, and also when the best candidate is more than twice the window + old, since labelling it with that window would misrepresent it. That second + case is what keeps the short columns blank when the refresh interval is + long, or when inference has held the event loop through several refreshes. + """ + + # Window 0 is the "now" column: difference against the previous scrape, + # whatever interval that turned out to be. + if window <= 0: + return history[-1] if history else None + + for ts, scalars in reversed(history): + age = now - ts + if age >= window: + return None if age > window * 2 else (ts, scalars) + return None + + +# A window wider than the session so far would print as a permanent blank, even +# though the counters hold the answer: the session's own history *is* the widest +# honest lookback available. So the first window the session has not lived +# through yet overflows into the whole history, relabelled with the span it +# actually covers (`~19.6h` under a `24h` header), and the windows beyond it +# stay blank rather than repeating it. The overflow column shrinks back to its +# nominal label once the session is old enough to fill it for real. +# +# The relabelling only fires when history is genuinely shorter than the window. +# A window that history *does* span but whose baseline was rejected as stale -- +# an event-loop stall pushing the last scrape past the `now` or `30s` column -- +# is a transient gap in one narrow column, not an overflow, and must not blank +# the wider columns behind it. +OVERFLOW_MIN_RATIO = 1.25 + + +def resolve_columns(history, now: float, columns): + """-> [(header, baseline or None)], one per column, applying the overflow.""" + + span = (now - history[0][0]) if history else 0.0 + out, overflowed, prev = [], False, 0.0 + for label, window in columns: + base = baseline_for(history, now, window) + if base is not None: + # The leftmost column has no nominal width -- it is however long the + # last refresh happened to take -- so it names its own interval. + # Watching that number drift above the refresh interval is the + # cheapest sign that inference is sitting on the event loop. + out.append((fmt_span(now - base[0]) if window <= 0 else label, base)) + elif history and window > span and not overflowed: + overflowed = True + # Too close to the column on its left to be worth a column of its + # own: it would read as two near-identical numbers. + if span >= prev * OVERFLOW_MIN_RATIO: + out.append(("~" + fmt_span(span), history[0])) + else: + out.append((label, None)) + else: + out.append((label, None)) + prev = window + return out + + +def prune_history(history, now: float, longest: float) -> None: + """Thin old samples in place, keeping the recent past at full resolution.""" + + kept, last_coarse = [], None + for ts, scalars in history: + age = now - ts + if age > longest * 1.1: + continue + if age <= HISTORY_FINE_AGE: + kept.append((ts, scalars)) + elif last_coarse is None or ts - last_coarse >= HISTORY_COARSE_SPACING: + kept.append((ts, scalars)) + last_coarse = ts + history[:] = kept + + +def row(c: C, label: str, value: str, note: str = "") -> str: + line = f" {c.key(label.ljust(26))} {c.val(value)}" + if note: + line += f" {c.dim(note)}" + return line + + +ANSI_RE = re.compile(r"\x1b\[[0-9;]*[A-Za-z]") +GUTTER = 3 + + +def vlen(s: str) -> int: + """Visible width, ignoring colour escapes — ljust() alone would pad by the byte count.""" + return len(ANSI_RE.sub("", s)) + + +class Block: + """One section, laid out only once its final column width is known. + + Rows are kept as (label, value, note) rather than formatted strings so the + label column can be sized to the block's own longest label. In two-column + mode that matters: a fixed 26-wide label field wastes a quarter of the + available room on sections whose labels are half that. + """ + + def __init__(self, title: str, full_width: bool = False, new_row: bool = False): + self.title = title + self.items: list = [] + # Wide, self-aligning content (the rate table, histogram bars) cannot be + # squeezed into a column and spans the terminal instead. + self.full_width = full_width + # Start a fresh row rather than filling the one in progress. Which + # sections are present varies with the model and the config, so a + # section that must sit beside the next one cannot rely on the count of + # sections before it coming out even. + self.new_row = new_row + + def row(self, label: str, value: str, note: str = ""): + self.items.append((label, value, note)) + + def raw(self, line: str): + self.items.append((None, line, "")) + + def __bool__(self): + return bool(self.items) + + def _labw(self) -> int: + return max([len(label) for label, _, _ in self.items if label] or [0]) + + def natural_width(self) -> int: + labw = self._labw() + widest = vlen(self.title) + 4 + for label, value, note in self.items: + if label is None: + widest = max(widest, vlen(value) + 2) + else: + w = 2 + labw + 2 + vlen(value) + (2 + vlen(note) if note else 0) + widest = max(widest, w) + return widest + + def lines(self, width: int, c: C) -> list: + rule = "─" * max(3, width - vlen(self.title) - 1) + out = [f"{c.head(self.title)} {c.dim(rule)}"] + labw = self._labw() + for label, value, note in self.items: + if label is None: + out.append(f" {value}") + continue + line = f" {c.key(label.ljust(labw))} {c.val(value)}" + if note: + line += f" {c.dim(note)}" + out.append(line) + return out + + +class Stack: + """Several sections sharing one column, laid out top to bottom. + + Packing works in whole columns, so a short section beside a tall one leaves + the space beneath it empty while the next section starts a row of its own. + Grouping the short ones into a single unit fills that gap instead. It is a + layout hint and nothing more: the sections keep their own titles, rules and + label widths, and a stack whose members are all absent vanishes with them. + """ + + def __init__(self, *blocks): + self.blocks = [b for b in blocks if b] + # A stack is only ever built from column-sized sections; if one of them + # turns out too wide for a cell, pack gives the whole stack its own row. + self.full_width = False + self.new_row = bool(self.blocks) and self.blocks[0].new_row + + def __bool__(self): + return bool(self.blocks) + + def natural_width(self) -> int: + return max(b.natural_width() for b in self.blocks) + + def lines(self, width: int, c: C) -> list: + out: list = [] + for b in self.blocks: + if out: + out.append("") + out.extend(b.lines(width, c)) + return out + + +def pack(blocks: list, width: int, max_cols: int) -> list: + """Group blocks into rows of side-by-side sections that fit the terminal. + + Greedy and order preserving, so sections stay where they are expected to be + rather than being reshuffled to fill space. A block that will not fit + alongside its predecessor starts a new row rather than being truncated. + """ + # Every column is the same width, so a block wider than its share would run + # into its neighbour however well the pair sums. Test each block against the + # cell, not the pair against the terminal. + cell = (width - GUTTER * (max_cols - 1)) // max_cols if max_cols > 1 else width + + rows, cur = [], [] + for b in blocks: + if not b: + continue + if b.full_width or max_cols == 1 or b.natural_width() > cell: + if cur: + rows.append(cur) + cur = [] + rows.append([b]) + continue + if b.new_row and cur: + rows.append(cur) + cur = [] + trial = cur + [b] + if len(trial) <= max_cols: + cur = trial + if len(cur) == max_cols: + rows.append(cur) + cur = [] + else: + if cur: + rows.append(cur) + cur = [b] + if cur: + rows.append(cur) + return rows + + +def compose(rows: list, width: int, c: C) -> list: + """Render packed rows, aligning every multi-column row on one column width.""" + paired = [b for r in rows if len(r) > 1 for b in r] + col_w = 0 + if paired: + widest = max(b.natural_width() for b in paired) + col_w = min(widest, (width - GUTTER) // 2) + + out: list = [] + for r in rows: + if len(r) == 1: + # full_width governs packing, not the rule: a wide block still ends + # its underline at its own content rather than trailing dashes to + # the edge of a very wide terminal. + b = r[0] + out.extend(b.lines(min(b.natural_width(), width), c)) + else: + cols = [b.lines(col_w, c) for b in r] + height = max(map(len, cols)) + padded = [cl + [""] * (height - len(cl)) for cl in cols] + for line_set in zip(*padded, strict=True): + line = "" + for i, cell in enumerate(line_set): + if i: + line += " " * (col_w + GUTTER - vlen(line)) + cell + else: + line = cell + out.append(line.rstrip()) + out.append("") + return out + + +def hist_block( + c: C, title: str, hist: dict, unit: str, width: int = 78, cap: float | None = None +) -> Block: + b = Block(title, full_width=True) + p50, p90, p99 = (quantile(hist, q, cap) for q in (0.5, 0.9, 0.99)) + b.row( + "mean / p50 / p90 / p99", + " · ".join(fmt_num(v, unit) for v in (mean(hist), p50, p90, p99)), + f"n={fmt_num(hist.get('count'))}", + ) + counts = per_bucket(hist) + counts = [(le, n) for le, n in counts if n > 0] + if not counts: + return b + peak = max(n for _, n in counts) + labw = max(len(fmt_le(le, unit)) for le, _ in counts) + barw = max(10, width - labw - 12) + for le, n in counts: + b.raw( + f"{c.dim(fmt_le(le, unit).rjust(labw))} {c.good(bar(n / peak, barw, c))} " + f"{c.dim(str(int(n)))}" + ) + return b + + +# ---------------------------------------------------------------- render ---- + + +def render( + scalars: dict, + hists: dict, + url: str, + c: C, + show_hists: bool, + history: list | None = None, + now: float = 0.0, + windows: list | None = None, + width: int = 100, + max_cols: int = 1, +) -> str: + g = scalars.get + # A 24-wide bar reads well down a single column but eats a third of a + # narrow one, and the number beside it carries the same information. + barw = 24 if max_cols == 1 else 14 + blocks: list = [] + + # --- live state + live = Block("Live") + blocks.append(live) + processing, deferred = g("requests_processing"), g("requests_deferred") + state = "idle" if not processing else f"{fmt_num(processing)} generating" + if deferred: + state += f", {fmt_num(deferred)} queued" + live.row("state", state) + + kv_ratio, kv_tok, kv_max = ( + g("kv_cache_usage_ratio"), + g("kv_cache_tokens"), + g("kv_cache_max_tokens"), + ) + if kv_ratio is not None: + paint = c.good if kv_ratio < 0.75 else (c.warn if kv_ratio < 0.9 else c.bad) + live.row( + "KV cache", + f"{paint(bar(kv_ratio, barw, c))} {fmt_num(kv_ratio, '%')}", + f"{fmt_num(kv_tok, 'tok')} / {fmt_num(kv_max, 'tok')}", + ) + + # Offloaded pages are headroom, not load: they cost RAM rather than VRAM and + # are what a displaced request comes back from. Shown next to the VRAM cache + # because the two together are the context the server can still resume. + off_max = g("kv_offload_max_tokens") + if off_max: + off_ratio = g("kv_offload_usage_ratio") or 0.0 + live.row( + "KV offload", + f"{c.good(bar(off_ratio, barw, c))} {fmt_num(off_ratio, '%')}", + f"{fmt_num(g('kv_offload_tokens'), 'tok')} / {fmt_num(off_max, 'tok')} in RAM", + ) + + if g("n_tokens_max") is not None: + live.row("largest ctx seen", fmt_num(g("n_tokens_max"), "tok")) + # The peaks the size histograms cannot show: a lone big completion sits + # above every percentile they report. + if g("tokens_predicted_max") is not None: + live.row("largest generation", fmt_num(g("tokens_predicted_max"), "tok")) + + # --- throughput + # Work done and time spent, for both phases, and deliberately no rate for + # either. Both times are summed per request, so under concurrency they cover + # more wall clock than actually elapsed, and tokens over such a span is the + # speed one stream saw while sharing the batch -- a figure that sags as load + # rises on unchanged hardware and is invariably misread as the model getting + # slower. Prefill would be worse still: a prefix cache hit takes tokens out + # of the count while leaving its lookup time in the span, so the ratio decays + # towards nonsense as reuse accumulates. Real rates are wall-clock ones, and + # they live in the staggered windows under --watch; for a per-stream speed + # worth comparing across configs, measure at concurrency 1 or use + # exllamav3's eval/perf.py. + thr = Block("Throughput (cumulative)") + blocks.append(thr) + thr.row( + "prefill work", + f"{fmt_num(g('prompt_tokens_total'), 'tok')} computed", + f"over {fmt_num(g('prompt_seconds_total'), 's')}", + ) + thr.row( + "decode work", + f"{fmt_num(g('tokens_predicted_total'), 'tok')} generated", + f"over {fmt_num(g('tokens_predicted_seconds_total'), 's')}", + ) + thr.row("requests finished", fmt_num(g("requests_total"))) + + # --- where the time goes + # The question this answers is which phase to buy speed in. A parallelism + # layout trades prefill against decode -- tensor parallel favours decode, + # pipeline parallel favours prefill -- and the right choice depends entirely + # on which one this server actually spends its time in, which is a property + # of the workload rather than of the hardware. + # + # The two phase counters are summed per request, so under concurrency they + # add up to more than the wall clock they ran in. That does not affect the + # split, which is what the decision turns on, but it does mean the share is + # of engine time rather than of elapsed time, and it is labelled as such + # when the two disagree. + prefill_s = g("prompt_seconds_total") or 0.0 + decode_s = g("tokens_predicted_seconds_total") or 0.0 + engine_s = prefill_s + decode_s + if engine_s > 0: + # process_start_time_seconds is epoch-based, following the convention, + # so it is compared against wall clock rather than the monotonic clock + # the rate windows use. + started = g("process_start_time_seconds") + uptime = (time.time() - started) if started else None + + tb = Block("Where the time goes", new_row=True) + blocks.append(tb) + + if uptime: + # Clamped, because summed request time can exceed the interval it + # ran in and a utilization above 100% reads as a bug rather than as + # concurrency, which the note beside it names explicitly + util = min(engine_s, uptime) / uptime + tb.row( + "engine busy", + f"{fmt_num(engine_s, 's')} of {fmt_num(uptime, 's')} up", + f"{fmt_num(util, '%')} utilization" + + ("" if engine_s <= uptime else c.dim(" · overlapped")), + ) + + for label, value in (("prefill", prefill_s), ("decode", decode_s)): + share = value / engine_s + tb.row( + label, + f"{c.dim(bar(share, barw, c))} {fmt_num(share, '%')}", + fmt_num(value, "s"), + ) + + # Queue time is not part of the split: a request waiting is the engine + # working on another one, so it would double-count. It is worth seeing + # because it is the part that more speed anywhere would relieve. + queued = (hists.get("request_queue_time_seconds") or {}).get("sum") + if queued: + tb.row("queued behind others", fmt_num(queued, "s"), c.dim("not engine time")) + + # The trade a parallelism layout presents, as one number. Tensor + # parallel buys decode and gives up prefill, pipeline parallel the + # reverse, and no fixed speedup factor describes either, so what is + # printed is the exchange rate between the two phases rather than the + # result of some assumed swap. + # + # Cutting prefill time by a fraction x saves x*p of the engine's time; + # inflating decode time by y costs y*d. Break-even is x*p == y*d, so one + # percent off the winning phase pays for exactly (p/d) percent onto the + # losing one. That ratio is the whole decision, and it is a property of + # the workload rather than of any particular hardware layout. + p, d = prefill_s / engine_s, decode_s / engine_s + win, lose = ("prefill", "decode") if p >= d else ("decode", "prefill") + hi, lo = (p, d) if p >= d else (d, p) + + if lo <= 0: + tb.row("exchange rate", f"{win} is all of it", c.dim(f"{lose} costs nothing")) + else: + rate = hi / lo + tb.row( + "exchange rate", + f"1% off {win} pays for {rate:,.0f}% onto {lose}" + if rate >= 100 + else f"1% off {win} pays for {rate:.1f}% onto {lose}", + c.dim(f"{fmt_num(hi, '%')} vs {fmt_num(lo, '%')}"), + ) + + # The marginal rate above is a linear reading and overstates the trade + # for large moves, so the exact break-even is given for a concrete one: + # after speeding the winning phase up by `factor`, the losing phase can + # fall to lo / (1 - hi/factor) of its current speed before the swap + # stops paying. Two factors, a cautious one and an ambitious one, bracket + # what a layout change realistically buys. + for factor in (1.5, 2.0) if lo > 0 else (): + headroom = 1.0 - hi / factor + if headroom <= 0: + continue + floor = lo / headroom + if floor >= 1.0: + verdict = c.bad(f"{lose} cannot give up anything") + else: + verdict = c.good(f"{lose} may fall {1 / floor:.0f}x") + tb.row( + f" at {factor:g}x {win}", + f"break-even at {fmt_num(floor, '%')} of {lose} speed", + verdict, + ) + + # --- spec decode and prefix cache + # Both are short, and "Where the time goes" beside them is not, so they are + # stacked into one column under it rather than each claiming a row. + sd = Block("Speculative decode") + if g("spec_decode_requests_total"): + acc = g("spec_decode_draft_acceptance_rate") + if acc is not None: + paint = c.good if acc > 0.7 else (c.warn if acc > 0.5 else c.bad) + sd.row( + "draft acceptance", + f"{paint(bar(acc, barw, c))} {fmt_num(acc, '%')}", + f"{fmt_num(g('spec_decode_num_accepted_tokens_total'), 'tok')} / " + f"{fmt_num(g('spec_decode_num_draft_tokens_total'), 'tok')}", + ) + tps = g("spec_decode_tokens_per_step") + if tps is not None: + sd.row("tokens / step", f"{tps:.3f}", f"+{(tps - 1) * 100:.0f}% vs. no drafter") + sd.row( + "decode steps", + fmt_num(g("spec_decode_num_decode_steps_total")), + f"{fmt_num(g('spec_decode_requests_total'))} req w/ drafter", + ) + + pc = Block("Prefix cache") + queries, hits = g("prefix_cache_queries"), g("prefix_cache_hits") + if queries: + hr = (hits or 0) / queries + paint = c.good if hr > 0.5 else (c.warn if hr > 0.2 else c.bad) + pc.row( + "hit rate", + f"{paint(bar(hr, barw, c))} {fmt_num(hr, '%')}", + f"{fmt_num(hits, 'tok')} / {fmt_num(queries, 'tok')}", + ) + pc.row("prefilled", fmt_num(g("prompt_tokens_total"), "tok"), "= queried − cached") + + blocks.append(Stack(sd, pc)) + + # --- KV offload + # A restore is already counted as a prefix cache hit above, since the server + # does not distinguish a page found in VRAM from one read back over PCIe. + # This section is that breakdown, plus what the reuse is costing in RAM. + if off_max: + off = Block("KV offload", new_row=True) + blocks.append(off) + + # The server does not count lookups that never reach RAM, so the reuse + # figure is taken against the prefix cache hits above: what share of the + # prompt the cache saved came back over PCIe rather than being found in + # VRAM. A low share is not a fault on its own — it means VRAM is serving + # the working set, which is the better outcome — so it is not painted as + # one. What it does bound is how much the offload cache is contributing. + restored = g("kv_offload_restored_tokens_total") or 0 + cache_hits = g("prefix_cache_hits") or 0 + share = restored / cache_hits if cache_hits else None + if share is not None: + off.row( + "share of cache hits", + f"{c.dim(bar(share, barw, c))} {fmt_num(share, '%')}", + f"{fmt_num(restored, 'tok')} of {fmt_num(cache_hits, 'tok')} from RAM", + ) + + # The whole capacity is pinned up front in the background, so max is the + # RAM this feature costs whatever the usage figure reads. + note = "" + cold = g("kv_offload_cold_allocs_total") + if cold: + note = c.warn(f"{fmt_num(cold)} stores pinned synchronously") + held, pinned = g("kv_offload_bytes"), g("kv_offload_max_bytes") + off.row("RAM in use", f"{fmt_num(held, 'B')} / {fmt_num(pinned, 'B')} pinned", note) + + # Once the cache is full every store has to evict something, so the + # eviction count climbing towards the store count says only that it is + # saturated -- which is the steady state of a cache that is being used, + # not a fault. Whether saturation is costing anything is a question about + # traffic: pages written out and never read back are the wasted ones, and + # bytes in each direction are counted exactly. + stores, evictions = g("kv_offload_stores_total"), g("kv_offload_evictions_total") + written = g("kv_offload_written_bytes_total") or 0 + read = g("kv_offload_read_bytes_total") or 0 + payback = read / written if written else None + if stores: + note = "" + if evictions and payback is not None and payback < 0.25: + note = c.warn("thrashing — raise sysmem_kv_cache") + deduped = g("kv_offload_deduped_stores_total") or 0 + detail = f"{fmt_num(stores)} stored, {fmt_num(evictions)} evicted" + if deduped: + detail += f", {fmt_num(deduped)} deduped" + off.row("churn", detail, note) + + off.row( + "traffic", + f"{fmt_num(read, 'B')} read · {fmt_num(written, 'B')} written", + f"{payback:.2f}x read back" if payback is not None else "", + ) + + # --- recurrent checkpoints (hybrid models only) + # Prompt reuse is capped at the longest prefix with both valid K/V pages and + # a matching recurrent checkpoint, so an undersized cache here silently + # defeats the KV cache above — and with offloading on, wastes the PCIe read + # that fetched the pages first. capped is that waste, in tokens. + rec_max = g("recurrent_cache_max_bytes") + capped = g("recurrent_capped_tokens_total") or 0 + if rec_max or capped: + rec = Block("Recurrent state") + blocks.append(rec) + rr = g("recurrent_cache_usage_ratio") or 0.0 + # High usage is not a problem in itself; eviction under pressure is, and + # that shows up in the capped row below. + rec.row( + "cache", + f"{c.good(bar(rr, barw, c))} {fmt_num(rr, '%')}", + f"{fmt_num(g('recurrent_cache_bytes'), 'B')} / {fmt_num(rec_max, 'B')}", + ) + ckpt = g("recurrent_checkpoint_bytes") + rec.row( + "checkpoints", + f"{fmt_num(g('recurrent_checkpoints'))} held", + f"{fmt_num(ckpt, 'B')} each" if ckpt else "", + ) + # Evictions on their own say nothing: a checkpoint whose anchor K/V page + # is already gone could never have been resumed, so dropping it is free, + # and the same for one reclaimed while idle. The costly ones are those + # dropped while their anchor page was still cached — those are the drops + # that become capped tokens below. + ev = g("recurrent_cache_evictions_total") or 0 + costly = g("recurrent_cache_live_kv_evictions_total") or 0 + free = (g("recurrent_cache_stranded_evictions_total") or 0) + ( + g("recurrent_cache_pruned_total") or 0 + ) + if ev: + note = c.warn(f"{fmt_num(costly)} dropped live") if costly else c.dim("all free") + rec.row("evictions", f"{fmt_num(ev)}, {fmt_num(free)} free", note) + + # The other direction: the K/V cache is the one under pressure, and + # evicting a page stranded the checkpoint anchored to it. Raising the + # recurrent budget cannot help with these. + by_kv = g("recurrent_stranded_by_kv_total") + if by_kv: + rec.row("stranded by KV eviction", fmt_num(by_kv), c.dim("KV cache is the constraint")) + + # The headline failure: valid K/V that was re-prefilled anyway. Whether a + # bigger budget would have prevented it is a separate question, and the + # answer is no unless something was actually dropped. Capping with an + # empty, never-evicted cache means no checkpoint covered those pages in + # the first place -- a prefix nothing ever generated past, or one whose + # checkpoint predates this process -- and no amount of RAM fixes that. + hits = g("prefix_cache_hits") or 0 + note = "" + if capped: + share = capped / (capped + hits) if (capped + hits) else 1.0 + under_pressure = bool(costly) or (bool(ev) and rr > 0.9) + if share <= 0.05: + note = c.dim("negligible") + elif under_pressure: + note = c.bad("raise sysmem_recurrent_cache") + elif by_kv: + note = c.warn("KV cache is the constraint, not this one") + else: + note = c.dim("no checkpoint covered them — not a budget problem") + rec.row("KV wasted by cap", fmt_num(capped, "tok"), note) + + # --- latency and size summaries + lat = [ + ("time to first token", "time_to_first_token_seconds", "s"), + ("queue wait", "request_queue_time_seconds", "s"), + ("prefill time", "request_prefill_time_seconds", "s"), + ("decode time", "request_decode_time_seconds", "s"), + ("end-to-end latency", "e2e_request_latency_seconds", "s"), + ] + # The size histograms have a companion peak counter, so their percentiles + # can be held to a value that was actually observed. The latency ones have + # no such counter and are left as the estimator reports them. + sizes = [ + ("prompt size", "request_prompt_tokens", "tok", g("n_tokens_max")), + ("generation size", "request_generation_tokens", "tok", g("tokens_predicted_max")), + ] + lat = [(lbl, key, unit, None) for lbl, key, unit in lat] + for title, spec in (("Latency (mean·p50·p90·p99)", lat), ("Size (mean·p50·p90·p99)", sizes)): + present = [(lbl, hists[k], u, cap) for lbl, k, u, cap in spec if k in hists] + if not present: + continue + b = Block(title) + blocks.append(b) + for lbl, h, u, cap in present: + qs = [quantile(h, q) for q in (0.5, 0.9, 0.99)] + capped = cap is not None and any(q is not None and q > cap for q in qs) + values = [mean(h)] + [quantile(h, q, cap) for q in (0.5, 0.9, 0.99)] + b.row( + lbl, + " · ".join(fmt_num(v, u) for v in values), + c.dim("capped at peak") if capped else "", + ) + + # --- windowed rates (--watch only) + # Counter deltas over elapsed wall clock. Nothing here divides by a + # per-request span, so prefix cache hits cannot skew it: a request that hits + # the cache contributes few tokens and little wall clock alike. This is + # throughput of the deployment, not speed of the model — an idle server + # reads near zero however fast its prefill is. + # + # Each column differences the current counters against the newest sample at + # least that old, and divides by the true elapsed time rather than the + # nominal window, so a scrape delayed by a busy event loop still yields a + # correct figure. Columns fill in from the left as the session gets longer; + # the widest one the session has not outlived yet covers all of history so + # far instead, under a `~`-prefixed header naming its true span. + if history and windows: + rates = Block("Rates over staggered windows", full_width=True) + blocks.append(rates) + + # The leftmost column is the delta since the previous scrape, headed by + # that interval. It is a liveness indicator rather than a throughput + # figure: the counters only advance when a request finishes, so a long + # prefill leaves every column at zero and then lands its whole prompt in + # one tick. The wider windows are the ones to read for a rate. + columns = resolve_columns(history, now, [("now", 0.0)] + list(windows)) + head = "".join(w.rjust(RATE_COL) for w, _ in columns) + rates.raw(f"{c.dim('window'.ljust(26))} {c.dim(head)}") + + # Byte counters are scaled to MB before rating, since fmt_rate is fixed + # width and PCIe traffic in raw bytes/s would render as "6000.0M". + for label, key, scale in ( + # "computed" is prefill actually performed; "cache hits" is prefill + # skipped. They sum to the rate prompt tokens are being ingested, + # and only the first is work the server did. Naming the second + # "cached" invited reading it as tokens going *into* the cache. + ("prompt computed (tok/s)", "prompt_tokens_total", 1.0), + ("prompt cache hits (tok/s)", "cached_tokens_total", 1.0), + ("generation (tok/s)", "tokens_predicted_total", 1.0), + ("requests (/s)", "requests_total", 1.0), + # Restored against the read/written pair below separates a cache + # that is idle from one that is being written to and never read + # back. Both read zero restores; only the second is a fault. + ("offload restored (tok/s)", "kv_offload_restored_tokens_total", 1.0), + ("offload read (MB/s)", "kv_offload_read_bytes_total", 1 / 1024**2), + ("offload written (MB/s)", "kv_offload_written_bytes_total", 1 / 1024**2), + ): + cur = g(key) + if cur is None: + continue + if key.startswith("kv_offload_") and not off_max: + continue + cells = [] + for _, base in columns: + if base is None or base[1].get(key) is None: + cells.append("—".rjust(RATE_COL)) + continue + ts, prev = base + cells.append(fmt_rate((cur - prev[key]) * scale / (now - ts)).rjust(RATE_COL)) + rates.raw(f"{c.key(label.ljust(26))} {c.val(''.join(cells))}") + + if show_hists: + for lbl, key, u, cap in lat + sizes: + if key in hists: + blocks.append(hist_block(c, f"Distribution — {lbl}", hists[key], u, cap=cap)) + if "spec_decode_acceptance_rate" in hists: + blocks.append( + hist_block( + c, + "Distribution — spec-decode acceptance rate", + hists["spec_decode_acceptance_rate"], + "", + ) + ) + + out = [c.head(f"TabbyAPI metrics {c.dim(url)}"), ""] + out.extend(compose(pack(blocks, width, max_cols), width, c)) + return "\n".join(out) + + +def digest(scalars: dict, hists: dict) -> dict: + out = {"scalars": scalars, "histograms": {}} + for name, h in hists.items(): + out["histograms"][name] = { + "count": h.get("count"), + "sum": h.get("sum"), + "mean": mean(h), + "p50": quantile(h, 0.5), + "p90": quantile(h, 0.9), + "p99": quantile(h, 0.99), + } + queries, hits = scalars.get("prefix_cache_queries"), scalars.get("prefix_cache_hits") + out["derived"] = {"prefix_cache_hit_rate": (hits / queries) if queries else None} + return out + + +def main() -> int: + ap = argparse.ArgumentParser(description="Pretty-print TabbyAPI /metrics.") + ap.add_argument("--url", help="full metrics URL (overrides --host/--port)") + ap.add_argument("--host", default="127.0.0.1") + ap.add_argument("--port", type=int, default=5000) + ap.add_argument("--watch", type=float, metavar="SEC", help="refresh every SEC seconds") + ap.add_argument("--hist", action="store_true", help="also draw bucket distributions") + ap.add_argument( + "--columns", + type=int, + default=0, + metavar="N", + help="section columns; 0 auto-fits the terminal (default 0)", + ) + ap.add_argument( + "--width", + type=int, + default=0, + metavar="COLS", + help="assume this terminal width instead of detecting it", + ) + ap.add_argument("--json", action="store_true", help="emit a JSON digest instead") + # Generous by default: a scrape competing with a large prefill has been + # measured waiting 8s, and waiting costs nothing when the server is idle. + ap.add_argument( + "--timeout", type=float, default=30.0, metavar="SEC", help="scrape timeout (default 30)" + ) + ap.add_argument( + "--windows", + type=parse_windows, + default=RATE_WINDOWS_DEFAULT, + metavar="LIST", + help="lookback windows for the --watch rate table, as a " + f"comma-separated duration list (default {RATE_WINDOWS_DEFAULT})", + ) + args = ap.parse_args() + + # argparse leaves a default string untouched, so parse it here when unset. + windows = args.windows + if isinstance(windows, str): + windows = parse_windows(windows) + + # A window shorter than the refresh interval can never be filled, so drop + # it rather than showing a permanently blank column. A window equal to the + # interval is dropped too: it differences against the previous scrape, so it + # only ever duplicates the "now" column. Keep the longest if that would + # empty the table. + if args.watch: + kept = [w for w in windows if w[1] > args.watch] + windows = kept or windows[-1:] + + if hasattr(sys.stdout, "reconfigure"): + # Line buffering too, or --watch writes nothing at all until it is + # killed when its output is a pipe rather than a terminal. + sys.stdout.reconfigure(encoding="utf-8", line_buffering=True) + + url = args.url or f"http://{args.host}:{args.port}/metrics" + c = C(color_on() and not args.json) + + def geometry(): + # Re-read on every frame so --watch reflows when the window is resized. + # Piped output has no size to detect, so fall back to a width that keeps + # two columns viable rather than to the 80 the OS reports. + width = args.width or shutil.get_terminal_size((160, 24)).columns + if args.columns: + return width, args.columns + # Two columns need room for the widest pair plus the gutter. Below that + # the sections stack, which is the old single-column layout. + return width, 2 if width >= 150 else 1 + + # Scrape history, so --watch can show rates over a range of lookbacks. + history: list = [] + + def once(tolerant: bool = False) -> int: + # TabbyAPI renders /metrics on the same event loop that drives the + # generator, and the generator holds the loop through each prefill + # chunk, so a scrape landing mid-inference waits for the chunk to + # finish. Measured on a 27B model at 4096 chunk size: 7ms idle, but a + # median of 6.8s and a peak of 8.1s during a 70k-token prefill. Hence + # the long default timeout, and hence --watch treating a failure as a + # skipped frame rather than a reason to quit. + started = time.monotonic() + try: + text = scrape(url, args.timeout) + except urllib.error.HTTPError as e: + print( + f"error: {url} returned HTTP {e.code}\n" + f"Fix: TabbyAPI serves /metrics only when network.enable_metrics is " + f"true in config.yml.", + file=sys.stderr, + ) + return 2 + except (urllib.error.URLError, OSError) as e: + if tolerant: + waited = time.monotonic() - started + print(c.dim(f" scrape failed after {waited:.1f}s ({e}) — retrying")) + return 0 + print( + f"error: cannot reach {url} ({e})\n" + f"Fix: check TabbyAPI is up — `curl -s {url} | head`", + file=sys.stderr, + ) + return 2 + + scalars, hists = parse(text) + if not scalars and not hists: + print(f"error: no tabbyapi: metrics found at {url}", file=sys.stderr) + return 2 + if args.json: + print(json.dumps(digest(scalars, hists), indent=2)) + else: + # Timestamp on receipt. The server renders its counters immediately + # before responding, so this is when the snapshot was taken, and a + # scrape delayed by a busy loop still yields a correct window. + now = time.monotonic() + width, cols = geometry() + print( + render( + scalars, + hists, + url, + c, + args.hist, + history, + now, + windows, + width=width, + max_cols=cols, + ) + ) + history.append((now, scalars)) + prune_history(history, now, windows[-1][1]) + blocked = now - started + if blocked > 1.0: + print( + c.dim( + f" this scrape waited {blocked:.1f}s for the server to " + f"come up for air (inference holds the event loop)" + ) + ) + return 0 + + if args.watch: + try: + while True: + sys.stdout.write("\033[H\033[J" if c.e else "") + rc = once(tolerant=True) + if rc: + return rc + print(c.dim(f" refreshing every {args.watch:g}s — Ctrl-C to stop")) + time.sleep(args.watch) + except KeyboardInterrupt: + return 0 + return once() + + +if __name__ == "__main__": + sys.exit(main())