diff --git a/Makefile b/Makefile
index d0961ef33..88ac9d37a 100755
--- a/Makefile
+++ b/Makefile
@@ -29,6 +29,7 @@ SM_LIBS += vditype
SM_LIBS += BaseISCSI
SM_LIBS += cleanup
SM_LIBS += lvutil
+SM_LIBS += lvmbackup
SM_LIBS += lvmcache
SM_LIBS += util
SM_LIBS += verifyVDIsOnSR
@@ -73,6 +74,7 @@ SM_LIBS += fcoelib
SM_LIBS += constants
SM_LIBS += cbtutil
SM_LIBS += sr_health_check
+SM_LIBS += backupmanager
UDEV_RULES = 65-multipath 55-xs-mpath-scsidev 57-usb 58-xapi 99-purestorage
MPATH_DAEMON = sm-multipath
diff --git a/drivers/LVMSR.py b/drivers/LVMSR.py
index 9c8fbf874..13204c592 100755
--- a/drivers/LVMSR.py
+++ b/drivers/LVMSR.py
@@ -572,6 +572,7 @@ def delete(self, uuid) -> None:
"for details.")
lvutil.removeVG(self.dconf['device'], self.vgname)
+ lvutil.LVM_METADATA_BACKUP.remove_vg_backups(self.vgname)
self._cleanup()
@override
diff --git a/drivers/LinstorSR.py b/drivers/LinstorSR.py
index 4b81e5321..0a3efbbca 100755
--- a/drivers/LinstorSR.py
+++ b/drivers/LinstorSR.py
@@ -824,16 +824,8 @@ def check_sr(self, sr_uuid) -> None:
# Applied only on the Linstor Controller, for reasons -> listed below.
if not LinstorVolumeManager.is_controller():
return
- # Validate and clean previous backups if necessary.
- # -> Needs access to backup files, available only on the Controller.
- LinstorVolumeManager.database_backup_validate_and_prune()
- # check_sr is launched on *all* hosts, but it turns out that
- # we do not want all of them to blindly generate concurrencing backups.
- # Hence we must choose one, either one is good, but there must be only one.
- # Apply throttling: only backup if last one is >1h old.
- # -> Needs access to backup files, available only on the Controller.
- if LinstorVolumeManager.get_database_backup_age() > LINSTOR_AUTO_BACKUP_DELAY:
- self.database_backup("auto")
+
+ self.database_backup("auto")
@override
@_locked_load
@@ -1598,6 +1590,19 @@ def database_backup(self, name: Literal["auto", "create", "delete", "snapshot"])
self._reconnect()
try:
assert self._linstor
+
+ if name == "auto":
+ # Validate and clean previous backups if necessary.
+ # -> Needs access to backup files, available only on the Controller.
+ self._linstor.database_backup_validate_and_prune()
+ # check_sr is launched on *all* hosts, but it turns out that
+ # we do not want all of them to blindly generate concurrencing backups.
+ # Hence we must choose one, either one is good, but there must be only one.
+ # Apply throttling: only backup if last one is >1h old.
+ # -> Needs access to backup files, available only on the Controller.
+ if self._linstor.get_database_backup_age() < LINSTOR_AUTO_BACKUP_DELAY:
+ return
+
self._linstor.database_backup(name)
except Exception as e:
util.SMlog(
diff --git a/drivers/backupmanager.py b/drivers/backupmanager.py
new file mode 100644
index 000000000..918023c33
--- /dev/null
+++ b/drivers/backupmanager.py
@@ -0,0 +1,244 @@
+#!/usr/bin/env python3
+#
+# Copyright (C) 2026 Vates SAS
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see .
+
+import itertools
+from abc import ABC, abstractmethod
+from datetime import datetime
+from enum import IntEnum
+from pathlib import Path
+
+import util
+
+from sm_typing import Generator, Generic, List, Optional, Tuple, TypeVar, Union
+
+T = TypeVar("T")
+
+BackupItem = Tuple[Path, datetime]
+
+# ==============================================================================
+
+class BackupManager(ABC, Generic[T]):
+ def __init__(
+ self,
+ *,
+ files_to_keep: int,
+ enabled: bool = True,
+ remove_expired_on_backup: bool = True
+ ) -> None:
+ self._files_to_keep = files_to_keep
+ self._enabled = enabled
+ self._remove_expired_on_backup = remove_expired_on_backup
+
+ @property
+ def _type_name(self) -> str:
+ return type(self).__name__
+
+ # ----------------------------------
+ # Backup file naming.
+ # ----------------------------------
+
+ @abstractmethod
+ def _backup_dir_path(self, ctx: T) -> Path:
+ """
+ Get the path to the directory where to create backups.
+
+ :param ctx: Implementation-specific context.
+ :return: The path to the backup directory.
+ """
+
+ @abstractmethod
+ def _format_backup_file_name(self, ctx: T, counter: Optional[int]) -> str:
+ """
+ Format the name of a new backup file.
+
+ :param ctx: Implementation-specific context.
+ :param counter: Optional counter to distinguish between
+ identically-named backup files.
+ :return: The name of the new backup file.
+ """
+
+ def _next_backup_file_path(self, ctx: T) -> Path:
+ """
+ Get the path to a new backup file.
+
+ :param ctx: Implementation-specific context.
+ :return: The path to the new backup file.
+ """
+ dst_dir = self._backup_dir_path(ctx)
+ file_path = dst_dir / self._format_backup_file_name(ctx, None)
+ counter = 1
+
+ while file_path.exists():
+ file_path = dst_dir / self._format_backup_file_name(ctx, counter)
+ counter += 1
+
+ return file_path
+
+ # ----------------------------------
+ # Backup file listing.
+ # ----------------------------------
+
+ @abstractmethod
+ def _backup_files(self, ctx: T) -> Generator[BackupItem, None, None]:
+ """
+ Get a generator to the visible backup files in the backup directory.
+
+ Each backup item is a tuple of the backup file path and its creation
+ date.
+
+ :param ctx: Implementation-specific context.
+ :return: The generator to the backup file items.
+ """
+
+ def _sorted_backup_files(
+ self,
+ ctx: T,
+ *,
+ reverse: bool = False
+ ) -> List[BackupItem]:
+ """
+ Get a list of backup files sorted by creation date.
+
+ :param ctx: Implementation-specific context.
+ :param reverse: Whether to reverse sorting.
+ :return: The list of sorted backup file items.
+ """
+ return sorted(self._backup_files(ctx),
+ reverse=reverse,
+ key=lambda p: p[1])
+
+ def _latest_backup_file(self, ctx: T) -> Union[BackupItem, Tuple[None, datetime]]:
+ """
+ Get the latest backup file, and its creation date.
+
+ Returns (None, timestamp(0)) when none are found.
+
+ :param ctx: Implementation-specific context.
+ :return: The latest backup file item.
+ """
+ return max(self._backup_files(ctx),
+ default=(None, datetime.fromtimestamp(0)),
+ key=lambda p: p[1])
+
+ def backup_age(self, ctx: T) -> float:
+ """
+ Get the latest backup age in seconds.
+
+ :param ctx: Implementation-specific context.
+ :return: The latest backup age in seconds.
+ """
+ return (datetime.now() - self._latest_backup_file(ctx)[1]).total_seconds()
+
+ # ----------------------------------
+ # Retention ops.
+ # ----------------------------------
+
+ class FileRetentionCheckResult(IntEnum):
+ # Perform the common checks on the file.
+ CHECK = 0
+ # Forcefully retain the file.
+ RETAIN = 1
+ # Forcefully remove the file.
+ REMOVE = 2
+
+ def _check_file_for_retention(self, _file_path: Path) -> FileRetentionCheckResult:
+ """
+ Check a backup file for a custom per-file retention policy.
+
+ :param _file_path: Path to the backup file to check.
+ :return: The result of the check.
+ """
+ return self.FileRetentionCheckResult.CHECK
+
+ def remove_expired(self, ctx: T) -> None:
+ """
+ Remove backup files considered expired based on the retention policy.
+
+ :param ctx: Implementation-specific context.
+ """
+ backup_files = self._sorted_backup_files(ctx)
+ older_files = backup_files[:-self._files_to_keep]
+ newer_files = backup_files[-self._files_to_keep:]
+
+ # These files are to be removed by default, unless forcefully retained.
+ older_files = [
+ file for file in older_files
+ if self._check_file_for_retention(file[0]) != self.FileRetentionCheckResult.RETAIN
+ ]
+
+ # These files are to be retained by default, unless forcefully removed.
+ newer_files = [
+ file for file in newer_files
+ if self._check_file_for_retention(file[0]) == self.FileRetentionCheckResult.REMOVE
+ ]
+
+ # If there are forcefully removed files, we need to keep some of the
+ # older files to reach the requested amount of files to keep.
+ if newer_files:
+ older_files = older_files[:-len(newer_files)]
+
+ for file_path, _ in itertools.chain(older_files, newer_files):
+ try:
+ file_path.unlink()
+ except OSError as e:
+ util.SMlog(
+ f"Unable to unlink backup file of type `{self._type_name}` at `{file_path}`: {e}",
+ priority=util.LOG_ERR
+ )
+
+ # ----------------------------------
+ # Backup ops.
+ # ----------------------------------
+
+ @abstractmethod
+ def _do_backup(self, ctx: T) -> Path:
+ """
+ Create a backup.
+
+ :param ctx: Implementation-specific context.
+ :return: The path to the new backup file.
+ """
+
+ def try_backup(self, ctx: T) -> bool:
+ """
+ Tries to create a backup.
+
+ This is a best-effort operation: if it fails, no exception is raised and
+ it is up to the caller to react according to the return value.
+
+ This also removes expired backup files if the retention policy allows it.
+
+ :param ctx: Implementation-specific context.
+ :return: True if a new backup was created, False otherwise.
+ """
+ if not self._enabled:
+ return False
+
+ try:
+ backup_file_path = self._do_backup(ctx)
+
+ if self._remove_expired_on_backup:
+ self.remove_expired(ctx)
+ except Exception as e:
+ util.SMlog(
+ f"Failed to create backup ({self._type_name} with context `{ctx}`): {e}",
+ priority=util.LOG_ERR
+ )
+
+ return False
+
+ util.SMlog(f"Backup of type {self._type_name} created in `{backup_file_path}`")
+ return True
diff --git a/drivers/linstorvolumemanager.py b/drivers/linstorvolumemanager.py
index 9ab3b0af8..f9ba279e3 100755
--- a/drivers/linstorvolumemanager.py
+++ b/drivers/linstorvolumemanager.py
@@ -18,7 +18,9 @@
from sm_typing import (
Any,
Dict,
+ Generator,
List,
+ Optional,
cast,
override,
)
@@ -33,9 +35,10 @@
import time
import util
import uuid
+from backupmanager import BackupItem, BackupManager
from datetime import datetime
+from enum import IntEnum
from pathlib import Path
-import contextlib
import zipfile
# Persistent prefix to add to RAW persistent volumes.
@@ -47,7 +50,8 @@
DATABASE_PATH = '/var/lib/linstor'
DATABASE_MKFS = 'mkfs.ext4'
DATABASE_BACKUP_DIR_MAIN = Path(DATABASE_PATH)
-DATABASE_BACKUP_DIR_SPARE = Path("/var/lib/linstor.d/db-backups")
+DATABASE_BACKUP_DIR_SPARE_RELATIVE = Path("../linstor.d/db-backups/")
+DATABASE_BACKUP_DIR_SPARE = (DATABASE_BACKUP_DIR_MAIN / DATABASE_BACKUP_DIR_SPARE_RELATIVE).resolve()
DATABASE_BACKUP_NAME_FORMAT = "linstor_database_backup-{}-{}"
DATABASE_BACKUP_RETENTION = 10
DATABASE_BACKUP_DATE_FORMAT = "%Y%m%d_%H%M%S"
@@ -258,9 +262,154 @@ def __init__(self, message, code=ERR_GENERIC):
def code(self):
return self._code
+
+# ==============================================================================
+
class LinstorDatabaseBackupError(Exception):
pass
+
+class LinstorDatabaseBackupContext:
+ class Location(IntEnum):
+ MAIN = 0
+ SPARE = 1
+
+ def __init__(
+ self,
+ location: "Location",
+ *,
+ name: str = ""
+ ) -> None:
+ self.location = location
+ self.name = name
+
+ @override
+ def __repr__(self) -> str:
+ return f"{type(self).__name__}({vars(self)})"
+
+
+class LinstorDatabaseBackup(BackupManager[LinstorDatabaseBackupContext]):
+ def __init__(self, linstor: "linstor.Linstor") -> None:
+ super().__init__(
+ files_to_keep=DATABASE_BACKUP_RETENTION,
+ remove_expired_on_backup=False
+ )
+
+ self._linstor = linstor
+
+ # ----------------------------------
+ # Backup file naming.
+ # ----------------------------------
+
+ @override
+ def _backup_dir_path(self, ctx: LinstorDatabaseBackupContext) -> Path:
+ if ctx.location == LinstorDatabaseBackupContext.Location.SPARE:
+ return DATABASE_BACKUP_DIR_SPARE
+ else:
+ return DATABASE_BACKUP_DIR_MAIN
+
+ @override
+ def _format_backup_file_name(
+ self,
+ ctx: LinstorDatabaseBackupContext,
+ counter: Optional[int]
+ ) -> str:
+ date = datetime.now().strftime(DATABASE_BACKUP_DATE_FORMAT)
+ filename = DATABASE_BACKUP_NAME_FORMAT.format(date, ctx.name)
+
+ if counter is not None:
+ return f"{filename}-{counter}.zip"
+
+ return f"{filename}.zip"
+
+ # ----------------------------------
+ # Backup file listing.
+ # ----------------------------------
+
+ @override
+ def _backup_files(
+ self,
+ ctx: LinstorDatabaseBackupContext
+ ) -> Generator[BackupItem, None, None]:
+ """
+ List all visible backup files in backup_dir_path.
+ DATABASE_BACKUP_DIR_MAIN is only available on the Linstor Controller.
+ DATABASE_BACKUP_DIR_SPARE will list backups previously made when the Host was the Linstor Controller.
+ This may not be useful information if it is not the Controller anymore.
+
+ :param ctx: Implementation-specific context.
+ :return: The generator to the backup file items.
+ """
+ backup_dir_path = self._backup_dir_path(ctx)
+
+ for path in backup_dir_path.glob(DATABASE_BACKUP_NAME_FORMAT.format(
+ "[0-9]" * 8 + "_" + "[0-9]" * 6, "*") + ".zip"):
+ try:
+ yield path, datetime.strptime(path.name.split("-")[1], DATABASE_BACKUP_DATE_FORMAT)
+ except (ValueError, IndexError):
+ continue
+
+ # ----------------------------------
+ # Retention ops.
+ # ----------------------------------
+
+ @classmethod
+ def _check_database_backup(cls, database_backup_path: Path) -> None:
+ """
+ Make some validation of a database backup zip-file.
+ Check its a valid zipfile, and CRC-test its content.
+ Check it contains a non-empty linstordb.mv.db file.
+ Always raises a LinstorDatabaseBackupError if checks failed.
+ """
+ try:
+ with zipfile.ZipFile(database_backup_path, mode="r") as archive:
+ if archive.testzip() is not None:
+ raise LinstorDatabaseBackupError("zip archive CRC failed")
+ linstordb = next((
+ f
+ for f in archive.filelist
+ if f.filename == "linstordb.mv.db"
+ ), None)
+ if not linstordb:
+ raise LinstorDatabaseBackupError("cannot find linstordb.mv.db")
+ if linstordb.file_size == 0:
+ raise LinstorDatabaseBackupError("linstordb.mv.db is empty")
+ except (FileNotFoundError, zipfile.BadZipFile, zipfile.LargeZipFile) as e:
+ raise LinstorDatabaseBackupError(e) from e
+
+ @override
+ def _check_file_for_retention(
+ self,
+ file_path: Path
+ ) -> BackupManager.FileRetentionCheckResult:
+ try:
+ self._check_database_backup(file_path)
+ except LinstorDatabaseBackupError as error:
+ util.SMlog(f"[database_backup] Check failed `{error}` [{file_path}]",
+ priority=util.LOG_ERR)
+
+ return self.FileRetentionCheckResult.REMOVE
+
+ return self.FileRetentionCheckResult.CHECK
+
+ # ----------------------------------
+ # Backup ops.
+ # ----------------------------------
+
+ @override
+ def _do_backup(self, ctx: LinstorDatabaseBackupContext) -> Path:
+ backup_file_path = self._next_backup_file_path(ctx)
+
+ if ctx.location == LinstorDatabaseBackupContext.Location.SPARE:
+ # Relative path are ok for a secondary backup filename:
+ # https://github.com/LINBIT/linstor-server/blob/3e9306a9d8215606544c64c50ced150625ee4926/controller/src/main/java/com/linbit/linstor/api/rest/v1/Controller.java#L408
+ self._linstor.controller_backupdb(str(DATABASE_BACKUP_DIR_SPARE_RELATIVE / backup_file_path.stem))
+ else:
+ self._linstor.controller_backupdb(backup_file_path.stem)
+
+ return backup_file_path
+
+
# ==============================================================================
# Note:
@@ -278,7 +427,7 @@ class LinstorVolumeManager(object):
'_linstor', '_uri', '_logger', '_redundancy',
'_base_group_name', '_group_name', '_ha_group_name',
'_volumes', '_storage_pools', '_storage_pools_time',
- '_kv_cache', '_resource_cache', '_volume_info_cache',
+ '_db_backup', '_kv_cache', '_resource_cache', '_volume_info_cache',
'_kv_cache_dirty', '_resource_cache_dirty', '_volume_info_cache_dirty',
'_resources_info_cache',
)
@@ -410,6 +559,7 @@ def __init__(
self._ha_group_name = self._build_ha_group_name(self._base_group_name)
self._volumes = set()
self._storage_pools_time = 0
+ self._db_backup = LinstorDatabaseBackup(self._linstor)
# To increase performance and limit request count to LINSTOR services,
# we use caches.
@@ -1409,6 +1559,7 @@ def destroy(self):
self._linstor = self._create_linstor_instance(
uri, keep_uri_unmodified=True
)
+ self._db_backup = LinstorDatabaseBackup(self._linstor)
# 4.3. Destroy database volume.
self._destroy_resource(DATABASE_VOLUME_NAME)
@@ -1793,46 +1944,38 @@ def get_database_path(self):
def is_controller(cls):
return cls._is_mounted(DATABASE_PATH)
- @classmethod
- def get_database_backup_age(cls):
+ def get_database_backup_age(self):
"""
Return the latest backup age in seconds.
If not called on the Controller, since backups are not available,
returns a huge value (a timestamp of now).
"""
- return (datetime.now() - cls._get_latest_database_backup()[1]).total_seconds()
+ return self._db_backup.backup_age(
+ LinstorDatabaseBackupContext(LinstorDatabaseBackupContext.Location.MAIN)
+ )
def database_backup(self, name=""):
- # Create new backup
- date = datetime.now().strftime(DATABASE_BACKUP_DATE_FORMAT)
- filename = DATABASE_BACKUP_NAME_FORMAT.format(date, name)
- self._linstor.controller_backupdb(filename)
- # Relative path are ok for a secondary backup filename:
- # https://github.com/LINBIT/linstor-server/blob/3e9306a9d8215606544c64c50ced150625ee4926/controller/src/main/java/com/linbit/linstor/api/rest/v1/Controller.java#L408
- self._linstor.controller_backupdb(f"../linstor.d/db-backups/{filename}")
- util.SMlog(f"[database_backup] Created: {filename}", priority=util.LOG_INFO)
+ self._db_backup.try_backup(
+ LinstorDatabaseBackupContext(LinstorDatabaseBackupContext.Location.MAIN, name=name)
+ )
- @classmethod
- def database_backup_validate_and_prune(cls):
+ self._db_backup.try_backup(
+ LinstorDatabaseBackupContext(LinstorDatabaseBackupContext.Location.SPARE, name=name)
+ )
+
+ def database_backup_validate_and_prune(self):
"""
Removes old backup based on two criterias:
- Validity of the zipfile done by self._check_database_backup.
- Number of valid files found, only the nth latest are kept.
"""
- for directory in (DATABASE_BACKUP_DIR_MAIN, DATABASE_BACKUP_DIR_SPARE):
- valid_backup_count = 0
- # Validate file and apply retention
- for database_backup_path, _ in cls._get_sorted_database_backup(directory):
- try:
- cls._check_database_backup(database_backup_path)
- valid_backup_count += 1
- if valid_backup_count < DATABASE_BACKUP_RETENTION:
- continue
- except LinstorDatabaseBackupError as error:
- util.SMlog(f"[database_backup] Check failed `{error}` [{database_backup_path}]",
- priority=util.LOG_ERR)
- with contextlib.suppress(OSError):
- os.unlink(database_backup_path)
+ self._db_backup.remove_expired(
+ LinstorDatabaseBackupContext(LinstorDatabaseBackupContext.Location.MAIN)
+ )
+
+ self._db_backup.remove_expired(
+ LinstorDatabaseBackupContext(LinstorDatabaseBackupContext.Location.SPARE)
+ )
@classmethod
def get_all_group_names(cls, base_name):
@@ -2711,67 +2854,6 @@ def _get_volume_properties(self, volume_uuid):
properties.namespace = self._build_volume_namespace(volume_uuid)
return properties
- @classmethod
- def _list_database_backups(cls, database_backup_dir):
- """
- List all visible backup files in database_backup_dir.
- DATABASE_BACKUP_DIR_MAIN is only available on the Linstor Controller.
- DATABASE_BACKUP_DIR_SPARE will list backups previously made when the Host was the Linstor Controller.
- This may not be useful information if it is not the Controller anymore.
- """
- for path in database_backup_dir.glob(DATABASE_BACKUP_NAME_FORMAT.format(
- "[0-9]" * 8 + "_" + "[0-9]" * 6, "*") + ".zip"):
- try:
- yield path, datetime.strptime(path.name.split("-")[1], DATABASE_BACKUP_DATE_FORMAT)
- except (ValueError, IndexError):
- continue
-
- @classmethod
- def _get_sorted_database_backup(cls, database_backup_dir):
- """
- Return list of backups in database_backup_dir, alongside their creation date.
- Sorted by date from the more recent to the older one.
- """
- return sorted(cls._list_database_backups(database_backup_dir),
- reverse=True,
- key=lambda p: p[1])
-
- @classmethod
- def _get_latest_database_backup(cls):
- """
- Return the latest backup in DATABASE_BACKUP_DIR_MAIN, and its creation date.
- Returns (None, timestamp(0)) when none are found.
- None will be found if it is not called on the Linstor Controller.
- (cf _list_database_backups)
- """
- return max(cls._list_database_backups(DATABASE_BACKUP_DIR_MAIN),
- default=(None, datetime.fromtimestamp(0)),
- key=lambda p: p[1])
-
- @classmethod
- def _check_database_backup(cls, database_backup_path):
- """
- Make some validation of a database backup zip-file.
- Check its a valid zipfile, and CRC-test its content.
- Check it contains a non-empty linstordb.mv.db file.
- Always raises a LinstorDatabaseBackupError if checks failed.
- """
- try:
- with zipfile.ZipFile(database_backup_path, mode="r") as archive:
- if archive.testzip() is not None:
- raise LinstorDatabaseBackupError("zip archive CRC failed")
- linstordb = next((
- f
- for f in archive.filelist
- if f.filename == "linstordb.mv.db"
- ), None)
- if not linstordb:
- raise LinstorDatabaseBackupError("cannot find linstordb.mv.db")
- if linstordb.file_size == 0:
- raise LinstorDatabaseBackupError("linstordb.mv.db is empty")
- except (FileNotFoundError, zipfile.BadZipFile, zipfile.LargeZipFile) as e:
- raise LinstorDatabaseBackupError(e) from e
-
@classmethod
def _build_sr_namespace(cls):
return '/{}/'.format(cls.NAMESPACE_SR)
diff --git a/drivers/lvmbackup.py b/drivers/lvmbackup.py
new file mode 100644
index 000000000..bfe2e14c4
--- /dev/null
+++ b/drivers/lvmbackup.py
@@ -0,0 +1,117 @@
+#!/usr/bin/env python3
+#
+# Copyright (C) 2026 Vates SAS
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see .
+
+import shutil
+import time
+from datetime import datetime
+from pathlib import Path
+
+import util
+from backupmanager import BackupItem, BackupManager
+
+from sm_typing import Final, Generator, Optional, override
+
+# ==============================================================================
+
+# Global parameters for LVM metadata backups.
+ENABLED: Final = True
+FILES_TO_KEEP: Final = 30
+
+BACKUP_FILE_NAME_DATE_FORMAT: Final = "%Y%m%d_%H%M%S"
+
+# Paths to the directories where the files to backup are (managed by LVM),
+# and where to copy them (managed by the SM).
+SRC_PATH: Final = Path("/etc/lvm/backup")
+DST_PATH: Final = Path("/etc/sm/lvm-backups")
+
+# ------------------------------------------------------------------------------
+
+class LVMMetadataBackup(BackupManager[str]):
+ def __init__(self) -> None:
+ super().__init__(files_to_keep=FILES_TO_KEEP, enabled=ENABLED)
+
+ # ----------------------------------
+ # Backup file naming.
+ # ----------------------------------
+
+ @override
+ def _backup_dir_path(self, ctx: str) -> Path:
+ return DST_PATH / ctx
+
+ @override
+ def _format_backup_file_name(self, _ctx: str, counter: Optional[int]) -> str:
+ formatted_time = time.strftime(BACKUP_FILE_NAME_DATE_FORMAT)
+
+ if counter is not None:
+ return f"{formatted_time}.{counter}.vg"
+
+ return f"{formatted_time}.vg"
+
+ # ----------------------------------
+ # Backup file listing.
+ # ----------------------------------
+
+ @override
+ def _backup_files(self, ctx: str) -> Generator[BackupItem, None, None]:
+ backup_dir_path = self._backup_dir_path(ctx)
+
+ try:
+ for file_path in backup_dir_path.iterdir():
+ try:
+ yield file_path, datetime.fromtimestamp(file_path.stat().st_mtime)
+ except OSError as e:
+ util.SMlog(
+ f"Unable to stat LVM metadata backup file `{file_path}`: {e}",
+ priority=util.LOG_ERR
+ )
+
+ continue
+ except OSError as e:
+ util.SMlog(
+ f"Unable to get files of LVM metadata backup directory `{backup_dir_path}`: {e}",
+ priority=util.LOG_ERR
+ )
+
+ # ----------------------------------
+ # Backup ops.
+ # ----------------------------------
+
+ @override
+ def _do_backup(self, ctx: str) -> Path:
+ dst_dir = self._backup_dir_path(ctx)
+ dst_dir.mkdir(parents=True, exist_ok=True)
+
+ source_file_path = SRC_PATH / ctx
+ backup_file_path = self._next_backup_file_path(ctx)
+
+ shutil.copyfile(source_file_path, backup_file_path)
+ return backup_file_path
+
+ def remove_vg_backups(self, vgname: str) -> None:
+ """
+ Remove the metadata backup files of a LVM volume group.
+
+ This is a best-effort operation: if it fails, no exception is raised.
+
+ :param str vgname: The name of the volume group to remove the backup files of.
+ """
+ try:
+ shutil.rmtree(self._backup_dir_path(vgname))
+ except Exception as e:
+ util.SMlog(
+ f"Failed to remove LVM metadata backups of `{vgname}`: {e}",
+ priority=util.LOG_ERR
+ )
diff --git a/drivers/lvutil.py b/drivers/lvutil.py
index e9c23c60e..a05adde80 100755
--- a/drivers/lvutil.py
+++ b/drivers/lvutil.py
@@ -22,6 +22,8 @@
import errno
import time
+from sm_typing import List
+
import scsiutil
from fairlock import Fairlock
import util
@@ -30,6 +32,7 @@
from constants import EXT_PREFIX, VG_LOCATION, VG_PREFIX
import lvmcache
import srmetadata
+from lvmbackup import LVMMetadataBackup
MDVOLUME_NAME = 'MGT'
VDI_UUID_TAG_PREFIX = 'vdi_'
@@ -72,8 +75,16 @@
LVM_COMMANDS = VG_COMMANDS.union(PV_COMMANDS, LV_COMMANDS, DM_COMMANDS)
+BACKUP_LVM_COMMANDS = frozenset({CMD_VGREMOVE, CMD_VGCHANGE, CMD_VGEXTEND,
+ CMD_LVCREATE, CMD_LVREMOVE, CMD_LVCHANGE,
+ CMD_LVRENAME, CMD_LVRESIZE, CMD_LVEXTEND})
+BACKUP_VCHANGE_IGNORED_OPTIONS = frozenset({"-ay", "-an", "--refresh",
+ "--config"})
+
LVM_LOCK = 'lvm'
+LVM_METADATA_BACKUP = LVMMetadataBackup()
+
def extract_vgname(str_in):
"""Search for and return a VG name
@@ -142,6 +153,36 @@ def decorated(*args, **kwargs):
return decorated
+def _try_backup_vg(lvm_cmd: str, lvm_args: List[str]) -> None:
+ """
+ Check whether the LVM command and arguments makes modifications to the
+ metadata of the target volume group. If so, the metadata of this volume
+ group is backed up.
+
+ :param lvm_cmd: The LVM command.
+ :param lvm_args: The arguments for the LVM command.
+ """
+ if lvm_cmd not in BACKUP_LVM_COMMANDS:
+ return
+
+ options = [arg for arg in lvm_args if arg.startswith("-")]
+ if lvm_cmd in [CMD_LVCHANGE, CMD_VGCHANGE] and \
+ all(opt in BACKUP_VCHANGE_IGNORED_OPTIONS for opt in options):
+ return
+
+ vgname = None
+
+ for arg in lvm_args:
+ vgname = extract_vgname(arg)
+ if vgname:
+ break
+
+ if not vgname:
+ return
+
+ LVM_METADATA_BACKUP.try_backup(vgname)
+
+
def cmd_lvm(cmd, pread_func=util.pread2, *args):
""" Construct and run the appropriate lvm command.
@@ -184,6 +225,8 @@ def cmd_lvm(cmd, pread_func=util.pread2, *args):
util.SMlog("CMD_LVM: Not all lvm arguments are of type 'str'")
return None
+ _try_backup_vg(lvm_cmd, lvm_args)
+
with Fairlock("devicemapper"):
start_time = time.time()
stdout = pread_func([os.path.join(LVM_BIN, lvm_cmd)] + lvm_args, * args)
diff --git a/mocks/linstor/__init__.py b/mocks/linstor/__init__.py
index 0f0f7c9c5..63ef7067e 100644
--- a/mocks/linstor/__init__.py
+++ b/mocks/linstor/__init__.py
@@ -1,2 +1,2 @@
class Linstor(object):
- pass
+ def controller_backupdb(self, backup_name: str): ...
diff --git a/tests/test_backupmanager.py b/tests/test_backupmanager.py
new file mode 100644
index 000000000..0ae1b0472
--- /dev/null
+++ b/tests/test_backupmanager.py
@@ -0,0 +1,301 @@
+#!/usr/bin/env python3
+#
+# Copyright (C) 2026 Vates SAS
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see .
+
+from sm_typing import override
+
+from datetime import datetime
+import unittest
+import unittest.mock as mock
+from pathlib import Path
+
+from backupmanager import BackupManager
+
+# ==============================================================================
+
+FILES_TO_KEEP = 5
+DST_PATH = Path("/tmp/backupmanager")
+
+# ------------------------------------------------------------------------------
+
+class MockBackup(BackupManager):
+ def __init__(self):
+ super().__init__(files_to_keep=FILES_TO_KEEP)
+
+ self._backup_file_list = [
+ (DST_PATH / str(i), datetime.fromtimestamp(i))
+ for i in range(FILES_TO_KEEP * 2)
+ ]
+
+ self._backup_file_list.insert(
+ 0,
+ (DST_PATH / f"retain.{0}", datetime.fromtimestamp(0))
+ )
+ self._backup_file_list.insert(
+ 0,
+ (DST_PATH / f"remove.{0}", datetime.fromtimestamp(0))
+ )
+
+ last_file_idx = FILES_TO_KEEP * 2 - 1
+ self._backup_file_list.append(
+ (DST_PATH / f"retain.{last_file_idx}",
+ datetime.fromtimestamp(last_file_idx))
+ )
+ self._backup_file_list.append(
+ (DST_PATH / f"remove.{last_file_idx}",
+ datetime.fromtimestamp(last_file_idx))
+ )
+
+ self._custom_file_retention_check_enabled = False
+
+ # ----------------------------------
+ # Backup file naming.
+ # ----------------------------------
+
+ @override
+ def _backup_dir_path(self, _ctx):
+ return DST_PATH
+
+ @override
+ def _format_backup_file_name(self, ctx, counter):
+ if counter is not None:
+ return f"{ctx}.{counter}"
+
+ return ctx
+
+ # ----------------------------------
+ # Backup file listing.
+ # ----------------------------------
+
+ @override
+ def _backup_files(self, _ctx):
+ return iter(self._backup_file_list)
+
+ # ----------------------------------
+ # Retention ops.
+ # ----------------------------------
+
+ @override
+ def _check_file_for_retention(self, file_path):
+ if not self._custom_file_retention_check_enabled:
+ return super()._check_file_for_retention(file_path)
+
+ if file_path.name.startswith("retain"):
+ return BackupManager.FileRetentionCheckResult.RETAIN
+ elif file_path.name.startswith("remove"):
+ return BackupManager.FileRetentionCheckResult.REMOVE
+
+ return BackupManager.FileRetentionCheckResult.CHECK
+
+ # ----------------------------------
+ # Backup ops.
+ # ----------------------------------
+
+ @override
+ def _do_backup(self, ctx):
+ latest_backup_file = self._latest_backup_file(ctx)
+ backup_time = int(latest_backup_file[1].timestamp()) + 1
+ backup_file_path = DST_PATH / str(backup_time)
+
+ self._backup_file_list.append(
+ (backup_file_path, datetime.fromtimestamp(backup_time))
+ )
+
+ return backup_file_path
+
+ # ----------------------------------
+ # Test knobs.
+ # ----------------------------------
+
+ @property
+ def backup_file_list(self):
+ return self._backup_file_list
+
+ def add_unsorted_backup_files(self):
+ self._backup_file_list.insert(
+ 0,
+ (DST_PATH / str(FILES_TO_KEEP), datetime.fromtimestamp(FILES_TO_KEEP))
+ )
+
+ self._backup_file_list.append(
+ (DST_PATH / "0", datetime.fromtimestamp(0))
+ )
+
+ def remove_all_backups(self):
+ self._backup_file_list.clear()
+
+ def remove_backup(self, path):
+ self._backup_file_list = [
+ file for file in self.backup_file_list
+ if file[0] != path
+ ]
+
+ def enable_custom_file_retention_check(self):
+ self._custom_file_retention_check_enabled = True
+
+ def disable_backups(self):
+ self._enabled = False
+
+ def disable_remove_expired_on_backup(self):
+ self._remove_expired_on_backup = False
+
+# ------------------------------------------------------------------------------
+
+class TestBackupManager(unittest.TestCase):
+ @override
+ def setUp(self):
+ self.backup = MockBackup()
+ self.backup_name = "GLaDOS_first_boot"
+
+ def test_next_backup_file_path(self):
+ self.assertEqual(
+ self.backup._next_backup_file_path(self.backup_name),
+ DST_PATH / self.backup_name
+ )
+
+ @mock.patch("pathlib.Path.exists")
+ def test_next_backup_file_path_with_counter(self, mock_path_exists):
+ mock_path_exists.side_effect = [
+ False,
+ True, False,
+ True, True, False,
+ ]
+
+ self.assertEqual(
+ self.backup._next_backup_file_path(self.backup_name),
+ DST_PATH / self.backup_name
+ )
+ self.assertEqual(
+ self.backup._next_backup_file_path(self.backup_name),
+ DST_PATH / f"{self.backup_name}.1"
+ )
+ self.assertEqual(
+ self.backup._next_backup_file_path(self.backup_name),
+ DST_PATH / f"{self.backup_name}.2"
+ )
+
+ def test_sorted_backup_files(self):
+ self.backup.add_unsorted_backup_files()
+
+ backup_files = self.backup._sorted_backup_files(self.backup_name)
+ self.assertEqual(backup_files[0][1], datetime.fromtimestamp(0))
+ self.assertEqual(backup_files[-1][1], datetime.fromtimestamp(9))
+
+ def test_sorted_backup_files_reversed(self):
+ self.backup.add_unsorted_backup_files()
+
+ backup_files = self.backup._sorted_backup_files(self.backup_name, reverse=True)
+ self.assertEqual(backup_files[0][1], datetime.fromtimestamp(9))
+ self.assertEqual(backup_files[-1][1], datetime.fromtimestamp(0))
+
+ def test_latest_backup_file(self):
+ self.backup.add_unsorted_backup_files()
+
+ backup_file = self.backup._latest_backup_file(self.backup_name)
+ self.assertEqual(backup_file[1], datetime.fromtimestamp(9))
+
+ def test_latest_backup_file_empty(self):
+ self.backup.remove_all_backups()
+
+ backup_file = self.backup._latest_backup_file(self.backup_name)
+ self.assertEqual(backup_file[0], None)
+
+ @mock.patch("backupmanager.datetime", autospec=True)
+ def test_backup_age(self, mock_datetime):
+ base_time = datetime.now()
+ mock_datetime.now.return_value = base_time
+
+ self.backup.add_unsorted_backup_files()
+
+ backup_age = self.backup.backup_age(self.backup_name)
+ self.assertEqual(
+ backup_age,
+ (base_time - datetime.fromtimestamp(9)).total_seconds()
+ )
+
+ def test_default_check_file_for_retention(self):
+ self.assertEqual(
+ self.backup._check_file_for_retention(DST_PATH / "0"),
+ BackupManager.FileRetentionCheckResult.CHECK
+ )
+
+ def test_remove_expired(self):
+ def side_effect_unlink(path):
+ self.backup.remove_backup(path)
+
+ with mock.patch("pathlib.Path.unlink", side_effect=side_effect_unlink, autospec=True):
+ self.backup.remove_expired(self.backup_name)
+
+ expected_files = [
+ (DST_PATH / "7", datetime.fromtimestamp(7)),
+ (DST_PATH / "8", datetime.fromtimestamp(8)),
+ (DST_PATH / "9", datetime.fromtimestamp(9)),
+ (DST_PATH / "retain.9", datetime.fromtimestamp(9)),
+ (DST_PATH / "remove.9", datetime.fromtimestamp(9)),
+ ]
+
+ self.assertListEqual(self.backup.backup_file_list, expected_files)
+
+ def test_remove_expired_with_custom_retention(self):
+ self.backup.enable_custom_file_retention_check()
+
+ def side_effect_unlink(path):
+ self.backup.remove_backup(path)
+
+ with mock.patch("pathlib.Path.unlink", side_effect=side_effect_unlink, autospec=True):
+ self.backup.remove_expired(self.backup_name)
+
+ expected_files = [
+ (DST_PATH / "retain.0", datetime.fromtimestamp(0)),
+ (DST_PATH / "6", datetime.fromtimestamp(6)),
+ (DST_PATH / "7", datetime.fromtimestamp(7)),
+ (DST_PATH / "8", datetime.fromtimestamp(8)),
+ (DST_PATH / "9", datetime.fromtimestamp(9)),
+ (DST_PATH / "retain.9", datetime.fromtimestamp(9)),
+ ]
+
+ self.assertListEqual(self.backup.backup_file_list, expected_files)
+
+ def test_do_backup(self):
+ self.assertEqual(
+ self.backup._do_backup(self.backup_name),
+ DST_PATH / "10"
+ )
+
+ @mock.patch.object(MockBackup, "_do_backup")
+ @mock.patch.object(MockBackup, "remove_expired")
+ def test_try_backup(self, mock_remove_expired, mock_do_backup):
+ self.assertTrue(self.backup.try_backup(self.backup_name))
+
+ mock_do_backup.assert_called_once()
+ mock_remove_expired.assert_called_once()
+
+ @mock.patch.object(MockBackup, "_do_backup")
+ @mock.patch.object(MockBackup, "remove_expired")
+ def test_try_backup_without_remove_expired(self, mock_remove_expired, mock_do_backup):
+ self.backup.disable_remove_expired_on_backup()
+ self.assertTrue(self.backup.try_backup(self.backup_name))
+
+ mock_do_backup.assert_called_once()
+ mock_remove_expired.assert_not_called()
+
+ @mock.patch.object(MockBackup, "_do_backup")
+ def test_try_backup_failure(self, mock_do_backup):
+ mock_do_backup.side_effect = OSError()
+ self.assertFalse(self.backup.try_backup(self.backup_name))
+
+ def test_try_backup_disabled(self):
+ self.backup.disable_backups()
+ self.assertFalse(self.backup.try_backup(self.backup_name))
diff --git a/tests/test_lvmbackup.py b/tests/test_lvmbackup.py
new file mode 100644
index 000000000..d96a999db
--- /dev/null
+++ b/tests/test_lvmbackup.py
@@ -0,0 +1,118 @@
+#!/usr/bin/env python3
+#
+# Copyright (C) 2026 Vates SAS
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see .
+
+from sm_typing import override
+
+import filecmp
+import os
+import tempfile
+import unittest
+import unittest.mock as mock
+from datetime import datetime
+from pathlib import Path
+
+from lvmbackup import LVMMetadataBackup
+
+# ==============================================================================
+
+class TestLvmBackup(unittest.TestCase):
+ MOCKED_FILES_TO_KEEP = 5
+ MOCKED_TIME = "20260908_155410"
+
+ @override
+ def setUp(self):
+ self.test_dir = tempfile.TemporaryDirectory()
+ self.addCleanup(self.test_dir.cleanup)
+
+ self.src_path = Path(self.test_dir.name) / "src"
+ self.dst_path = Path(self.test_dir.name) / "dst"
+ self.src_path.mkdir(parents=True)
+ self.dst_path.mkdir(parents=True)
+
+ mock.patch("lvmbackup.DST_PATH", self.dst_path).start()
+ mock.patch("lvmbackup.SRC_PATH", self.src_path).start()
+ mock.patch("lvmbackup.FILES_TO_KEEP", self.MOCKED_FILES_TO_KEEP).start()
+ self.addCleanup(mock.patch.stopall)
+
+ self.vgname = "VG_XenStorage-8c1e6de6-d35c-755d-82a2-1e7d1c903190"
+ with (self.src_path / self.vgname).open("w") as f:
+ f.write(f"Some metadata for VG {self.vgname}")
+
+ self.backup = LVMMetadataBackup()
+
+ def test_backup_dir_path(self):
+ self.assertEqual(
+ self.backup._backup_dir_path(self.vgname),
+ self.dst_path / self.vgname
+ )
+
+ @mock.patch("time.strftime")
+ def test_format_backup_file_name(self, mock_strftime):
+ mock_strftime.return_value = self.MOCKED_TIME
+
+ self.assertEqual(
+ self.backup._format_backup_file_name(self.vgname, None),
+ f"{self.MOCKED_TIME}.vg"
+ )
+
+ @mock.patch("time.strftime")
+ def test_format_backup_file_name_with_counter(self, mock_strftime):
+ mock_strftime.return_value = self.MOCKED_TIME
+
+ self.assertEqual(
+ self.backup._format_backup_file_name(self.vgname, 2),
+ f"{self.MOCKED_TIME}.2.vg"
+ )
+
+ def test_backup_files(self):
+ backup_file_path = self.backup._next_backup_file_path(self.vgname)
+ backup_file_path.parent.mkdir(parents=True, exist_ok=True)
+
+ with backup_file_path.open("w") as f:
+ f.write(f"Some metadata for VG {self.vgname}")
+
+ os.utime(backup_file_path, (42, 42))
+
+ expected_backup_files = [(backup_file_path, datetime.fromtimestamp(42))]
+
+ backup_files = list(self.backup._backup_files(self.vgname))
+ self.assertListEqual(backup_files, expected_backup_files)
+
+ @mock.patch("lvmbackup.LVMMetadataBackup._next_backup_file_path")
+ def test_do_backup(self, mock_next_backup_file_path):
+ backup_file_path = self.dst_path / self.vgname / "do_backup.vg"
+ mock_next_backup_file_path.return_value = backup_file_path
+
+ self.assertEqual(self.backup._do_backup(self.vgname), backup_file_path)
+ self.assertTrue(backup_file_path.exists())
+ self.assertTrue(filecmp.cmp(self.src_path / self.vgname, backup_file_path, shallow=False))
+
+ @mock.patch("lvmbackup.LVMMetadataBackup._backup_dir_path")
+ def test_remove_vg_backups(self, mock_backup_dir_path):
+ mock_backup_dir_path.return_value = self.dst_path
+
+ self.assertTrue(self.dst_path.parent.is_dir())
+ self.assertTrue(self.dst_path.is_dir())
+
+ self.backup.remove_vg_backups(self.vgname)
+
+ self.assertTrue(self.dst_path.parent.is_dir())
+ self.assertFalse(self.dst_path.is_dir())
+
+ @mock.patch("shutil.rmtree")
+ def test_remove_vg_backups_failure(self, mock_rmtree):
+ mock_rmtree.side_effect = FileNotFoundError()
+ self.backup.remove_vg_backups(self.vgname)
diff --git a/tests/test_lvutil.py b/tests/test_lvutil.py
index b707eb35b..7db52691d 100644
--- a/tests/test_lvutil.py
+++ b/tests/test_lvutil.py
@@ -371,7 +371,7 @@ def test_pvs_in_vg(self, mock_smlog, mock_cmd_lvm):
result = lvutil.getPVsInVG("vg1")
self.assertEqual(result, ["pv1", "pv2"])
mock_smlog.assert_called_once_with("PVs in VG vg1: ['pv1', 'pv2']")
-
+
def test_no_pvs(self, mock_smlog, mock_cmd_lvm):
# Test when no PVs are returned
mock_cmd_lvm.return_value = ""
@@ -418,3 +418,110 @@ def test_check_PV_SCSI_IDs_success(self, mock_get_scsi_id, mock_smlog, mock_cmd_
mock_get_scsi_id.side_effect = ['36001405fdd436fa7f854cd685fc3b1fd']
lvutil.checkPVScsiIds('VG_XenStorage-401d198b-60ab-1f21-1359-bd4f127b8f38',
'36001405fdd436fa7f854cd685fc3b1fd')
+
+
+@mock.patch("lvutil.LVM_METADATA_BACKUP", autospec=True)
+class TestCmdLvmBackup(unittest.TestCase):
+ def test_vgcreate_does_not_back_up(self, mock_lvmbackup):
+ lvutil._try_backup_vg(
+ lvutil.CMD_VGCREATE,
+ ["--metadatasize", "10M", TEST_VG, "/dev/sda1"]
+ )
+
+ mock_lvmbackup.try_backup.assert_not_called()
+
+ def test_vgremove_does_back_up(self, mock_lvmbackup):
+ lvutil._try_backup_vg(lvutil.CMD_VGREMOVE, [TEST_VG])
+ mock_lvmbackup.try_backup.assert_called_once_with(TEST_VG)
+
+ def test_vgextend_does_back_up(self, mock_lvmbackup):
+ lvutil._try_backup_vg(
+ lvutil.CMD_VGEXTEND,
+ [TEST_VG, "/dev/sdb1"]
+ )
+
+ mock_lvmbackup.try_backup.assert_called_once_with(TEST_VG)
+
+ def test_lvcreate_does_back_up(self, mock_lvmbackup):
+ lvutil._try_backup_vg(
+ lvutil.CMD_LVCREATE,
+ ["-n", "vol", "-L", "100", TEST_VG]
+ )
+
+ mock_lvmbackup.try_backup.assert_called_once_with(TEST_VG)
+
+ def test_lvremove_does_back_up(self, mock_lvmbackup):
+ lvutil._try_backup_vg(lvutil.CMD_LVREMOVE, ["-f", TEST_VOL])
+ mock_lvmbackup.try_backup.assert_called_once_with(TEST_VG)
+
+ def test_lvremove_with_config_does_back_up(self, mock_lvmbackup):
+ lvutil._try_backup_vg(
+ lvutil.CMD_LVREMOVE,
+ ["-f", TEST_VOL, "--config", "devices{}"]
+ )
+
+ mock_lvmbackup.try_backup.assert_called_once_with(TEST_VG)
+
+ def test_lvrename_does_back_up(self, mock_lvmbackup):
+ lvutil._try_backup_vg(
+ lvutil.CMD_LVRENAME,
+ [TEST_VOL, "new_name"]
+ )
+
+ mock_lvmbackup.try_backup.assert_called_once_with(TEST_VG)
+
+ def test_lvresize_does_back_up(self, mock_lvmbackup):
+ lvutil._try_backup_vg(
+ lvutil.CMD_LVRESIZE,
+ ["-L", "200", TEST_VOL]
+ )
+
+ mock_lvmbackup.try_backup.assert_called_once_with(TEST_VG)
+
+ def test_lvextend_does_back_up(self, mock_lvmbackup):
+ lvutil._try_backup_vg(
+ lvutil.CMD_LVEXTEND,
+ ["-L", "+100", TEST_VOL]
+ )
+
+ mock_lvmbackup.try_backup.assert_called_once_with(TEST_VG)
+
+ def test_vgchange_activate_does_not_backup(self, mock_lvmbackup):
+ lvutil._try_backup_vg(lvutil.CMD_VGCHANGE, ["-an", TEST_VG])
+ mock_lvmbackup.try_backup.assert_not_called()
+
+ def test_vgchange_activate_with_config_does_not_backup(self, mock_lvmbackup):
+ lvutil._try_backup_vg(
+ lvutil.CMD_VGCHANGE,
+ ["-an", "--config", "devices{}", TEST_VG]
+ )
+
+ mock_lvmbackup.try_backup.assert_not_called()
+
+ def test_lvchange_activate_does_not_backup(self, mock_lvmbackup):
+ lvutil._try_backup_vg(lvutil.CMD_LVCHANGE, ["-an", TEST_VOL])
+ mock_lvmbackup.try_backup.assert_not_called()
+
+ def test_lvchange_perm_does_backup(self, mock_lvmbackup):
+ lvutil._try_backup_vg(lvutil.CMD_LVCHANGE, ["-p", "r", TEST_VOL])
+ mock_lvmbackup.try_backup.assert_called_once_with(TEST_VG)
+
+ def test_vgchange_perm_with_config_does_backup(self, mock_lvmbackup):
+ lvutil._try_backup_vg(
+ lvutil.CMD_LVCHANGE,
+ ["-p", "r", "--config", "devices{}", TEST_VOL]
+ )
+
+ mock_lvmbackup.try_backup.assert_called_once_with(TEST_VG)
+
+ def test_vgs_does_not_back_up(self, mock_lvmbackup):
+ lvutil._try_backup_vg(lvutil.CMD_VGS, ["--readonly", TEST_VG])
+ mock_lvmbackup.try_backup.assert_not_called()
+
+ def test_pvcreate_does_not_backup(self, mock_lvmbackup):
+ lvutil._try_backup_vg(
+ lvutil.CMD_PVCREATE,
+ ["-ff", "-y", "--metadatasize", "10M", "/dev/sda1"]
+ )
+
+ mock_lvmbackup.try_backup.assert_not_called()