#!/usr/bin/env python3 """Collect a non-secret host capacity observation over SSH. Does not dump cloud-init metadata wholesale (that blob can contain admin_pass). Only named keys are read. Usage: scripts/observe-host-capacity.py Railiance01 scripts/observe-host-capacity.py Railiance01 > docs/evidence/resource-hosteurope-railiance01/observations/....json """ from __future__ import annotations import json import os import subprocess import sys from datetime import datetime, timezone REMOTE = r""" set -euo pipefail python3 - <<'PY' import json, os, subprocess, shutil def sh(cmd): return subprocess.check_output(cmd, shell=True, text=True).strip() def cloud(key): if not shutil.which("cloud-init"): return None try: out = subprocess.check_output( ["cloud-init", "query", key], text=True, stderr=subprocess.DEVNULL ).strip() except subprocess.CalledProcessError: return None if not out or out.lower() in {"none", "null"}: return None return out mem = open("/proc/meminfo").read().split() kv = {} for i, tok in enumerate(mem): if tok.endswith(":") and i + 1 < len(mem): try: kv[tok[:-1]] = int(mem[i + 1]) except ValueError: pass st = os.statvfs("/") k3s = {} if shutil.which("k3s"): raw = subprocess.check_output( ["sudo", "k3s", "kubectl", "get", "node", "-o", "json"], text=True ) node = json.loads(raw)["items"][0] cap = node["status"]["capacity"] alloc = node["status"]["allocatable"] k3s = { "node_name": node["metadata"]["name"], "capacity": { "cpu": cap.get("cpu"), "memory": cap.get("memory"), "ephemeral_storage": cap.get("ephemeral-storage"), "pods": cap.get("pods"), }, "allocatable": { "cpu": alloc.get("cpu"), "memory": alloc.get("memory"), "ephemeral_storage": alloc.get("ephemeral-storage"), "pods": alloc.get("pods"), }, } try: top = subprocess.check_output( ["sudo", "k3s", "kubectl", "top", "node", "--no-headers"], text=True, stderr=subprocess.DEVNULL, ).split() # NAME CPU(cores) CPU% MEMORY(bytes) MEMORY% if len(top) >= 5: k3s["observed"] = { "cpu": top[1], "cpu_pct": top[2], "memory": top[3], "memory_pct": top[4], } except subprocess.CalledProcessError: k3s["observed"] = None print(json.dumps({ "hostname": sh("hostname -f"), "os_pretty": sh("grep ^PRETTY_NAME= /etc/os-release").split("=", 1)[1].strip('"'), "nproc": int(sh("nproc")), "cpu_model": sh("lscpu | awk -F: '/Model name/ {print $2; exit}'").strip(), "hypervisor": sh("lscpu | awk -F: '/Hypervisor vendor/ {print $2; exit}'").strip() or None, "mem_total_kib": kv.get("MemTotal"), "mem_available_kib": kv.get("MemAvailable"), "swap_total_kib": kv.get("SwapTotal"), "root_total_bytes": st.f_frsize * st.f_blocks, "root_used_bytes": st.f_frsize * (st.f_blocks - st.f_bfree), "root_avail_bytes": st.f_frsize * st.f_bavail, "block_devices": sh("lsblk -dn -o NAME,SIZE,TYPE,MODEL"), "loadavg": open("/proc/loadavg").read().split()[:3], "uptime_seconds": float(open("/proc/uptime").read().split()[0]), "instance_type": cloud("ds.ec2_metadata.instance-type"), "instance_id_ec2": cloud("ds.ec2_metadata.instance-id"), "instance_uuid": cloud("ds.meta_data.uuid"), "instance_name": cloud("ds.meta_data.name"), "product_family": cloud("ds.meta_data.meta.app"), "product_role": cloud("ds.meta_data.meta.role"), "local_hostname": cloud("ds.ec2_metadata.local-hostname"), "project_id": cloud("ds.meta_data.project_id"), "k3s": k3s or None, })) PY """ def inventory_ssh(name: str) -> tuple[str, str]: import yaml path = os.path.join(os.path.dirname(__file__), "..", "inventory", "servers.yaml") with open(path) as f: servers = yaml.safe_load(f).get("servers") or [] for s in servers: if s.get("name") == name: return s["ip"], s.get("ssh_user", "tegwick") raise SystemExit(f"unknown inventory host: {name}") def main() -> int: if len(sys.argv) != 2: print("Usage: observe-host-capacity.py ", file=sys.stderr) return 2 name = sys.argv[1] ip, user = inventory_ssh(name) observed_at = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") raw = subprocess.check_output( ["ssh", "-o", "BatchMode=yes", "-o", "ConnectTimeout=15", f"{user}@{ip}", REMOTE], text=True, ) host = json.loads(raw) rec = { "schema_version": "1.0", "record_type": "host_capacity_observation", "resource_id": "resource:hosteurope:railiance01" if name == "Railiance01" else f"resource:host:{name.lower()}", "inventory_name": name, "observed_at": observed_at, "observer": "railiance-infra/scripts/observe-host-capacity.py", "source_commands": [ "nproc / lscpu /proc/meminfo /proc/loadavg statvfs(/)", "cloud-init query ", "sudo k3s kubectl get node -o json", "sudo k3s kubectl top node", ], "layers": { "provider_declared": { "instance_type": host.get("instance_type"), "instance_uuid": host.get("instance_uuid"), "instance_id_ec2": host.get("instance_id_ec2"), "instance_name": host.get("instance_name"), "product_family": host.get("product_family"), "product_role": host.get("product_role"), "local_hostname": host.get("local_hostname"), "project_id": host.get("project_id"), "note": "Declared by the provider metadata service. Not a booked SKU confirmation.", }, "operating_system": { "hostname": host["hostname"], "os": host["os_pretty"], "cpu": { "value": host["nproc"], "unit": "vCPU", "kind": "usable", "model": host.get("cpu_model"), "hypervisor": host.get("hypervisor"), }, "memory": { "total_kib": host["mem_total_kib"], "available_kib": host["mem_available_kib"], "unit": "KiB", "kind": "usable", }, "swap": { "total_kib": host["swap_total_kib"], "unit": "KiB", "kind": "observed", "note": "OS swapfile, not a provider disk grant", }, "root_filesystem": { "total_bytes": host["root_total_bytes"], "used_bytes": host["root_used_bytes"], "avail_bytes": host["root_avail_bytes"], "kind": "provisioned", }, "block_devices": host["block_devices"], "loadavg": host["loadavg"], "uptime_seconds": host["uptime_seconds"], }, "kubernetes": host.get("k3s"), }, "unknown": [ { "field": "traffic_allowance", "classification": "unknown", "owner": "operator", "source": "Host Europe control panel or contract", } ], } json.dump(rec, sys.stdout, indent=2) sys.stdout.write("\n") return 0 if __name__ == "__main__": raise SystemExit(main())