"""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" retries: int = 2 # Wiederholungen bei belegten Sperren retry_delay: int = 60 # Wartezeit dazwischen pause_between: int = 0 # Pause zwischen zwei Gaesten def validate(self): problems = [] if self.retries < 0: problems.append("[global] 'retries' darf nicht negativ sein") if self.retry_delay < 1: problems.append("[global] 'retry_delay' muss mindestens 1 Sekunde betragen") if self.pause_between < 0: problems.append("[global] 'pause_between' darf nicht negativ sein") 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", } # Kommentar am Zeilenende abschneiden ("keep_time = 7d # eine Woche"). # Absichtlich nicht ueber configparser: in Beschreibungs-Vorlagen soll ein '#' # erhalten bleiben, deshalb wird dieser Schnitt dort nicht angewandt. _INLINE_COMMENT = re.compile(r"\s+[#;].*$", re.DOTALL) # Schluessel, deren Wert unangetastet bleibt (freier Text). _VERBATIM_KEYS = {"description", "description_template"} def _strip_comment(value): return _INLINE_COMMENT.sub("", str(value)).strip() 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 raw = section[key] if key in _VERBATIM_KEYS else _strip_comment(section[key]) try: setattr(result, target or key, parser_fn(raw)) 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()) take("retries", lambda v: int(str(v).strip())) take("retry_delay", lambda v: parse_duration(v, default_unit="s")) take("pause_between", lambda v: parse_duration(v, default_unit="s")) # 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", "retries", "retry_delay", "pause_between"): 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] if key in _VERBATIM_KEYS else _strip_comment(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("retries = %d" % g.retries) lines.append("retry_delay = %s" % format_duration(g.retry_delay, zero="60s")) lines.append("pause_between = %s" % format_duration(g.pause_between, zero="0")) 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))