alex/sots-re
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

tools: live screen wall + contact sheet for the five lab guests

Browse source 
The lab went from one Windows guest to five and there was no way to see what
they were all doing without issuing a QEMU screendump per guest by hand.

- tools/vmwatch.py: always-on HTTP service serving an auto-refreshing wall of
  live guest screens. Runs on spicy as vmwatch.service; browse it at
  http://192.168.3.201:8140/. Click a tile for that guest full size. Guests are
  discovered from /etc/pve/qemu-server by matching sots-re, so clones appear and
  vanish on their own. A stopped, paused or unreachable guest gets a labelled
  placeholder tile carrying the monitor's own error, never a broken image or a
  500. Python 3 stdlib only.
- tools/vmwatch-install.sh: install/update/uninstall the unit on the host.
- tools/vmshot.py: one-shot contact sheet, and --one <id> for a full-size grab.
  Pulls frames from the vmwatch service when it is up (0.5s) and falls back to
  ssh + qm monitor when it is not (4s).
- guides/lab-screen-wall.md: how to use both, and why.

Capture goes over each guest's QMP socket rather than forking qm: qm is a Perl
program, and one fork per guest per tick cost ~90% of a host core and a 728 MB
cgroup peak. Direct QMP is 0.33 CPU-seconds per 88s and 23 MB RSS. QEMU 11 here
dumps PNG natively; the fallback PPM encoder was verified pixel-identical to
QEMU's own on a real framebuffer.

Read-only throughout: screendump does not perturb the guest (method-rule 19),
so reading VM 140's screen is not an experiment and does not take its lock.
This commit is contained in:
alex 2026-09-08 16:53:04 -04:00
parent 2e935b8491
commit 962cdb6981
4 changed files with 1204 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

63
guides/lab-screen-wall.md Normal file
View file

@ -0,0 +1,63 @@
# Watching the lab guests
The lab is five Windows guests now. This is how to see all of them at once.
## The live wall
<http://192.168.3.201:8140/>
Leave the tab open. Tiles refresh every 5 s; click a tile for that guest full size.
Each tile carries the VM id, name, status and the capture timestamp. A guest that is
stopped, paused or unreachable shows a labelled `NO SIGNAL` tile with the monitor's
own error text, never a broken image.
Served by `vmwatch.service` on **spicy** (`/opt/vmwatch/vmwatch.py`, systemd, enabled
at boot). Source of truth is `tools/vmwatch.py` in this repo; redeploy with:
tools/vmwatch-install.sh spicy # install or update
tools/vmwatch-install.sh spicy --uninstall
Nothing was installed on spicy beyond that file and its unit — the service is Python 3
standard library only (spicy has no ImageMagick, no netpbm, no Pillow).
## The one-shot contact sheet
tools/vmshot.py # all guests -> dumps/vmshot/sheet-<ts>.png + latest.png
tools/vmshot.py --one 140 # one guest, full size
tools/vmshot.py 144 145 --cols 2
tools/vmshot.py --ssh # bypass the service, capture over SSH
`vmshot` pulls frames from the vmwatch service when it is reachable (~0.5 s, no extra
load on the guests) and falls back to SSH + `qm monitor` when it is not (~4 s). Output
lands in `dumps/vmshot/`, which is gitignored.
## Why this mechanism
Both tools capture with QEMU `screendump`, reached over each guest's
`/var/run/qemu-server/<id>.qmp` socket — the same socket and command `qm monitor` uses.
* It needs **no guest agent, no guest network and nothing installed in the guest**.
* It does **not perturb the guest**. That is the point (method-rule 19: an instrument
that perturbs the thing it measures has already cost this campaign once). Reading
VM 140's screen is therefore *not* an experiment and does **not** take 140's
exclusivity lock.
* The campaign board already records that `qm monitor` screendump is more reliable
than the in-guest click helper's `shot`.
## Two things worth knowing
**Do not fork `qm` in a loop.** The first version of vmwatch shelled out to
`qm monitor` once per guest per 5 s tick. `qm` is a Perl program: that cost **~90 % of
a host core and a 728 MB cgroup peak**. Talking to the QMP socket directly — same
socket, same command, no fork — brought it to **0.33 CPU-seconds per 88 s and 23 MB
RSS**, roughly a 300× reduction. The fleet list comes from `/etc/pve/qemu-server/*.conf`
plus `query-status` for the same reason. `qm` remains the fallback path only.
**QEMU 11 on spicy dumps PNG natively** (`screendump <file> -f png`), so no PPM
conversion is needed. `vmwatch.ppm_to_png` exists as a fallback for an older QEMU and
was checked against QEMU's own encoder on a real 1024×768 framebuffer: pixel-identical.
Guests are discovered dynamically by matching `sots-re` in the guest name, so clones
added or destroyed later appear and vanish on their own — nothing to edit.
Both tools are strictly read-only. Neither starts, stops, resets nor reconfigures a VM.

444
tools/vmshot.py Executable file
View file

