first commit

This commit is contained in:
duffyduck
2026-07-30 16:52:58 +02:00
commit 87f1b4d555
17 changed files with 3558 additions and 0 deletions
+268
View File
@@ -0,0 +1,268 @@
# pvesnap — automatische Snapshots für Proxmox VE
Ein systemd-Dienst, der nach frei definierbaren Zeitplänen Snapshots von VMs und
Containern anlegt und alte Snapshots nach einer einstellbaren Vorhaltezeit wieder
löscht. Alles wird über **eine INI-Datei** gesteuert, die sich auch bequem in einem
**ncurses-Editor** bearbeiten lässt.
* Beliebig viele **Gruppen** — z. B. „diese VMs stündlich, jene täglich, der Rest monatlich“
* **VM-Auswahl** nach VMID, Name (mit `*`-Mustern), Tag, Pool oder einfach „alle“
* **Vorhaltezeit** je Gruppe nach Anzahl *und/oder* Alter
* Jeder Snapshot bekommt eine **Beschreibung**, an der er sich wiedererkennen lässt
* Läuft cluster-weit (alle Nodes) über `pvesh`
* Nur Python-Standardbibliothek — keine zusätzlichen Pakete nötig
---
## Installation
Auf dem Proxmox-Host als root:
```bash
git clone <dieses-repo> pvesnap && cd pvesnap
./install.sh
```
Das Skript legt an:
| Pfad | Inhalt |
|---|---|
| `/usr/lib/pvesnap/` | Programmcode |
| `/usr/local/bin/pvesnap` | Startbefehl |
| `/etc/pvesnap/pvesnap.conf` | Konfiguration (wird bei Updates **nicht** überschrieben) |
| `/var/lib/pvesnap/state.json` | merkt sich die letzten Läufe |
| `/etc/systemd/system/pvesnap.service` | systemd-Unit |
In der mitgelieferten Beispielkonfiguration sind alle Gruppen auf `enabled = no`
gesetzt — es passiert also erst etwas, wenn du eine Gruppe aktivierst.
**Deinstallation:**
```bash
./uninstall.sh # Programm und Dienst entfernen, Konfiguration bleibt
./uninstall.sh --purge # zusätzlich /etc/pvesnap und /var/lib/pvesnap löschen
```
Bereits angelegte Snapshots werden dabei **nie** angerührt.
---
## Konfiguration im ncurses-Editor
```bash
pvesnap config
```
* **Gruppenliste** — `Enter` bearbeiten, `n` neu, `c` kopieren, `d` löschen,
`Leertaste` an/aus, `g` globale Einstellungen, `v` Übersicht welche VM in
welcher Gruppe landet, `s` speichern, `q` Ende
* **Gruppe bearbeiten** — Feld auswählen, `Enter` ändern, `Leertaste` umschalten.
Ganz unten steht live, auf wie viele Gäste die Auswahl gerade zutrifft.
* **VM-Auswahl** (`v`) — Liste aller VMs und Container mit Node, Status und Tags;
`Leertaste` auswählen, `a` alle, `n` keine, `i` umkehren, `/` filtern,
`Enter` übernehmen.
> Beim Speichern wird die INI-Datei neu geschrieben. Eigene Kommentare und ein
> `[defaults]`-Abschnitt gehen dabei verloren (die Werte bleiben inhaltlich
> erhalten, sie stehen danach in jeder Gruppe einzeln). Eine Sicherung wird als
> `pvesnap.conf.bak` abgelegt.
---
## Konfiguration von Hand
```ini
[global]
prefix = auto # Namenspräfix aller pvesnap-Snapshots
check_interval = 60s # wie oft der Dienst nach Fälligem schaut
state_file = /var/lib/pvesnap/state.json
log_level = INFO
task_timeout = 15m
run_on_start = no # beim Start sofort einen Durchlauf machen?
dry_run = no # yes = nichts wirklich tun, nur protokollieren
description = pvesnap | Gruppe: {group} | erstellt: {datetime} | Vorhaltezeit: {keep_time}
[defaults] # Vorgaben für alle Gruppen (optional)
skip_stopped = no
[group:stündlich]
interval = 1h
align = yes # an der Uhr ausrichten: 00:00, 01:00, ...
keep_count = 24 # höchstens 24 Snapshots je VM
keep_time = 2d # nichts älter als 2 Tage
tags = stuendlich # alle VMs mit diesem Tag
skip_stopped = yes
[group:täglich]
schedule = daily
at = 02:30
keep_count = 14
keep_time = 21d
names = web-*, db-*
[group:monatlich]
schedule = monthly
day_of_month = 1
at = 04:00
keep_count = 6
all = yes
exclude_tags = nosnap
```
Nach jeder Änderung:
```bash
pvesnap check # Konfiguration prüfen
systemctl reload pvesnap # Dienst übernimmt die Änderung ohne Neustart
```
### Zeitplan
Entweder ein Intervall …
```ini
interval = 30m # 30m, 1h, 6h, 2d12h, 1w ...
align = yes # yes: an der Uhr ausgerichtet (00:00, 00:30, 01:00, ...)
# no: 30 Minuten nach dem letzten Lauf
```
… oder ein fester Termin:
| `schedule` | zusätzliche Angaben | Beispiel |
|---|---|---|
| `hourly` | `minute = 15` | jede Stunde um :15 |
| `daily` | `at = 02:30` | täglich um 02:30 |
| `weekly` | `at`, `day_of_week = so` | sonntags um 03:00 |
| `monthly` | `at`, `day_of_month = 1` | am 1. jedes Monats |
| `yearly` | `at`, `day_of_month`, `month` | einmal jährlich |
`day_of_month = 31` wird in kürzeren Monaten automatisch auf den letzten Tag
gezogen. Läuft der Host zum Termin nicht, wird der Lauf beim nächsten Start
nachgeholt.
### Vorhaltezeit
```ini
keep_count = 24 # höchstens 24 Snapshots je VM und Gruppe (0 = unbegrenzt)
keep_time = 7d # nichts älter als 7 Tage (0 = unbegrenzt)
keep_min = 1 # so viele bleiben in jedem Fall stehen
```
Beides ist kombinierbar — gelöscht wird, was *eine* der beiden Grenzen reißt.
Mindestens eine der beiden Angaben muss gesetzt sein, sonst würde die Zahl der
Snapshots unbegrenzt wachsen (`pvesnap check` weist darauf hin).
### Welche VMs?
Alle Angaben wirken als **ODER**, Ausschlüsse gewinnen immer:
```ini
all = yes # alle VMs und Container
vmids = 100,101,105-110 # nach ID, auch Bereiche
names = web-*, db-0? # nach Name, mit Platzhaltern
tags = prod, wichtig # nach Proxmox-Tag
pools = Kunden # nach Proxmox-Pool
types = qemu # nur VMs (bzw. lxc = nur Container; leer = beides)
exclude_vmids = 999
exclude_names = *-test
exclude_tags = nosnap
```
### Snapshot-Optionen
```ini
vmstate = yes # RAM mitsichern (nur QEMU und nur bei laufender VM)
skip_stopped = yes # gestoppte Gäste überspringen
description = Sicherung von {name} ({vmid}) vom {date}
```
Platzhalter der Beschreibung: `{group}` `{group_slug}` `{vmid}` `{name}` `{node}`
`{type}` `{pool}` `{tags}` `{date}` `{time}` `{datetime}` `{timestamp}`
`{keep_time}` `{keep_count}` `{schedule}`
---
## Namensschema und Sicherheit
```
auto-taeglich-20260730-023000
│ │ │ └── Uhrzeit
│ │ └────────── Datum
│ └──────────────────── Kurzname der Gruppe
└─────────────────────────── prefix aus [global]
```
pvesnap löscht **ausschließlich** Snapshots, deren Name exakt auf dieses Muster
passt und deren Gruppen-Kurzname zur jeweiligen Gruppe gehört. Von Hand oder von
anderen Werkzeugen angelegte Snapshots bleiben garantiert unangetastet.
Umlaute und Sonderzeichen im Gruppennamen werden für den Kurznamen umgeschrieben
(`täglich``taeglich`). Wenn zwei Gruppen denselben Kurznamen ergäben, meldet
das `pvesnap check`.
---
## Befehle
```bash
pvesnap status # Übersicht: Gruppen, letzte und nächste Läufe
pvesnap vms # alle VMs und die Gruppen, in denen sie stecken
pvesnap list # vorhandene pvesnap-Snapshots (-a = auch fremde)
pvesnap check # Konfiguration prüfen
pvesnap config # ncurses-Editor
pvesnap run # jetzt fällige Gruppen ausführen
pvesnap run --force -g täglich # eine Gruppe sofort ausführen
pvesnap run --force --dry-run # Probelauf, ändert nichts
pvesnap prune -g stündlich # nur aufräumen, keine neuen Snapshots
systemctl status pvesnap
systemctl reload pvesnap # Konfiguration neu einlesen (SIGHUP)
journalctl -u pvesnap -f # Protokoll mitlesen
```
Ein manueller `pvesnap run` ist auch möglich, während der Dienst läuft — beide
teilen sich eine Sperre und kommen sich nicht in die Quere.
Mit `-c /pfad/zur.conf` lässt sich jederzeit eine andere Konfiguration verwenden,
z. B. zum Ausprobieren.
---
## Hinweise aus der Praxis
* **Snapshot ist kein Backup.** Snapshots liegen auf demselben Storage wie die VM.
Für echte Sicherungen zusätzlich `vzdump` / Proxmox Backup Server verwenden.
* **Speicherplatz:** Jeder Snapshot hält alte Blöcke fest. Bei schreibfreudigen
VMs lieber kurze Vorhaltezeiten wählen und den Storage im Auge behalten.
* **Nicht jedes Storage kann Snapshots** — LVM-thick und Verzeichnis-Storage mit
`raw`-Images können es nicht. Solche VMs melden einen Fehler im Protokoll; die
übrigen laufen normal weiter.
* **`vmstate = yes`** friert die VM kurz ein und braucht Platz in Höhe des
zugewiesenen RAM. Für regelmäßige Läufe meist unnötig.
* **Gruppe umbenennen:** Der neue Name bekommt einen eigenen Kurznamen. Snapshots
unter dem alten Namen werden dann nicht mehr automatisch aufgeräumt — vorher
aufräumen lassen oder die alten Snapshots von Hand entfernen.
* **Zeitumstellung:** Es wird mit lokaler Zeit gerechnet. Ein täglicher Termin um
02:30 kann in der Umstellungsnacht ausfallen oder doppelt anstehen.
---
## Aufbau
```
pvesnap/
config.py INI lesen und schreiben, Gruppen- und Globaleinstellungen
schedule.py Berechnung des nächsten Termins
naming.py Namensschema und Beschreibungs-Vorlagen
proxmox.py pvesh-Anbindung (Inventar, Snapshots anlegen/löschen)
engine.py Auswahl der Gäste, Anlegen, Aufräumen
daemon.py Hauptschleife, Signale, Sperren
state.py merkt sich die letzten Läufe
tui.py ncurses-Editor
cli.py Kommandozeile
config/pvesnap.conf.example
systemd/pvesnap.service
install.sh uninstall.sh
```
+112
View File
@@ -0,0 +1,112 @@
# =====================================================================
# pvesnap - Konfiguration
# =====================================================================
#
# Nach jeder Aenderung: systemctl reload pvesnap
# Pruefen: pvesnap check
# Komfortabel bearbeiten: pvesnap config (ncurses-Editor)
#
# Zeitangaben ueberall: 30m 1h 6h 2d12h 1w (0 = unbegrenzt)
# ---------------------------------------------------------------------
[global]
# Praefix aller von pvesnap angelegten Snapshots. Nur Snapshots, die mit
# diesem Praefix beginnen, werden jemals automatisch geloescht - von Hand
# angelegte Snapshots bleiben also garantiert unangetastet.
# Namensschema: <prefix>-<gruppe>-<JJJJMMTT>-<HHMMSS>
prefix = auto
# Wie oft der Dienst prueft, ob eine Gruppe faellig ist.
check_interval = 60s
# Merkt sich, wann welche Gruppe zuletzt gelaufen ist.
state_file = /var/lib/pvesnap/state.json
log_level = INFO
# log_file = /var/log/pvesnap.log
# Maximale Wartezeit auf einen einzelnen Snapshot-Task in Proxmox.
task_timeout = 15m
# yes = beim Dienststart sofort einen Durchlauf machen,
# no = auf den naechsten regulaeren Termin warten.
run_on_start = no
# yes = nichts wirklich anlegen/loeschen, nur protokollieren.
dry_run = no
# Beschreibung, die an jedem Snapshot haengt (in der Proxmox-Oberflaeche
# sichtbar). Platzhalter:
# {group} {group_slug} {vmid} {name} {node} {type} {pool} {tags}
# {date} {time} {datetime} {timestamp} {keep_time} {keep_count} {schedule}
description = pvesnap | Gruppe: {group} | erstellt: {datetime} | Vorhaltezeit: {keep_time} | max: {keep_count}
# ---------------------------------------------------------------------
# [defaults] gilt als Vorgabe fuer *alle* Gruppen und kann in jeder
# Gruppe einzeln ueberschrieben werden.
# ---------------------------------------------------------------------
[defaults]
enabled = yes
skip_stopped = no
vmstate = no
# ---------------------------------------------------------------------
# Gruppen - beliebig viele, Abschnittsname: [group:NAME]
# ---------------------------------------------------------------------
#
# Zeitplan (entweder ... oder):
# interval = 30m alle 30 Minuten
# align = yes an der Uhr ausrichten (00:00, 00:30, 01:00 ...)
# oder
# schedule = hourly + minute = 15
# schedule = daily + at = 02:30
# schedule = weekly + at = 03:00, day_of_week = so
# schedule = monthly + at = 04:00, day_of_month = 1
# schedule = yearly + at = 04:00, day_of_month = 1, month = 1
#
# Vorhaltezeit (beides kombinierbar, 0 = unbegrenzt):
# keep_count = 24 hoechstens 24 Snapshots je VM
# keep_time = 7d nichts aelter als 7 Tage
# keep_min = 1 so viele bleiben immer stehen
#
# Auswahl der Gaeste (alles kombinierbar, wirkt als ODER):
# all = yes
# vmids = 100,101,105-110
# names = web-*, db-*
# tags = prod, wichtig
# pools = Kunden
# types = qemu (oder lxc; leer = beides)
# exclude_vmids = 999
# exclude_names = *-test
# exclude_tags = nosnap
# ---------------------------------------------------------------------
[group:stuendlich]
enabled = no
interval = 1h
align = yes
keep_count = 24
keep_time = 2d
tags = stuendlich
skip_stopped = yes
[group:taeglich]
enabled = no
schedule = daily
at = 02:30
keep_count = 14
keep_time = 21d
names = web-*, db-*
[group:monatlich]
enabled = no
schedule = monthly
day_of_month = 1
at = 04:00
keep_count = 6
keep_time = 400d
all = yes
exclude_tags = nosnap
description = Monatssicherung {name} ({vmid}) vom {date}
Executable
+133
View File
@@ -0,0 +1,133 @@
#!/bin/bash
# ---------------------------------------------------------------------
# pvesnap - Installation
#
# ./install.sh installieren, Dienst aktivieren und starten
# ./install.sh --no-start installieren, aber Dienst nicht starten
# ./install.sh --force auch ohne erkanntes Proxmox VE installieren
# ---------------------------------------------------------------------
set -euo pipefail
LIB_DIR="/usr/lib/pvesnap"
BIN="/usr/local/bin/pvesnap"
CONF_DIR="/etc/pvesnap"
CONF="${CONF_DIR}/pvesnap.conf"
STATE_DIR="/var/lib/pvesnap"
DOC_DIR="/usr/share/doc/pvesnap"
UNIT="/etc/systemd/system/pvesnap.service"
SRC="$(cd "$(dirname "$0")" && pwd)"
START=1
FORCE=0
for arg in "$@"; do
case "$arg" in
--no-start) START=0 ;;
--force) FORCE=1 ;;
-h|--help) sed -n '2,8p' "$0" | sed 's/^#[[:space:]]\{0,1\}//'; exit 0 ;;
*) echo "Unbekannte Option: $arg" >&2; exit 1 ;;
esac
done
info() { printf '\033[1;32m==>\033[0m %s\n' "$*"; }
warn() { printf '\033[1;33m==>\033[0m %s\n' "$*" >&2; }
fail() { printf '\033[1;31m==>\033[0m %s\n' "$*" >&2; exit 1; }
[ "$(id -u)" -eq 0 ] || fail "Bitte als root ausfuehren (sudo ./install.sh)."
command -v python3 >/dev/null 2>&1 || fail "python3 wird benoetigt."
python3 - <<'PY' || fail "Python 3.7 oder neuer wird benoetigt."
import sys
sys.exit(0 if sys.version_info >= (3, 7) else 1)
PY
if ! command -v pvesh >/dev/null 2>&1; then
if [ "$FORCE" -eq 1 ]; then
warn "pvesh nicht gefunden - installiere trotzdem (--force)."
START=0
else
fail "pvesh nicht gefunden. pvesnap gehoert auf einen Proxmox-VE-Host.
Mit --force laesst sich die Installation trotzdem erzwingen."
fi
fi
command -v systemctl >/dev/null 2>&1 || fail "systemd wird benoetigt."
[ -d "${SRC}/pvesnap" ] || fail "Verzeichnis 'pvesnap' nicht gefunden (Aufruf aus dem Projektordner?)."
# --- Programm ---------------------------------------------------------
info "Installiere Programm nach ${LIB_DIR}"
rm -rf "${LIB_DIR}/pvesnap"
install -d -m 0755 "${LIB_DIR}"
cp -r "${SRC}/pvesnap" "${LIB_DIR}/pvesnap"
find "${LIB_DIR}/pvesnap" -name '__pycache__' -type d -exec rm -rf {} + 2>/dev/null || true
chmod -R go-w "${LIB_DIR}/pvesnap"
info "Installiere Startbefehl ${BIN}"
cat > "${BIN}" <<EOF
#!/bin/sh
# von install.sh erzeugt
PYTHONPATH="${LIB_DIR}\${PYTHONPATH:+:\$PYTHONPATH}" exec /usr/bin/python3 -m pvesnap "\$@"
EOF
chmod 0755 "${BIN}"
# --- Konfiguration ----------------------------------------------------
install -d -m 0755 "${CONF_DIR}" "${STATE_DIR}" "${DOC_DIR}"
if [ -f "${CONF}" ]; then
info "Vorhandene Konfiguration bleibt unveraendert: ${CONF}"
install -m 0644 "${SRC}/config/pvesnap.conf.example" "${CONF_DIR}/pvesnap.conf.example"
else
info "Lege Beispielkonfiguration an: ${CONF}"
install -m 0640 "${SRC}/config/pvesnap.conf.example" "${CONF}"
install -m 0644 "${SRC}/config/pvesnap.conf.example" "${CONF_DIR}/pvesnap.conf.example"
NEW_CONFIG=1
fi
if [ -f "${SRC}/README.md" ]; then
install -m 0644 "${SRC}/README.md" "${DOC_DIR}/README.md"
fi
# --- systemd ----------------------------------------------------------
info "Installiere systemd-Unit ${UNIT}"
install -m 0644 "${SRC}/systemd/pvesnap.service" "${UNIT}"
systemctl daemon-reload
info "Pruefe Konfiguration"
if ! "${BIN}" --config "${CONF}" check; then
warn "Die Konfiguration ist noch nicht vollstaendig - Dienst wird nicht gestartet."
START=0
fi
if systemctl enable pvesnap.service >/dev/null 2>&1; then
info "Dienst beim Systemstart aktiviert"
else
warn "Dienst konnte nicht aktiviert werden - bitte 'systemctl enable pvesnap' pruefen."
fi
if [ "$START" -eq 1 ]; then
info "Starte Dienst"
systemctl restart pvesnap.service
sleep 1
systemctl --no-pager --lines=5 status pvesnap.service || true
else
warn "Dienst wurde nicht gestartet."
fi
cat <<EOF
Fertig.
Konfiguration bearbeiten : pvesnap config (ncurses-Editor)
nano ${CONF}
Konfiguration pruefen : pvesnap check
Uebersicht : pvesnap status
VMs und Gruppen : pvesnap vms
Testlauf ohne Aenderung : pvesnap run --force --dry-run
Nach Aenderungen : systemctl reload pvesnap
Protokoll : journalctl -u pvesnap -f
EOF
if [ "${NEW_CONFIG:-0}" -eq 1 ]; then
cat <<EOF
Hinweis: In der Beispielkonfiguration sind alle Gruppen mit "enabled = no"
angelegt. Erst nach dem Aktivieren einer Gruppe werden Snapshots erstellt.
EOF
fi
+8
View File
@@ -0,0 +1,8 @@
"""pvesnap - automatische Snapshots fuer Proxmox VE.
Gruppenbasierte Zeitplanung und Vorhaltezeiten, konfiguriert ueber eine
INI-Datei, betrieben als systemd-Dienst.
"""
__version__ = "1.0.0"
__all__ = ["__version__"]
+6
View File
@@ -0,0 +1,6 @@
import sys
from .cli import main
if __name__ == "__main__":
sys.exit(main())
+430
View File
@@ -0,0 +1,430 @@
"""Kommandozeile von pvesnap."""
from __future__ import annotations
import argparse
import logging
import os
import sys
from datetime import datetime
from . import __version__
from .config import DEFAULT_CONFIG_PATH, ConfigError, load_config
from .daemon import Daemon, SingleInstanceLock
from .engine import run_group, select_guests
from .naming import parse_name
from .proxmox import Proxmox, ProxmoxError, pvesh_available
from .schedule import describe, next_due
from .state import State
from .util import format_duration, truncate
log = logging.getLogger("pvesnap")
# ---------------------------------------------------------------------------
# Hilfsfunktionen
# ---------------------------------------------------------------------------
def setup_logging(level, log_file=""):
root = logging.getLogger("pvesnap")
root.setLevel(getattr(logging, str(level).upper(), logging.INFO))
for handler in list(root.handlers):
root.removeHandler(handler)
# Unter systemd steht der Zeitstempel schon im Journal.
plain = os.environ.get("JOURNAL_STREAM") or os.environ.get("INVOCATION_ID")
fmt = "%(levelname)s %(message)s" if plain else "%(asctime)s %(levelname)s %(message)s"
stream = logging.StreamHandler(sys.stderr)
stream.setFormatter(logging.Formatter(fmt, datefmt="%Y-%m-%d %H:%M:%S"))
root.addHandler(stream)
if log_file:
try:
handler = logging.FileHandler(log_file, encoding="utf-8")
handler.setFormatter(logging.Formatter(
"%(asctime)s %(levelname)s %(message)s", datefmt="%Y-%m-%d %H:%M:%S"))
root.addHandler(handler)
except OSError as exc:
root.warning("Logdatei %s nicht beschreibbar: %s", log_file, exc)
def _load(args):
config = load_config(args.config)
problems = config.validate()
if problems:
raise ConfigError("Konfiguration fehlerhaft:\n - %s" % "\n - ".join(problems))
if args.dry_run:
config.globals.dry_run = True
setup_logging("DEBUG" if args.verbose else config.globals.log_level,
config.globals.log_file)
return config
def _table(rows, headers):
if not rows:
return ""
widths = [len(h) for h in headers]
for row in rows:
for index, cell in enumerate(row):
widths[index] = max(widths[index], len(str(cell)))
lines = [" ".join(h.ljust(widths[i]) for i, h in enumerate(headers)).rstrip()]
lines.append(" ".join("-" * widths[i] for i in range(len(headers))))
for row in rows:
lines.append(" ".join(str(cell).ljust(widths[i])
for i, cell in enumerate(row)).rstrip())
return "\n".join(lines)
def _age(moment, now):
seconds = int((now - moment).total_seconds())
if seconds < 0:
return "in " + format_duration(-seconds, zero="0s")
return format_duration(seconds, zero="0s")
def _require_pve():
if not pvesh_available():
raise ProxmoxError("'pvesh' nicht gefunden - dieser Befehl muss auf einem "
"Proxmox-VE-Host ausgefuehrt werden.")
# ---------------------------------------------------------------------------
# Befehle
# ---------------------------------------------------------------------------
def cmd_daemon(args):
config = _load(args)
daemon = Daemon(args.config, dry_run=args.dry_run)
daemon.config = config
daemon.run()
return 0
def cmd_run(args):
config = _load(args)
_require_pve()
state = State(config.globals.state_file).load()
now = datetime.now()
if args.group:
wanted = []
for name in args.group:
group = config.group(name)
if group is None:
print("Unbekannte Gruppe: %s" % name, file=sys.stderr)
return 2
wanted.append(group)
else:
wanted = [g for g in config.groups if g.enabled]
if not args.force:
due = []
for group in wanted:
last = state.last_run(group.name)
if last is None or next_due(group, last, now) <= now:
due.append(group)
wanted = due
if not wanted:
print("Zurzeit ist keine Gruppe faellig. (--force erzwingt den Lauf)")
return 0
proxmox = Proxmox(dry_run=config.globals.dry_run,
task_timeout=config.globals.task_timeout)
guests = proxmox.inventory()
failed = False
with SingleInstanceLock(config.globals.lock_file, wait=600,
busy_message="Ein anderer pvesnap-Lauf ist gerade aktiv. "
"Bitte spaeter erneut versuchen."):
for group in wanted:
result = run_group(proxmox, config, group, now=datetime.now(),
prune=not args.no_prune, guests=guests)
if not config.globals.dry_run:
state.record_run(group.name, datetime.now(),
created=len(result.created),
deleted=len(result.deleted), errors=result.errors)
print(result.summary())
failed = failed or bool(result.errors)
state.save()
return 1 if failed else 0
def cmd_prune(args):
config = _load(args)
_require_pve()
groups = []
for name in (args.group or [g.name for g in config.groups]):
group = config.group(name)
if group is None:
print("Unbekannte Gruppe: %s" % name, file=sys.stderr)
return 2
groups.append(group)
proxmox = Proxmox(dry_run=config.globals.dry_run,
task_timeout=config.globals.task_timeout)
guests = proxmox.inventory()
failed = False
with SingleInstanceLock(config.globals.lock_file, wait=600,
busy_message="Ein anderer pvesnap-Lauf ist gerade aktiv. "
"Bitte spaeter erneut versuchen."):
for group in groups:
result = run_group(proxmox, config, group, now=datetime.now(),
create=False, prune=True, guests=guests)
print(result.summary())
failed = failed or bool(result.errors)
return 1 if failed else 0
def cmd_status(args):
config = _load(args)
state = State(config.globals.state_file).load()
now = datetime.now()
guests = None
if pvesh_available():
try:
guests = Proxmox().inventory()
except ProxmoxError as exc:
print("Hinweis: Inventar nicht lesbar (%s)\n" % exc, file=sys.stderr)
print("pvesnap %s - Konfiguration: %s" % (__version__, config.path))
print("Praefix: %s Zustand: %s%s\n"
% (config.globals.prefix, config.globals.state_file,
" [TESTLAUF aktiv]" if config.globals.dry_run else ""))
rows = []
for group in config.groups:
last = state.last_run(group.name)
info = state.info(group.name)
if group.enabled:
try:
upcoming = next_due(group, last, now)
next_text = upcoming.strftime("%d.%m. %H:%M")
if upcoming <= now:
next_text = "faellig"
except ValueError:
next_text = "?"
else:
next_text = "-"
matched = "?" if guests is None else str(len(select_guests(group, guests)))
rows.append([
group.name,
"ja" if group.enabled else "nein",
describe(group),
"%s / %s" % (group.keep_count or "unbegr.", format_duration(group.keep_time)),
matched,
last.strftime("%d.%m. %H:%M") if last else "nie",
next_text,
info.get("status", "-"),
])
if rows:
print(_table(rows, ["Gruppe", "Aktiv", "Zeitplan", "Behalte (Anz./Zeit)",
"VMs", "Letzter Lauf", "Naechster", "Status"]))
else:
print("Keine Gruppen konfiguriert.")
errors = [(name, info.get("errors") or [])
for name, info in state.data.get("groups", {}).items()]
errors = [(name, msgs) for name, msgs in errors if msgs]
if errors:
print("\nLetzte Fehler:")
for name, messages in errors:
for message in messages:
print(" [%s] %s" % (name, message))
return 0
def cmd_list(args):
config = _load(args)
_require_pve()
proxmox = Proxmox()
guests = proxmox.inventory()
now = datetime.now()
prefix = config.globals.prefix
if args.group:
groups = []
for name in args.group:
group = config.group(name)
if group is None:
print("Unbekannte Gruppe: %s" % name, file=sys.stderr)
return 2
groups.append(group)
else:
groups = config.groups
interesting = {}
for group in groups:
for guest in select_guests(group, guests):
interesting[guest.vmid] = guest
rows = []
for vmid in sorted(interesting):
guest = interesting[vmid]
try:
snapshots = proxmox.list_snapshots(guest)
except ProxmoxError as exc:
print("%s: %s" % (guest.label, exc), file=sys.stderr)
continue
for snap in sorted(snapshots, key=lambda s: s.snaptime, reverse=True):
parsed = parse_name(prefix, snap.name)
if parsed is None and not args.all:
continue
created = (datetime.fromtimestamp(snap.snaptime) if snap.snaptime
else (parsed or {}).get("created"))
rows.append([
guest.vmid,
truncate(guest.name, 20),
snap.name,
(parsed or {}).get("slug", "-"),
created.strftime("%d.%m.%y %H:%M") if created else "?",
_age(created, now) if created else "?",
truncate(snap.description.replace("\n", " "), 44),
])
if rows:
print(_table(rows, ["VMID", "Name", "Snapshot", "Gruppe", "Erstellt",
"Alter", "Beschreibung"]))
print("\n%d Snapshot(s)%s" % (len(rows), "" if args.all else " von pvesnap"))
else:
print("Keine passenden Snapshots gefunden.")
return 0
def cmd_vms(args):
config = _load(args)
_require_pve()
guests = Proxmox().inventory()
assignment = {}
for group in config.groups:
for guest in select_guests(group, guests):
assignment.setdefault(guest.vmid, []).append(group.name)
rows = [[guest.vmid,
"LXC" if guest.type == "lxc" else "VM",
truncate(guest.name, 28),
guest.node,
guest.status,
",".join(guest.tags) or "-",
guest.pool or "-",
", ".join(assignment.get(guest.vmid, [])) or "-"]
for guest in guests]
if rows:
print(_table(rows, ["VMID", "Typ", "Name", "Node", "Status", "Tags",
"Pool", "Gruppen"]))
else:
print("Keine VMs/Container gefunden.")
return 0
def cmd_check(args):
try:
config = load_config(args.config)
except ConfigError as exc:
print(str(exc), file=sys.stderr)
return 1
problems = config.validate()
if problems:
print("Konfiguration fehlerhaft:", file=sys.stderr)
for problem in problems:
print(" - %s" % problem, file=sys.stderr)
return 1
print("Konfiguration in Ordnung: %d Gruppe(n)." % len(config.groups))
for group in config.groups:
print(" - %s: %s, behalte %s / %s%s"
% (group.name, describe(group),
group.keep_count or "unbegrenzt", format_duration(group.keep_time),
"" if group.enabled else " [deaktiviert]"))
return 0
def cmd_config(args):
from .tui import run_editor
return run_editor(args.config)
# ---------------------------------------------------------------------------
# Argumente
# ---------------------------------------------------------------------------
def build_parser():
parser = argparse.ArgumentParser(
prog="pvesnap",
description="Automatische Proxmox-Snapshots mit Gruppen, Zeitplan und Vorhaltezeit.")
parser.add_argument("-c", "--config", default=os.environ.get("PVESNAP_CONFIG",
DEFAULT_CONFIG_PATH),
help="Pfad zur INI-Datei (Vorgabe: %(default)s)")
parser.add_argument("-n", "--dry-run", action="store_true",
help="nichts wirklich anlegen oder loeschen, nur anzeigen")
parser.add_argument("-v", "--verbose", action="store_true", help="ausfuehrliche Ausgabe")
parser.add_argument("-V", "--version", action="version",
version="pvesnap %s" % __version__)
sub = parser.add_subparsers(dest="command")
p_daemon = sub.add_parser("daemon", help="Dienst im Vordergrund starten (fuer systemd)")
p_daemon.set_defaults(func=cmd_daemon)
p_run = sub.add_parser("run", help="faellige Gruppen jetzt ausfuehren")
p_run.add_argument("-g", "--group", action="append", help="nur diese Gruppe (mehrfach moeglich)")
p_run.add_argument("-f", "--force", action="store_true",
help="unabhaengig vom Zeitplan ausfuehren")
p_run.add_argument("--no-prune", action="store_true", help="nicht aufraeumen")
p_run.set_defaults(func=cmd_run)
p_prune = sub.add_parser("prune", help="nur alte Snapshots aufraeumen")
p_prune.add_argument("-g", "--group", action="append", help="nur diese Gruppe")
p_prune.set_defaults(func=cmd_prune)
p_status = sub.add_parser("status", help="Uebersicht ueber Gruppen und Termine")
p_status.set_defaults(func=cmd_status)
p_list = sub.add_parser("list", help="vorhandene Snapshots anzeigen")
p_list.add_argument("-g", "--group", action="append", help="nur diese Gruppe")
p_list.add_argument("-a", "--all", action="store_true",
help="auch fremde/manuelle Snapshots anzeigen")
p_list.set_defaults(func=cmd_list)
p_vms = sub.add_parser("vms", help="alle VMs/Container und ihre Gruppen anzeigen")
p_vms.set_defaults(func=cmd_vms)
p_check = sub.add_parser("check", help="Konfiguration pruefen")
p_check.set_defaults(func=cmd_check)
for name in ("config", "edit"):
p_config = sub.add_parser(name, help="Konfiguration im ncurses-Editor bearbeiten")
p_config.set_defaults(func=cmd_config)
return parser
def main(argv=None):
parser = build_parser()
args = parser.parse_args(argv)
if not getattr(args, "func", None):
args.command = "status"
args.func = cmd_status
setup_logging("DEBUG" if args.verbose else "INFO")
try:
return args.func(args)
except ConfigError as exc:
print(str(exc), file=sys.stderr)
return 2
except ProxmoxError as exc:
print("Proxmox-Fehler: %s" % exc, file=sys.stderr)
return 3
except RuntimeError as exc:
print(str(exc), file=sys.stderr)
return 4
except KeyboardInterrupt:
print("\nAbgebrochen.", file=sys.stderr)
return 130
if __name__ == "__main__": # pragma: no cover
sys.exit(main())
+501
View File
@@ -0,0 +1,501 @@
"""Einlesen und Schreiben der pvesnap-INI-Datei."""
from __future__ import annotations
import configparser
import os
import re
from dataclasses import dataclass, field, replace
from .util import (format_bool, format_duration, format_vmid_list, parse_bool,
parse_duration, parse_list, parse_vmid_list, slugify)
DEFAULT_CONFIG_PATH = "/etc/pvesnap/pvesnap.conf"
GROUP_SECTION_RE = re.compile(r"^(?:group|gruppe)\s*[:\.]\s*(.+)$", re.IGNORECASE)
SCHEDULE_KINDS = ("hourly", "daily", "weekly", "monthly", "yearly")
WEEKDAYS = {
"mon": 0, "monday": 0, "mo": 0, "montag": 0,
"tue": 1, "tuesday": 1, "di": 1, "dienstag": 1, "tues": 1,
"wed": 2, "wednesday": 2, "mi": 2, "mittwoch": 2,
"thu": 3, "thursday": 3, "do": 3, "donnerstag": 3, "thur": 3, "thurs": 3,
"fri": 4, "friday": 4, "fr": 4, "freitag": 4,
"sat": 5, "saturday": 5, "sa": 5, "samstag": 5,
"sun": 6, "sunday": 6, "so": 6, "sonntag": 6,
}
WEEKDAY_NAMES = ["Mo", "Di", "Mi", "Do", "Fr", "Sa", "So"]
DEFAULT_DESCRIPTION = (
"pvesnap | Gruppe: {group} | erstellt: {datetime} | "
"Vorhaltezeit: {keep_time} | max: {keep_count}"
)
class ConfigError(Exception):
"""Fehler beim Lesen oder Pruefen der Konfiguration."""
def parse_weekday(text):
value = str(text).strip().lower()
if value in WEEKDAYS:
return WEEKDAYS[value]
if value.isdigit():
number = int(value)
if 0 <= number <= 6:
return number
if number == 7:
return 6
raise ValueError("ungueltiger Wochentag: %r" % text)
def parse_time_of_day(text):
"""'02:30' -> (2, 30). Auch '2', '2:5', '02:30:00' werden akzeptiert."""
value = str(text).strip()
match = re.match(r"^(\d{1,2})(?::(\d{1,2}))?(?::(\d{1,2}))?$", value)
if not match:
raise ValueError("ungueltige Uhrzeit: %r (erwartet HH:MM)" % text)
hour = int(match.group(1))
minute = int(match.group(2) or 0)
if hour > 23 or minute > 59:
raise ValueError("ungueltige Uhrzeit: %r" % text)
return hour, minute
@dataclass
class Group:
"""Eine Snapshot-Gruppe: Zeitplan + Vorhaltezeit + VM-Auswahl."""
name: str
enabled: bool = True
# --- Zeitplan -------------------------------------------------------
interval: int = 0 # Sekunden; > 0 = intervallbasiert
schedule: str = "" # hourly/daily/weekly/monthly/yearly
at: str = "00:00" # Uhrzeit fuer daily/weekly/monthly/yearly
minute: int = 0 # Minute fuer schedule = hourly
day_of_week: int = 0 # 0 = Montag (weekly)
day_of_month: int = 1 # 1..31 (monthly/yearly)
month: int = 1 # 1..12 (yearly)
align: bool = True # Intervalle an der Uhr ausrichten
# --- Vorhaltezeit ---------------------------------------------------
keep_count: int = 0 # max. Anzahl je VM (0 = unbegrenzt)
keep_time: int = 0 # max. Alter in Sekunden (0 = unbegrenzt)
keep_min: int = 0 # so viele bleiben immer stehen
# --- Auswahl der Gaeste ---------------------------------------------
all: bool = False
vmids: list = field(default_factory=list)
names: list = field(default_factory=list) # Glob-Muster
tags: list = field(default_factory=list)
pools: list = field(default_factory=list)
types: list = field(default_factory=list) # qemu / lxc
exclude_vmids: list = field(default_factory=list)
exclude_names: list = field(default_factory=list)
exclude_tags: list = field(default_factory=list)
# --- Snapshot-Optionen ----------------------------------------------
vmstate: bool = False # RAM mitsichern (nur QEMU, nur laufend)
skip_stopped: bool = False # gestoppte Gaeste ueberspringen
description: str = "" # Vorlage; leer = globale Vorlage
@property
def slug(self):
"""Kurzform des Gruppennamens, taucht im Snapshot-Namen auf."""
return slugify(self.name)
def selects_anything(self):
return bool(self.all or self.vmids or self.names or self.tags or self.pools)
def validate(self):
problems = []
if not self.name.strip():
problems.append("Gruppe ohne Namen")
if self.interval <= 0 and not self.schedule:
problems.append("[%s] weder 'interval' noch 'schedule' gesetzt" % self.name)
if self.interval > 0 and self.schedule:
problems.append("[%s] 'interval' und 'schedule' schliessen sich aus" % self.name)
if self.schedule and self.schedule not in SCHEDULE_KINDS:
problems.append("[%s] unbekannter Zeitplan %r (erlaubt: %s)"
% (self.name, self.schedule, ", ".join(SCHEDULE_KINDS)))
if self.interval > 0 and self.interval < 60:
problems.append("[%s] 'interval' muss mindestens 60s betragen" % self.name)
if not self.selects_anything():
problems.append("[%s] keine VMs ausgewaehlt (all/vmids/names/tags/pools)" % self.name)
if self.keep_count <= 0 and self.keep_time <= 0:
problems.append("[%s] weder 'keep_count' noch 'keep_time' gesetzt - "
"Snapshots wuerden nie geloescht" % self.name)
for kind in self.types:
if kind not in ("qemu", "lxc"):
problems.append("[%s] unbekannter Typ %r (erlaubt: qemu, lxc)" % (self.name, kind))
try:
parse_time_of_day(self.at)
except ValueError as exc:
problems.append("[%s] %s" % (self.name, exc))
if not 1 <= self.day_of_month <= 31:
problems.append("[%s] 'day_of_month' muss zwischen 1 und 31 liegen" % self.name)
return problems
@dataclass
class GlobalConfig:
prefix: str = "auto"
check_interval: int = 60
state_file: str = "/var/lib/pvesnap/state.json"
log_level: str = "INFO"
log_file: str = ""
dry_run: bool = False
task_timeout: int = 900
run_on_start: bool = False
description: str = DEFAULT_DESCRIPTION
lock_file: str = "/run/pvesnap.lock"
def validate(self):
problems = []
if not re.match(r"^[A-Za-z][A-Za-z0-9]{0,15}$", self.prefix):
problems.append("[global] 'prefix' muss mit einem Buchstaben beginnen und darf "
"nur Buchstaben/Ziffern enthalten (max. 16 Zeichen)")
if self.check_interval < 5:
problems.append("[global] 'check_interval' muss mindestens 5 Sekunden betragen")
if self.log_level.upper() not in ("DEBUG", "INFO", "WARNING", "ERROR"):
problems.append("[global] unbekannter 'log_level': %s" % self.log_level)
return problems
@dataclass
class Config:
globals: GlobalConfig = field(default_factory=GlobalConfig)
groups: list = field(default_factory=list)
defaults: dict = field(default_factory=dict)
path: str = DEFAULT_CONFIG_PATH
def group(self, name):
for group in self.groups:
if group.name.lower() == str(name).lower():
return group
return None
def validate(self):
from .naming import effective_slug
problems = list(self.globals.validate())
seen = {}
for group in self.groups:
problems.extend(group.validate())
# Der Kurzname wird im Snapshot-Namen ggf. gekuerzt - Kollisionen
# muessen also auf der gekuerzten Form geprueft werden.
slug = effective_slug(self.globals.prefix, group.slug)
if slug in seen:
problems.append("Gruppen %r und %r ergeben denselben Kurznamen %r - "
"bitte unterscheidbarer benennen" % (seen[slug], group.name, slug))
seen[slug] = group.name
return problems
# ---------------------------------------------------------------------------
# Lesen
# ---------------------------------------------------------------------------
_GROUP_KEYS = {
"enabled", "interval", "schedule", "at", "minute", "day_of_week", "day_of_month",
"month", "align", "keep_count", "keep_time", "keep_min", "all", "vmids", "names",
"tags", "pools", "types", "exclude_vmids", "exclude_names", "exclude_tags",
"vmstate", "skip_stopped", "description",
}
# Deutsche Schreibweisen als Synonyme, damit die INI lesbar bleibt.
_KEY_ALIASES = {
"aktiv": "enabled", "aktiviert": "enabled",
"intervall": "interval",
"zeitplan": "schedule",
"uhrzeit": "at", "zeit": "at",
"wochentag": "day_of_week",
"monatstag": "day_of_month",
"monat": "month",
"anzahl": "keep_count", "max_anzahl": "keep_count", "behalte_anzahl": "keep_count",
"vorhaltezeit": "keep_time", "behalte_zeit": "keep_time", "max_alter": "keep_time",
"mindestens": "keep_min",
"alle": "all",
"namen": "names",
"typen": "types",
"beschreibung": "description",
"ausschluss_vmids": "exclude_vmids",
"ausschluss_namen": "exclude_names",
"gestoppte_ueberspringen": "skip_stopped",
}
def _canonical_key(key):
key = key.strip().lower().replace("-", "_")
return _KEY_ALIASES.get(key, key)
def _normalize_section(section):
return {_canonical_key(k): v for k, v in section.items()}
def load_config(path=DEFAULT_CONFIG_PATH):
"""Liest die INI-Datei und gibt eine geprueft-parsierte Config zurueck."""
if not os.path.exists(path):
raise ConfigError("Konfigurationsdatei nicht gefunden: %s" % path)
parser = configparser.ConfigParser(interpolation=None)
parser.optionxform = str # Gross-/Kleinschreibung selbst behandeln
try:
with open(path, "r", encoding="utf-8") as handle:
parser.read_file(handle)
except (configparser.Error, OSError) as exc:
raise ConfigError("Konfiguration nicht lesbar (%s): %s" % (path, exc))
config = Config(path=path)
errors = []
# [global]
for name in parser.sections():
if name.strip().lower() in ("global", "allgemein", "main"):
config.globals = _parse_global(_normalize_section(parser[name]), errors)
break
# [defaults] - Vorgabewerte fuer alle Gruppen
defaults = {}
for name in parser.sections():
if name.strip().lower() in ("defaults", "default", "vorgaben", "standard"):
defaults = _normalize_section(parser[name])
break
config.defaults = dict(defaults)
# [group:NAME]
for name in parser.sections():
match = GROUP_SECTION_RE.match(name.strip())
if not match:
continue
group_name = match.group(1).strip()
merged = dict(defaults)
merged.update(_normalize_section(parser[name]))
config.groups.append(_parse_group(group_name, merged, errors))
if errors:
raise ConfigError("Fehler in %s:\n - %s" % (path, "\n - ".join(errors)))
return config
def _parse_global(section, errors):
result = GlobalConfig()
def take(key, parser_fn, target=None):
if key not in section:
return
try:
setattr(result, target or key, parser_fn(section[key]))
except ValueError as exc:
errors.append("[global] %s: %s" % (key, exc))
take("prefix", lambda v: str(v).strip())
take("check_interval", lambda v: parse_duration(v, default_unit="s"))
take("state_file", lambda v: str(v).strip())
take("log_level", lambda v: str(v).strip().upper())
take("log_file", lambda v: str(v).strip())
take("lock_file", lambda v: str(v).strip())
take("dry_run", parse_bool)
take("task_timeout", lambda v: parse_duration(v, default_unit="s"))
take("run_on_start", parse_bool)
take("description", lambda v: str(v).strip())
# Synonyme
if "description_template" in section:
result.description = str(section["description_template"]).strip()
if "testlauf" in section:
try:
result.dry_run = parse_bool(section["testlauf"])
except ValueError as exc:
errors.append("[global] testlauf: %s" % exc)
for key in section:
if key not in ("prefix", "check_interval", "state_file", "log_level", "log_file",
"lock_file", "dry_run", "task_timeout", "run_on_start", "description",
"description_template", "testlauf"):
errors.append("[global] unbekannter Schluessel: %s" % key)
return result
def _parse_group(name, section, errors):
group = Group(name=name)
prefix = "[group:%s]" % name
def take(key, parser_fn, target=None):
if key not in section:
return
raw = section[key]
try:
setattr(group, target or key, parser_fn(raw))
except ValueError as exc:
errors.append("%s %s: %s" % (prefix, key, exc))
take("enabled", parse_bool)
take("interval", lambda v: parse_duration(v, default_unit="m"))
take("schedule", lambda v: str(v).strip().lower())
take("at", lambda v: "%02d:%02d" % parse_time_of_day(v))
take("minute", lambda v: int(str(v).strip()))
take("day_of_week", parse_weekday)
take("day_of_month", lambda v: int(str(v).strip()))
take("month", lambda v: int(str(v).strip()))
take("align", parse_bool)
take("keep_count", lambda v: max(0, int(str(v).strip())))
take("keep_time", lambda v: parse_duration(v, default_unit="d"))
take("keep_min", lambda v: max(0, int(str(v).strip())))
take("all", parse_bool)
take("vmids", parse_vmid_list)
take("names", parse_list)
take("tags", lambda v: [t.lower() for t in parse_list(v)])
take("pools", parse_list)
take("types", lambda v: [t.lower() for t in parse_list(v)])
take("exclude_vmids", parse_vmid_list)
take("exclude_names", parse_list)
take("exclude_tags", lambda v: [t.lower() for t in parse_list(v)])
take("vmstate", parse_bool)
take("skip_stopped", parse_bool)
take("description", lambda v: str(v).strip())
# Bequemlichkeit: "schedule = 1h" wird als Intervall verstanden.
if group.schedule and group.schedule not in SCHEDULE_KINDS and not group.interval:
try:
group.interval = parse_duration(group.schedule, default_unit="m")
group.schedule = ""
except ValueError:
pass
for key in section:
if key not in _GROUP_KEYS:
errors.append("%s unbekannter Schluessel: %s" % (prefix, key))
return group
# ---------------------------------------------------------------------------
# Schreiben (wird vom ncurses-Editor benutzt)
# ---------------------------------------------------------------------------
_HEADER = """\
# pvesnap - Konfiguration
#
# Diese Datei wurde vom Konfigurationseditor geschrieben ("pvesnap config").
# Sie kann jederzeit auch von Hand bearbeitet werden.
#
# Zeitangaben: 30m, 1h, 6h, 2d12h, 1w ... (0 / "nie" = unbegrenzt)
# Nach Aenderungen: systemctl reload pvesnap
"""
def _group_to_lines(group, globals_):
lines = ["[group:%s]" % group.name]
lines.append("enabled = %s" % format_bool(group.enabled))
if group.schedule:
lines.append("schedule = %s" % group.schedule)
if group.schedule == "hourly":
lines.append("minute = %d" % group.minute)
else:
lines.append("at = %s" % group.at)
if group.schedule == "weekly":
lines.append("day_of_week = %s" % WEEKDAY_NAMES[group.day_of_week].lower())
if group.schedule in ("monthly", "yearly"):
lines.append("day_of_month = %d" % group.day_of_month)
if group.schedule == "yearly":
lines.append("month = %d" % group.month)
else:
lines.append("interval = %s" % format_duration(group.interval, zero="1h"))
lines.append("align = %s" % format_bool(group.align))
lines.append("keep_count = %d" % group.keep_count)
lines.append("keep_time = %s" % format_duration(group.keep_time, zero="0"))
if group.keep_min:
lines.append("keep_min = %d" % group.keep_min)
if group.all:
lines.append("all = yes")
if group.vmids:
lines.append("vmids = %s" % format_vmid_list(group.vmids))
if group.names:
lines.append("names = %s" % ", ".join(group.names))
if group.tags:
lines.append("tags = %s" % ", ".join(group.tags))
if group.pools:
lines.append("pools = %s" % ", ".join(group.pools))
if group.types:
lines.append("types = %s" % ", ".join(group.types))
if group.exclude_vmids:
lines.append("exclude_vmids = %s" % format_vmid_list(group.exclude_vmids))
if group.exclude_names:
lines.append("exclude_names = %s" % ", ".join(group.exclude_names))
if group.exclude_tags:
lines.append("exclude_tags = %s" % ", ".join(group.exclude_tags))
if group.vmstate:
lines.append("vmstate = yes")
if group.skip_stopped:
lines.append("skip_stopped = yes")
if group.description and group.description != globals_.description:
lines.append("description = %s" % group.description)
return lines
def dump_config(config):
"""Erzeugt den kompletten INI-Text zu einer Config."""
g = config.globals
lines = [_HEADER, "[global]"]
lines.append("prefix = %s" % g.prefix)
lines.append("check_interval = %s" % format_duration(g.check_interval, zero="60s"))
lines.append("state_file = %s" % g.state_file)
if g.lock_file != GlobalConfig.lock_file:
lines.append("lock_file = %s" % g.lock_file)
lines.append("log_level = %s" % g.log_level)
if g.log_file:
lines.append("log_file = %s" % g.log_file)
lines.append("task_timeout = %s" % format_duration(g.task_timeout, zero="900s"))
lines.append("run_on_start = %s" % format_bool(g.run_on_start))
lines.append("dry_run = %s" % format_bool(g.dry_run))
lines.append("description = %s" % g.description)
lines.append("")
for group in config.groups:
lines.extend(_group_to_lines(group, g))
lines.append("")
return "\n".join(lines).rstrip() + "\n"
def save_config(config, path=None):
"""Schreibt die Config atomar; legt vorher eine .bak-Kopie an."""
target = path or config.path
directory = os.path.dirname(os.path.abspath(target))
os.makedirs(directory, exist_ok=True)
if os.path.exists(target):
try:
with open(target, "r", encoding="utf-8") as handle:
previous = handle.read()
with open(target + ".bak", "w", encoding="utf-8") as handle:
handle.write(previous)
except OSError:
pass # Backup ist Kuer, kein Muss
tmp = target + ".tmp"
with open(tmp, "w", encoding="utf-8") as handle:
handle.write(dump_config(config))
handle.flush()
os.fsync(handle.fileno())
os.replace(tmp, target)
return target
def clone_group(group, new_name):
return replace(group, name=new_name,
vmids=list(group.vmids), names=list(group.names),
tags=list(group.tags), pools=list(group.pools),
types=list(group.types), exclude_vmids=list(group.exclude_vmids),
exclude_names=list(group.exclude_names),
exclude_tags=list(group.exclude_tags))
+232
View File
@@ -0,0 +1,232 @@
"""Der eigentliche Dienst: Zeitplan ueberwachen und Gruppen ausfuehren."""
from __future__ import annotations
import errno
import fcntl
import logging
import os
import signal
import threading
import time
from datetime import datetime
from .config import ConfigError, load_config
from .engine import run_group
from .proxmox import Proxmox, ProxmoxError
from .schedule import next_due
from .state import State
log = logging.getLogger("pvesnap.daemon")
class SingleInstanceLock:
"""Dateisperre.
Zwei Sperren sind im Spiel:
* <lock_file>.daemon haelt der Dienst waehrend seiner gesamten Laufzeit,
damit er nicht doppelt startet.
* <lock_file> wird nur waehrend der eigentlichen Snapshot-Arbeit
gehalten - so kann man jederzeit auch von Hand `pvesnap run` aufrufen,
ohne dass sich die Laeufe in die Quere kommen.
"""
def __init__(self, path, wait=0, busy_message=None):
self.path = path
self.wait = wait
self.busy_message = busy_message or "pvesnap laeuft bereits (Sperrdatei %s)" % path
self._handle = None
def __enter__(self):
# Die Datei anzulegen muss sofort klappen - fehlende Rechte sind ein
# harter Fehler und kein "gerade belegt".
try:
os.makedirs(os.path.dirname(os.path.abspath(self.path)), exist_ok=True)
self._handle = open(self.path, "w")
except OSError as exc:
self._handle = None
raise RuntimeError("Sperrdatei %s nicht nutzbar: %s" % (self.path, exc))
deadline = time.monotonic() + max(0, self.wait)
while True:
try:
fcntl.flock(self._handle, fcntl.LOCK_EX | fcntl.LOCK_NB)
self._handle.write("%d\n" % os.getpid())
self._handle.flush()
return self
except OSError as exc:
busy = exc.errno in (errno.EAGAIN, errno.EWOULDBLOCK, errno.EACCES)
if not busy or time.monotonic() >= deadline:
self._handle.close()
self._handle = None
if busy:
raise RuntimeError(self.busy_message)
raise RuntimeError("Sperrdatei %s nicht nutzbar: %s" % (self.path, exc))
time.sleep(1.0)
def __exit__(self, *_exc):
if self._handle:
try:
fcntl.flock(self._handle, fcntl.LOCK_UN)
except OSError:
pass
self._handle.close()
self._handle = None
return False
class Daemon:
def __init__(self, config_path, dry_run=False):
self.config_path = config_path
self.dry_run_override = dry_run
self.config = None
self.state = None
self._stop = threading.Event()
self._reload = threading.Event()
self._wake = threading.Event() # bricht das Warten bei Signalen ab
# -- Signale ----------------------------------------------------------
def install_signal_handlers(self):
signal.signal(signal.SIGTERM, self._on_stop)
signal.signal(signal.SIGINT, self._on_stop)
signal.signal(signal.SIGHUP, self._on_reload)
def _on_stop(self, signum, _frame):
log.info("Signal %s empfangen - beende nach dem aktuellen Durchlauf", signum)
self._stop.set()
self._wake.set()
def _on_reload(self, _signum, _frame):
log.info("SIGHUP empfangen - Konfiguration wird neu geladen")
self._reload.set()
self._wake.set()
# -- Konfiguration ----------------------------------------------------
def load(self):
config = load_config(self.config_path)
problems = config.validate()
if problems:
raise ConfigError("Konfiguration fehlerhaft:\n - %s" % "\n - ".join(problems))
if self.dry_run_override:
config.globals.dry_run = True
self.config = config
if self.state is None or self.state.path != config.globals.state_file:
self.state = State(config.globals.state_file).load()
return config
def _reload_if_requested(self):
if not self._reload.is_set():
return
self._reload.clear()
try:
self.load()
log.info("Konfiguration neu geladen: %d Gruppe(n)", len(self.config.groups))
except ConfigError as exc:
log.error("Neuladen fehlgeschlagen, behalte alte Konfiguration: %s", exc)
# -- Hauptschleife ----------------------------------------------------
def run(self):
self.load()
self.install_signal_handlers()
globals_ = self.config.globals
log.info("pvesnap gestartet (%d Gruppe(n), Praefix '%s'%s)",
len(self.config.groups), globals_.prefix,
", TESTLAUF" if globals_.dry_run else "")
daemon_lock = SingleInstanceLock(
globals_.lock_file + ".daemon",
busy_message="Der pvesnap-Dienst laeuft bereits (%s.daemon)" % globals_.lock_file)
with daemon_lock:
while not self._stop.is_set():
self._wake.clear()
self._reload_if_requested()
try:
self.tick()
except ProxmoxError as exc:
log.error("Proxmox nicht erreichbar: %s", exc)
except Exception: # pragma: no cover - Dienst darf nie sterben
log.exception("Unerwarteter Fehler im Durchlauf")
self._sleep()
if self.state:
self.state.save()
log.info("pvesnap beendet")
def tick(self, now=None):
"""Einmal alle Gruppen pruefen und faellige ausfuehren."""
now = now or datetime.now()
config = self.config
state = self.state
state.prune_unknown([g.name for g in config.groups])
due = []
for group in config.groups:
if not group.enabled:
continue
last = state.last_run(group.name)
if last is None and not config.globals.run_on_start:
# Erster Start: nicht sofort feuern, sondern auf den naechsten
# regulaeren Termin warten.
state.record_run(group.name, now, created=0, deleted=0)
log.info("Gruppe '%s': erster Termin am %s", group.name,
next_due(group, now, now).strftime("%Y-%m-%d %H:%M:%S"))
continue
if last is None or next_due(group, last, now) <= now:
due.append(group)
if not due:
state.save()
return []
proxmox = Proxmox(dry_run=config.globals.dry_run,
task_timeout=config.globals.task_timeout)
results = []
# Die Arbeitssperre wird nur waehrend des Laufs gehalten, damit
# `pvesnap run` von Hand weiterhin moeglich bleibt.
work_lock = SingleInstanceLock(
config.globals.lock_file, wait=300,
busy_message="Ein anderer pvesnap-Lauf ist noch aktiv - "
"dieser Durchlauf wird uebersprungen")
try:
with work_lock:
guests = proxmox.inventory(refresh=True)
for group in due:
if self._stop.is_set():
break
result = run_group(proxmox, config, group, now=datetime.now(),
guests=guests)
state.record_run(group.name, datetime.now(),
created=len(result.created),
deleted=len(result.deleted), errors=result.errors)
results.append(result)
except RuntimeError as exc:
log.warning("%s", exc)
state.save()
return results
def _sleep(self):
"""Bis zum naechsten Termin schlafen, hoechstens aber check_interval."""
interval = self.config.globals.check_interval
now = datetime.now()
wait = float(interval)
for group in self.config.groups:
if not group.enabled:
continue
last = self.state.last_run(group.name)
if last is None:
continue
try:
remaining = (next_due(group, last, now) - now).total_seconds()
except ValueError:
continue
wait = min(wait, max(1.0, remaining))
# Auf Signale reagieren wir sofort - das Event bricht das Warten ab.
# Geleert wird es am Anfang des Schleifendurchlaufs, damit ein Signal
# waehrend tick() nicht verloren geht.
self._wake.wait(max(1.0, min(wait, float(interval))))
+181
View File
@@ -0,0 +1,181 @@
"""Kernlogik: Gaeste auswaehlen, Snapshots anlegen, alte Snapshots aufraeumen."""
from __future__ import annotations
import fnmatch
import logging
from dataclasses import dataclass, field
from datetime import datetime
from .naming import build_name, effective_slug, parse_name, render_description
from .proxmox import ProxmoxError
log = logging.getLogger("pvesnap.engine")
# ---------------------------------------------------------------------------
# Auswahl der Gaeste
# ---------------------------------------------------------------------------
def _matches_any_glob(text, patterns):
lowered = (text or "").lower()
return any(fnmatch.fnmatch(lowered, pattern.lower()) for pattern in patterns)
def select_guests(group, guests):
"""Alle Gaeste, auf die die Auswahlregeln der Gruppe zutreffen."""
wanted_vmids = set(group.vmids)
wanted_tags = set(group.tags)
wanted_pools = set(p.lower() for p in group.pools)
excluded_vmids = set(group.exclude_vmids)
excluded_tags = set(group.exclude_tags)
selected = []
for guest in guests:
if group.types and guest.type not in group.types:
continue
if group.all:
hit = True
else:
hit = (guest.vmid in wanted_vmids
or (group.names and _matches_any_glob(guest.name, group.names))
or (wanted_tags and wanted_tags.intersection(guest.tags))
or (wanted_pools and guest.pool.lower() in wanted_pools))
if not hit:
continue
if guest.vmid in excluded_vmids:
continue
if group.exclude_names and _matches_any_glob(guest.name, group.exclude_names):
continue
if excluded_tags and excluded_tags.intersection(guest.tags):
continue
selected.append(guest)
return selected
# ---------------------------------------------------------------------------
# Vorhaltezeit
# ---------------------------------------------------------------------------
def managed_snapshots(prefix, group, snapshots):
"""Nur die Snapshots, die pvesnap fuer *diese* Gruppe angelegt hat."""
slug = effective_slug(prefix, group.slug)
result = []
for snap in snapshots:
parsed = parse_name(prefix, snap.name)
if not parsed or parsed["slug"] != slug:
continue
created = (datetime.fromtimestamp(snap.snaptime) if snap.snaptime
else parsed["created"])
result.append((snap, created))
result.sort(key=lambda item: item[1], reverse=True) # neueste zuerst
return result
def plan_prune(prefix, group, snapshots, now):
"""Welche Snapshots sollen weg? Gibt eine Liste von (Snapshot, Grund) zurueck."""
doomed = []
for index, (snap, created) in enumerate(managed_snapshots(prefix, group, snapshots)):
if index < group.keep_min:
continue
age = (now - created).total_seconds()
if group.keep_count > 0 and index >= group.keep_count:
doomed.append((snap, "Anzahl > %d" % group.keep_count))
elif group.keep_time > 0 and age > group.keep_time:
doomed.append((snap, "aelter als Vorhaltezeit"))
return doomed
# ---------------------------------------------------------------------------
# Ausfuehrung
# ---------------------------------------------------------------------------
@dataclass
class GroupResult:
group: str
matched: int = 0
created: list = field(default_factory=list)
deleted: list = field(default_factory=list)
skipped: list = field(default_factory=list)
errors: list = field(default_factory=list)
@property
def ok(self):
return not self.errors
def summary(self):
return ("Gruppe '%s': %d Gast/Gaeste, %d Snapshot(s) angelegt, "
"%d geloescht, %d Fehler"
% (self.group, self.matched, len(self.created),
len(self.deleted), len(self.errors)))
def run_group(proxmox, config, group, now=None, create=True, prune=True, guests=None):
"""Legt fuer eine Gruppe Snapshots an und raeumt alte weg."""
now = now or datetime.now()
prefix = config.globals.prefix
result = GroupResult(group=group.name)
if guests is None:
guests = proxmox.inventory()
selected = select_guests(group, guests)
result.matched = len(selected)
if not selected:
log.warning("Gruppe '%s': keine passenden Gaeste gefunden", group.name)
return result
snapshot_name = build_name(prefix, group.slug, now)
template = group.description or config.globals.description
for guest in selected:
if group.skip_stopped and not guest.running:
result.skipped.append("%s (gestoppt)" % guest.label)
log.info("Gruppe '%s': %s uebersprungen (gestoppt)", group.name, guest.label)
continue
if create:
description = render_description(template, group, guest, now, prefix)
try:
proxmox.create_snapshot(
guest, snapshot_name, description,
vmstate=group.vmstate and guest.running,
)
result.created.append("%s:%s" % (guest.vmid, snapshot_name))
log.info("Gruppe '%s': Snapshot '%s' fuer %s angelegt",
group.name, snapshot_name, guest.label)
except ProxmoxError as exc:
message = "%s: Snapshot fehlgeschlagen: %s" % (guest.label, exc)
result.errors.append(message)
log.error("Gruppe '%s': %s", group.name, message)
continue # ohne neuen Snapshot nicht aufraeumen
if prune:
try:
_prune_guest(proxmox, config, group, guest, now, result)
except ProxmoxError as exc:
message = "%s: Aufraeumen fehlgeschlagen: %s" % (guest.label, exc)
result.errors.append(message)
log.error("Gruppe '%s': %s", group.name, message)
log.info(result.summary())
return result
def _prune_guest(proxmox, config, group, guest, now, result):
prefix = config.globals.prefix
snapshots = proxmox.list_snapshots(guest)
for snap, reason in plan_prune(prefix, group, snapshots, now):
try:
proxmox.delete_snapshot(guest, snap.name)
result.deleted.append("%s:%s" % (guest.vmid, snap.name))
log.info("Gruppe '%s': Snapshot '%s' von %s geloescht (%s)",
group.name, snap.name, guest.label, reason)
except ProxmoxError as exc:
message = "%s: '%s' nicht loeschbar: %s" % (guest.label, snap.name, exc)
result.errors.append(message)
log.error("Gruppe '%s': %s", group.name, message)
+88
View File
@@ -0,0 +1,88 @@
"""Namensschema der von pvesnap verwalteten Snapshots.
Ein Name sieht so aus: auto-taeglich-20260730-023000
^^^^ ^^^^^^^^ ^^^^^^^^ ^^^^^^
| | Datum Uhrzeit
| Kurzname der Gruppe
Praefix aus [global]
Nur Snapshots, die exakt auf dieses Muster passen, werden von pvesnap
angefasst. Von Hand angelegte Snapshots bleiben damit garantiert unberuehrt.
"""
from __future__ import annotations
import re
from datetime import datetime
# Proxmox erlaubt fuer Snapshot-Namen nur begrenzte Laenge - wir bleiben
# bewusst deutlich darunter.
MAX_NAME_LEN = 40
_STAMP_LEN = len("-20260730-023000")
def max_slug_len(prefix):
return max(1, MAX_NAME_LEN - len(prefix) - _STAMP_LEN)
def effective_slug(prefix, slug):
"""Der Kurzname, wie er tatsaechlich im Snapshot-Namen landet."""
return slug[: max_slug_len(prefix)]
def build_name(prefix, slug, moment):
return "%s-%s-%s" % (prefix, effective_slug(prefix, slug),
moment.strftime("%Y%m%d-%H%M%S"))
def name_pattern(prefix, slug=None):
part = re.escape(effective_slug(prefix, slug)) if slug else r"[A-Za-z0-9]+"
return re.compile(r"^%s-(%s)-(\d{8})-(\d{6})$" % (re.escape(prefix), part))
def parse_name(prefix, name):
"""Zerlegt einen Snapshot-Namen; gibt None zurueck, wenn er nicht zu uns gehoert."""
match = name_pattern(prefix).match(str(name))
if not match:
return None
slug, date_part, time_part = match.groups()
try:
created = datetime.strptime(date_part + time_part, "%Y%m%d%H%M%S")
except ValueError:
return None
return {"slug": slug, "created": created}
def render_description(template, group, guest, moment, prefix):
"""Setzt die Platzhalter der Beschreibungs-Vorlage ein."""
from .util import format_duration # lokal, um Zyklen zu vermeiden
values = {
"prefix": prefix,
"group": group.name,
"group_slug": group.slug,
"vmid": guest.vmid,
"name": guest.name or "",
"type": "LXC" if guest.type == "lxc" else "VM",
"node": guest.node or "",
"pool": guest.pool or "",
"tags": ",".join(guest.tags),
"date": moment.strftime("%Y-%m-%d"),
"time": moment.strftime("%H:%M:%S"),
"datetime": moment.strftime("%Y-%m-%d %H:%M:%S"),
"timestamp": int(moment.timestamp()),
"keep_time": format_duration(group.keep_time),
"keep_count": group.keep_count if group.keep_count else "unbegrenzt",
"schedule": _schedule_text(group),
}
try:
text = str(template).format(**values)
except (KeyError, IndexError, ValueError):
# Unbekannter Platzhalter: lieber eine brauchbare Beschreibung als ein Abbruch.
text = "pvesnap | Gruppe: %s | erstellt: %s" % (group.name, values["datetime"])
return text.strip()[:1024]
def _schedule_text(group):
from .schedule import describe
return describe(group)
+202
View File
@@ -0,0 +1,202 @@
"""Duenne Schicht um `pvesh` - Inventar lesen, Snapshots anlegen/loeschen.
Bewusst ueber `pvesh` statt `qm`/`pct`: damit funktioniert alles auch fuer
Gaeste, die auf einem anderen Node des Clusters laufen.
"""
from __future__ import annotations
import json
import logging
import shutil
import subprocess
import time
from dataclasses import dataclass, field
log = logging.getLogger("pvesnap.proxmox")
PVESH = "/usr/bin/pvesh"
class ProxmoxError(Exception):
"""Ein pvesh-Aufruf ist fehlgeschlagen."""
@dataclass
class Guest:
vmid: int
name: str = ""
type: str = "qemu" # qemu | lxc
node: str = ""
status: str = "unknown"
tags: list = field(default_factory=list)
pool: str = ""
@property
def running(self):
return self.status == "running"
@property
def label(self):
return "%s %d (%s)" % ("LXC" if self.type == "lxc" else "VM",
self.vmid, self.name or "?")
@dataclass
class Snapshot:
name: str
description: str = ""
snaptime: int = 0
parent: str = ""
def pvesh_available():
return shutil.which(PVESH) is not None or shutil.which("pvesh") is not None
def _pvesh_binary():
return shutil.which(PVESH) or shutil.which("pvesh") or PVESH
class Proxmox:
"""Zugriff auf die Proxmox-API ueber das Kommando `pvesh`."""
def __init__(self, dry_run=False, task_timeout=900, command_timeout=60):
self.dry_run = dry_run
self.task_timeout = task_timeout
self.command_timeout = command_timeout
self._inventory_cache = None
# -- unterste Ebene ---------------------------------------------------
def _run(self, args, timeout=None):
command = [_pvesh_binary()] + args
log.debug("pvesh: %s", " ".join(command))
try:
proc = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
timeout=timeout or self.command_timeout)
except FileNotFoundError:
raise ProxmoxError("'pvesh' nicht gefunden - laeuft pvesnap wirklich "
"auf einem Proxmox-VE-Host?")
except subprocess.TimeoutExpired:
raise ProxmoxError("Zeitueberschreitung bei: %s" % " ".join(command))
stdout = proc.stdout.decode("utf-8", "replace").strip()
stderr = proc.stderr.decode("utf-8", "replace").strip()
if proc.returncode != 0:
raise ProxmoxError((stderr or stdout or "unbekannter Fehler").splitlines()[0])
return stdout, stderr
def _json(self, args):
stdout, _ = self._run(args + ["--output-format", "json"])
if not stdout:
return None
try:
return json.loads(stdout)
except ValueError:
raise ProxmoxError("unerwartete Antwort von pvesh: %s" % stdout[:200])
# -- Inventar ---------------------------------------------------------
def inventory(self, refresh=False):
"""Alle VMs und Container des Clusters."""
if self._inventory_cache is not None and not refresh:
return self._inventory_cache
data = self._json(["get", "/cluster/resources", "--type", "vm"]) or []
guests = []
for entry in data:
if entry.get("type") not in ("qemu", "lxc"):
continue
raw_tags = entry.get("tags") or ""
tags = [t.strip().lower() for t in str(raw_tags).replace(",", ";").split(";")
if t.strip()]
guests.append(Guest(
vmid=int(entry.get("vmid")),
name=entry.get("name") or "",
type=entry.get("type"),
node=entry.get("node") or "",
status=entry.get("status") or "unknown",
tags=tags,
pool=entry.get("pool") or "",
))
guests.sort(key=lambda g: g.vmid)
self._inventory_cache = guests
return guests
@staticmethod
def local_node():
import socket
return socket.gethostname().split(".")[0]
# -- Snapshots --------------------------------------------------------
def _base_path(self, guest):
return "/nodes/%s/%s/%d/snapshot" % (guest.node, guest.type, guest.vmid)
def list_snapshots(self, guest):
data = self._json(["get", self._base_path(guest)]) or []
result = []
for entry in data:
name = entry.get("name")
if not name or name == "current":
continue # "current" ist kein echter Snapshot
result.append(Snapshot(
name=name,
description=(entry.get("description") or "").strip(),
snaptime=int(entry.get("snaptime") or 0),
parent=entry.get("parent") or "",
))
result.sort(key=lambda s: s.snaptime)
return result
def create_snapshot(self, guest, name, description="", vmstate=False):
args = ["create", self._base_path(guest), "--snapname", name]
if description:
args += ["--description", description]
if vmstate and guest.type == "qemu":
args += ["--vmstate", "1"]
if self.dry_run:
log.info("[TESTLAUF] wuerde Snapshot anlegen: %s -> %s", guest.label, name)
return None
stdout, _ = self._run(args)
return self._wait_task(guest.node, stdout, "Snapshot %s fuer %s" % (name, guest.label))
def delete_snapshot(self, guest, name):
if self.dry_run:
log.info("[TESTLAUF] wuerde Snapshot loeschen: %s -> %s", guest.label, name)
return None
stdout, _ = self._run(["delete", "%s/%s" % (self._base_path(guest), name)])
return self._wait_task(guest.node, stdout,
"Loeschen von %s bei %s" % (name, guest.label))
# -- Task-Verfolgung --------------------------------------------------
def _wait_task(self, node, output, what):
"""pvesh liefert bei Snapshot-Aktionen eine UPID; darauf warten wir."""
upid = self._extract_upid(output)
if not upid:
return None
deadline = time.time() + self.task_timeout
delay = 0.5
while True:
status = self._json(["get", "/nodes/%s/tasks/%s/status" % (node, upid)]) or {}
if status.get("status") == "stopped":
exit_status = status.get("exitstatus") or "unbekannt"
if exit_status != "OK":
raise ProxmoxError("%s fehlgeschlagen: %s" % (what, exit_status))
return upid
if time.time() > deadline:
raise ProxmoxError("%s: Zeitueberschreitung nach %ds (Task laeuft weiter: %s)"
% (what, self.task_timeout, upid))
time.sleep(delay)
delay = min(delay * 1.5, 5.0)
@staticmethod
def _extract_upid(output):
for line in (output or "").splitlines():
candidate = line.strip().strip('"')
if candidate.startswith("UPID:"):
return candidate
return None
+133
View File
@@ -0,0 +1,133 @@
"""Berechnung der naechsten Ausfuehrungszeit einer Gruppe (lokale Zeit)."""
from __future__ import annotations
import calendar
from datetime import datetime, timedelta
from .config import WEEKDAY_NAMES, parse_time_of_day
from .util import format_duration
_EPOCH = datetime(1970, 1, 1)
def _midnight(moment):
return moment.replace(hour=0, minute=0, second=0, microsecond=0)
def _aligned_after(moment, interval):
"""Naechster an der Uhr ausgerichteter Zeitpunkt nach `moment`.
Intervalle, die glatt in einen Tag passen (5m, 15m, 1h, 6h, 12h), werden an
Mitternacht verankert - 6h ergibt also 00:00, 06:00, 12:00, 18:00.
Alles andere wird an der Epoche verankert.
"""
if interval <= 86400 and 86400 % interval == 0:
anchor = _midnight(moment)
else:
anchor = _EPOCH
elapsed = (moment - anchor).total_seconds()
steps = int(elapsed // interval) + 1
return anchor + timedelta(seconds=steps * interval)
def _at_time(day, hour, minute):
return day.replace(hour=hour, minute=minute, second=0, microsecond=0)
def _clamp_day(year, month, day):
return min(day, calendar.monthrange(year, month)[1])
def next_due(group, last_run, now):
"""Wann ist die Gruppe das naechste Mal faellig?
`last_run` ist der letzte tatsaechliche Lauf (oder None). Liegt das
Ergebnis in der Vergangenheit, ist die Gruppe sofort faellig - so werden
Laeufe nachgeholt, die waehrend eines Neustarts ausgefallen sind.
"""
if group.interval > 0:
if last_run is None:
base = now
else:
base = last_run
if group.align:
return _aligned_after(base, group.interval)
return base + timedelta(seconds=group.interval)
base = last_run if last_run is not None else now - timedelta(seconds=1)
kind = group.schedule
if kind == "hourly":
minute = max(0, min(59, group.minute))
candidate = base.replace(minute=minute, second=0, microsecond=0)
while candidate <= base:
candidate += timedelta(hours=1)
return candidate
hour, minute = parse_time_of_day(group.at)
if kind == "daily":
candidate = _at_time(base, hour, minute)
while candidate <= base:
candidate += timedelta(days=1)
return candidate
if kind == "weekly":
candidate = _at_time(base, hour, minute)
shift = (group.day_of_week - candidate.weekday()) % 7
candidate += timedelta(days=shift)
while candidate <= base:
candidate += timedelta(days=7)
return candidate
if kind == "monthly":
year, month = base.year, base.month
for _ in range(60):
day = _clamp_day(year, month, group.day_of_month)
candidate = _at_time(datetime(year, month, day), hour, minute)
if candidate > base:
return candidate
month += 1
if month > 12:
month, year = 1, year + 1
raise ValueError("kein monatlicher Termin ermittelbar")
if kind == "yearly":
year = base.year
month = max(1, min(12, group.month))
for _ in range(5):
day = _clamp_day(year, month, group.day_of_month)
candidate = _at_time(datetime(year, month, day), hour, minute)
if candidate > base:
return candidate
year += 1
raise ValueError("kein jaehrlicher Termin ermittelbar")
raise ValueError("unbekannter Zeitplan: %r" % kind)
def is_due(group, last_run, now, run_on_start=False):
if not group.enabled:
return False
if last_run is None and run_on_start:
return True
return next_due(group, last_run, now) <= now
def describe(group):
"""Zeitplan der Gruppe als kurzer, lesbarer Text."""
if group.interval > 0:
text = "alle %s" % format_duration(group.interval)
return text + (" (an der Uhr ausgerichtet)" if group.align else "")
if group.schedule == "hourly":
return "stuendlich zur Minute %02d" % group.minute
if group.schedule == "daily":
return "taeglich um %s" % group.at
if group.schedule == "weekly":
return "woechentlich %s um %s" % (WEEKDAY_NAMES[group.day_of_week], group.at)
if group.schedule == "monthly":
return "monatlich am %d. um %s" % (group.day_of_month, group.at)
if group.schedule == "yearly":
return "jaehrlich am %d.%d. um %s" % (group.day_of_month, group.month, group.at)
return "kein Zeitplan"
+90
View File
@@ -0,0 +1,90 @@
"""Persistenter Zustand: wann lief welche Gruppe zuletzt?"""
from __future__ import annotations
import json
import logging
import os
from datetime import datetime
log = logging.getLogger("pvesnap.state")
_TIME_FORMAT = "%Y-%m-%dT%H:%M:%S"
class State:
def __init__(self, path):
self.path = path
self.data = {"version": 1, "groups": {}}
self._dirty = False
# -- Laden / Speichern ------------------------------------------------
def load(self):
if not os.path.exists(self.path):
return self
try:
with open(self.path, "r", encoding="utf-8") as handle:
data = json.load(handle)
if isinstance(data, dict) and isinstance(data.get("groups"), dict):
self.data = data
self.data.setdefault("version", 1)
except (OSError, ValueError) as exc:
log.warning("Zustandsdatei %s nicht lesbar (%s) - starte mit leerem Zustand",
self.path, exc)
return self
def save(self, force=False):
if not self._dirty and not force:
return
directory = os.path.dirname(os.path.abspath(self.path))
try:
os.makedirs(directory, exist_ok=True)
tmp = self.path + ".tmp"
with open(tmp, "w", encoding="utf-8") as handle:
json.dump(self.data, handle, indent=2, ensure_ascii=False)
handle.write("\n")
handle.flush()
os.fsync(handle.fileno())
os.replace(tmp, self.path)
self._dirty = False
except OSError as exc:
log.error("Zustand konnte nicht gespeichert werden (%s): %s", self.path, exc)
# -- Zugriff ----------------------------------------------------------
def _entry(self, group_name):
return self.data["groups"].setdefault(group_name, {})
def last_run(self, group_name):
raw = self._entry(group_name).get("last_run")
if not raw:
return None
try:
return datetime.strptime(raw, _TIME_FORMAT)
except ValueError:
return None
def record_run(self, group_name, moment, created=0, deleted=0, errors=None):
entry = self._entry(group_name)
entry["last_run"] = moment.strftime(_TIME_FORMAT)
entry["created"] = created
entry["deleted"] = deleted
entry["errors"] = list(errors or [])
entry["status"] = "error" if errors else "ok"
entry["total_runs"] = int(entry.get("total_runs") or 0) + 1
self._dirty = True
def info(self, group_name):
return dict(self._entry(group_name))
def forget(self, group_name):
if group_name in self.data["groups"]:
del self.data["groups"][group_name]
self._dirty = True
def prune_unknown(self, known_names):
known = set(known_names)
for name in [n for n in self.data["groups"] if n not in known]:
del self.data["groups"][name]
self._dirty = True
+889
View File
@@ -0,0 +1,889 @@
"""ncurses-Editor fuer die pvesnap-Konfiguration.
Aufruf: pvesnap config
"""
from __future__ import annotations
import curses
import os
from .config import (WEEKDAY_NAMES, Config, ConfigError, Group, clone_group,
load_config, parse_time_of_day, save_config)
from .naming import effective_slug
from .proxmox import Proxmox, ProxmoxError, pvesh_available
from .schedule import describe
from .util import (format_duration, format_vmid_list, parse_duration,
parse_list, parse_vmid_list, truncate)
SCHEDULE_CHOICES = [
("interval", "Intervall (z.B. alle 30 Minuten)"),
("hourly", "stuendlich"),
("daily", "taeglich"),
("weekly", "woechentlich"),
("monthly", "monatlich"),
("yearly", "jaehrlich"),
]
TYPE_CHOICES = [
([], "alle (VMs und Container)"),
(["qemu"], "nur VMs (QEMU/KVM)"),
(["lxc"], "nur Container (LXC)"),
]
C_HEADER, C_FOOTER, C_SEL, C_WARN, C_OK, C_DIM = 1, 2, 3, 4, 5, 6
# ---------------------------------------------------------------------------
# Zeichen-Helfer
# ---------------------------------------------------------------------------
def _put(win, y, x, text, attr=0):
"""addstr, das bei zu kleinem Terminal nicht abstuerzt."""
height, width = win.getmaxyx()
if y < 0 or y >= height or x >= width:
return
try:
win.addstr(y, x, str(text)[: max(0, width - x - 1)], attr)
except curses.error:
pass
def _fill(win, y, text, attr):
height, width = win.getmaxyx()
if y < 0 or y >= height:
return
try:
win.addstr(y, 0, str(text).ljust(width - 1)[: width - 1], attr)
except curses.error:
pass
def _read_key(win):
try:
return win.get_wch()
except curses.error:
return None
except KeyboardInterrupt:
return "\x1b"
def _is_enter(key):
return key in ("\n", "\r", curses.KEY_ENTER)
def _is_escape(key):
return key == "\x1b"
def _is_backspace(key):
return key in ("\x7f", "\b", "\x08", curses.KEY_BACKSPACE)
class Editor:
def __init__(self, path):
self.path = path
self.config = None
self.dirty = False
self._guests = None
self._inventory_error = ""
# -- Daten ------------------------------------------------------------
def load(self):
if os.path.exists(self.path):
self.config = load_config(self.path)
else:
self.config = Config(path=self.path)
self.dirty = True
def guests(self, refresh=False):
"""Inventar der VMs; wird nur bei Bedarf einmal geholt."""
if self._guests is not None and not refresh:
return self._guests
self._guests = []
self._inventory_error = ""
if not pvesh_available():
self._inventory_error = ("'pvesh' nicht gefunden - VM-Liste nur auf einem "
"Proxmox-Host verfuegbar.")
return self._guests
try:
self._guests = Proxmox().inventory(refresh=True)
except ProxmoxError as exc:
self._inventory_error = str(exc)
return self._guests
# -- Rahmen -----------------------------------------------------------
def _frame(self, win, title, subtitle, keys):
win.erase()
height, width = win.getmaxyx()
_fill(win, 0, " pvesnap - %s" % title, curses.color_pair(C_HEADER) | curses.A_BOLD)
marker = "*" if self.dirty else " "
_fill(win, 1, " %s%s" % (marker, subtitle), curses.color_pair(C_DIM))
_fill(win, height - 1, " " + keys, curses.color_pair(C_FOOTER))
return height, width
def _message(self, win, text, error=False):
height, _ = win.getmaxyx()
attr = curses.color_pair(C_WARN if error else C_OK) | curses.A_BOLD
_fill(win, height - 2, " " + str(text) + " [Taste druecken]", attr)
win.refresh()
_read_key(win)
def _confirm(self, win, question):
height, _ = win.getmaxyx()
_fill(win, height - 2, " %s [j/n]" % question,
curses.color_pair(C_WARN) | curses.A_BOLD)
win.refresh()
while True:
key = _read_key(win)
if key in ("j", "J", "y", "Y"):
return True
if key in ("n", "N") or _is_escape(key):
return False
def _prompt(self, win, label, initial=""):
"""Einzeilige Eingabe am unteren Rand. None = abgebrochen."""
height, width = win.getmaxyx()
buffer = list(str(initial))
curses.curs_set(1)
try:
while True:
text = "".join(buffer)
prompt = " %s: " % label
visible = width - len(prompt) - 2
shown = text[-visible:] if visible > 0 and len(text) > visible else text
_fill(win, height - 2, prompt + shown, curses.color_pair(C_SEL))
try:
win.move(height - 2, min(width - 2, len(prompt) + len(shown)))
except curses.error:
pass
win.refresh()
key = _read_key(win)
if key is None:
continue
if _is_enter(key):
return "".join(buffer).strip()
if _is_escape(key):
return None
if _is_backspace(key):
if buffer:
buffer.pop()
continue
if key == "\x15": # Strg-U
buffer = []
continue
if isinstance(key, str) and key.isprintable():
buffer.append(key)
finally:
curses.curs_set(0)
def _choose(self, win, title, entries, current=0):
"""Kleines Auswahlfenster. entries: Liste von (wert, beschriftung)."""
height, width = win.getmaxyx()
box_h = min(len(entries) + 4, height - 2)
box_w = min(max([len(title)] + [len(e[1]) for e in entries]) + 8, width - 2)
top = max(0, (height - box_h) // 2)
left = max(0, (width - box_w) // 2)
box = curses.newwin(box_h, box_w, top, left)
box.keypad(True)
index = max(0, min(current, len(entries) - 1))
while True:
box.erase()
box.box()
_put(box, 0, 2, " %s " % title, curses.A_BOLD)
for row, (_value, label) in enumerate(entries[: box_h - 4]):
attr = curses.color_pair(C_SEL) if row == index else 0
_put(box, row + 2, 2, " " + label.ljust(box_w - 5), attr)
box.refresh()
key = _read_key(box)
if key in (curses.KEY_UP, "k"):
index = (index - 1) % len(entries)
elif key in (curses.KEY_DOWN, "j"):
index = (index + 1) % len(entries)
elif _is_enter(key):
return entries[index][0]
elif _is_escape(key) or key == "q":
return None
# -- Hauptschleife ----------------------------------------------------
def run(self, stdscr):
curses.curs_set(0)
if hasattr(curses, "set_escdelay"):
curses.set_escdelay(25)
try:
curses.start_color()
curses.use_default_colors()
curses.init_pair(C_HEADER, curses.COLOR_WHITE, curses.COLOR_BLUE)
curses.init_pair(C_FOOTER, curses.COLOR_BLACK, curses.COLOR_WHITE)
curses.init_pair(C_SEL, curses.COLOR_BLACK, curses.COLOR_CYAN)
curses.init_pair(C_WARN, curses.COLOR_WHITE, curses.COLOR_RED)
curses.init_pair(C_OK, curses.COLOR_BLACK, curses.COLOR_GREEN)
curses.init_pair(C_DIM, -1, -1)
except curses.error:
pass
stdscr.keypad(True)
self.screen_groups(stdscr)
# -- Bildschirm: Gruppenliste ----------------------------------------
def screen_groups(self, win):
index = 0
while True:
groups = self.config.groups
index = max(0, min(index, max(0, len(groups) - 1)))
height, width = self._frame(
win, "Gruppen",
"Datei: %s" % self.path,
"Enter Bearbeiten | n Neu | c Kopieren | d Loeschen | Leer An/Aus | "
"g Global | v VMs | s Speichern | q Ende")
_put(win, 3, 2, "Gruppe".ljust(18) + "Aktiv " + "Zeitplan".ljust(30)
+ "Behalte".ljust(20) + "Auswahl", curses.A_BOLD)
_put(win, 4, 2, "-" * max(0, width - 4), curses.color_pair(C_DIM))
if not groups:
_put(win, 6, 4, "Noch keine Gruppe angelegt - mit 'n' eine erste anlegen.")
_put(win, 8, 4, "Beispiel: 'stuendlich' alle 1h, 24 Snapshots behalten;")
_put(win, 9, 4, " 'taeglich' um 02:30, 14 Tage Vorhaltezeit.")
visible = max(1, height - 8)
start = max(0, min(index - visible // 2, max(0, len(groups) - visible)))
for row, group in enumerate(groups[start:start + visible]):
position = start + row
attr = curses.color_pair(C_SEL) if position == index else 0
keep = "%s / %s" % (group.keep_count or "unbegr.",
format_duration(group.keep_time))
line = ("%s%s%s%s%s"
% (truncate(group.name, 17).ljust(18),
("ja" if group.enabled else "nein").ljust(7),
truncate(describe(group), 29).ljust(30),
truncate(keep, 19).ljust(20),
truncate(self._selection_text(group), max(4, width - 80))))
_fill_row(win, 5 + row, 2, line, width, attr)
problems = self.config.validate()
if problems:
_fill(win, height - 2, " %d Hinweis(e) - 'p' zeigt Details"
% len(problems), curses.color_pair(C_WARN))
win.refresh()
key = _read_key(win)
if key in (curses.KEY_UP, "k"):
index -= 1
elif key in (curses.KEY_DOWN, "j"):
index += 1
elif key == curses.KEY_HOME:
index = 0
elif key == curses.KEY_END:
index = len(groups) - 1
elif _is_enter(key) and groups:
self.screen_group(win, groups[index])
elif key == "n":
self._new_group(win)
index = len(self.config.groups) - 1
elif key == "c" and groups:
name = self._prompt(win, "Name der Kopie", groups[index].name + "-kopie")
if name:
self.config.groups.append(clone_group(groups[index], name))
self.dirty = True
elif key == "d" and groups:
if self._confirm(win, "Gruppe '%s' wirklich loeschen?" % groups[index].name):
del self.config.groups[index]
self.dirty = True
elif key == " " and groups:
groups[index].enabled = not groups[index].enabled
self.dirty = True
elif key == "g":
self.screen_globals(win)
elif key == "v":
self.screen_overview(win)
elif key == "p":
self._show_problems(win)
elif key == "s":
self._save(win)
elif key in ("q", "Q") or _is_escape(key):
if self.dirty and not self._confirm(win, "Ungespeicherte Aenderungen verwerfen?"):
continue
return
def _selection_text(self, group):
parts = []
if group.all:
parts.append("alle VMs")
if group.vmids:
parts.append("IDs: " + format_vmid_list(group.vmids))
if group.names:
parts.append("Namen: " + ", ".join(group.names))
if group.tags:
parts.append("Tags: " + ", ".join(group.tags))
if group.pools:
parts.append("Pools: " + ", ".join(group.pools))
return "; ".join(parts) or "(nichts ausgewaehlt)"
def _new_group(self, win):
name = self._prompt(win, "Name der neuen Gruppe", "")
if not name:
return
if self.config.group(name):
self._message(win, "Eine Gruppe mit diesem Namen gibt es bereits.", error=True)
return
group = Group(name=name, interval=3600, keep_count=24, keep_time=2 * 86400)
self.config.groups.append(group)
self.dirty = True
self.screen_group(win, group)
def _show_problems(self, win):
problems = self.config.validate()
height, width = self._frame(win, "Hinweise zur Konfiguration", self.path,
"beliebige Taste = zurueck")
if not problems:
_put(win, 4, 4, "Alles in Ordnung.", curses.color_pair(C_OK))
for row, problem in enumerate(problems[: height - 6]):
_put(win, 4 + row, 4, truncate("- " + problem, width - 6))
win.refresh()
_read_key(win)
def _save(self, win):
problems = self.config.validate()
if problems:
if not self._confirm(win, "%d Hinweis(e) - trotzdem speichern?" % len(problems)):
return
try:
target = save_config(self.config, self.path)
self.dirty = False
self._message(win, "Gespeichert: %s (systemctl reload pvesnap nicht vergessen)"
% target)
except OSError as exc:
self._message(win, "Speichern fehlgeschlagen: %s" % exc, error=True)
# -- Bildschirm: eine Gruppe -----------------------------------------
def screen_group(self, win, group):
index = 0
top = 0
while True:
fields = self._group_fields(group)
index = max(0, min(index, len(fields) - 1))
while fields[index][0] == "-" and index < len(fields) - 1:
index += 1
height, width = self._frame(
win, "Gruppe: %s" % group.name,
"Kurzname im Snapshot: %s-%s-JJJJMMTT-HHMMSS"
% (self.config.globals.prefix,
effective_slug(self.config.globals.prefix, group.slug)),
"Enter Aendern | Leer Umschalten | v VM-Auswahl | q zurueck")
visible = height - 5
if index < top:
top = index
if index >= top + visible:
top = index - visible + 1
for row in range(visible):
position = top + row
if position >= len(fields):
break
kind, label, value, _handler = fields[position]
if kind == "-":
_put(win, 3 + row, 2, "-- %s " % label + "-" * max(0, width - len(label) - 8),
curses.color_pair(C_DIM) | curses.A_BOLD)
continue
attr = curses.color_pair(C_SEL) if position == index else 0
_fill_row(win, 3 + row, 2,
truncate(label, 30).ljust(32) + truncate(value, max(4, width - 40)),
width, attr)
win.refresh()
key = _read_key(win)
if key in (curses.KEY_UP, "k"):
index = self._step(fields, index, -1)
elif key in (curses.KEY_DOWN, "j"):
index = self._step(fields, index, +1)
elif key == curses.KEY_NPAGE:
index = self._step(fields, min(index + 10, len(fields) - 1), +1, land=True)
elif key == curses.KEY_PPAGE:
index = self._step(fields, max(index - 10, 0), -1, land=True)
elif _is_enter(key) or key == " ":
handler = fields[index][3]
if handler:
handler(win)
self.dirty = True
elif key == "v":
self._pick_vms(win, group)
elif key in ("q", "Q") or _is_escape(key):
return
@staticmethod
def _step(fields, index, direction, land=False):
position = index if land else index + direction
while 0 <= position < len(fields) and fields[position][0] == "-":
position += direction
if position < 0 or position >= len(fields):
return index
return position
def _group_fields(self, group):
"""Liste von (art, beschriftung, angezeigter_wert, handler)."""
fields = []
def text_field(label, getter, setter, hint=""):
def handler(win):
value = self._prompt(win, hint or label, getter())
if value is not None:
setter(value)
fields.append(("f", label, getter(), handler))
def bool_field(label, getter, setter):
def handler(_win):
setter(not getter())
fields.append(("f", label, "ja" if getter() else "nein", handler))
def int_field(label, getter, setter, zero_text=""):
def handler(win):
raw = self._prompt(win, label, str(getter()))
if raw is None:
return
try:
setter(max(0, int(raw.strip() or 0)))
except ValueError:
self._message(win, "Bitte eine ganze Zahl eingeben.", error=True)
value = getter()
fields.append(("f", label, zero_text if (value == 0 and zero_text) else str(value),
handler))
def duration_field(label, getter, setter):
def handler(win):
raw = self._prompt(win, label + " (z.B. 30m, 6h, 7d, 0=unbegrenzt)",
format_duration(getter(), zero="0"))
if raw is None:
return
try:
setter(parse_duration(raw, default_unit="m"))
except ValueError as exc:
self._message(win, str(exc), error=True)
fields.append(("f", label, format_duration(getter()), handler))
def list_field(label, getter, setter, hint=""):
def handler(win):
raw = self._prompt(win, hint or label, ", ".join(getter()))
if raw is not None:
setter(parse_list(raw))
fields.append(("f", label, ", ".join(getter()) or "-", handler))
# --- Allgemein ---
fields.append(("-", "Allgemein", "", None))
def set_name(value):
if value:
group.name = value
text_field("Name", lambda: group.name, set_name)
bool_field("Aktiv", lambda: group.enabled,
lambda v: setattr(group, "enabled", v))
# --- Zeitplan ---
fields.append(("-", "Zeitplan", "", None))
current_kind = group.schedule or "interval"
def change_kind(win):
values = [c[0] for c in SCHEDULE_CHOICES]
chosen = self._choose(win, "Zeitplan", SCHEDULE_CHOICES,
values.index(current_kind))
if chosen is None:
return
if chosen == "interval":
group.schedule = ""
group.interval = group.interval or 3600
else:
group.schedule = chosen
group.interval = 0
fields.append(("f", "Art des Zeitplans",
dict(SCHEDULE_CHOICES)[current_kind], change_kind))
if current_kind == "interval":
duration_field("Intervall", lambda: group.interval,
lambda v: setattr(group, "interval", max(60, v)))
bool_field("An der Uhr ausrichten", lambda: group.align,
lambda v: setattr(group, "align", v))
elif current_kind == "hourly":
int_field("Minute (0-59)", lambda: group.minute,
lambda v: setattr(group, "minute", min(59, v)))
else:
def set_time(value):
try:
group.at = "%02d:%02d" % parse_time_of_day(value)
except ValueError:
pass
text_field("Uhrzeit (HH:MM)", lambda: group.at, set_time)
if current_kind == "weekly":
def change_day(win):
entries = [(i, name) for i, name in enumerate(
["Montag", "Dienstag", "Mittwoch", "Donnerstag",
"Freitag", "Samstag", "Sonntag"])]
chosen = self._choose(win, "Wochentag", entries, group.day_of_week)
if chosen is not None:
group.day_of_week = chosen
fields.append(("f", "Wochentag",
WEEKDAY_NAMES[group.day_of_week], change_day))
if current_kind in ("monthly", "yearly"):
int_field("Tag im Monat (1-31)", lambda: group.day_of_month,
lambda v: setattr(group, "day_of_month", max(1, min(31, v))))
if current_kind == "yearly":
int_field("Monat (1-12)", lambda: group.month,
lambda v: setattr(group, "month", max(1, min(12, v))))
fields.append(("f", "Naechster Termin (Vorschau)", _preview_next(group), None))
# --- Vorhaltezeit ---
fields.append(("-", "Vorhaltezeit", "", None))
int_field("Anzahl behalten (je VM)", lambda: group.keep_count,
lambda v: setattr(group, "keep_count", v), zero_text="unbegrenzt")
duration_field("Maximales Alter", lambda: group.keep_time,
lambda v: setattr(group, "keep_time", v))
int_field("Mindestens behalten", lambda: group.keep_min,
lambda v: setattr(group, "keep_min", v), zero_text="0 (keine Untergrenze)")
# --- Auswahl ---
fields.append(("-", "Welche VMs?", "", None))
bool_field("Alle VMs/Container", lambda: group.all,
lambda v: setattr(group, "all", v))
fields.append(("f", "VMs aus Liste waehlen ...",
format_vmid_list(group.vmids) or "-",
lambda win: self._pick_vms(win, group)))
list_field("Namensmuster (z.B. web-*)", lambda: group.names,
lambda v: setattr(group, "names", v))
list_field("Tags", lambda: group.tags,
lambda v: setattr(group, "tags", [t.lower() for t in v]))
list_field("Pools", lambda: group.pools,
lambda v: setattr(group, "pools", v))
def change_types(win):
labels = [(index, text) for index, (_v, text) in enumerate(TYPE_CHOICES)]
current = 0
for index, (value, _text) in enumerate(TYPE_CHOICES):
if sorted(value) == sorted(group.types):
current = index
chosen = self._choose(win, "Welche Gasttypen?", labels, current)
if chosen is not None:
group.types = list(TYPE_CHOICES[chosen][0])
current_type_text = "alle (VMs und Container)"
for value, text in TYPE_CHOICES:
if sorted(value) == sorted(group.types):
current_type_text = text
fields.append(("f", "Gasttypen", current_type_text, change_types))
def set_exclude_vmids(win):
raw = self._prompt(win, "Ausgeschlossene VMIDs (z.B. 900,905-910)",
format_vmid_list(group.exclude_vmids))
if raw is None:
return
try:
group.exclude_vmids = parse_vmid_list(raw)
except ValueError as exc:
self._message(win, str(exc), error=True)
fields.append(("f", "Ausgeschlossene VMIDs",
format_vmid_list(group.exclude_vmids) or "-", set_exclude_vmids))
list_field("Ausgeschlossene Namensmuster", lambda: group.exclude_names,
lambda v: setattr(group, "exclude_names", v))
list_field("Ausgeschlossene Tags", lambda: group.exclude_tags,
lambda v: setattr(group, "exclude_tags", [t.lower() for t in v]))
matched = self._match_count(group)
fields.append(("f", "Trifft aktuell zu auf", matched, None))
# --- Optionen ---
fields.append(("-", "Snapshot-Optionen", "", None))
bool_field("RAM mitsichern (nur laufende VMs)", lambda: group.vmstate,
lambda v: setattr(group, "vmstate", v))
bool_field("Gestoppte ueberspringen", lambda: group.skip_stopped,
lambda v: setattr(group, "skip_stopped", v))
text_field("Beschreibung (Vorlage)",
lambda: group.description or self.config.globals.description,
lambda v: setattr(group, "description", v),
hint="Beschreibung, Platzhalter: {group} {vmid} {name} {datetime} "
"{keep_time} {keep_count}")
return fields
def _match_count(self, group):
guests = self.guests()
if not guests:
return self._inventory_error or "unbekannt"
from .engine import select_guests
selected = select_guests(group, guests)
if not selected:
return "0 Gaeste"
preview = ", ".join("%d" % g.vmid for g in selected[:8])
return "%d Gast/Gaeste (%s%s)" % (len(selected), preview,
" ..." if len(selected) > 8 else "")
# -- Bildschirm: VM-Auswahl ------------------------------------------
def _pick_vms(self, win, group):
guests = self.guests()
if not guests:
self._message(win, self._inventory_error or "Keine VMs gefunden.", error=True)
return
chosen = set(group.vmids)
index, top, filter_text = 0, 0, ""
while True:
shown = [g for g in guests if self._matches_filter(g, filter_text)]
index = max(0, min(index, max(0, len(shown) - 1)))
height, width = self._frame(
win, "VMs fuer Gruppe '%s'" % group.name,
"%d ausgewaehlt%s" % (len(chosen),
(" Filter: " + filter_text) if filter_text else ""),
"Leer Auswaehlen | a Alle | n Keine | i Umkehren | / Filter | "
"Enter Uebernehmen | q Abbruch")
_put(win, 3, 2, " VMID Typ Name".ljust(46) + "Node".ljust(14)
+ "Status".ljust(10) + "Tags", curses.A_BOLD)
visible = max(1, height - 6)
if index < top:
top = index
if index >= top + visible:
top = index - visible + 1
for row in range(visible):
position = top + row
if position >= len(shown):
break
guest = shown[position]
attr = curses.color_pair(C_SEL) if position == index else 0
mark = "[x]" if guest.vmid in chosen else "[ ]"
line = ("%s %s %s %s%s%s%s"
% (mark,
str(guest.vmid).rjust(6),
("LXC" if guest.type == "lxc" else "VM ").ljust(4),
truncate(guest.name or "-", 26).ljust(28),
truncate(guest.node, 12).ljust(14),
truncate(guest.status, 9).ljust(10),
truncate(",".join(guest.tags) or "-", 20)))
_fill_row(win, 4 + row, 2, line, width, attr)
win.refresh()
key = _read_key(win)
if key in (curses.KEY_UP, "k"):
index -= 1
elif key in (curses.KEY_DOWN, "j"):
index += 1
elif key == curses.KEY_NPAGE:
index += visible
elif key == curses.KEY_PPAGE:
index -= visible
elif key == curses.KEY_HOME:
index = 0
elif key == curses.KEY_END:
index = len(shown) - 1
elif key == " " and shown:
vmid = shown[index].vmid
if vmid in chosen:
chosen.discard(vmid)
else:
chosen.add(vmid)
index += 1
elif key == "a":
chosen.update(g.vmid for g in shown)
elif key == "n":
chosen.difference_update(g.vmid for g in shown)
elif key == "i":
for guest in shown:
if guest.vmid in chosen:
chosen.discard(guest.vmid)
else:
chosen.add(guest.vmid)
elif key == "/":
entered = self._prompt(win, "Filter (Name, VMID oder Tag)", filter_text)
filter_text = entered or ""
index = top = 0
elif key == "r":
self.guests(refresh=True)
guests = self._guests
elif _is_enter(key):
group.vmids = sorted(chosen)
self.dirty = True
return
elif key in ("q", "Q") or _is_escape(key):
return
index = max(0, min(index, max(0, len(shown) - 1)))
@staticmethod
def _matches_filter(guest, text):
if not text:
return True
needle = text.lower()
return (needle in (guest.name or "").lower()
or needle in str(guest.vmid)
or needle in " ".join(guest.tags)
or needle in (guest.pool or "").lower()
or needle in (guest.node or "").lower())
# -- Bildschirm: globale Einstellungen -------------------------------
def screen_globals(self, win):
globals_ = self.config.globals
index = 0
while True:
fields = [
("Praefix im Snapshot-Namen", globals_.prefix, "text", "prefix"),
("Pruefintervall des Dienstes", format_duration(globals_.check_interval),
"duration", "check_interval"),
("Zustandsdatei", globals_.state_file, "text", "state_file"),
("Protokollstufe", globals_.log_level, "level", "log_level"),
("Zusaetzliche Logdatei", globals_.log_file or "-", "text", "log_file"),
("Zeitlimit je Snapshot-Task", format_duration(globals_.task_timeout),
"duration", "task_timeout"),
("Beim Start sofort ausfuehren", "ja" if globals_.run_on_start else "nein",
"bool", "run_on_start"),
("Testlauf (nichts wirklich tun)", "ja" if globals_.dry_run else "nein",
"bool", "dry_run"),
("Standard-Beschreibung", globals_.description, "text", "description"),
]
index = max(0, min(index, len(fields) - 1))
height, width = self._frame(win, "Globale Einstellungen", self.path,
"Enter Aendern | q zurueck")
for row, (label, value, _kind, _key) in enumerate(fields):
attr = curses.color_pair(C_SEL) if row == index else 0
_fill_row(win, 4 + row, 2,
truncate(label, 32).ljust(34) + truncate(str(value), max(4, width - 42)),
width, attr)
_put(win, 6 + len(fields), 2,
"Platzhalter der Beschreibung: {group} {vmid} {name} {node} {type} "
"{datetime} {date} {time} {keep_time} {keep_count} {schedule}",
curses.color_pair(C_DIM))
win.refresh()
key = _read_key(win)
if key in (curses.KEY_UP, "k"):
index -= 1
elif key in (curses.KEY_DOWN, "j"):
index += 1
elif _is_enter(key) or key == " ":
label, _value, kind, attribute = fields[index]
self._edit_global(win, label, kind, attribute)
elif key in ("q", "Q") or _is_escape(key):
return
def _edit_global(self, win, label, kind, attribute):
globals_ = self.config.globals
if kind == "bool":
setattr(globals_, attribute, not getattr(globals_, attribute))
self.dirty = True
return
if kind == "level":
entries = [(lvl, lvl) for lvl in ("DEBUG", "INFO", "WARNING", "ERROR")]
chosen = self._choose(win, "Protokollstufe", entries)
if chosen:
globals_.log_level = chosen
self.dirty = True
return
if kind == "duration":
raw = self._prompt(win, label, format_duration(getattr(globals_, attribute),
zero="0"))
if raw is None:
return
try:
setattr(globals_, attribute, parse_duration(raw, default_unit="s"))
self.dirty = True
except ValueError as exc:
self._message(win, str(exc), error=True)
return
raw = self._prompt(win, label, getattr(globals_, attribute))
if raw is None:
return
setattr(globals_, attribute, raw)
self.dirty = True
# -- Bildschirm: Uebersicht ------------------------------------------
def screen_overview(self, win):
from .engine import select_guests
guests = self.guests()
top = 0
while True:
height, width = self._frame(win, "Uebersicht: welche VM in welcher Gruppe?",
self._inventory_error or "%d Gaeste" % len(guests),
"Pfeiltasten Blaettern | r Neu laden | q zurueck")
_put(win, 3, 2, "VMID".ljust(8) + "Name".ljust(26) + "Status".ljust(10) + "Gruppen",
curses.A_BOLD)
rows = []
for guest in guests:
names = [g.name for g in self.config.groups if guest in select_guests(g, [guest])]
rows.append((guest, names))
visible = max(1, height - 6)
top = max(0, min(top, max(0, len(rows) - visible)))
for row in range(visible):
position = top + row
if position >= len(rows):
break
guest, names = rows[position]
attr = 0 if names else curses.color_pair(C_DIM) | curses.A_DIM
_put(win, 4 + row, 2,
truncate(str(guest.vmid), 7).ljust(8)
+ truncate(guest.name or "-", 24).ljust(26)
+ truncate(guest.status, 9).ljust(10)
+ truncate(", ".join(names) or "(keine Gruppe)", max(4, width - 48)),
attr)
win.refresh()
key = _read_key(win)
if key == curses.KEY_DOWN:
top += 1
elif key == curses.KEY_UP:
top -= 1
elif key == curses.KEY_NPAGE:
top += visible
elif key == curses.KEY_PPAGE:
top -= visible
elif key == "r":
guests = self.guests(refresh=True)
elif key in ("q", "Q") or _is_escape(key):
return
top = max(0, top)
def _preview_next(group):
"""Zeigt den naechsten Termin, damit man die Einstellung sofort pruefen kann."""
from datetime import datetime
from .schedule import next_due
try:
moment = next_due(group, None, datetime.now())
except ValueError:
return "-"
tage = ["Mo", "Di", "Mi", "Do", "Fr", "Sa", "So"]
return "%s %s (%s)" % (tage[moment.weekday()],
moment.strftime("%d.%m.%Y %H:%M"), describe(group))
def _fill_row(win, y, x, text, width, attr):
_put(win, y, x, str(text).ljust(max(0, width - x - 1))[: max(0, width - x - 1)], attr)
def run_editor(path):
editor = Editor(path)
try:
editor.load()
except ConfigError as exc:
print(str(exc))
return 2
curses.wrapper(editor.run)
if editor.dirty:
print("Achtung: Aenderungen wurden NICHT gespeichert.")
return 1
return 0
+162
View File
@@ -0,0 +1,162 @@
"""Kleine Hilfsfunktionen fuer pvesnap (Dauer-Parsing, Slugs, Listen)."""
from __future__ import annotations
import re
import unicodedata
_DURATION_UNITS = {
"s": 1, "sec": 1, "secs": 1, "second": 1, "seconds": 1,
"sek": 1, "sekunde": 1, "sekunden": 1,
"m": 60, "min": 60, "mins": 60, "minute": 60, "minutes": 60, "minuten": 60,
"h": 3600, "hour": 3600, "hours": 3600, "std": 3600, "stunde": 3600, "stunden": 3600,
"d": 86400, "day": 86400, "days": 86400, "t": 86400, "tag": 86400, "tage": 86400,
"w": 604800, "week": 604800, "weeks": 604800, "woche": 604800, "wochen": 604800,
"mo": 2592000, "month": 2592000, "months": 2592000, "monat": 2592000, "monate": 2592000,
"y": 31536000, "year": 31536000, "years": 31536000,
"j": 31536000, "jahr": 31536000, "jahre": 31536000,
}
_DURATION_TOKEN = re.compile(r"(\d+(?:[.,]\d+)?)\s*([a-zäöü]*)")
_NEVER = {"0", "never", "nie", "unbegrenzt", "unlimited", "off", "aus", "keine", "-"}
def parse_duration(text, default_unit="s"):
"""'2d12h' -> 216000. 0 / 'nie' / 'unbegrenzt' -> 0 (= keine Begrenzung)."""
if text is None:
raise ValueError("leere Zeitangabe")
s = str(text).strip().lower()
if not s:
raise ValueError("leere Zeitangabe")
if s in _NEVER:
return 0
total = 0.0
pos = 0
found = False
for match in _DURATION_TOKEN.finditer(s):
if match.start() != pos:
raise ValueError("ungueltige Zeitangabe: %r" % text)
pos = match.end()
value = float(match.group(1).replace(",", "."))
unit = match.group(2) or default_unit
if unit not in _DURATION_UNITS:
raise ValueError("unbekannte Zeiteinheit %r in %r" % (unit, text))
total += value * _DURATION_UNITS[unit]
found = True
if not found or pos != len(s):
raise ValueError("ungueltige Zeitangabe: %r" % text)
return int(round(total))
def format_duration(seconds, zero="unbegrenzt"):
"""216000 -> '2d12h'. Ergebnis ist wieder von parse_duration lesbar."""
seconds = int(seconds or 0)
if seconds <= 0:
return zero
parts = []
for unit, size in (("w", 604800), ("d", 86400), ("h", 3600), ("m", 60), ("s", 1)):
if seconds >= size:
count, seconds = divmod(seconds, size)
parts.append("%d%s" % (count, unit))
return "".join(parts)
def slugify(text, maxlen=20):
"""Gruppenname -> kurzer, PVE-tauglicher Bezeichner (nur a-z0-9, beginnt mit Buchstabe)."""
# Umlaute vor der Normalisierung ersetzen - NFKD wuerde sie sonst zerlegen.
lowered = str(text).lower()
for umlaut, replacement in (("ä", "ae"), ("ö", "oe"), ("ü", "ue"), ("ß", "ss")):
lowered = lowered.replace(umlaut, replacement)
ascii_only = unicodedata.normalize("NFKD", lowered).encode("ascii", "ignore").decode("ascii")
slug = re.sub(r"[^a-z0-9]+", "", ascii_only)
if not slug:
slug = "grp"
if not slug[0].isalpha():
slug = "g" + slug
return slug[:maxlen]
_TRUE = {"1", "yes", "y", "true", "on", "ja", "j", "an", "ein", "enabled", "aktiv"}
_FALSE = {"0", "no", "n", "false", "off", "nein", "aus", "disabled", "inaktiv", ""}
def parse_bool(text, default=None):
if isinstance(text, bool):
return text
value = str(text).strip().lower()
if value in _TRUE:
return True
if value in _FALSE:
return default if (value == "" and default is not None) else False
raise ValueError("ungueltiger Ja/Nein-Wert: %r" % text)
def format_bool(value):
return "yes" if value else "no"
def parse_list(text):
"""Kommagetrennte (oder per Semikolon/Whitespace getrennte) Liste -> Liste von Strings."""
if not text:
return []
if isinstance(text, (list, tuple)):
items = list(text)
else:
items = re.split(r"[,;\s]+", str(text))
return [item.strip() for item in items if item and item.strip()]
def parse_vmid_list(text):
"""'100,101,105-108' -> [100, 101, 105, 106, 107, 108]."""
result = []
for token in parse_list(text):
match = re.match(r"^(\d+)\s*-\s*(\d+)$", token)
if match:
start, end = int(match.group(1)), int(match.group(2))
if end < start:
start, end = end, start
if end - start > 100000:
raise ValueError("VMID-Bereich zu gross: %r" % token)
result.extend(range(start, end + 1))
elif token.isdigit():
result.append(int(token))
else:
raise ValueError("ungueltige VMID: %r" % token)
return sorted(set(result))
def format_vmid_list(vmids):
"""[100,101,102,105] -> '100-102,105' (Umkehrung von parse_vmid_list)."""
ids = sorted(set(int(v) for v in vmids))
if not ids:
return ""
chunks = []
start = previous = ids[0]
for vmid in ids[1:]:
if vmid == previous + 1:
previous = vmid
continue
chunks.append((start, previous))
start = previous = vmid
chunks.append((start, previous))
out = []
for begin, end in chunks:
if end - begin >= 2:
out.append("%d-%d" % (begin, end))
else:
out.extend(str(v) for v in range(begin, end + 1))
return ",".join(out)
def truncate(text, width):
text = str(text)
if width <= 0:
return ""
if len(text) <= width:
return text
if width <= 1:
return text[:width]
return text[: width - 1] + ""
+32
View File
@@ -0,0 +1,32 @@
[Unit]
Description=pvesnap - automatische Proxmox-Snapshots
Documentation=file:/usr/share/doc/pvesnap/README.md
After=network-online.target pve-cluster.service pvedaemon.service
Wants=pve-cluster.service pvedaemon.service
[Service]
Type=simple
ExecStart=/usr/bin/python3 -m pvesnap --config /etc/pvesnap/pvesnap.conf daemon
ExecReload=/bin/kill -HUP $MAINPID
Environment=PYTHONPATH=/usr/lib/pvesnap
Environment=PYTHONUNBUFFERED=1
Restart=on-failure
RestartSec=30
TimeoutStopSec=300
# pvesh benoetigt root-Rechte.
User=root
StateDirectory=pvesnap
RuntimeDirectory=pvesnap
# Moderate Absicherung - der Dienst muss auf pvesh und /etc/pve zugreifen.
NoNewPrivileges=yes
PrivateTmp=yes
ProtectHome=yes
ProtectSystem=full
ProtectKernelTunables=yes
ProtectControlGroups=yes
RestrictRealtime=yes
[Install]
WantedBy=multi-user.target
Executable
+91
View File
@@ -0,0 +1,91 @@
#!/bin/bash
# ---------------------------------------------------------------------
# pvesnap - Deinstallation
#
# ./uninstall.sh Programm und Dienst entfernen,
# Konfiguration und Zustand bleiben erhalten
# ./uninstall.sh --purge zusaetzlich /etc/pvesnap und /var/lib/pvesnap
# entfernen
#
# Bereits angelegte Snapshots werden NIE angeruehrt.
# ---------------------------------------------------------------------
set -euo pipefail
LIB_DIR="/usr/lib/pvesnap"
BIN="/usr/local/bin/pvesnap"
CONF_DIR="/etc/pvesnap"
STATE_DIR="/var/lib/pvesnap"
DOC_DIR="/usr/share/doc/pvesnap"
UNIT="/etc/systemd/system/pvesnap.service"
PURGE=0
YES=0
for arg in "$@"; do
case "$arg" in
--purge) PURGE=1 ;;
-y|--yes) YES=1 ;;
-h|--help) sed -n '2,11p' "$0" | sed 's/^#[[:space:]]\{0,1\}//'; exit 0 ;;
*) echo "Unbekannte Option: $arg" >&2; exit 1 ;;
esac
done
info() { printf '\033[1;32m==>\033[0m %s\n' "$*"; }
warn() { printf '\033[1;33m==>\033[0m %s\n' "$*" >&2; }
fail() { printf '\033[1;31m==>\033[0m %s\n' "$*" >&2; exit 1; }
[ "$(id -u)" -eq 0 ] || fail "Bitte als root ausfuehren (sudo ./uninstall.sh)."
if [ "$YES" -eq 0 ]; then
if [ "$PURGE" -eq 1 ]; then
printf 'pvesnap inklusive Konfiguration und Zustand entfernen? [j/N] '
else
printf 'pvesnap entfernen (Konfiguration bleibt erhalten)? [j/N] '
fi
read -r answer
case "$answer" in
j|J|y|Y) ;;
*) echo "Abgebrochen."; exit 0 ;;
esac
fi
if systemctl list-unit-files 2>/dev/null | grep -q '^pvesnap\.service'; then
info "Stoppe und deaktiviere Dienst"
systemctl stop pvesnap.service 2>/dev/null || true
systemctl disable pvesnap.service 2>/dev/null || true
fi
if [ -f "$UNIT" ]; then
info "Entferne $UNIT"
rm -f "$UNIT"
fi
systemctl daemon-reload 2>/dev/null || true
systemctl reset-failed pvesnap.service 2>/dev/null || true
if [ -e "$BIN" ]; then
info "Entferne $BIN"
rm -f "$BIN"
fi
if [ -d "$LIB_DIR" ]; then
info "Entferne $LIB_DIR"
rm -rf "$LIB_DIR"
fi
rm -rf "$DOC_DIR"
rm -f /run/pvesnap.lock /run/pvesnap.lock.daemon
if [ "$PURGE" -eq 1 ]; then
info "Entferne Konfiguration und Zustand"
rm -rf "$CONF_DIR" "$STATE_DIR"
else
warn "Behalten: $CONF_DIR und $STATE_DIR (mit --purge ebenfalls entfernen)"
fi
cat <<EOF
pvesnap wurde entfernt.
Bereits vorhandene Snapshots bleiben bestehen. Auflisten und loeschen lassen
sie sich weiterhin in der Proxmox-Oberflaeche oder mit:
qm listsnapshot <VMID> / qm delsnapshot <VMID> <NAME>
pct listsnapshot <VMID> / pct delsnapshot <VMID> <NAME>
EOF