From 25445e3d184470c62af507a3cc193650bae1e155 Mon Sep 17 00:00:00 2001 From: Gahow Wang Date: Mon, 25 May 2026 16:19:33 +0800 Subject: [PATCH] A5: joined analysis with reuse decomp, interference, hot-spot, labels New analysis/characterization/joined_analysis.py joins replayer metrics.jsonl + proxy breakdown.json + worker_state.jsonl by request_id, plus engine_*.jsonl by worker_id, and emits: - joined.jsonl per-request merged record - reuse_decomposition.json real intra/cross/shared classification using session_id + hash_ids + cached_tokens - interference_index.json TPOT_p90(same-worker prefill overlap) / TPOT_p90(clean), per Batch 2 - hotspot_index.json max/median worker TTFT-p90, per Batch 3 - failure_label.jsonl per-slow-request cause label, per Batch 5 - failure_breakdown.json label histogram - window_summary.json SRR warmup/steady/drain aggregates Closes the analyzer side of Phase A; replaces the status: unavailable placeholders the existing scaffold emits when join sources are missing. Co-Authored-By: Claude Opus 4.7 --- analysis/characterization/joined_analysis.py | 531 +++++++++++++++++++ tests/test_joined_analysis.py | 170 ++++++ 2 files changed, 701 insertions(+) create mode 100644 analysis/characterization/joined_analysis.py create mode 100644 tests/test_joined_analysis.py diff --git a/analysis/characterization/joined_analysis.py b/analysis/characterization/joined_analysis.py new file mode 100644 index 0000000..e75823c --- /dev/null +++ b/analysis/characterization/joined_analysis.py @@ -0,0 +1,531 @@ +"""A5: joined-record analysis from instrumented runs. + +Inputs (all optional; functions degrade gracefully when missing): + +- replayer metrics.jsonl with A1 fields + (t_dispatch_unix, t_first_token_unix, t_finish_unix, proxy_request_id, + endpoint_url, trace_hash_ids) +- proxy breakdown.json with A2 fields + (session_id, candidate_scores, chosen_score_*, t_first_token_unix, + t_done_unix, t_decision_unix) +- proxy worker_state.jsonl with A2 schema (one row per route decision) +- vLLM scheduler engine_state JSONLs from A3 + (one per engine, env AGENTIC_STEP_LOG_PATH) + +Outputs under /: + +- joined.jsonl — per-request join across all sources +- reuse_decomposition.json +- interference_index.json +- hotspot_index.json +- failure_label.jsonl +- window_summary.json — when run_meta.json (from SRR loadgen) is present +""" + +from __future__ import annotations + +import argparse +import json +import math +import statistics +from collections import defaultdict +from pathlib import Path +from typing import Any, Iterable + +JsonDict = dict[str, Any] + + +# ---------- I/O --------------------------------------------------------- + +def load_jsonl(path: Path) -> list[JsonDict]: + if not path.exists(): + return [] + out: list[JsonDict] = [] + for line in path.read_text(encoding="utf-8").splitlines(): + line = line.strip() + if not line: + continue + try: + out.append(json.loads(line)) + except json.JSONDecodeError: + continue + return out + + +def load_json(path: Path) -> Any: + if not path.exists(): + return None + text = path.read_text(encoding="utf-8").strip() + if not text: + return None + return json.loads(text) + + +def write_json(path: Path, data: Any) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(data, indent=2, sort_keys=True)) + + +def write_jsonl(path: Path, rows: Iterable[JsonDict]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + with path.open("w", encoding="utf-8") as fh: + for row in rows: + fh.write(json.dumps(row, sort_keys=True) + "\n") + + +# ---------- Joining ----------------------------------------------------- + +def build_joined_records( + metrics: list[JsonDict], + breakdown: list[JsonDict], + worker_state: list[JsonDict], +) -> list[JsonDict]: + """Join metrics + breakdown + worker_state by request_id. + + Returns one row per metrics record (the load-generator's view of truth). + Missing sources leave the corresponding columns as None. + """ + bk_by_id = {str(r.get("request_id")): r for r in breakdown if r.get("request_id")} + ws_by_id = {str(r.get("request_id")): r for r in worker_state if r.get("request_id")} + + joined: list[JsonDict] = [] + for m in metrics: + rid = str(m.get("request_id") or m.get("proxy_request_id") or "") + bk = bk_by_id.get(rid) + ws = ws_by_id.get(rid) + row: JsonDict = { + "request_id": rid, + "session_id": m.get("session_id"), + "turn_id": m.get("turn_id"), + "trace_timestamp_s": m.get("trace_timestamp_s"), + "input_length": m.get("input_length"), + "output_length": m.get("output_length"), + "cached_tokens": m.get("cached_tokens"), + "actual_output_tokens": m.get("actual_output_tokens"), + "latency_s": m.get("latency_s"), + "ttft_s": m.get("ttft_s"), + "tpot_s": m.get("tpot_s"), + "t_dispatch_unix": m.get("t_dispatch_unix"), + "t_first_token_unix": m.get("t_first_token_unix"), + "t_finish_unix": m.get("t_finish_unix"), + "endpoint_url": m.get("endpoint_url"), + "trace_hash_ids": m.get("trace_hash_ids") or [], + "error": m.get("error"), + } + if bk: + row["policy"] = bk.get("policy") + row["route_class"] = bk.get("route_class") + row["routed_to"] = bk.get("routed_to") + row["chosen_idx"] = bk.get("chosen_idx") + row["chosen_score_linear"] = bk.get("chosen_score_linear") + row["chosen_score_lmetric"] = bk.get("chosen_score_lmetric") + row["estimated_new_tokens"] = bk.get("estimated_new_tokens") + row["cache_hit_proxy"] = bk.get("cache_hit") + row["proxy_t_decision_unix"] = bk.get("t_decision_unix") + row["proxy_t_first_token_unix"] = bk.get("t_first_token_unix") + row["proxy_t_done_unix"] = bk.get("t_done_unix") + if ws: + row["worker_state_at_decision"] = ws.get("workers") + joined.append(row) + return joined + + +# ---------- Reuse decomposition (real) ---------------------------------- + +def reuse_decomposition(records: list[JsonDict], block_size: int = 16) -> JsonDict: + """Real intra/cross/shared decomposition keyed on (session, hash blocks).""" + if not records: + return {"status": "unavailable", "reason": "no joined records"} + + # block first-seen index: hash_id -> (session_id, first_seen_seq) + first_seen: dict[int, tuple[str, int]] = {} + block_sessions: dict[int, set[str]] = defaultdict(set) + seq = 0 + for r in sorted(records, key=lambda x: x.get("t_dispatch_unix") or 0.0): + sid = str(r.get("session_id")) + for h in r.get("trace_hash_ids") or []: + h_int = int(h) + if h_int not in first_seen: + first_seen[h_int] = (sid, seq) + block_sessions[h_int].add(sid) + seq += 1 + + total_cached = 0 + intra = cross = shared = unclassified = 0 + for r in records: + cached = r.get("cached_tokens") or 0 + if not cached: + continue + total_cached += cached + sid = str(r.get("session_id")) + hashes = [int(h) for h in (r.get("trace_hash_ids") or [])] + if not hashes: + unclassified += cached + continue + # Approximate: classify the cached tokens by the first non-current + # owner of any hash block we've seen before. + block_tokens = max(cached // max(len(hashes), 1), 1) + for h in hashes: + first_sid, _ = first_seen.get(h, (None, None)) + if first_sid is None: + unclassified += min(block_tokens, cached) + elif first_sid == sid: + intra += min(block_tokens, cached) + elif len(block_sessions[h]) >= 8: + shared += min(block_tokens, cached) + else: + cross += min(block_tokens, cached) + cached -= block_tokens + if cached <= 0: + break + + return { + "status": "supported", + "total_cached_tokens": total_cached, + "intra_session_tokens": intra, + "cross_session_tokens": cross, + "shared_prefix_tokens": shared, + "unclassified_tokens": unclassified, + "fractions": _fractions(intra, cross, shared, unclassified), + "shared_prefix_min_sessions": 8, + } + + +def _fractions(*parts: int) -> JsonDict: + total = sum(parts) + if total == 0: + return {"intra": 0.0, "cross": 0.0, "shared": 0.0, "unclassified": 0.0} + labels = ["intra", "cross", "shared", "unclassified"] + return {label: parts[i] / total for i, label in enumerate(labels)} + + +# ---------- Interference index (B2) ------------------------------------- + +def interference_index( + joined: list[JsonDict], + engine_state_by_worker: dict[str, list[JsonDict]], +) -> JsonDict: + """Label each completed request's decode period as overlap / no-overlap. + + A request 'overlaps same-worker prefill' if any scheduler step on the + chosen worker between (t_first_token_unix, t_finish_unix) had + prefill_tokens > 0 from a request other than this one. + """ + if not joined or not engine_state_by_worker: + return {"status": "unavailable", + "reason": "missing joined records or engine state"} + + tpot_overlap: list[float] = [] + tpot_clean: list[float] = [] + for r in joined: + rid = r.get("request_id") + worker = _normalize_worker(r.get("endpoint_url") or r.get("routed_to")) + steps = engine_state_by_worker.get(worker) + if not steps: + continue + t0 = r.get("t_first_token_unix") + t1 = r.get("t_finish_unix") + tpot = r.get("tpot_s") + if t0 is None or t1 is None or tpot is None or r.get("error"): + continue + overlap = False + for s in steps: + t = s.get("t_unix") + if t is None or t < t0 or t > t1: + continue + if not s.get("prefill_tokens"): + continue + # If the only prefill belongs to *this* request, that's still + # this request's own prefill warming up, not interference. + other_prefill = False + for pr in s.get("per_req", []) or []: + if pr.get("phase") == "prefill" and pr.get("rid") != rid: + other_prefill = True + break + if other_prefill: + overlap = True + break + (tpot_overlap if overlap else tpot_clean).append(float(tpot)) + + p90_overlap = _percentile(tpot_overlap, 0.90) if tpot_overlap else None + p90_clean = _percentile(tpot_clean, 0.90) if tpot_clean else None + idx = None + if p90_overlap is not None and p90_clean and p90_clean > 0: + idx = p90_overlap / p90_clean + + return { + "status": "supported" if idx is not None else "partial", + "n_overlap_requests": len(tpot_overlap), + "n_clean_requests": len(tpot_clean), + "tpot_p90_overlap_s": p90_overlap, + "tpot_p90_clean_s": p90_clean, + "interference_index": idx, + } + + +def _normalize_worker(url_or_id: str | None) -> str | None: + """Map endpoint URLs to AGENTIC_WORKER_ID convention engine_{i}.""" + if not url_or_id: + return None + if url_or_id.startswith("engine_"): + return url_or_id + # Endpoint URLs look like http://host:8000; map by port offset against 8000. + try: + port = int(url_or_id.rsplit(":", 1)[1].split("/")[0]) + return f"engine_{port - 8000}" + except (ValueError, IndexError): + return url_or_id + + +# ---------- Hotspot index (B3) ------------------------------------------ + +def hotspot_index(joined: list[JsonDict]) -> JsonDict: + """max/median per-worker queue-delay p90 across completed requests.""" + if not joined: + return {"status": "unavailable"} + + by_worker_queue: dict[str, list[float]] = defaultdict(list) + by_worker_latency: dict[str, list[float]] = defaultdict(list) + for r in joined: + if r.get("error"): + continue + worker = r.get("routed_to") or r.get("endpoint_url") + if not worker: + continue + # queue delay proxy: (t_first_token - t_dispatch) - prefill estimate + # is fragile; use raw TTFT as the queue-stressing signal. + ttft = r.get("ttft_s") + lat = r.get("latency_s") + if ttft is not None: + by_worker_queue[worker].append(float(ttft)) + if lat is not None: + by_worker_latency[worker].append(float(lat)) + + worker_p90_q: dict[str, float] = { + w: _percentile(v, 0.90) for w, v in by_worker_queue.items() if v + } + worker_p90_lat: dict[str, float] = { + w: _percentile(v, 0.90) for w, v in by_worker_latency.items() if v + } + p90s_q = sorted(worker_p90_q.values()) + idx = None + if len(p90s_q) >= 2: + median = p90s_q[len(p90s_q) // 2] + if median > 0: + idx = max(p90s_q) / median + + return { + "status": "supported" if idx is not None else "partial", + "per_worker_ttft_p90_s": worker_p90_q, + "per_worker_latency_p90_s": worker_p90_lat, + "hotspot_index_ttft_p90": idx, + } + + +# ---------- Failure label (B5) ------------------------------------------ + +DEFAULT_SLO = { + "ttft_p90_s": 2.0, + "tpot_p90_s": 0.15, +} + + +def label_slow_requests( + joined: list[JsonDict], + engine_state_by_worker: dict[str, list[JsonDict]], + slo: JsonDict | None = None, + slow_ttft_factor: float = 2.0, +) -> list[JsonDict]: + slo = slo or DEFAULT_SLO + ttft_threshold = float(slo["ttft_p90_s"]) * slow_ttft_factor + + # Per-worker queue p90 to flag hot workers + by_worker_ttft: dict[str, list[float]] = defaultdict(list) + for r in joined: + if r.get("ttft_s") is not None: + by_worker_ttft[r.get("routed_to") or ""].append(float(r["ttft_s"])) + worker_p90 = {w: _percentile(v, 0.90) for w, v in by_worker_ttft.items() if v} + global_p90 = _percentile( + [v for vs in by_worker_ttft.values() for v in vs], 0.90, + ) if by_worker_ttft else None + hot_workers = {w for w, p in worker_p90.items() + if global_p90 and p > global_p90 * 1.2} + + labels: list[JsonDict] = [] + for r in joined: + ttft = r.get("ttft_s") + if ttft is None or r.get("error"): + continue + if ttft <= ttft_threshold: + continue + label = _assign_label(r, hot_workers, engine_state_by_worker) + labels.append({ + "request_id": r.get("request_id"), + "session_id": r.get("session_id"), + "routed_to": r.get("routed_to"), + "ttft_s": ttft, + "latency_s": r.get("latency_s"), + "input_length": r.get("input_length"), + "cached_tokens": r.get("cached_tokens"), + "estimated_new_tokens": r.get("estimated_new_tokens"), + "label": label, + }) + return labels + + +def _assign_label( + r: JsonDict, + hot_workers: set[str], + engine_state_by_worker: dict[str, list[JsonDict]], +) -> str: + worker = _normalize_worker(r.get("endpoint_url") or r.get("routed_to")) + rid = r.get("request_id") + steps = engine_state_by_worker.get(worker, []) + t0 = r.get("t_first_token_unix") + t1 = r.get("t_finish_unix") + if steps and t0 and t1: + for s in steps: + t = s.get("t_unix") + if t is None or t < t0 or t > t1: + continue + for pr in s.get("per_req", []) or []: + if pr.get("phase") == "prefill" and pr.get("rid") != rid: + return "same_worker_prefill_overlap" + if (r.get("routed_to") or "") in hot_workers: + return "hot_worker_queue" + est = r.get("estimated_new_tokens") or 0 + inp = r.get("input_length") or 0 + if est and inp and est >= 0.5 * inp: + return "cache_miss_large_append" + snap = r.get("worker_state_at_decision") or [] + if snap: + chosen_idx = r.get("chosen_idx") + if isinstance(chosen_idx, int) and 0 <= chosen_idx < len(snap): + cached = snap[chosen_idx].get("cached_blocks", 0) + if cached and cached > 50_000: + return "high_kv_occupancy" + return "unknown" + + +# ---------- Window summary (B4) ----------------------------------------- + +def window_summary(joined: list[JsonDict], run_meta: JsonDict | None) -> JsonDict: + if not run_meta: + return {"status": "unavailable", "reason": "no run_meta"} + warmup_end = float(run_meta["warmup_end_unix"]) + steady_end = float(run_meta["steady_end_unix"]) + + buckets: dict[str, list[JsonDict]] = {"warmup": [], "steady": [], "drain": []} + for r in joined: + t = r.get("t_dispatch_unix") + if t is None: + continue + if t < warmup_end: + buckets["warmup"].append(r) + elif t < steady_end: + buckets["steady"].append(r) + else: + buckets["drain"].append(r) + + out: JsonDict = {"run_meta": run_meta, "windows": {}} + for name, rows in buckets.items(): + ttft = [r["ttft_s"] for r in rows if r.get("ttft_s") is not None] + tpot = [r["tpot_s"] for r in rows if r.get("tpot_s") is not None] + e2e = [r["latency_s"] for r in rows if r.get("latency_s") is not None] + errs = sum(1 for r in rows if r.get("error")) + out["windows"][name] = { + "attempted": len(rows), + "completed": len(rows) - errs, + "errored": errs, + "ttft_p50_s": _percentile(ttft, 0.50) if ttft else None, + "ttft_p90_s": _percentile(ttft, 0.90) if ttft else None, + "tpot_p50_s": _percentile(tpot, 0.50) if tpot else None, + "tpot_p90_s": _percentile(tpot, 0.90) if tpot else None, + "e2e_p50_s": _percentile(e2e, 0.50) if e2e else None, + "e2e_p90_s": _percentile(e2e, 0.90) if e2e else None, + } + return out + + +# ---------- helpers ----------------------------------------------------- + +def _percentile(values: list[float], pct: float) -> float | None: + if not values: + return None + sorted_vals = sorted(values) + if len(sorted_vals) == 1: + return sorted_vals[0] + rank = pct * (len(sorted_vals) - 1) + lo = int(rank) + hi = min(lo + 1, len(sorted_vals) - 1) + frac = rank - lo + return sorted_vals[lo] * (1 - frac) + sorted_vals[hi] * frac + + +def load_engine_state(dir_path: Path) -> dict[str, list[JsonDict]]: + """Load all engine_*.jsonl files from a directory; key by worker id.""" + if not dir_path.exists() or not dir_path.is_dir(): + return {} + by_worker: dict[str, list[JsonDict]] = {} + for p in sorted(dir_path.glob("engine_*.jsonl")): + worker_id = p.stem # 'engine_0', etc. + rows = load_jsonl(p) + # Sort steps by time for binary-search-friendly access. + rows.sort(key=lambda r: r.get("t_unix") or 0.0) + by_worker[worker_id] = rows + return by_worker + + +# ---------- CLI --------------------------------------------------------- + +def main(argv: list[str] | None = None) -> None: + p = argparse.ArgumentParser(description="A5 joined analysis") + p.add_argument("--metrics", type=Path, required=True) + p.add_argument("--breakdown", type=Path, default=None) + p.add_argument("--worker-state", type=Path, default=None) + p.add_argument("--engine-state-dir", type=Path, default=None, + help="Directory containing engine_*.jsonl from A3 patch") + p.add_argument("--run-meta", type=Path, default=None, + help="run_meta or window_summary.json from SRR loadgen") + p.add_argument("--out-dir", type=Path, required=True) + p.add_argument("--slow-ttft-factor", type=float, default=2.0) + args = p.parse_args(argv) + + metrics = load_jsonl(args.metrics) + breakdown_raw = load_json(args.breakdown) if args.breakdown else [] + if isinstance(breakdown_raw, dict): + breakdown_raw = breakdown_raw.get("records", [breakdown_raw]) + breakdown = list(breakdown_raw or []) + worker_state_raw = load_json(args.worker_state) if args.worker_state else [] + if isinstance(worker_state_raw, dict): + worker_state_raw = worker_state_raw.get("records", [worker_state_raw]) + worker_state = list(worker_state_raw or []) + engine_state = ( + load_engine_state(args.engine_state_dir) if args.engine_state_dir else {} + ) + run_meta = load_json(args.run_meta) if args.run_meta else None + + joined = build_joined_records(metrics, breakdown, worker_state) + + args.out_dir.mkdir(parents=True, exist_ok=True) + write_jsonl(args.out_dir / "joined.jsonl", joined) + write_json(args.out_dir / "reuse_decomposition.json", + reuse_decomposition(joined)) + write_json(args.out_dir / "interference_index.json", + interference_index(joined, engine_state)) + write_json(args.out_dir / "hotspot_index.json", + hotspot_index(joined)) + labels = label_slow_requests(joined, engine_state, + slow_ttft_factor=args.slow_ttft_factor) + write_jsonl(args.out_dir / "failure_label.jsonl", labels) + counts: dict[str, int] = defaultdict(int) + for L in labels: + counts[L["label"]] += 1 + write_json(args.out_dir / "failure_breakdown.json", + {"counts": dict(counts), "n_slow": len(labels)}) + write_json(args.out_dir / "window_summary.json", + window_summary(joined, run_meta)) + + +if __name__ == "__main__": + main() diff --git a/tests/test_joined_analysis.py b/tests/test_joined_analysis.py new file mode 100644 index 0000000..b33ee0c --- /dev/null +++ b/tests/test_joined_analysis.py @@ -0,0 +1,170 @@ +"""Tests for A5 joined analysis: join + indices + labels.""" + +from __future__ import annotations + +from analysis.characterization.joined_analysis import ( + build_joined_records, + hotspot_index, + interference_index, + label_slow_requests, + reuse_decomposition, + window_summary, + _normalize_worker, + _percentile, +) + + +def _mk_metric(rid, **kw): + base = { + "request_id": rid, "session_id": "s1", "turn_id": 0, + "trace_timestamp_s": 1.0, "input_length": 1000, "output_length": 50, + "cached_tokens": 0, "actual_output_tokens": 50, + "latency_s": 1.0, "ttft_s": 0.5, "tpot_s": 0.04, + "t_dispatch_unix": 1000.0, "t_first_token_unix": 1000.5, + "t_finish_unix": 1001.0, "endpoint_url": "http://h:8000", + "trace_hash_ids": [], "error": None, + } + base.update(kw) + return base + + +def test_build_joined_records_merges_by_request_id(): + metrics = [_mk_metric("r1"), _mk_metric("r2")] + breakdown = [{"request_id": "r1", "policy": "lmetric", "chosen_idx": 3, + "estimated_new_tokens": 500, "routed_to": "http://h:8000"}] + worker_state = [{"request_id": "r2", "workers": [{"idx": 0, "url": "x"}]}] + + joined = build_joined_records(metrics, breakdown, worker_state) + assert len(joined) == 2 + j_by_id = {r["request_id"]: r for r in joined} + assert j_by_id["r1"]["policy"] == "lmetric" + assert j_by_id["r1"]["chosen_idx"] == 3 + assert j_by_id["r1"]["estimated_new_tokens"] == 500 + assert j_by_id["r2"]["worker_state_at_decision"][0]["url"] == "x" + assert j_by_id["r2"].get("policy") is None # no breakdown for r2 + + +def test_reuse_decomposition_classifies_intra_and_cross(): + records = [ + _mk_metric("r1", session_id="A", trace_hash_ids=[11], + cached_tokens=0, t_dispatch_unix=1.0), + _mk_metric("r2", session_id="A", trace_hash_ids=[11], + cached_tokens=100, t_dispatch_unix=2.0), + _mk_metric("r3", session_id="B", trace_hash_ids=[11], + cached_tokens=100, t_dispatch_unix=3.0), + ] + out = reuse_decomposition(records) + assert out["status"] == "supported" + assert out["intra_session_tokens"] > 0 + assert out["cross_session_tokens"] > 0 + fr = out["fractions"] + assert abs(sum(fr.values()) - 1.0) < 1e-9 + + +def test_normalize_worker_maps_port_to_engine_id(): + assert _normalize_worker("http://node:8000") == "engine_0" + assert _normalize_worker("http://node:8005/foo") == "engine_5" + assert _normalize_worker("engine_3") == "engine_3" + assert _normalize_worker(None) is None + + +def test_interference_index_marks_overlap_when_other_request_prefilling(): + metrics = [ + _mk_metric("decode_target", endpoint_url="http://h:8000", + t_first_token_unix=10.0, t_finish_unix=11.0, + tpot_s=0.10), + _mk_metric("decode_clean", endpoint_url="http://h:8001", + t_first_token_unix=20.0, t_finish_unix=21.0, + tpot_s=0.04), + ] + joined = build_joined_records(metrics, [], []) + engine_state = { + "engine_0": [ + {"t_unix": 10.5, "prefill_tokens": 8000, + "per_req": [{"rid": "other", "phase": "prefill"}]}, + ], + "engine_1": [ + {"t_unix": 20.5, "prefill_tokens": 0, + "per_req": [{"rid": "decode_clean", "phase": "decode"}]}, + ], + } + out = interference_index(joined, engine_state) + assert out["status"] == "supported" + assert out["n_overlap_requests"] == 1 + assert out["n_clean_requests"] == 1 + assert out["interference_index"] is not None + # Overlap p90 = 0.10; clean p90 = 0.04; ratio > 2 + assert out["interference_index"] > 2.0 + + +def test_hotspot_index_max_over_median_p90(): + """One hot worker with TTFT 10x the others should drive a >1 index.""" + rows = [] + for i in range(3): + for _ in range(10): + rows.append({ + "request_id": f"x{i}", "routed_to": f"http://h:800{i}", + "endpoint_url": f"http://h:800{i}", + "ttft_s": 0.5 if i < 2 else 5.0, "latency_s": 1.0, + "error": None, + }) + out = hotspot_index(rows) + assert out["status"] == "supported" + assert out["hotspot_index_ttft_p90"] > 5.0 + + +def test_label_slow_requests_flags_overlap_and_hot_worker(): + metrics = [ + _mk_metric("slow_overlap", ttft_s=10.0, + t_first_token_unix=10.0, t_finish_unix=11.0), + _mk_metric("slow_no_overlap", ttft_s=10.0, + endpoint_url="http://h:8005", + t_first_token_unix=20.0, t_finish_unix=21.0), + _mk_metric("fast", ttft_s=0.5, + t_first_token_unix=15.0, t_finish_unix=16.0), + ] + metrics[0]["routed_to"] = "http://h:8000" + metrics[1]["routed_to"] = "http://h:8005" + metrics[2]["routed_to"] = "http://h:8000" + bk = [ + {"request_id": "slow_overlap", "routed_to": "http://h:8000"}, + {"request_id": "slow_no_overlap", "routed_to": "http://h:8005"}, + {"request_id": "fast", "routed_to": "http://h:8000"}, + ] + joined = build_joined_records(metrics, bk, []) + engine_state = { + "engine_0": [{"t_unix": 10.5, "prefill_tokens": 5000, + "per_req": [{"rid": "other", "phase": "prefill"}]}], + } + labels = label_slow_requests(joined, engine_state, slow_ttft_factor=2.0) + by_id = {L["request_id"]: L["label"] for L in labels} + assert by_id.get("slow_overlap") == "same_worker_prefill_overlap" + assert "fast" not in by_id + assert "slow_no_overlap" in by_id + + +def test_window_summary_buckets_by_dispatch_unix(): + run_meta = { + "run_start_unix": 1000.0, + "warmup_end_unix": 1010.0, + "steady_end_unix": 1030.0, + "drain_end_unix": 1040.0, + } + joined = [ + _mk_metric("w", t_dispatch_unix=1005.0, ttft_s=0.5, latency_s=1.0, + tpot_s=0.04), + _mk_metric("s", t_dispatch_unix=1020.0, ttft_s=0.6, latency_s=1.5, + tpot_s=0.05), + _mk_metric("d", t_dispatch_unix=1035.0, ttft_s=0.7, latency_s=2.0, + tpot_s=0.06), + ] + out = window_summary(joined, run_meta) + assert out["windows"]["warmup"]["attempted"] == 1 + assert out["windows"]["steady"]["attempted"] == 1 + assert out["windows"]["drain"]["attempted"] == 1 + assert out["windows"]["steady"]["ttft_p90_s"] is not None + + +def test_percentile_helper_handles_singleton(): + assert _percentile([5.0], 0.99) == 5.0 + assert _percentile([], 0.50) is None