@ -0,0 +1,444 @@
#!/usr/bin/env -S uv run --quiet --with pillow python3
"""vmshot -- one-shot screenshot contact sheet for the SOTS lab Windows guests.
Why this exists
---------------
The lab went from one Windows guest (VM 140) to five (140, 141, 144, 145, 146).
There was no way to see what they were all doing without issuing a QEMU
``screendump`` per guest by hand. This grabs every guest in parallel and lays
them out on a single labelled contact sheet.
Mechanism
---------
``qm monitor <id> <<< "screendump <file> -f png"`` on the Proxmox host. This
reads the guest framebuffer straight out of QEMU:
* no guest agent, no guest network, nothing installed inside the guest;
* it does not perturb the guest at all -- relevant to method-rule 19, an
instrument that perturbs the thing it measures. Reading VM 140's screen is
therefore *not* an experiment and does not take VM 140's exclusivity lock;
* the campaign board already records that ``qm monitor`` screendump is more
reliable than the in-guest click helper's ``shot``.
spicy runs pve-manager 9.2 / QEMU 11, whose ``screendump`` takes ``-f png``
natively, so no PPM conversion and no netpbm/ImageMagick is needed on the host
(none is installed there, and this tool installs nothing). The PNGs are written
to a per-run temp dir on the host, streamed back inside a base64 tar in the same
SSH round trip, and the temp dir is removed before the SSH call returns -- so
nothing is left behind on spicy even if this script is killed.
Relationship to vmwatch
-----------------------
``tools/vmwatch.py`` is the live version of this: an always-on service on spicy
serving an auto-refreshing wall at http://192.168.3.201:8140/. When that
service is reachable this script pulls its already-captured frames over HTTP
instead of opening its own SSH session -- same frames, no extra load on the
guests, and it works from anywhere on the LAN without SSH. If the service is
down it falls back to the SSH + ``qm monitor`` path described below, so the CLI
never depends on the service being up.
Usage
-----
tools/vmshot.py # contact sheet of every sots-* guest
tools/vmshot.py --open # ... and open it in the default viewer
tools/vmshot.py 140 141 # only these ids
tools/vmshot.py --one 140 # full-size single guest, no sheet
tools/vmshot.py --cols 3 --width 900
Output lands in ``~/sots-re/dumps/vmshot/`` (``dumps/`` is gitignored).
A guest that is stopped, paused, or whose screendump fails gets a labelled
placeholder tile with the host-side error rather than aborting the sheet.
"""
from __future__ import annotations
import argparse
import base64
import io
import json
import os
import re
import subprocess
import urllib.error
import urllib.request
import sys
import tarfile
import time
from datetime import datetime
from pathlib import Path
from PIL import Image, ImageDraw, ImageFont
HOST = "spicy"
SERVICE = os.environ.get("VMWATCH_URL", "http://192.168.3.201:8140")
OUT_DIR = Path.home() / "sots-re" / "dumps" / "vmshot"
FLEET_PATTERN = re.compile(r"sots", re.I)
FONT_PATH = "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf"
FONT_BOLD = "/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf"
# Tile chrome
LABEL_H = 34
PAD = 10
BG = (24, 26, 30)
LABEL_BG = (38, 42, 50)
LABEL_BG_BAD = (74, 34, 34)
FG = (232, 234, 238)
FG_DIM = (150, 156, 166)
BORDER = (60, 66, 78)
PLACEHOLDER_BG = (44, 30, 30)
def ssh(cmd: str, timeout: int = 90) -> subprocess.CompletedProcess:
return subprocess.run(
["ssh", "-o", "BatchMode=yes", HOST, cmd],
capture_output=True,
timeout=timeout,
)
def service_state(timeout: float = 3.0) -> list[dict] | None:
"""Fleet state from a reachable vmwatch service, or None if it is not up."""
try:
with urllib.request.urlopen(f"{SERVICE}/api/state", timeout=timeout) as r:
return json.load(r)
except (urllib.error.URLError, OSError, ValueError, json.JSONDecodeError):
return None
def service_grab(ids: list[int], state: list[dict]) -> dict[int, dict]:
"""Pull already-captured frames from the vmwatch service. No guest load."""
by_id = {g["id"]: g for g in state}
out: dict[int, dict] = {}
for vid in ids:
g = by_id.get(vid)
if g is None:
out[vid] = {"png": None, "status": "absent", "log": "not known to vmwatch"}
continue
png = None
if g.get("ok"):
try:
with urllib.request.urlopen(
f"{SERVICE}/shot/{vid}.png?t={g.get('ts', '')}", timeout=10
) as r:
png = r.read()
except (urllib.error.URLError, OSError):
png = None
out[vid] = {
"png": png,
"status": g.get("status", "?"),
"log": g.get("error", "") if not png else "",
}
return out
def discover_fleet() -> list[tuple[int, str, str]]:
"""Return [(vmid, name, status)] for every VM whose name matches sots-*."""
r = ssh("qm list", timeout=30)
if r.returncode != 0:
sys.exit(f"vmshot: `qm list` on {HOST} failed: {r.stderr.decode().strip()}")
fleet = []
for line in r.stdout.decode().splitlines()[1:]:
parts = line.split()
if len(parts) < 3 or not parts[0].isdigit():
continue
vmid, name, status = int(parts[0]), parts[1], parts[2]
if FLEET_PATTERN.search(name):
fleet.append((vmid, name, status))
return sorted(fleet)
# Remote script: dump every requested guest in parallel, tar the results back,
# always clean up. Per-guest stderr is kept so a failure explains itself.
REMOTE = r"""
set -u
D=$(mktemp -d /tmp/vmshot.XXXXXX)
trap 'rm -rf "$D"' EXIT INT TERM
for id in %(ids)s; do
(
st=$(qm status "$id" 2>&1 | awk '{print $2}')
echo "$st" > "$D/$id.status"
timeout %(t)s qm monitor "$id" <<< "screendump $D/$id.png -f png" \
> "$D/$id.log" 2>&1
# qm monitor exits 0 even when the monitor command errors; the monitor
# echoes the failure into the log, so treat "no file" as the real test.
[ -s "$D/$id.png" ] || echo "no framebuffer written" >> "$D/$id.log"
) &
done
wait
tar -C "$D" -cf - . | base64 -w0
"""
def grab(ids: list[int], per_vm_timeout: int = 20) -> dict[int, dict]:
"""Screendump each id on the host; return {id: {'png': bytes|None, 'status', 'log'}}."""
remote = REMOTE % {"ids": " ".join(str(i) for i in ids), "t": per_vm_timeout}
r = ssh(f"bash -s <<'VMSHOT_EOF'\n{remote}\nVMSHOT_EOF", timeout=per_vm_timeout + 60)
if r.returncode != 0 and not r.stdout.strip():
sys.exit(f"vmshot: remote grab failed: {r.stderr.decode().strip()[:500]}")
out: dict[int, dict] = {i: {"png": None, "status": "?", "log": ""} for i in ids}
try:
blob = base64.b64decode(r.stdout.strip())
with tarfile.open(fileobj=io.BytesIO(blob), mode="r:") as tf:
for m in tf.getmembers():
name = Path(m.name).name
if not m.isfile():
continue
stem, _, ext = name.rpartition(".")
if not stem.isdigit():
continue
vid = int(stem)
if vid not in out:
continue
data = tf.extractfile(m).read()
if ext == "png":
out[vid]["png"] = data
elif ext == "status":
out[vid]["status"] = data.decode(errors="replace").strip() or "?"
elif ext == "log":
out[vid]["log"] = data.decode(errors="replace").strip()
except Exception as e: # noqa: BLE001 - a mangled tar must not lose the whole sheet
for v in out.values():
v["log"] = v["log"] or f"could not unpack remote payload: {e}"
return out
def _font(size: int, bold: bool = False) -> ImageFont.FreeTypeFont:
try:
return ImageFont.truetype(FONT_BOLD if bold else FONT_PATH, size)
except OSError:
return ImageFont.load_default()
def _clean_monitor_log(log: str) -> str:
"""Strip the monitor banner/prompt noise, keep the actual complaint."""
keep = []
for ln in log.splitlines():
ln = ln.replace("\x1b", "").strip()
ln = re.sub(r"^(QEMU \d[\w.]* monitor.*|qm> ?)", "", ln).strip()
if not ln or ln.startswith("Entering QEMU Monitor") or ln.startswith("Type 'help'"):
continue
keep.append(ln)
return "; ".join(keep)[:180]
def make_tile(vmid: int, name: str, info: dict, width: int, stamp: str) -> Image.Image:
png = info.get("png")
status = info.get("status", "?")
err = _clean_monitor_log(info.get("log", ""))
if png:
shot = Image.open(io.BytesIO(png)).convert("RGB")
native = f"{shot.width}x{shot.height}"
h = max(1, round(shot.height * width / shot.width))
shot = shot.resize((width, h), Image.LANCZOS)
detail = f"{status} {native} {stamp}"
bad = False
else:
h = round(width * 3 / 4)
shot = Image.new("RGB", (width, h), PLACEHOLDER_BG)
d = ImageDraw.Draw(shot)
msg = err or f"no screendump ({status})"
d.text(
(width // 2, h // 2 - 12),
"NO SIGNAL",
font=_font(max(18, width // 22), bold=True),
fill=(210, 120, 120),
anchor="mm",
)
f = _font(max(10, width // 60))
# crude wrap
line, lines = "", []
for word in msg.split():
if len(line) + len(word) + 1 > 52:
lines.append(line)
line = word
else:
line = f"{line} {word}".strip()
lines.append(line)
for i, ln in enumerate(lines[:4]):
d.text(
(width // 2, h // 2 + 18 + i * 15),
ln,
font=f,
fill=(190, 150, 150),
anchor="mm",
)
detail = f"{status} {stamp}"
bad = True
tile = Image.new("RGB", (width, h + LABEL_H), LABEL_BG_BAD if bad else LABEL_BG)
tile.paste(shot, (0, LABEL_H))
d = ImageDraw.Draw(tile)
f_id, f_name, f_detail = _font(16, bold=True), _font(14), _font(11)
y = LABEL_H // 2
d.text((8, y), str(vmid), font=f_id, fill=FG, anchor="lm")
name_x = 8 + round(d.textlength(str(vmid), font=f_id)) + 10
# A narrow tile must not let the two labels overlap: drop the timestamp
# first, then ellipsise the guest name to whatever room is left.
for candidate in (detail, detail.rsplit(" ", 1)[0], status):
detail_w = d.textlength(candidate, font=f_detail)
if name_x + d.textlength(name, font=f_name) + 12 + detail_w + 8 <= width:
detail = candidate
break
else:
detail = status
detail_w = d.textlength(detail, font=f_detail)
room = width - 8 - detail_w - 12 - name_x
shown = name
while shown and d.textlength(shown + "…", font=f_name) > room:
shown = shown[:-1]
if shown != name:
shown = (shown + "…") if shown else ""
if room > 0 and shown:
d.text((name_x, y), shown, font=f_name, fill=FG, anchor="lm")
d.text((width - 8, y), detail, font=f_detail, fill=FG_DIM, anchor="rm")
d.rectangle([0, 0, width - 1, h + LABEL_H - 1], outline=BORDER)
return tile
def contact_sheet(fleet, shots, tile_w: int, cols: int, stamp: str) -> Image.Image:
tiles = [make_tile(vid, name, shots.get(vid, {}), tile_w, stamp) for vid, name, _ in fleet]
rows = (len(tiles) + cols - 1) // cols
row_h = [
max((t.height for t in tiles[r * cols : (r + 1) * cols]), default=0) for r in range(rows)
]
header = 40
W = PAD + cols * (tile_w + PAD)
H = header + PAD + sum(h + PAD for h in row_h)
sheet = Image.new("RGB", (W, H), BG)
d = ImageDraw.Draw(sheet)
live = sum(1 for v in shots.values() if v.get("png"))
d.text((PAD, header // 2), "SOTS lab guests", font=_font(18, bold=True), fill=FG, anchor="lm")
d.text(
(W - PAD, header // 2),
f"{live}/{len(fleet)} framebuffers captured {stamp} host={HOST}",
font=_font(12),
fill=FG_DIM,
anchor="rm",
)
y = header + PAD
for r in range(rows):
x = PAD
for t in tiles[r * cols : (r + 1) * cols]:
sheet.paste(t, (x, y))
x += tile_w + PAD
y += row_h[r] + PAD
return sheet
def main() -> int:
ap = argparse.ArgumentParser(description="Contact sheet of the SOTS lab VM screens.")
ap.add_argument("ids", nargs="*", type=int, help="VM ids (default: every sots-* guest)")
ap.add_argument("--one", type=int, metavar="ID", help="save one guest full-size, no sheet")
ap.add_argument("--cols", type=int, default=3)
ap.add_argument("--width", type=int, default=760, help="tile width in px")
ap.add_argument("--out", type=Path, help="output png path")
ap.add_argument("--timeout", type=int, default=20, help="per-guest screendump timeout (s)")
ap.add_argument("--keep-raw", action="store_true", help="also save each guest's raw png")
ap.add_argument("--json", action="store_true", help="print a machine-readable result line")
ap.add_argument("--open", action="store_true", help="xdg-open the result")
ap.add_argument(
"--ssh",
action="store_true",
help="always capture over SSH, even if the vmwatch service is reachable",
)
args = ap.parse_args()
OUT_DIR.mkdir(parents=True, exist_ok=True)
t0 = time.time()
state = None if args.ssh else service_state()
if state is not None:
names = {g["id"]: g["name"] for g in state}
all_ids = [g["id"] for g in state]
source = SERVICE
else:
discovered = discover_fleet()
names = {v: n for v, n, _ in discovered}
all_ids = [v for v, _, _ in discovered]
source = f"ssh {HOST}"
if args.one is not None:
wanted = [args.one]
elif args.ids:
wanted = args.ids
else:
wanted = all_ids
if not wanted:
sys.exit(f"vmshot: no sots-* guests found via {source}")
fleet = [(v, names.get(v, f"vm{v}"), "") for v in wanted]
shots = service_grab(wanted, state) if state is not None else grab(wanted, args.timeout)
now = datetime.now()
stamp = now.strftime("%Y-%m-%d %H:%M:%S")
tag = now.strftime("%Y%m%d-%H%M%S")
if args.keep_raw or args.one is not None:
for vid, info in shots.items():
if info.get("png"):
p = OUT_DIR / f"vm{vid}-{tag}.png"
p.write_bytes(info["png"])
if args.one is not None:
info = shots.get(args.one, {})
if not info.get("png"):
print(
f"vmshot: no framebuffer from {args.one}: "
f"{_clean_monitor_log(info.get('log', '')) or info.get('status', '?')}",
file=sys.stderr,
)
return 1
out = args.out or (OUT_DIR / f"vm{args.one}-{tag}.png")
out.write_bytes(info["png"])
else:
sheet = contact_sheet(fleet, shots, args.width, args.cols, stamp)
out = args.out or (OUT_DIR / f"sheet-{tag}.png")
sheet.save(out)
if args.out is None: # only the default location keeps a `latest` alias
(OUT_DIR / "latest.png").write_bytes(out.read_bytes())
live = sum(1 for v in shots.values() if v.get("png"))
if args.json:
print(
json.dumps(
{
"out": str(out),
"captured": live,
"requested": len(wanted),
"source": source,
"seconds": round(time.time() - t0, 1),
"guests": {
str(v): {
"name": names.get(v, ""),
"status": i.get("status"),
"ok": bool(i.get("png")),
"error": _clean_monitor_log(i.get("log", "")) or None,
}
for v, i in sorted(shots.items())
},
},
indent=2,
)
)
else:
for v, i in sorted(shots.items()):
mark = "ok " if i.get("png") else "FAIL"
note = "" if i.get("png") else " " + (_clean_monitor_log(i.get("log", "")) or "?")
print(f" {mark} {v:<5} {names.get(v, ''):<20} {i.get('status', '?')}{note}")
print(
f"vmshot: {live}/{len(wanted)} captured in {time.time() - t0:.1f}s "
f"via {source} -> {out}"
)
if args.open:
subprocess.run(["xdg-open", str(out)], check=False)
return 0 if live else 2
if __name__ == "__main__":
raise SystemExit(main())

60
tools/vmwatch-install.sh Executable file
View file

@ -0,0 +1,60 @@
#!/usr/bin/env bash
# Install / update the vmwatch fishtank service on the Proxmox host.
#
# What this puts on spicy (and nothing else -- no packages, no pip):
# /opt/vmwatch/vmwatch.py the service (Python 3 stdlib only)
# /etc/systemd/system/vmwatch.service
#
# Usage: tools/vmwatch-install.sh [host] (default host: spicy)
# tools/vmwatch-install.sh spicy --uninstall
set -euo pipefail
HOST="${1:-spicy}"
SRC="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)/vmwatch.py"
PORT="${VMWATCH_PORT:-8140}"
INTERVAL="${VMWATCH_INTERVAL:-5}"
if [[ "${2:-}" == "--uninstall" ]]; then
ssh "$HOST" 'systemctl disable --now vmwatch.service 2>/dev/null || true
rm -f /etc/systemd/system/vmwatch.service
rm -rf /opt/vmwatch
systemctl daemon-reload
echo "vmwatch removed from $(hostname)"'
exit 0
fi
echo "installing vmwatch on $HOST (port $PORT, interval ${INTERVAL}s)"
ssh "$HOST" "mkdir -p /opt/vmwatch"
scp -q "$SRC" "$HOST:/opt/vmwatch/vmwatch.py"
ssh "$HOST" "chmod 0755 /opt/vmwatch/vmwatch.py"
ssh "$HOST" "cat > /etc/systemd/system/vmwatch.service" <<UNIT
[Unit]
Description=vmwatch - live screen wall for the SOTS lab VMs
Documentation=file:///opt/vmwatch/vmwatch.py
After=network-online.target pve-guests.service
Wants=network-online.target
[Service]
Type=simple
ExecStart=/usr/bin/python3 /opt/vmwatch/vmwatch.py --port ${PORT} --interval ${INTERVAL}
Restart=always
RestartSec=3
# read-only instrument: it only runs \`qm list\` and \`qm monitor ... screendump\`
NoNewPrivileges=yes
ProtectHome=yes
PrivateTmp=no
StandardOutput=journal
StandardError=journal
[Install]
WantedBy=multi-user.target
UNIT
# `enable --now` will not restart an already-running unit, so restart explicitly
ssh "$HOST" "systemctl daemon-reload \
&& systemctl enable vmwatch.service >/dev/null \
&& systemctl restart vmwatch.service && sleep 2 && systemctl is-active vmwatch.service"
IP=$(ssh "$HOST" "hostname -I | tr ' ' '\n' | grep -E '^192\.168\.' | head -1")
echo
echo "http://${IP}:${PORT}/"

637
tools/vmwatch.py Executable file
View file

@ -0,0 +1,637 @@
#!/usr/bin/env python3
"""vmwatch -- a fishtank for the SOTS lab Windows guests.
A small always-on HTTP service that serves an auto-refreshing page of live
guest screens. Leave the tab open and watch the lab.
http://192.168.3.201:8140/
Where it runs
-------------
On the Proxmox host **spicy** itself, because it needs `qm`. Installed as
``/opt/vmwatch/vmwatch.py`` with a systemd unit ``vmwatch.service``. The
canonical copy lives here in ``~/sots-re/tools/`` -- edit here, then reinstall
with ``tools/vmwatch-install.sh``.
Dependencies: **Python 3 standard library only.** Nothing is installed on
spicy beyond this file and its unit -- no pip, no ImageMagick, no netpbm (none
of which are present there).
Mechanism
---------
A QMP ``screendump`` sent to each guest's ``/var/run/qemu-server/<id>.qmp``
unix socket -- the same socket, and the same command, that
``qm monitor <id> <<< "screendump <file> -f png"`` uses, minus the Perl. That
detail is not cosmetic: forking ``qm`` once per guest per poll interval cost
about 90%% of a core on the host, which is not acceptable on a box that is also
running the guests. ``qm monitor`` remains the fallback when the socket is
absent. Either way this reads the guest framebuffer straight out of QEMU:
* no guest agent, no guest network, nothing running inside the guest;
* it does **not** perturb the guest -- relevant to method-rule 19, an
instrument that perturbs the thing it measures has already burned this
campaign once. Reading VM 140's screen is therefore not an experiment and
does not take VM 140's exclusivity lock;
* the campaign board already records that ``qm monitor`` screendump is more
reliable than the in-guest click helper's ``shot``.
spicy runs pve-manager 9.2 / QEMU 11, whose ``screendump`` accepts ``-f png``
natively. On an older QEMU without ``-f``, this falls back to a PPM dump and a
hand-rolled PPM->PNG encoder (zlib + a stored-filter IDAT), so no external
image tooling is needed either way. Temp files are read into memory and
unlinked immediately; nothing accumulates in /tmp.
Politeness
----------
A single background poller grabs every guest on a fixed interval and serves
every browser from that one in-memory copy, so N open tabs still cost one
screendump per guest per interval. The poller goes idle when no client has
asked for anything in IDLE_AFTER seconds, and wakes on the next request.
Routes
------
GET / the wall
GET /one/<id> one guest, full size
GET /api/state JSON: per-guest id, name, status, ok, error, age
GET /shot/<id>.png the latest cached framebuffer for that guest
GET /healthz plaintext ok
Read-only: this service never starts, stops, resets or reconfigures anything.
"""
from __future__ import annotations
import argparse
import html
import json
import os
import re
import socket
import struct
import subprocess
import sys
import tempfile
import threading
import time
import zlib
from concurrent.futures import ThreadPoolExecutor
from http import HTTPStatus
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
FLEET_PATTERN = re.compile(r"sots-re", re.I)
DEFAULT_PORT = 8140
DEFAULT_INTERVAL = 5.0 # seconds between framebuffer grabs
FLEET_TTL = 20.0 # seconds between `qm list` refreshes
IDLE_AFTER = 120.0 # stop polling if nobody has looked for this long
DUMP_TIMEOUT = 15 # per-guest screendump timeout, seconds
# --------------------------------------------------------------------------
# host-side capture
# --------------------------------------------------------------------------
CONF_DIR = "/etc/pve/qemu-server"
QMP_DIR = "/var/run/qemu-server"
def _run(cmd: list[str], timeout: int, stdin: str | None = None) -> subprocess.CompletedProcess:
return subprocess.run(
cmd, input=stdin, capture_output=True, text=True, timeout=timeout, check=False
)
class QmpError(Exception):
pass
class Qmp:
"""Minimal QMP client over the per-VM unix socket PVE already exposes.
This is the same socket `qm monitor` uses, opened the same way (connect,
negotiate, one command, close). Going straight to it instead of shelling
out to `qm` matters: `qm` is a Perl program and forking one per guest per
poll interval cost ~90%% of a core on the host. Falls back to `qm monitor`
if the socket is missing.
"""
def __init__(self, vmid: int, timeout: float = 10.0):
self.path = os.path.join(QMP_DIR, f"{vmid}.qmp")
self.timeout = timeout
self.sock: socket.socket | None = None
self.buf = b""
def __enter__(self) -> Qmp:
s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
s.settimeout(self.timeout)
s.connect(self.path)
self.sock = s
self._read_json() # greeting
self.execute("qmp_capabilities")
return self
def __exit__(self, *exc) -> None:
if self.sock:
try:
self.sock.close()
except OSError:
pass
def _read_json(self) -> dict:
while True:
nl = self.buf.find(b"\n")
if nl >= 0:
line, self.buf = self.buf[:nl], self.buf[nl + 1 :]
if line.strip():
msg = json.loads(line)
if "event" in msg: # async event: not our reply
continue
return msg
continue
chunk = self.sock.recv(65536) # type: ignore[union-attr]
if not chunk:
raise QmpError("monitor closed the connection")
self.buf += chunk
def execute(self, cmd: str, **args) -> dict:
payload = {"execute": cmd}
if args:
payload["arguments"] = args
self.sock.sendall(json.dumps(payload).encode() + b"\n") # type: ignore[union-attr]
reply = self._read_json()
if "error" in reply:
raise QmpError(reply["error"].get("desc", json.dumps(reply["error"]))[:200])
return reply.get("return", {})
def _conf_name(vmid: int) -> str | None:
"""Guest name from its PVE config, ignoring the [snapshot] sections."""
try:
with open(os.path.join(CONF_DIR, f"{vmid}.conf"), encoding="utf-8") as fh:
for line in fh:
if line.startswith("["): # snapshot section begins
break
if line.startswith("name:"):
return line.split(":", 1)[1].strip()
except OSError:
return None
return None
def list_fleet() -> list[dict]:
"""[{id, name, status}] for every VM whose name matches sots-re.
Read from /etc/pve/qemu-server + the QMP sockets rather than `qm list`,
which is another Perl fork. Falls back to `qm list` if that is not
readable (e.g. running somewhere that is not a PVE node).
"""
fleet = []
try:
entries = os.listdir(CONF_DIR)
except OSError:
entries = []
for entry in entries:
stem, _, ext = entry.rpartition(".")
if ext != "conf" or not stem.isdigit():
continue
vmid = int(stem)
name = _conf_name(vmid)
if not name or not FLEET_PATTERN.search(name):
continue
status = "stopped"
if os.path.exists(os.path.join(QMP_DIR, f"{vmid}.qmp")):
try:
with Qmp(vmid, timeout=5) as q:
status = q.execute("query-status").get("status", "running")
except (OSError, QmpError, ValueError):
status = "unreachable"
fleet.append({"id": vmid, "name": name, "status": status})
if fleet:
return sorted(fleet, key=lambda g: g["id"])
try:
r = _run(["qm", "list"], timeout=20)
except (subprocess.TimeoutExpired, OSError):
return []
for line in r.stdout.splitlines()[1:]:
parts = line.split()
if len(parts) < 3 or not parts[0].isdigit():
continue
if FLEET_PATTERN.search(parts[1]):
fleet.append({"id": int(parts[0]), "name": parts[1], "status": parts[2]})
return sorted(fleet, key=lambda g: g["id"])
def ppm_to_png(data: bytes) -> bytes:
"""Minimal binary-P6 PPM -> PNG. Only used if QEMU lacks screendump -f png."""
if not data.startswith(b"P6"):
raise ValueError("not a binary P6 PPM")
# header: P6, then width height maxval, whitespace separated, # comments allowed
fields, pos = [], 2
while len(fields) < 3:
while pos < len(data) and data[pos : pos + 1].isspace():
pos += 1
if data[pos : pos + 1] == b"#":
while pos < len(data) and data[pos] != 0x0A:
pos += 1
continue
start = pos
while pos < len(data) and not data[pos : pos + 1].isspace():
pos += 1
fields.append(int(data[start:pos]))
pos += 1 # single whitespace byte after maxval
w, h, maxval = fields
if maxval != 255:
raise ValueError(f"unsupported PPM maxval {maxval}")
px = data[pos : pos + w * h * 3]
stride = w * 3
raw = bytearray()
for y in range(h):
raw.append(0) # filter type 0 (None)
raw += px[y * stride : (y + 1) * stride]
def chunk(tag: bytes, payload: bytes) -> bytes:
return (
struct.pack(">I", len(payload))
+ tag
+ payload
+ struct.pack(">I", zlib.crc32(tag + payload) & 0xFFFFFFFF)
)
return (
b"\x89PNG\r\n\x1a\n"
+ chunk(b"IHDR", struct.pack(">IIBBBBB", w, h, 8, 2, 0, 0, 0))
+ chunk(b"IDAT", zlib.compress(bytes(raw), 6))
+ chunk(b"IEND", b"")
)
def screendump(vmid: int, png_native: bool = True) -> tuple[bytes | None, str]:
"""Grab one guest framebuffer. Returns (png_bytes, error_message); never raises."""
suffix = ".png" if png_native else ".ppm"
fd, path = tempfile.mkstemp(prefix=f"vmwatch-{vmid}-", suffix=suffix, dir="/tmp")
os.close(fd)
os.unlink(path) # QEMU creates it itself; we only wanted a unique name
err = ""
try:
if os.path.exists(os.path.join(QMP_DIR, f"{vmid}.qmp")):
args = {"filename": path}
if png_native:
args["format"] = "png"
with Qmp(vmid, timeout=DUMP_TIMEOUT) as q:
q.execute("screendump", **args)
else:
# not running, or not a PVE node: fall back to the CLI monitor
cmd = f"screendump {path}" + (" -f png" if png_native else "")
r = _run(["qm", "monitor", str(vmid)], timeout=DUMP_TIMEOUT, stdin=cmd + "\n")
# `qm monitor` exits 0 even when the monitor command itself errors;
# it echoes the failure to stdout, so "no file" is the real test.
err = _clean_monitor_text(r.stdout + r.stderr)
except FileNotFoundError:
_unlink(path)
return None, "guest is not running (no monitor socket)"
except (socket.timeout, subprocess.TimeoutExpired):
_unlink(path)
return None, "screendump timed out"
except (QmpError, OSError, ValueError) as e:
_unlink(path)
return None, str(e) or f"{type(e).__name__}"
try:
with open(path, "rb") as fh:
blob = fh.read()
except OSError:
blob = b""
finally:
_unlink(path)
if not blob:
return None, err or "no framebuffer written"
if png_native and blob.startswith(b"\x89PNG"):
return blob, ""
try:
return ppm_to_png(blob), ""
except Exception as e: # noqa: BLE001
return None, f"could not decode framebuffer: {e}"
def _unlink(path: str) -> None:
try:
os.unlink(path)
except OSError:
pass
def _clean_monitor_text(text: str) -> str:
keep = []
for ln in text.replace("\x1b", "").splitlines():
ln = re.sub(r"^(QEMU \d[\w.]* monitor.*|qm> ?)", "", ln.strip()).strip()
if not ln or ln.startswith(("Entering QEMU Monitor", "Type 'help'", "screendump ")):
continue
keep.append(ln)
return "; ".join(keep)[:200]
# --------------------------------------------------------------------------
# poller
# --------------------------------------------------------------------------
class Tank:
def __init__(self, interval: float):
self.interval = interval
self.lock = threading.Lock()
self.shots: dict[int, dict] = {} # id -> {png, error, ts}
self.fleet: list[dict] = []
self.fleet_ts = 0.0
self.last_client = time.time()
self.png_native = True
self.cycles = 0
self.started = time.time()
self.wake = threading.Event()
def touch(self) -> None:
idle = time.time() - self.last_client > IDLE_AFTER
self.last_client = time.time()
if idle:
self.wake.set() # first look after a nap: refresh right away
def get_fleet(self) -> list[dict]:
with self.lock:
fresh = time.time() - self.fleet_ts < FLEET_TTL and self.fleet
if fresh:
return list(self.fleet)
fleet = list_fleet()
with self.lock:
if fleet or not self.fleet:
self.fleet, self.fleet_ts = fleet, time.time()
return list(self.fleet)
def cycle(self) -> None:
fleet = self.get_fleet()
if not fleet:
return
with ThreadPoolExecutor(max_workers=max(1, len(fleet))) as pool:
results = list(pool.map(lambda g: (g["id"], screendump(g["id"], self.png_native)), fleet))
now = time.time()
with self.lock:
for vmid, (png, err) in results:
prev = self.shots.get(vmid, {})
self.shots[vmid] = {
"png": png if png else prev.get("png"),
"fresh": bool(png),
"error": err,
"ts": now if png else prev.get("ts", 0.0),
}
live = {g["id"] for g in fleet}
for gone in set(self.shots) - live:
del self.shots[gone]
self.cycles += 1
# one-time downgrade if this QEMU has no `-f png`
if self.png_native and all(not p for _, (p, _) in results):
errs = " ".join(e for _, (_, e) in results)
if "format" in errs.lower() or "invalid" in errs.lower() or "-f" in errs:
self.png_native = False
print("vmwatch: QEMU lacks screendump -f png; falling back to PPM", flush=True)
def loop(self) -> None:
while True:
if time.time() - self.last_client <= IDLE_AFTER:
try:
self.cycle()
except Exception as e: # noqa: BLE001 - the poller must never die
print(f"vmwatch: cycle error: {e}", file=sys.stderr, flush=True)
self.wake.wait(self.interval)
else:
self.wake.wait(2.0)
self.wake.clear()
def state(self) -> list[dict]:
now = time.time()
out = []
with self.lock:
shots = dict(self.shots)
for g in self.get_fleet():
s = shots.get(g["id"], {})
out.append(
{
**g,
"ok": bool(s.get("png")),
"fresh": bool(s.get("fresh")),
"error": s.get("error") or "",
"age": round(now - s["ts"], 1) if s.get("ts") else None,
"ts": time.strftime("%H:%M:%S", time.localtime(s["ts"])) if s.get("ts") else "",
}
)
return out
def png(self, vmid: int) -> bytes | None:
with self.lock:
return (self.shots.get(vmid) or {}).get("png")
# --------------------------------------------------------------------------
# page
# --------------------------------------------------------------------------
PAGE = """<!doctype html>
<html lang="en"><head><meta charset="utf-8">
<meta name="viewport" content="width=device-width,initial-scale=1">
<title>%(title)s</title>
<link rel="icon" href="data:image/svg+xml,%%3Csvg xmlns='http://www.w3.org/2000/svg' viewBox='0 0 16 16'%%3E%%3Crect width='16' height='16' rx='2' fill='%%232a2f39'/%%3E%%3Crect x='2' y='3' width='12' height='8' rx='1' fill='%%2368c07a'/%%3E%%3Crect x='6' y='12' width='4' height='1.5' fill='%%2398a0ad'/%%3E%%3C/svg%%3E">
<style>
:root{--bg:#16181c;--tile:#22262e;--tile2:#2a2f39;--line:#3a4150;--fg:#e8eaee;
--dim:#98a0ad;--bad:#4a2222;--badfg:#e08a8a;--ok:#68c07a}
*{box-sizing:border-box}
body{margin:0;background:var(--bg);color:var(--fg);
font:13px/1.4 "Segoe UI",system-ui,-apple-system,sans-serif}
header{display:flex;align-items:center;gap:12px;padding:8px 14px;
background:#1c1f25;border-bottom:1px solid var(--line);
position:sticky;top:0;z-index:5}
header h1{font-size:14px;font-weight:600;margin:0;letter-spacing:.2px}
header .sp{flex:1}
header .meta{color:var(--dim);font-size:11px;font-variant-numeric:tabular-nums}
.dot{width:7px;height:7px;border-radius:50%%;background:var(--ok);display:inline-block;
margin-right:5px;transition:opacity .3s}
.dot.blink{opacity:.25}
main{display:grid;gap:10px;padding:10px;
grid-template-columns:repeat(auto-fill,minmax(%(minw)spx,1fr))}
main.one{grid-template-columns:1fr;padding:10px}
.tile{background:var(--tile);border:1px solid var(--line);border-radius:5px;
overflow:hidden;text-decoration:none;color:inherit;display:block}
.tile.bad{background:var(--bad);border-color:#6a3030}
.bar{display:flex;align-items:center;gap:8px;padding:5px 9px;background:var(--tile2);
border-bottom:1px solid var(--line)}
.tile.bad .bar{background:#3a1e1e;border-bottom-color:#6a3030}
.id{font-weight:700;font-size:13px;font-variant-numeric:tabular-nums}
.nm{color:var(--fg);font-size:12px}
.st{margin-left:auto;color:var(--dim);font-size:11px;font-variant-numeric:tabular-nums;
white-space:nowrap}
.shot{display:block;width:100%%;aspect-ratio:4/3;object-fit:contain;background:#000}
.ph{display:flex;flex-direction:column;align-items:center;justify-content:center;gap:6px;
aspect-ratio:4/3;background:#2c1e1e;color:var(--badfg);text-align:center;padding:14px}
.ph b{font-size:15px;letter-spacing:1.5px}
.ph span{font-size:11px;color:#b89696;max-width:34em;word-break:break-word}
a.back{color:var(--dim);text-decoration:none;font-size:12px}
a.back:hover{color:var(--fg)}
</style></head><body>
<header>
<span class="dot" id="dot"></span>
<h1>%(title)s</h1>
%(nav)s
<span class="sp"></span>
<span class="meta" id="meta">connecting...</span>
</header>
<main id="wall" class="%(cls)s"></main>
<script>
const ONLY = %(only)s, INTERVAL = %(interval)s;
const wall = document.getElementById('wall'), meta = document.getElementById('meta'),
dot = document.getElementById('dot');
const esc = s => String(s).replace(/[&<>"]/g, c =>
({'&':'&amp;','<':'&lt;','>':'&gt;','"':'&quot;'}[c]));
function render(guests){
const want = guests.map(g => String(g.id)).join(',');
if (wall.dataset.keys !== want){ wall.innerHTML = ''; wall.dataset.keys = want;
for (const g of guests){
const el = document.createElement(ONLY ? 'div' : 'a');
el.className = 'tile'; el.id = 'g' + g.id;
if (!ONLY) el.href = '/one/' + g.id;
el.innerHTML = '<div class="bar"><span class="id"></span><span class="nm"></span>'
+ '<span class="st"></span></div><div class="body"></div>';
wall.appendChild(el);
}
}
for (const g of guests){
const el = document.getElementById('g' + g.id); if (!el) continue;
el.classList.toggle('bad', !g.ok);
el.querySelector('.id').textContent = g.id;
el.querySelector('.nm').textContent = g.name;
el.querySelector('.st').textContent =
g.status + (g.ok ? ' ' + g.ts + (g.fresh ? '' : ' (stale)') : ' no signal');
const body = el.querySelector('.body');
if (g.ok){
let img = body.querySelector('img');
if (!img){ body.innerHTML = ''; img = document.createElement('img');
img.className = 'shot'; img.alt = 'VM ' + g.id;
// an undecodable frame must degrade to the placeholder, not a broken icon
img.onerror = () => { el.classList.add('bad'); body.innerHTML =
'<div class="ph"><b>NO SIGNAL</b><span>frame did not decode</span></div>'; };
body.appendChild(img); }
// cache-buster keyed to capture time: no refetch unless the frame changed
const next = '/shot/' + g.id + '.png?t=' + encodeURIComponent(g.ts);
if (img.getAttribute('src') !== next) img.src = next;
} else {
body.innerHTML = '<div class="ph"><b>NO SIGNAL</b><span>'
+ esc(g.error || g.status) + '</span></div>';
}
}
}
async function tick(){
try{
const r = await fetch('/api/state', {cache:'no-store'});
let guests = await r.json();
if (ONLY) guests = guests.filter(g => g.id == ONLY);
render(guests);
const live = guests.filter(g => g.ok).length;
meta.textContent = live + '/' + guests.length + ' framebuffers · '
+ new Date().toLocaleTimeString() + ' · every ' + INTERVAL + 's · spicy';
dot.style.background = 'var(--ok)';
} catch(e){
meta.textContent = 'lost contact with vmwatch — retrying';
dot.style.background = '#c05a5a';
}
dot.classList.add('blink'); setTimeout(() => dot.classList.remove('blink'), 300);
}
tick(); setInterval(tick, INTERVAL * 1000);
</script></body></html>
"""
class Handler(BaseHTTPRequestHandler):
server_version = "vmwatch"
tank: Tank = None # type: ignore[assignment]
def log_message(self, fmt, *a): # quieter than the default one-line-per-image
pass
def _send(self, code, ctype, body: bytes, cache: str = "no-store"):
self.send_response(code)
self.send_header("Content-Type", ctype)
self.send_header("Content-Length", str(len(body)))
self.send_header("Cache-Control", cache)
self.end_headers()
try:
self.wfile.write(body)
except (BrokenPipeError, ConnectionResetError):
pass
def _page(self, only: int | None):
title = f"SOTS guest {only}" if only else "SOTS lab guests"
nav = '<a class="back" href="/">&larr; all guests</a>' if only else ""
body = PAGE % {
"title": html.escape(title),
"nav": nav,
"cls": "one" if only else "",
"only": only if only else "null",
"interval": self.tank.interval,
"minw": 900 if only else 460,
}
self._send(HTTPStatus.OK, "text/html; charset=utf-8", body.encode())
def do_GET(self): # noqa: N802
path = self.path.split("?", 1)[0]
self.tank.touch()
if path == "/":
return self._page(None)
if m := re.fullmatch(r"/one/(\d+)/?", path):
return self._page(int(m.group(1)))
if path == "/api/state":
return self._send(
HTTPStatus.OK, "application/json", json.dumps(self.tank.state()).encode()
)
if m := re.fullmatch(r"/shot/(\d+)\.png", path):
png = self.tank.png(int(m.group(1)))
if png:
# immutable: the URL carries the capture timestamp
return self._send(HTTPStatus.OK, "image/png", png, "max-age=30")
return self._send(HTTPStatus.NOT_FOUND, "text/plain", b"no framebuffer\n")
if path == "/healthz":
t = self.tank
up = int(time.time() - t.started)
return self._send(
HTTPStatus.OK,
"text/plain",
f"ok guests={len(t.get_fleet())} cycles={t.cycles} uptime={up}s\n".encode(),
)
return self._send(HTTPStatus.NOT_FOUND, "text/plain", b"not found\n")
def main() -> int:
ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
ap.add_argument("--port", type=int, default=DEFAULT_PORT)
ap.add_argument("--bind", default="0.0.0.0")
ap.add_argument("--interval", type=float, default=DEFAULT_INTERVAL)
args = ap.parse_args()
tank = Tank(args.interval)
Handler.tank = tank
threading.Thread(target=tank.loop, daemon=True, name="poller").start()
srv = ThreadingHTTPServer((args.bind, args.port), Handler)
srv.daemon_threads = True
print(
f"vmwatch: http://{args.bind}:{args.port}/ interval={args.interval}s "
f"guests={[g['id'] for g in tank.get_fleet()]}",
flush=True,
)
try:
srv.serve_forever()
except KeyboardInterrupt:
pass
return 0
if __name__ == "__main__":
raise SystemExit(main())