#!/usr/bin/env python3 """Prometheus exporter for Apple silicon, built on powermetrics and vm_stat. Purpose: publish power draw and unified-memory use for Track M in the shape Prometheus scrapes. Apple ships no Prometheus exporter, and the tools that do report power on a Mac are command-line ones, so this runs them and republishes two numbers at /metrics. Platform: mac (Apple silicon). Not for the other tracks: Tracks S and N use NVIDIA's DCGM exporter, Track X uses rocm-exporter.py in this directory. Minimum memory: 8 GB, which is the service being watched; this process needs almost none. Assumes: macOS, with `powermetrics`, `vm_stat` and `sysctl` in their usual places. powermetrics needs root: run this exporter with sudo, or run it without and accept that the power gauge will be absent while the memory gauges still work. Apple publishes the powermetrics manual only as a manual page on the machine itself, so before trusting the sampler names below, read `man powermetrics` on your own Mac and confirm them. Nothing here has been executed on Apple hardware by this course. Unified memory has no separate video memory to report, so "accelerator memory" here is the machine's memory: total from `sysctl hw.memsize`, and in-use computed from vm_stat as active plus wired plus compressed pages. That is a choice, it is stated on the dashboard, and it is the number that decides whether a model fits. Usage: sudo python3 mac-exporter.py --once sudo python3 mac-exporter.py --port 9402 python3 mac-exporter.py --port 9402 --no-power (memory only, no root needed) """ import argparse import os import re import shutil import subprocess import sys from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer ACCELERATOR = "apple" # powermetrics prints power as lines such as "CPU Power: 1234 mW". The combined figure is # preferred where it appears; otherwise the parts are added up. Confirm the sampler names # against `man powermetrics` before relying on this. POWER_LINE = re.compile(r"^\s*(combined|cpu|gpu|ane|package)\s+power:\s*([\d.]+)\s*mw", re.IGNORECASE | re.MULTILINE) VM_STAT_LINE = re.compile(r"^(.*?):\s+(\d+)\.?$", re.MULTILINE) PAGE_SIZE_HINT = re.compile(r"page size of (\d+) bytes") state = {"scrapes": 0, "errors": 0, "last_error": ""} def run(cmd, timeout=20): if shutil.which(cmd[0]) is None: state["last_error"] = f"{cmd[0]} is not on PATH" return None try: out = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout, check=False) except (OSError, subprocess.TimeoutExpired) as exc: state["last_error"] = f"running {cmd[0]} failed: {exc}" return None if out.returncode != 0: state["last_error"] = f"{cmd[0]} exited {out.returncode}: {out.stderr.strip()[:200]}" return None return out.stdout def read_power(sample_ms): """One powermetrics sample, in watts, or None with the reason recorded.""" text = run(["powermetrics", "--samplers", "cpu_power,gpu_power", "-n", "1", "-i", str(sample_ms)], timeout=sample_ms / 1000 + 20) if text is None: return None parts = {name.lower(): float(value) for name, value in POWER_LINE.findall(text)} if not parts: state["last_error"] = ("powermetrics ran but printed no power line this script " "recognises; run with --once and read its output") return None for combined in ("combined", "package"): if combined in parts: return parts[combined] / 1000.0 total = sum(parts.get(k, 0.0) for k in ("cpu", "gpu", "ane")) return total / 1000.0 if total else None def read_memory(): """Total and in-use memory in bytes, from sysctl and vm_stat.""" total_text = run(["sysctl", "-n", "hw.memsize"]) vm_text = run(["vm_stat"]) if total_text is None or vm_text is None: return None, None try: total = int(total_text.strip()) except ValueError: state["last_error"] = "sysctl hw.memsize did not return a number" return None, None hint = PAGE_SIZE_HINT.search(vm_text) page = int(hint.group(1)) if hint else 4096 counts = {name.strip().lower(): int(value) for name, value in VM_STAT_LINE.findall(vm_text)} def pages(*keys): for key in keys: for name, value in counts.items(): if name.startswith(key): return value return 0 used_pages = (pages("pages active") + pages("pages wired down", "pages wired") + pages("pages occupied by compressor", "pages stored in compressor")) return total, used_pages * page def collect(sample_ms, want_power): rows = [] power = read_power(sample_ms) if want_power else None if want_power and power is None: state["errors"] += 1 elif power is not None: rows.append(("local_llm_accelerator_power_watts", power)) total, used = read_memory() if total is None: state["errors"] += 1 else: rows.append(("local_llm_accelerator_memory_total_bytes", float(total))) rows.append(("local_llm_accelerator_memory_used_bytes", float(used))) return rows def render(rows): state["scrapes"] += 1 lines = [ "# HELP local_llm_accelerator_power_watts Package power draw reported by powermetrics.", "# TYPE local_llm_accelerator_power_watts gauge", "# HELP local_llm_accelerator_memory_used_bytes Unified memory active, wired and compressed.", "# TYPE local_llm_accelerator_memory_used_bytes gauge", "# HELP local_llm_accelerator_memory_total_bytes Unified memory fitted.", "# TYPE local_llm_accelerator_memory_total_bytes gauge", "# HELP local_llm_exporter_scrape_errors_total Failed collections since start.", "# TYPE local_llm_exporter_scrape_errors_total counter", "# HELP local_llm_exporter_scrapes_total Collections served since start.", "# TYPE local_llm_exporter_scrapes_total counter", ] for name, value in rows: lines.append(f'{name}{{accelerator="{ACCELERATOR}"}} {value}') lines.append(f"local_llm_exporter_scrape_errors_total {state['errors']}") lines.append(f"local_llm_exporter_scrapes_total {state['scrapes']}") return "\n".join(lines) + "\n" def make_handler(sample_ms, want_power): class Handler(BaseHTTPRequestHandler): def do_GET(self): # noqa: N802 - the name is fixed by BaseHTTPRequestHandler if self.path.split("?")[0] != "/metrics": self.send_response(404) self.send_header("Content-Type", "text/plain; charset=utf-8") self.end_headers() self.wfile.write(b"Nothing here. The metrics are at /metrics.\n") return body = render(collect(sample_ms, want_power)).encode("utf-8") self.send_response(200) self.send_header("Content-Type", "text/plain; version=0.0.4; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def log_message(self, fmt, *args): if os.environ.get("EXPORTER_ACCESS_LOG"): sys.stderr.write("%s - %s\n" % (self.address_string(), fmt % args)) return Handler def print_once(sample_ms, want_power): if want_power and os.geteuid() != 0: print("powermetrics normally needs root. Re-run with sudo, or pass --no-power.", file=sys.stderr) rows = collect(sample_ms, want_power) if state["last_error"]: print(f"note: {state['last_error']}", file=sys.stderr) if not rows: return 1 print("What Prometheus would receive:") print(render(rows)) return 0 def main(): parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) parser.add_argument("--port", type=int, default=9402, help="port to serve /metrics on") parser.add_argument("--host", default="127.0.0.1", help="address to bind; keep the default unless you know why not") parser.add_argument("--sample-ms", type=int, default=500, help="how long each powermetrics sample takes, in milliseconds") parser.add_argument("--no-power", action="store_true", help="skip powermetrics and publish memory only, without root") parser.add_argument("--once", action="store_true", help="collect once, print what would be served, and exit") args = parser.parse_args() want_power = not args.no_power if args.once: sys.exit(print_once(args.sample_ms, want_power)) server = ThreadingHTTPServer((args.host, args.port), make_handler(args.sample_ms, want_power)) print(f"mac-exporter listening on http://{args.host}:{args.port}/metrics") if want_power: print("Each scrape runs powermetrics, which needs root and takes a moment. Keep the") print("Prometheus scrape interval at fifteen seconds or longer.") try: server.serve_forever() except KeyboardInterrupt: print("\nstopped") finally: server.server_close() if __name__ == "__main__": main()