Files
proxmox-snapshot-service/pvesnap/curses_util.py
T
duffyduckandClaude Opus 5.5 74be5aee59 Platz vor jedem Snapshot pruefen, bei Mangel alte loeschen, per Mail warnen
Volle Ceph-Pools legen alle VMs darauf lahm. Vor jedem Snapshot gilt jetzt
je Storage des Gastes: frei >= sofortiger Verbrauch + space_min_free
(Vorgabe 20 %). Die Reserve bleibt also auch nach dem Snapshot frei.

Reicht es nicht, wird der aelteste pvesnap-Snapshot des Gastes geloescht und
neu geprueft, bis es reicht. Sind nur noch die neuesten space_keep (Vorgabe
7) uebrig, entfaellt der Snapshot und der naechste Gast kommt dran.

Gezaehlt wird je Gast ueber alle Gruppen, anders als bei der Vorhaltezeit.
Auf dem Testhost haette eine Grenze je Gruppe nie gegriffen: keep_count = 5
liegt unter 7, und 13 GiB steckten in einer Gruppe, die es nicht mehr gibt.

Sofortiger Verbrauch: Dicke ZFS-Zvols muessen ihre Reservierung auch nach
dem Snapshot garantieren, das kostet alles seit dem letzten Snapshot
Geschriebene. Exakt ist das refreservation - usedbyrefreservation; die Werte
kommen per zfs get, bei anderen Nodes per ssh ueber die IP aus dem
Clusterstatus, weil Node-Namen nicht ueberall aufloesen. Dazu kommt der RAM
bei vmstate. --dry-run zeigt die Rechnung je Storage.

Ceph und ZFS geben Platz erst im Hintergrund frei. Deshalb wird nach jedem
Loeschen gewartet, solange der Platz waechst, und erst nach space_settle
(Vorgabe 30s) ohne Zuwachs weitergemacht. Ohne das wurde auf Ceph ein Snapshot
zu viel geloescht. Laesst sich der Platz nicht feststellen, wird nichts
geloescht und der Snapshot trotzdem versucht.

Neu ist [mail]: eine Alarm-Mail je Gruppe und Lauf, wenn ein Snapshot wegen
Platzmangels ausfaellt. STARTTLS oder SSL/TLS mit Zertifikatspruefung, ohne
Rueckfall auf Klartext. Einzurichten im Editor (Taste m) oder von Hand, Test
mit pvesnap mail-test. Liegt ein Passwort in der Datei, setzt der Editor die
Rechte auf 600, auch fuer die .bak.

Das Handbuch beschreibt, wie man Ceph (Ratios) und ZFS (Notreserve) frueher
"voll" sagen laesst und was im Notfall zu tun ist. Ausserdem war
"pvesnap run --dry-run" ueberall falsch: Die Option ist global und muss vor
dem Unterbefehl stehen.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-01 10:27:21 +01:00

239 lines
7.7 KiB
Python

"""Gemeinsame curses-Bausteine fuer Konfigurationseditor und Explorer."""
from __future__ import annotations
import curses
C_HEADER, C_FOOTER, C_SEL, C_WARN, C_OK, C_DIM, C_DIR, C_MARK, C_MSG = range(1, 10)
def init_colors():
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)
# Gruen als Schriftfarbe, nicht als Hintergrund. Schwarz auf Gruen sah
# im Terminal matschig aus - und mit A_BOLD wird aus Schwarz (Farbe 0)
# Hellschwarz (Farbe 8), also Grau. Grau auf Gruen kann man nicht lesen.
curses.init_pair(C_OK, curses.COLOR_GREEN, -1)
curses.init_pair(C_DIM, -1, -1)
curses.init_pair(C_DIR, curses.COLOR_CYAN, -1)
curses.init_pair(C_MARK, curses.COLOR_YELLOW, -1)
# Meldungsleiste: Weiss auf Blau bleibt auch fett gut lesbar, weil die
# Schrift dabei heller wird statt dunkler.
curses.init_pair(C_MSG, curses.COLOR_WHITE, curses.COLOR_BLUE)
except curses.error:
pass
# ---------------------------------------------------------------------------
# Zeichnen
# ---------------------------------------------------------------------------
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 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)
# ---------------------------------------------------------------------------
# Tasten
# ---------------------------------------------------------------------------
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)
def keybar(width, items):
"""Tastenleiste als Liste von Zeilen, passend zur Breite des Terminals.
Reicht der Platz nicht fuer eine Zeile, wird auf zwei Zeilen umgebrochen;
bei ganz schmalen Terminals bleiben nur noch die Tasten uebrig.
"""
def join(entries, index):
return " | ".join("%s %s" % (entry[0], entry[index]) for entry in entries)
long_text = join(items, 1)
if len(long_text) + 2 <= width:
return [long_text]
half = (len(items) + 1) // 2
first, second = join(items[:half], 1), join(items[half:], 1)
if max(len(first), len(second)) + 2 <= width:
return [first, second]
short_text = join(items, 2)
if len(short_text) + 2 <= width:
return [short_text]
first, second = join(items[:half], 2), join(items[half:], 2)
if max(len(first), len(second)) + 2 <= width:
return [first, second]
return [" ".join(entry[0] for entry in items)]
# ---------------------------------------------------------------------------
# Dialoge
# ---------------------------------------------------------------------------
def message(win, text, error=False):
height, _ = win.getmaxyx()
attr = curses.color_pair(C_WARN if error else C_MSG) | curses.A_BOLD
fill(win, height - 2, " " + str(text) + " [Taste druecken]", attr)
win.refresh()
read_key(win)
def confirm(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 ask(win, question, answers, default=None):
"""Mehrfachauswahl in einer Zeile. `answers` ist z.B. 'jnak'."""
height, _ = win.getmaxyx()
fill(win, height - 2, " %s" % question, curses.color_pair(C_WARN) | curses.A_BOLD)
win.refresh()
while True:
key = read_key(win)
if isinstance(key, str) and key.lower() in answers:
return key.lower()
if is_escape(key):
return default
def prompt(win, label, initial="", secret=False):
"""Einzeilige Eingabe am unteren Rand. None = abgebrochen.
Mit `secret` erscheinen statt der Zeichen Sternchen.
"""
height, width = win.getmaxyx()
buffer = list(str(initial))
curses.curs_set(1)
try:
while True:
text = "".join(buffer)
label_text = " %s: " % label
visible = width - len(label_text) - 2
shown = text[-visible:] if visible > 0 and len(text) > visible else text
if secret:
shown = "*" * len(shown)
fill(win, height - 2, label_text + shown, curses.color_pair(C_SEL))
try:
win.move(height - 2, min(width - 2, len(label_text) + 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(win, title, entries, current=0):
"""Kleines Auswahlfenster. entries: Liste von (wert, beschriftung)."""
if not entries:
return None
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))
scroll = 0
rows = max(1, box_h - 4)
while True:
if index < scroll:
scroll = index
if index >= scroll + rows:
scroll = index - rows + 1
box.erase()
box.box()
put(box, 0, 2, " %s " % title, curses.A_BOLD)
for row in range(rows):
position = scroll + row
if position >= len(entries):
break
attr = curses.color_pair(C_SEL) if position == index else 0
put(box, row + 2, 2, " " + entries[position][1].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 key == curses.KEY_NPAGE:
index = min(len(entries) - 1, index + rows)
elif key == curses.KEY_PPAGE:
index = max(0, index - rows)
elif is_enter(key):
return entries[index][0]
elif is_escape(key) or key == "q":
return None