#!/usr/bin/env python3 # pve_mapper.py # Frägt die Proxmox-Cluster-Infrastruktur sowie Docker-Container ab # und generiert einen strukturierten digitalen Zwilling (JSON). import json import subprocess import os import sys HOST = "80.90.43.178" SSH_KEY = os.path.expanduser("~/.ssh/id_rsa") def run_ssh(host, command, user="root"): """Führt einen Befehl über SSH aus und gibt das Ergebnis zurück.""" ssh_cmd = [ "ssh", "-i", SSH_KEY, "-o", "StrictHostKeyChecking=no", "-o", "UserKnownHostsFile=/dev/null", f"{user}@{host}", command ] try: res = subprocess.run(ssh_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, timeout=15) if res.returncode == 0: return res.stdout.strip() else: return None except subprocess.TimeoutExpired: return None except Exception: return None def main(): print("=== 1. Fräge Proxmox Cluster-Ressourcen ab ===") pve_raw = run_ssh(HOST, "pvesh get /cluster/resources --output-format json") if not pve_raw: print("Fehler: Konnte keine Verbindung zum Proxmox-Host herstellen oder pvesh ausführen.") sys.exit(1) try: resources = json.loads(pve_raw) except Exception as e: print(f"Fehler beim Parsen der Proxmox-Ressourcen: {e}") sys.exit(1) twin = { "nodes": [], "vms": [], "lxcs": [], "docker_hosts": {} } # Bekannte Docker-Hosts für tiefere Inspektion known_docker_hosts = { "80.90.43.178": "pve-host", "10.190.20.10": "venus-traefik", "10.190.20.50": "venus-rustdesk" } for res in resources: res_type = res.get("type") res_name = res.get("name") res_node = res.get("node") vmid = res.get("vmid") status = res.get("status") if res_type == "node": twin["nodes"].append({ "name": res.get("node"), "status": res.get("status"), "cpu_cores": res.get("maxcpu"), "memory_max_gb": round(res.get("maxmem", 0) / (1024**3), 2) if res.get("maxmem") else 0 }) continue if res_type not in ["qemu", "lxc"]: continue item = { "vmid": vmid, "name": res_name, "node": res_node, "status": status, "cpu_cores": res.get("maxcpu"), "memory_max_gb": round(res.get("maxmem", 0) / (1024**3), 2) if res.get("maxmem") else 0, "ips": [] } if status == "running": # IP-Adressen ermitteln if res_type == "qemu": ip_raw = run_ssh(HOST, f"pvesh get /nodes/{res_node}/qemu/{vmid}/agent/network-get-interfaces --output-format json") if ip_raw: try: interfaces = json.loads(ip_raw).get("result", []) for iface in interfaces: for addr in iface.get("ip-addresses", []): ip = addr.get("ip-address") if ip and not ip.startswith("127.") and not ip.startswith("172.") and ":" not in ip: item["ips"].append(ip) except Exception: pass elif res_type == "lxc": ip_raw = run_ssh(HOST, f"pvesh get /nodes/{res_node}/lxc/{vmid}/interfaces --output-format json") if ip_raw: try: interfaces = json.loads(ip_raw) for iface in interfaces: for addr in iface.get("ip-addresses", []): ip = addr.get("ip-address") if ip and not ip.startswith("127.") and not ip.startswith("172.") and ":" not in ip: item["ips"].append(ip) except Exception: pass if res_type == "qemu": twin["vms"].append(item) else: twin["lxcs"].append(item) print("=== 2. Fräge Docker-Container auf aktiven Hosts ab ===") for ip, label in known_docker_hosts.items(): print(f"Inspektion von Docker-Host: {label} ({ip})") # Port 22 oder ProxyJump-Auswertung # PVE-Host direkt, VMs über SSH-Routing (das lokale ssh-config kümmert sich um das Proxy-Routing) target = ip if label == "pve-host": docker_cmd = "docker ps --format '{{json .}}'" else: # Wir verwenden die in ~/.ssh/config definierten Host-Namen (venus-traefik, venus-rustdesk) target = label docker_cmd = "docker ps --format '{{json .}}'" docker_raw = run_ssh(target, docker_cmd) if docker_raw: twin["docker_hosts"][label] = { "ip": ip, "containers": [] } # Mehrere JSON-Objekte zeilenweise parsen for line in docker_raw.split("\n"): if not line.strip(): continue try: c = json.loads(line) twin["docker_hosts"][label]["containers"].append({ "id": c.get("ID"), "name": c.get("Names"), "image": c.get("Image"), "status": c.get("Status"), "ports": c.get("Ports") }) except Exception: pass # Speicherpfad definieren state_dir = "/Volumes/omv-150/MayaDO/Infrastruktur/agent-bootstrap/state" os.makedirs(state_dir, exist_ok=True) out_path = os.path.join(state_dir, "digital_twin.json") with open(out_path, "w", encoding="utf-8") as f: json.dump(twin, f, indent=2, ensure_ascii=False) print(f"=== 3. Digitaler Zwilling erfolgreich geschrieben nach: {out_path} ===") # Generiere Markdown-Zusammenfassung für Graphify md_lines = [ "# Digitaler Zwilling: Proxmox & Docker Cluster-Topologie", "", "Dieses Dokument beschreibt die physische und virtuelle Infrastruktur von MayaDO.", "" ] md_lines.append("## Physische Cluster-Nodes") for n in twin["nodes"]: md_lines.append(f"### Node: {n['name']}") md_lines.append(f"- **Status**: {n['status']}") md_lines.append(f"- **CPU Kerne**: {n['cpu_cores']}") md_lines.append(f"- **Maximaler Arbeitsspeicher**: {n['memory_max_gb']} GB") md_lines.append("") md_lines.append("## Virtuelle Maschinen (Qemu/VM)") for vm in twin["vms"]: md_lines.append(f"### VM {vm['vmid']}: {vm['name']}") md_lines.append(f"- **Node**: {vm['node']}") md_lines.append(f"- **Status**: {vm['status']}") md_lines.append(f"- **CPU Kerne**: {vm['cpu_cores']}") md_lines.append(f"- **RAM Limit**: {vm['memory_max_gb']} GB") if vm["ips"]: md_lines.append(f"- **IP-Adressen**: {', '.join(vm['ips'])}") md_lines.append("") md_lines.append("## LXC-Container") for lxc in twin["lxcs"]: md_lines.append(f"### LXC {lxc['vmid']}: {lxc['name']}") md_lines.append(f"- **Node**: {lxc['node']}") md_lines.append(f"- **Status**: {lxc['status']}") md_lines.append(f"- **CPU Kerne**: {lxc['cpu_cores']}") md_lines.append(f"- **RAM Limit**: {lxc['memory_max_gb']} GB") if lxc["ips"]: md_lines.append(f"- **IP-Adressen**: {', '.join(lxc['ips'])}") md_lines.append("") md_lines.append("## Docker-Dienste und -Container") for host_label, host_info in twin["docker_hosts"].items(): md_lines.append(f"### Docker-Host: {host_label} (IP: {host_info['ip']})") if not host_info["containers"]: md_lines.append("- *Keine laufenden Container gefunden*") for c in host_info["containers"]: md_lines.append(f"#### Container: {c['name']}") md_lines.append(f" - **ID**: `{c['id']}`") md_lines.append(f" - **Image**: `{c['image']}`") md_lines.append(f" - **Status**: {c['status']}") if c['ports']: md_lines.append(f" - **Ports**: `{c['ports']}`") md_lines.append("") md_path = os.path.join(state_dir, "digital_twin.md") with open(md_path, "w", encoding="utf-8") as f: f.write("\n".join(md_lines)) print(f"=== 4. Markdown-Zusammenfassung geschrieben nach: {md_path} ===") if __name__ == "__main__": main()