#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ gpu-analyzer.py -- Pod Efficiency Analyzer, local system probe BY: Arthur Rasmusson -- open.pod-efficiency.tools Inspects the system it runs on (CPU, RAM, NVIDIA GPUs, NFS mounts, disk), estimates the inference-serving uplift available with Inferra by Lightbits Labs (KV-cache offload to fast NVMe), and uploads the statistics to https://api.pod-efficiency.tools/api/v1/stats to enrich your PDF report. Estimates are derived from Lightbits Labs' published LightInferra benchmarks (4x NVIDIA L40S, ScaleFlux NVMe): https://lightbitslabs.com/blog/introducing-lightinferra-280x-improved-ai-token-economy-by-lightbits-labs/ Usage: python3 gpu-analyzer.py # analyze + upload python3 gpu-analyzer.py --no-upload # analyze only, nothing leaves this box Only Python 3 standard library is used. Nothing else is read or sent. """ import argparse import hashlib import json import math import os import platform import shutil import socket import subprocess import zipfile import subprocess import sys import urllib.parse import urllib.request API_URL = "https://api.pod-efficiency.tools/api/v1/stats" BENCH_SOURCE = ("https://lightbitslabs.com/blog/introducing-lightinferra-" "280x-improved-ai-token-economy-by-lightbits-labs/") FARMGPU_SOURCE = "https://blog.farmgpu.com/kv-cache-lightbits-scaleflux/" # Published turn-2 measurements, 4x L40S rig (per-GPU figures = rig / 4). RIG_GPUS = 4 BENCHMARKS = [ {"model": "Qwen2.5-7B-Instruct-1M", "ctx": 100000, "tps_base": 5.0, "tps_inf": 95.0, "ttft_base_s": None, "ttft_inf_s": None}, {"model": "DeepSeek-R1-Distill-Llama-70B-FP8", "ctx": 131000, "tps_base": 1.7, "tps_inf": 18.7, "ttft_base_s": 70.8, "ttft_inf_s": 0.465}, {"model": "Llama-4-Scout-17B-16E-FP8", "ctx": 400000, "tps_base": None, "tps_inf": None, "ttft_base_s": 103.0, "ttft_inf_s": 0.457}, {"model": "Qwen2.5-7B-Instruct-1M", "ctx": 1010000, "tps_base": 0.3, "tps_inf": 27.2, "ttft_base_s": 372.3, "ttft_inf_s": 1.3}, ] REUSE_SHARE = 0.70 # assumed share of multi-turn requests hitting reusable KV cache class C: PLUM = "\033[35m" GOLD = "\033[33m" GREEN = "\033[32m" RED = "\033[31m" BOLD = "\033[1m" DIM = "\033[2m" END = "\033[0m" @classmethod def off(cls): for name in ("PLUM", "GOLD", "GREEN", "RED", "BOLD", "DIM", "END"): setattr(cls, name, "") def run(cmd): try: out = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, timeout=20) return out.stdout.decode(errors="replace").strip() except Exception: return "" def collect_cpu(): model, cores = "unknown", os.cpu_count() or 0 try: with open("/proc/cpuinfo") as f: for line in f: if line.lower().startswith("model name"): model = line.split(":", 1)[1].strip() break except OSError: pass return {"model": model, "logical_cores": cores} def collect_mem_gib(): try: with open("/proc/meminfo") as f: for line in f: if line.startswith("MemTotal:"): return round(int(line.split()[1]) / 1048576.0, 1) except OSError: pass return None def collect_gpus(): if not shutil.which("nvidia-smi"): return [] out = run(["nvidia-smi", "--query-gpu=name,memory.total,utilization.gpu,driver_version", "--format=csv,noheader,nounits"]) gpus = [] for line in out.splitlines(): parts = [p.strip() for p in line.split(",")] if len(parts) >= 4: gpus.append({"name": parts[0], "memory_mib": _to_num(parts[1]), "utilization_pct": _to_num(parts[2]), "driver": parts[3]}) return gpus def _to_num(s): try: return float(s) except ValueError: return None def collect_nfs(): mounts = [] try: with open("/proc/mounts") as f: for line in f: dev, mnt, fstype = line.split()[:3] if fstype.startswith("nfs"): mounts.append({"device": dev, "mountpoint": mnt, "type": fstype}) except OSError: pass return mounts def collect_disk(): try: du = shutil.disk_usage("/") return {"total_gib": round(du.total / 2**30, 1), "free_gib": round(du.free / 2**30, 1)} except OSError: return None def capacity_multiplier(k, s=REUSE_SHARE): return 1.0 / ((1.0 - s) + s / k) if k > 1 else 1.0 def fmt_ttft(v): if v is None: return "--" if v >= 60: return "%.1f min" % (v / 60) if v >= 1: return "%.1f s" % v return "%d ms" % round(v * 1000) def print_report(info): p = print w = 78 p("") p(C.PLUM + C.BOLD + "=" * w + C.END) p(C.PLUM + C.BOLD + " POD EFFICIENCY ANALYZER".ljust(w - 24) + "by Arthur Rasmusson " + C.END) p(C.DIM + " Inferra by Lightbits Labs -- KV-cache offload uplift estimate" + C.END) p(C.PLUM + C.BOLD + "=" * w + C.END) p(C.BOLD + "\n Your system" + C.END) p(" Host : %s (%s %s)" % (info["hostname"], info["os"], info["kernel"])) p(" CPU : %s (%s logical cores)" % (info["cpu"]["model"], info["cpu"]["logical_cores"])) p(" Memory : %s GiB" % info["memory_gib"]) if info["disk"]: p(" Disk (/) : %s GiB total, %s GiB free" % (info["disk"]["total_gib"], info["disk"]["free_gib"])) if info["gpus"]: for i, g in enumerate(info["gpus"]): p(" GPU %-6s : %s, %.0f MiB, util %.0f%%, driver %s" % ( "#%d" % i, g["name"], g["memory_mib"] or 0, g["utilization_pct"] or 0, g["driver"])) else: p(" GPU : " + C.GOLD + "none detected (nvidia-smi not found or no devices)" + C.END) if info["nfs_mounts"]: for m in info["nfs_mounts"]: p(" NFS : %s on %s (%s)" % (m["device"], m["mountpoint"], m["type"])) else: p(" NFS : no NFS mounts") ngpus = max(len(info["gpus"]), 1) unit = "detected GPU(s)" if info["gpus"] else "GPU (hypothetical)" p(C.BOLD + "\n Estimated multi-turn serving uplift with Inferra on your %d %s" % (ngpus, unit) + C.END) p(C.DIM + " Assumes %d%% of requests reuse KV cache; M = 1/((1-s)+s/k) applied to" % (REUSE_SHARE * 100) + " published per-hit speedups." + C.END) p("") hdr = " %-38s %10s %12s %12s" % ("Workload (published benchmark)", "Speedup", "tok/s now", "with Inferra") p(C.PLUM + hdr + C.END) p(" " + "-" * (w - 8)) for b in info["uplift"]["workloads"]: p(" %-38s %9.0fx %12.1f %12.1f" % ( "%s @ %dK ctx" % (b["model"].split("-Instruct")[0][:26], b["ctx"] // 1000), b["effective_multiplier"], b["fleet_tps_base"], b["fleet_tps_inf"])) p("") for b in info["uplift"]["ttft"]: p(" TTFT @ %4dK ctx : %s -> %s (%dx faster)" % ( b["ctx"] // 1000, fmt_ttft(b["ttft_base_s"]), fmt_ttft(b["ttft_inf_s"]), round(b["ttft_base_s"] / b["ttft_inf_s"]))) p(" TTFT @ 10M ctx : 1,154x faster on Llama-4-Scout (FarmGPU study), plus") p(" 3x more requests/GPU and 65% lower infrastructure cost.") m100k = info["uplift"]["workloads"][0]["effective_multiplier"] p(C.GREEN + C.BOLD + "\n Bottom line" + C.END) p(" The same %d GPU(s) could serve ~%.1fx more multi-turn inference at 100K" % (ngpus, m100k)) p(" context, or today's load could run on ~%d GPU(s) instead of %d." % ( max(math.ceil(ngpus / m100k), 1), ngpus)) p(" Interactive context extends to 1M+ tokens (sub-1.5 s turn-2 TTFT).") p(C.DIM + "\n Sources: %s" % BENCH_SOURCE + C.END) p(C.DIM + " %s" % FARMGPU_SOURCE + C.END) p(C.DIM + " Estimates only -- validate with a proof of concept on your workload." + C.END) p(C.PLUM + C.BOLD + "=" * w + C.END) def build_payload(): gpus = collect_gpus() ngpus = max(len(gpus), 1) workloads, ttft = [], [] for b in BENCHMARKS: if b["tps_base"]: k = b["tps_inf"] / b["tps_base"] m = capacity_multiplier(k) per_gpu = b["tps_base"] / RIG_GPUS workloads.append({ "model": b["model"], "ctx": b["ctx"], "published_speedup": round(k, 1), "effective_multiplier": round(m, 1), "fleet_tps_base": round(per_gpu * ngpus, 1), "fleet_tps_inf": round(per_gpu * ngpus * m, 1), }) if b["ttft_base_s"]: ttft.append({"ctx": b["ctx"], "ttft_base_s": b["ttft_base_s"], "ttft_inf_s": b["ttft_inf_s"]}) return { "analyzer_version": "0.4.0", "hostname": socket.gethostname(), "os": "%s %s" % (platform.system(), platform.release()), "kernel": platform.version(), "python": platform.python_version(), "cpu": collect_cpu(), "memory_gib": collect_mem_gib(), "disk": collect_disk(), "gpus": gpus, "nfs_mounts": collect_nfs(), "uplift": {"reuse_share": REUSE_SHARE, "workloads": workloads, "ttft": ttft, "source": BENCH_SOURCE}, } # ================= .pop generation ================= # A Performance Optimization Package: the run, the machine it ran on, and the raw # harness output, in one container with a digest over all of it. # # Signing shells out to gpg. The probe stays stdlib-only on purpose (people pipe # it into a shell and should be able to read it first), and stdlib has no # asymmetric crypto, so gpg is the honest option. Without gpg the package is # still written and still carries its digest, marked unsigned. An unsigned real # measurement beats no measurement. POP_KEY_UID = "pod-efficiency-analyzer (.pop signing) " def _cmd(args, stdin=None): try: p = subprocess.run(args, input=stdin, capture_output=True, timeout=60) return p.returncode, p.stdout, p.stderr except Exception: return 1, b"", b"" def _read(path, limit=4000): try: with open(path, "r", errors="replace") as f: return f.read(limit).strip() except Exception: return None def collect_state(): """Everything the browser cannot see: firmware, kernel, storage, drivers.""" def dmi(name): return _read("/sys/class/dmi/id/" + name, 200) nvme = [] try: for d in sorted(os.listdir("/sys/class/nvme")): nvme.append({ "device": d, "model": _read("/sys/class/nvme/%s/model" % d, 120), "firmware": _read("/sys/class/nvme/%s/firmware_rev" % d, 60), "serial_present": bool(_read("/sys/class/nvme/%s/serial" % d, 60)), }) except Exception: pass mounts = [] try: with open("/proc/mounts") as f: for line in f: p = line.split() if len(p) >= 3 and p[2] in ("ext4", "xfs", "btrfs", "nfs", "nfs4", "zfs"): mounts.append({"target": p[1], "fstype": p[2]}) except Exception: pass rc, out, _ = _cmd(["nvidia-smi", "--query-gpu=driver_version,vbios_version", "--format=csv,noheader"]) driver = out.decode(errors="replace").strip().splitlines()[:1] if rc == 0 else [] return { "kernel": platform.release(), "kernel_full": platform.version(), "os": _read("/etc/os-release", 600), "firmware": { "bios_vendor": dmi("bios_vendor"), "bios_version": dmi("bios_version"), "bios_date": dmi("bios_date"), "board": dmi("board_name"), "product": dmi("product_name"), "uefi": os.path.isdir("/sys/firmware/efi"), }, "nvidia": {"driver_vbios": driver}, "nvme": nvme, "filesystems": mounts[:40], "cpu_governor": _read("/sys/devices/system/cpu/cpu0/cpufreq/scaling_governor", 40), "thp": _read("/sys/kernel/mm/transparent_hugepage/enabled", 80), } def _http(url, timeout=2.0): try: with urllib.request.urlopen(url, timeout=timeout) as r: return r.read().decode("utf-8", "replace") except Exception: return None SERVING_PORTS = (8000, 8080, 30000, 8001, 9000) # The Prometheus names that carry what the model actually needs: traffic volume, # cache occupancy and prefix reuse. vLLM and SGLang both export in this shape. METRIC_KEYS = ( "vllm:prompt_tokens_total", "vllm:generation_tokens_total", "vllm:gpu_cache_usage_perc", "vllm:cpu_cache_usage_perc", "vllm:gpu_prefix_cache_hit_rate", "vllm:gpu_prefix_cache_hits_total", "vllm:gpu_prefix_cache_queries_total", "vllm:num_requests_running", "vllm:num_requests_waiting", "vllm:num_preemptions_total", "vllm:request_success_total", "sglang:prompt_tokens_total", "sglang:generation_tokens_total", "sglang:cache_hit_rate", "sglang:num_running_reqs", "sglang:token_usage", ) def _parse_prom(text): out = {} for line in text.splitlines(): if not line or line[0] == "#": continue name = line.split("{")[0].split(" ")[0] if name not in METRIC_KEYS: continue try: out.setdefault(name, 0.0) out[name] += float(line.rsplit(" ", 1)[1]) except Exception: pass return out def collect_serving(): """A running vLLM or SGLang is the only place the real traffic numbers live: token totals, cache occupancy, prefix hits. Everything else is a guess.""" for port in SERVING_PORTS: base = "http://127.0.0.1:%d" % port text = _http(base + "/metrics") if not text or ("vllm:" not in text and "sglang:" not in text): continue stack = "sglang" if "sglang:" in text else "vllm" m = _parse_prom(text) models = _http(base + "/v1/models") model_id = None try: model_id = (json.loads(models)["data"][0]["id"]) if models else None except Exception: pass hits = m.get("vllm:gpu_prefix_cache_hits_total") qs = m.get("vllm:gpu_prefix_cache_queries_total") derived = (hits / qs) if (hits and qs) else m.get("vllm:gpu_prefix_cache_hit_rate") if derived is None: derived = m.get("sglang:cache_hit_rate") pt = m.get("vllm:prompt_tokens_total") or m.get("sglang:prompt_tokens_total") gt = m.get("vllm:generation_tokens_total") or m.get("sglang:generation_tokens_total") return { "detected": True, "stack": stack, "port": port, "model": model_id, "prompt_tokens_total": pt, "generation_tokens_total": gt, "input_output_ratio": (pt / gt) if (pt and gt) else None, "prefix_cache_hit_rate": derived, "gpu_cache_usage": m.get("vllm:gpu_cache_usage_perc"), "cpu_cache_usage": m.get("vllm:cpu_cache_usage_perc"), "requests_running": m.get("vllm:num_requests_running") or m.get("sglang:num_running_reqs"), "requests_waiting": m.get("vllm:num_requests_waiting"), "preemptions_total": m.get("vllm:num_preemptions_total"), "raw": m, } return {"detected": False, "note": "No vLLM or SGLang metrics endpoint answered on %s. Traffic volume, " "cache occupancy and prefix reuse are unavailable, so the model falls " "back to estimating them." % (", ".join(str(p) for p in SERVING_PORTS))} def collect_faults(): """Page faults and swap pressure: the cheapest signal that a cache tier is thrashing rather than serving.""" out = {} try: with open("/proc/vmstat") as f: for line in f: k, _, v = line.partition(" ") if k in ("pgfault", "pgmajfault", "pswpin", "pswpout", "pgsteal_kswapd", "numa_hit", "numa_miss", "numa_foreign"): out[k] = int(v) except Exception: pass try: with open("/proc/pressure/memory") as f: out["pressure_memory"] = f.read().strip() with open("/proc/pressure/io") as f: out["pressure_io"] = f.read().strip() except Exception: pass return out def collect_fabric(): """Storage fabric: the RDMA and NIC side that Inferra's prefetch depends on, plus the PCIe width that bounds every NVMe read.""" rdma = [] try: for d in sorted(os.listdir("/sys/class/infiniband")): base = "/sys/class/infiniband/" + d rdma.append({ "device": d, "fw_ver": _read(base + "/fw_ver", 40), "hca_type": _read(base + "/hca_type", 40), "board_id": _read(base + "/board_id", 60), "rate": _read(base + "/ports/1/rate", 40), "state": _read(base + "/ports/1/state", 40), "link_layer": _read(base + "/ports/1/link_layer", 30), }) except Exception: pass nics = [] try: for d in sorted(os.listdir("/sys/class/net")): if d == "lo": continue sp = _read("/sys/class/net/%s/speed" % d, 20) nics.append({"iface": d, "speed_mbps": sp, "operstate": _read("/sys/class/net/%s/operstate" % d, 20)}) except Exception: pass pcie = [] try: for d in sorted(os.listdir("/sys/class/nvme")): dev = os.path.realpath("/sys/class/nvme/" + d + "/device") pcie.append({"device": d, "link_speed": _read(dev + "/current_link_speed", 30), "link_width": _read(dev + "/current_link_width", 10)}) except Exception: pass return {"rdma": rdma, "nics": nics[:12], "nvme_pcie": pcie} def collect_topology(): """NVLink layout decides what tensor parallelism actually costs.""" rc, out, _ = _cmd(["nvidia-smi", "topo", "-m"]) topo = out.decode("utf-8", "replace").strip() if rc == 0 else None rc2, out2, _ = _cmd(["nvidia-smi", "--query-gpu=name,memory.total,ecc.mode.current,persistence_mode," "power.limit,clocks.max.sm,compute_cap", "--format=csv,noheader"]) detail = out2.decode("utf-8", "replace").strip().splitlines() if rc2 == 0 else [] numa = None try: numa = len([d for d in os.listdir("/sys/devices/system/node") if d.startswith("node")]) except Exception: pass rc3, out3, _ = _cmd(["nvcc", "--version"]) cuda = None if rc3 == 0: for ln in out3.decode(errors="replace").splitlines(): if "release" in ln: cuda = ln.strip() return {"nvidia_smi_topo": topo, "gpu_detail": detail, "numa_nodes": numa, "cuda": cuda} SERVING_HINTS = ("vllm", "sglang", "text-generation", "trtllm", "tensorrt_llm", "lmdeploy") def collect_serving_proc(): """The serving command line is where the whole configuration actually lives: quantization, context, parallelism, batching, prefix caching. No API exposes it, so read it from /proc. Own-user processes only, no root needed.""" found = [] try: pids = [d for d in os.listdir("/proc") if d.isdigit()] except Exception: return found for pid in pids: try: with open("/proc/%s/cmdline" % pid, "rb") as f: argv = f.read().decode("utf-8", "replace").split("\x00") except Exception: continue joined = " ".join(argv).lower() if not any(h in joined for h in SERVING_HINTS): continue if "gpu-analyzer" in joined: continue args = {} for i, a in enumerate(argv): if a.startswith("--"): key = a[2:].split("=")[0] if "=" in a: args[key] = a.split("=", 1)[1] elif i + 1 < len(argv) and argv[i + 1] and not argv[i + 1].startswith("--"): args[key] = argv[i + 1] else: args[key] = True rss = None try: with open("/proc/%s/status" % pid) as f: for line in f: if line.startswith("VmRSS:"): rss = line.split()[1] + " kB" except Exception: pass found.append({ "pid": int(pid), "engine": ("sglang" if "sglang" in joined else "vllm" if "vllm" in joined else "trtllm" if "trtllm" in joined or "tensorrt_llm" in joined else "lmdeploy" if "lmdeploy" in joined else "other"), "rss": rss, "args": {k: v for k, v in args.items() if k in ( "model", "served-model-name", "quantization", "kv-cache-dtype", "dtype", "max-model-len", "max-num-seqs", "max-num-batched-tokens", "tensor-parallel-size", "pipeline-parallel-size", "data-parallel-size", "gpu-memory-utilization", "block-size", "swap-space", "enable-prefix-caching", "no-enable-prefix-caching", "enable-chunked-prefill", "num-scheduler-steps", "speculative-model", "num-speculative-tokens", "speculative-config", "enforce-eager", "distributed-executor-backend", "port", "cpu-offload-gb", "mem-fraction-static", "chunked-prefill-size", "schedule-conservativeness", )}, }) if len(found) >= 4: break return found def collect_pyenv(): """Which build of the serving stack is actually installed.""" out = {} for mod in ("vllm", "torch", "transformers", "sglang", "flashinfer", "xformers"): rc, o, _ = _cmd([sys.executable, "-c", "import %s,sys;sys.stdout.write(getattr(%s,'__version__','?'))" % (mod, mod)]) if rc == 0 and o: out[mod] = o.decode(errors="replace").strip()[:40] return out def collect_block(): """Queue depth, scheduler and readahead bound NVMe throughput as hard as the device does, and they are routinely left at defaults that do not suit it.""" devs = [] try: for d in sorted(os.listdir("/sys/block")): if not (d.startswith("nvme") or d.startswith("sd")): continue q = "/sys/block/%s/queue/" % d devs.append({ "device": d, "scheduler": _read(q + "scheduler", 80), "nr_requests": _read(q + "nr_requests", 12), "read_ahead_kb": _read(q + "read_ahead_kb", 12), "max_sectors_kb": _read(q + "max_sectors_kb", 12), "rotational": _read(q + "rotational", 4), "nomerges": _read(q + "nomerges", 4), "size_512b": _read("/sys/block/%s/size" % d, 24), }) except Exception: pass stats = {} try: with open("/proc/diskstats") as f: for line in f: p = line.split() if len(p) >= 14 and (p[2].startswith("nvme")): stats[p[2]] = {"reads": int(p[3]), "read_sectors": int(p[5]), "writes": int(p[7]), "write_sectors": int(p[9]), "io_ms": int(p[12])} except Exception: pass return {"devices": devs, "diskstats": stats} def collect_memory_detail(): """NUMA distances stand in for access latency, which cannot be probed from a stdlib script without a microbenchmark we would not trust anyway. Hugepages and memlock are here because RDMA fails quietly without them.""" dist = {} try: for n in sorted(os.listdir("/sys/devices/system/node")): if n.startswith("node") and n[4:].isdigit(): dist[n] = _read("/sys/devices/system/node/%s/distance" % n, 120) except Exception: pass huge = {} try: for k in ("nr_hugepages", "nr_overcommit_hugepages"): huge[k] = _read("/sys/kernel/mm/hugepages/hugepages-2048kB/" + k, 20) huge["1G_pages"] = _read("/sys/kernel/mm/hugepages/hugepages-1048576kB/nr_hugepages", 20) except Exception: pass meminfo = {} try: with open("/proc/meminfo") as f: for line in f: k, _, v = line.partition(":") if k in ("MemTotal", "MemAvailable", "HugePages_Total", "HugePages_Free", "Hugepagesize", "SwapTotal", "Dirty", "Writeback", "Mlocked"): meminfo[k] = v.strip() except Exception: pass memlock = None try: import resource memlock = resource.getrlimit(resource.RLIMIT_MEMLOCK) memlock = "unlimited" if memlock[0] == -1 else str(memlock[0]) except Exception: pass dimms = None rc, out, _ = _cmd(["dmidecode", "-t", "memory"]) if rc == 0 and out: speeds, types = set(), set() for line in out.decode(errors="replace").splitlines(): line = line.strip() if line.startswith("Configured Memory Speed:") and "Unknown" not in line: speeds.add(line.split(":", 1)[1].strip()) if line.startswith("Type:") and "Unknown" not in line: types.add(line.split(":", 1)[1].strip()) dimms = {"types": sorted(types)[:4], "configured_speeds": sorted(speeds)[:4]} return {"numa_distance": dist, "hugepages": huge, "meminfo": meminfo, "memlock_rlimit": memlock, "dimms": dimms, "note": "Access latency is not probed: a stdlib microbenchmark would be noise. " "NUMA distance is the honest proxy."} def collect_fs_detail(): """Mount options change durability and throughput more than most people expect.""" out = [] try: with open("/proc/mounts") as f: for line in f: p = line.split() if len(p) >= 4 and p[2] in ("ext4", "xfs", "btrfs", "nfs", "nfs4", "zfs", "overlay"): entry = {"target": p[1], "fstype": p[2], "options": p[3][:200], "source_kind": ("nvme" if "nvme" in p[0] else "net" if ":" in p[0] else "other")} if p[2] == "xfs": rc, o, _ = _cmd(["xfs_info", p[1]]) if rc == 0: entry["xfs_info"] = o.decode(errors="replace")[:600] out.append(entry) except Exception: pass return out[:40] def collect_secureboot(): sb = None try: for f in os.listdir("/sys/firmware/efi/efivars"): if f.startswith("SecureBoot-"): with open("/sys/firmware/efi/efivars/" + f, "rb") as fh: data = fh.read() sb = bool(data[-1]) if data else None except Exception: pass return {"secure_boot": sb, "efi": os.path.isdir("/sys/firmware/efi"), "tpm": os.path.exists("/dev/tpm0") or os.path.exists("/dev/tpmrm0")} def collect_runtime(): """Container and scheduling context: a probe inside a pod sees a different machine from the one the model assumes.""" cg = _read("/proc/1/cgroup", 400) in_container = bool(cg and ("docker" in cg or "kubepods" in cg or "containerd" in cg)) \ or os.path.exists("/.dockerenv") return { "in_container": in_container, "kubernetes": bool(os.environ.get("KUBERNETES_SERVICE_HOST")), "cgroup_hint": (cg or "")[:200], "uptime_s": (lambda v: float(v.split()[0]) if v else None)(_read("/proc/uptime", 40)), "loadavg": _read("/proc/loadavg", 60), "isolcpus": _read("/sys/devices/system/cpu/isolated", 80), "nohz_full": _read("/sys/devices/system/cpu/nohz_full", 80), } def collect_gpu_sample(seconds=3): """A short live sample: what the GPUs are actually doing right now, as opposed to what they could do.""" rc, out, _ = _cmd(["nvidia-smi", "--query-gpu=index,utilization.gpu,utilization.memory,memory.used," "memory.total,temperature.gpu,power.draw,clocks.sm,ecc.errors.uncorrected.volatile.total", "--format=csv,noheader,nounits", "-l", "1", "-c", str(seconds)]) if rc != 0: rc, out, _ = _cmd(["nvidia-smi", "--query-gpu=index,utilization.gpu,memory.used,memory.total", "--format=csv,noheader,nounits"]) return out.decode("utf-8", "replace").strip().splitlines()[:64] if rc == 0 else [] def gpg_available(): return shutil.which("gpg") is not None def pop_key_fpr(create=False): """Fingerprint of this machine's .pop signing key, creating it if asked.""" if not gpg_available(): return None rc, out, _ = _cmd(["gpg", "--batch", "--with-colons", "--list-secret-keys", POP_KEY_UID]) if rc == 0: for line in out.decode(errors="replace").splitlines(): if line.startswith("fpr:"): return line.split(":")[9] if not create: return None params = ("%%no-protection\nKey-Type: eddsa\nKey-Curve: ed25519\n" "Name-Real: pod-efficiency-analyzer\nName-Comment: .pop signing\n" "Name-Email: pop@localhost\nExpire-Date: 0\n%%commit\n") % () rc, _, err = _cmd(["gpg", "--batch", "--gen-key", "-"], stdin=params.encode()) if rc != 0: return None return pop_key_fpr(create=False) def sign_digest(digest_text, fpr): rc, out, _ = _cmd(["gpg", "--batch", "--yes", "--armor", "--detach-sign", "--local-user", fpr], stdin=digest_text.encode()) return out.decode(errors="replace") if rc == 0 else None def convert_harness(path): """Wrap third-party benchmark output unmodified. We never rewrite results.""" try: with open(path, "rb") as f: raw = f.read() except Exception: return None, None name = os.path.basename(path) low = name.lower() harness = ("guidellm" if "guidellm" in low else "vllm-bench" if "vllm" in low else "genai-perf" if "genai" in low else "mlperf" if "mlperf" in low else "unknown") return harness, raw POP_API = "https://api.pod-efficiency.tools/api/v1/pop" def upload_pop(path, session=None): with open(path, "rb") as f: raw = f.read() url = POP_API + (("?session=" + urllib.parse.quote(session)) if session else "") req = urllib.request.Request(url, data=raw, headers={"Content-Type": "application/zip"}) with urllib.request.urlopen(req, timeout=30) as resp: return json.loads(resp.read().decode()) def ask_yes(prompt): """Default yes, but never assume it when there is no one to ask.""" if not sys.stdin.isatty(): return False try: a = input(prompt).strip().lower() except (EOFError, KeyboardInterrupt): print() return False return a in ("", "y", "yes") def build_pop(info, harness_path=None, sign=False): state = collect_state() gpus = info.get("gpus") or [] files = {} harness, raw = (None, None) if harness_path: harness, raw = convert_harness(harness_path) if raw is not None: files["results/" + os.path.basename(harness_path)] = raw measured = raw is not None # Resolve the signing key FIRST. The manifest records whether the package is # signed, so it has to be final before the digest is taken, or the signature # would cover a manifest that never shipped. signed_by = pop_key_fpr(create=True) if sign else None files["manifest.json"] = json.dumps({ "format": "pop", "format_version": "0.1", "source": "measured" if measured else "probed", "attested": bool(signed_by), "signed_by": signed_by, "producer": {"name": "gpu-analyzer.py", "version": info.get("analyzer_version")}, "note": ("Harness output under results/ is byte-for-byte as produced; metrics are " "derived from it, never hand-entered." if measured else "System state probed from this machine. No benchmark was supplied, so " "results/ is empty and this package describes hardware, not performance. " "Re-run with --harness FILE to include a measured run."), }, indent=2).encode() files["system.json"] = json.dumps({ "source": "probed", "hostname_present": True, "cpu": info.get("cpu"), "memory_gib": info.get("memory_gib"), "gpus": gpus, "disk": info.get("disk"), "nfs_mounts": info.get("nfs_mounts"), "state": state, "fabric": collect_fabric(), "topology": collect_topology(), "faults": collect_faults(), "block": collect_block(), "memory_detail": collect_memory_detail(), "filesystems_detail": collect_fs_detail(), "secureboot": collect_secureboot(), "runtime": collect_runtime(), "gpu_sample": collect_gpu_sample(), }, indent=2).encode() serving = collect_serving() procs = collect_serving_proc() files["config.json"] = json.dumps({ "serving": serving, "processes": procs, "python_packages": collect_pyenv(), "note": "processes[].args is the real serving configuration: quantization, context, " "parallelism, batching and prefix caching, read from the command line because " "no engine exposes it over HTTP.", }, indent=2).encode() files["traffic.json"] = json.dumps({ "source": "vllm/sglang metrics" if serving.get("detected") else "unavailable", "prompt_tokens_total": serving.get("prompt_tokens_total"), "generation_tokens_total": serving.get("generation_tokens_total"), "input_output_ratio": serving.get("input_output_ratio"), "prefix_cache_hit_rate": serving.get("prefix_cache_hit_rate"), "gpu_cache_usage": serving.get("gpu_cache_usage"), "page_faults": collect_faults(), }, indent=2).encode() files["provenance.json"] = json.dumps({ "branch": os.environ.get("INFERRA_BRANCH"), "commit": os.environ.get("INFERRA_COMMIT"), "committed_at": os.environ.get("INFERRA_COMMITTED_AT"), "build": os.environ.get("INFERRA_BUILD"), "release": os.environ.get("INFERRA_RELEASE"), "note": "Populated from INFERRA_* environment variables when run against a build.", }, indent=2).encode() files["metrics.json"] = json.dumps({ "basis": "measured" if measured else "none", "harness": harness, "uplift_estimate": info.get("uplift"), "note": "Estimates here come from published benchmarks, not from this machine.", }, indent=2).encode() digest_lines = [] for name in sorted(files): digest_lines.append("%s %s" % (hashlib.sha256(files[name]).hexdigest(), name)) digest_text = "\n".join(digest_lines) + "\n" files["attestation/digest.txt"] = digest_text.encode() # Signature and public key sit OUTSIDE the digest by definition: a detached # signature cannot be one of the things it signs. if signed_by: sig = sign_digest(digest_text, signed_by) if sig: files["attestation/signature.asc"] = sig.encode() rc, pub, _ = _cmd(["gpg", "--batch", "--armor", "--export", signed_by]) if rc == 0: files["attestation/pubkey.asc"] = pub else: signed_by = None return files, signed_by, measured def write_pop(files, path): with zipfile.ZipFile(path, "w", zipfile.ZIP_STORED) as z: for name in sorted(files): z.writestr(name, files[name]) return path def upload(payload): data = json.dumps(payload).encode() req = urllib.request.Request(API_URL, data=data, headers={"Content-Type": "application/json"}) with urllib.request.urlopen(req, timeout=15) as resp: return json.loads(resp.read().decode()) def main(): ap = argparse.ArgumentParser(description="Pod Efficiency Analyzer -- local probe") ap.add_argument("--no-upload", action="store_true", help="analyze only; send nothing to api.pod-efficiency.tools") ap.add_argument("--json", action="store_true", help="print raw JSON payload too") ap.add_argument("--pop", metavar="PATH", nargs="?", const="auto", help="write a .pop package (Performance Optimization Package)") ap.add_argument("--harness", metavar="FILE", help="include this benchmark output (guidellm, vllm-bench, genai-perf, " "mlperf) unmodified under results/") ap.add_argument("--sign", action="store_true", help="sign the .pop with gpg, generating a signing key on first use") ap.add_argument("--session", metavar="ID", help="the session id shown next to the command on the website, so the " "package is offered back to that browser session") args = ap.parse_args() if not sys.stdout.isatty(): C.off() info = build_payload() print_report(info) if args.json: print(json.dumps(info, indent=2)) if args.pop: path = args.pop if path == "auto": path = "pod-efficiency-%s.pop" % socket.gethostname().split(".")[0] files, fpr, measured = build_pop(info, args.harness, args.sign) write_pop(files, path) kind = "measured" if measured else "probed" print("\n Wrote %s (%s, %d files)" % (path, kind, len(files))) if args.sign: if fpr: print(C.GREEN + " Signed by %s" % fpr + C.END) print(" Public key is inside the package at attestation/pubkey.asc") else: print(C.RED + " Not signed: gpg unavailable or key generation failed." + C.END) print(" The package still carries its digest, so tampering is detectable.") if not measured: print(" No --harness given, so this describes hardware, not performance.") if args.no_upload: print(" --no-upload set: the package stays on this machine.\n") else: print("\n This package describes your hardware and, if a harness was given,") print(" its measured performance. It contains no credentials and no identity.") if args.session: print(" It will be linked to session %s so the site can offer it back to you." % args.session) if ask_yes(" Upload it to api.pod-efficiency.tools? [Y/n] "): try: r = upload_pop(path, args.session) print(C.GREEN + " Uploaded. id %s" % r.get("id") + C.END) except Exception as exc: print(C.RED + " Upload failed (%s). The file is still at %s" % (exc, path) + C.END) else: print(" Not uploaded. The file is at %s\n" % path) if args.no_upload: print("\n --no-upload set: nothing was sent.\n") return print("\n Uploading statistics to %s ..." % API_URL) try: resp = upload(info) print(C.GREEN + " Uploaded OK. Report id: %s" % resp.get("id") + C.END) print(" Reference this id on open.pod-efficiency.tools when generating your PDF.\n") except Exception as exc: print(C.RED + " Upload failed (%s). Re-run with --no-upload to skip." % exc + C.END) sys.exit(1) if __name__ == "__main__": main()