datasources: Added LVM snapshot support

awit_backstep/datasources/lvm.py

+"""AWIT Backstep LVM datasource."""
+#
+# SPDX-License-Identifier: GPL-3.0-or-later
+#
+# Copyright (C) 2019-2021, AllWorldIT.
+#
+# 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 <https://www.gnu.org/licenses/>.
+
+import subprocess  # nosec
+from datetime import datetime
+from typing import Dict, List, Optional
+
+from ..notifiers import NotifierList
+from ..util import chunks, get_file_size
+from . import BackupArchiveDevices, BackupSet, DatasourceError, DatasourcePluginBase
+
+__VERSION__ = "0.0.1"
+
+
+class LVMBackupArchive(BackupArchiveDevices):
+    """LVM backup archive class."""
+
+    _lvm_volumes: List[str]
+    _snapshots: Dict[str, str]
+
+    def __init__(self, archive_name: str, lvm_volume_list: List[str], notifiers: Optional[NotifierList] = None):  # noqa: E1101
+        """
+        Inititalize the object.
+
+        :param archive_name: Backup archive name
+        :type archive_name: str
+        :param lvm_volume_list: LVM volume list
+        :type lvm_volume_list: str
+        """
+
+        # Use super to initialize the base class
+        super().__init__(name=archive_name, notifiers=notifiers)
+
+        # Save our LVM volume list
+        self._lvm_volumes = lvm_volume_list
+        # Snapshots of the disks we're going to backup
+        self._snapshots = {}
+
+        # Output info about the LVM volumes we're going to backup
+        self.notifiers.info(f'Adding LVM volumes to archive "{archive_name}":')
+        for lvm_volume in self.lvm_volumes:
+            self.notifiers.info(f"  - {lvm_volume}")
+
+    def pre_backup(self) -> None:
+        """Pre-backup method, where we snapshot the disks."""
+
+        # Snapshot volumes
+        self._create_snapshots()
+
+        self.notifiers.debug("Adding snapshot paths to archive:")
+        for _, snapshot_device in self._snapshots.items():
+            self.notifiers.debug(f"  - {snapshot_device}")
+            self.add_path(snapshot_device)
+
+    def post_backup(self) -> None:
+        """Post-backup method, where we recover the snapshots."""
+
+        # Commit our snapshot afterwards
+        self._remove_snapshot()
+
+    def _create_snapshots(self) -> None:
+        """Create a snapshots."""
+
+        # Work out a snapshot suffix to use
+        snapshot_suffix = "-" + datetime.now().strftime("awitbackstep-snapshot-%Y%m%d%H%M%S")
+
+        self.notifiers.info("LVM snapshotting started")
+
+        # Loop with disks and add them to the snapshot
+        for lvm_volume in self.lvm_volumes:
+
+            snapshot_lv = f"{lvm_volume}{snapshot_suffix}"
+            snapshot_device = f"/dev/{snapshot_lv}"
+
+            # Grab LVM volume size
+            lvm_volume_size = int(get_file_size(f"/dev/{lvm_volume}") / 1048576)
+            # Work out minimum size, or 5GB
+            snapshot_size_min = max([int(lvm_volume_size * 0.05), 5 * 1024])
+            # Work out maximum size or 50GB
+            snapshot_size = min([snapshot_size_min, 50 * 1024])
+
+            snapshot_args = [
+                "lvcreate",
+                "--snapshot",
+                lvm_volume,
+                "--size",
+                f"{snapshot_size}M",
+                "--name",
+                snapshot_lv,
+            ]
+
+            # Add snapshot to our list
+            self._snapshots[lvm_volume] = snapshot_device
+
+            self.notifiers.info(f"  - Snapshotting '{lvm_volume}' as '{snapshot_lv}' with {snapshot_size}MB of space")
+
+            self.notifiers.debug(f"Running: {snapshot_args}")
+            try:
+                # Create process and monitor status
+                process = subprocess.Popen(snapshot_args, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)  # nosec
+                # Loop with output
+                for line in chunks(process.stdout):
+                    self.notifiers.info(f" - {line}")
+            except KeyboardInterrupt:
+                self.notifiers.error("LVM snapshotting failed")
+                raise DatasourceError(f'Keyboard interrupt while snapshotting LVM volume "{lvm_volume}"') from None
+            except subprocess.CalledProcessError as err:
+                self.notifiers.error("LVM snapshotting failed")
+                raise DatasourceError(
+                    f'Error snapshotting LVM volume "{lvm_volume}", ' f"exited with code {err.returncode}"
+                ) from None
+            except OSError as err:
+                self.notifiers.error("LVM snapshotting failed")
+                raise DatasourceError(f'OS error snapshotting LVM volume "{lvm_volume}", ' f"exited with => {err}") from None
+            # Check result code
+            process.communicate()
+            result_code = process.poll()
+            if result_code:
+                self.notifiers.error("LVM snapshotting failed")
+                raise DatasourceError(f'Error snapshotting LVM volume "{lvm_volume}", ' f"exited with code {result_code}") from None
+
+        self.notifiers.info("LVM snapshotting done")
+
+    def _remove_snapshot(self) -> None:
+        """Remove a LVM snapshot."""
+
+        # Loop with snapshotted devices
+        for _, snapshot_device in sorted(self._snapshots.items()):
+            # Try do a commit
+            try:
+                self.notifiers.info(f'Starting LVM snapshot removal for "{snapshot_device}"')
+                lvremove_args = ["lvremove", "--force", snapshot_device]
+                self.notifiers.debug(f"Running: {lvremove_args}")
+                # Create process and monitor status
+                process = subprocess.Popen(lvremove_args, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)  # nosec
+                # Loop with output
+                for line in chunks(process.stdout, delim="\n\r"):
+                    self.notifiers.info(f" - {line}")
+            except KeyboardInterrupt:
+                self.notifiers.error("LVM snapshot removal failed")
+                raise DatasourceError(f'Keyboard interrupt during LVM snapshot removal for "{snapshot_device}"') from None
+            except subprocess.CalledProcessError as err:
+                self.notifiers.error("LVM snapshot removal failed")
+                raise DatasourceError(
+                    f'Error running LVM snapshot removal for "{snapshot_device}", ' f"exited with code {err.returncode}"
+                ) from None
+            except OSError as err:
+                self.notifiers.error("LVM snapshot removal failed")
+                raise DatasourceError(
+                    f'OS error running LVM snapshot removal for "{snapshot_device}", ' f"exited with => {err}"
+                ) from None
+            # Check result code
+            process.communicate()
+            result_code = process.poll()
+            if result_code:
+                self.notifiers.error("LVM snapshot removal failed")
+                raise DatasourceError(
+                    f'Error running LVM snapshot removal for "{snapshot_device}", ' f"exited with code {result_code}"
+                ) from None
+
+            self.notifiers.info(f'Completed snapshot removal for "{snapshot_device}"')
+
+    @property
+    def lvm_volumes(self) -> List[str]:
+        """Return a list of our LVM volumes."""
+        return sorted(self._lvm_volumes)
+
+
+class LVMPlugin(DatasourcePluginBase):
+    """LVM Plugin."""
+
+    _name = "lvm"
+
+    def get_backup_set(self, backup_items: List[str], backup_name: Optional[str]) -> BackupSet:
+        """Parse the backup items into backup archives and paths."""
+
+        # Call the parent object to set things up
+        super().get_backup_set(backup_items=backup_items, backup_name=backup_name)
+
+        # Create backup set
+        backup_set = BackupSet()
+
+        for volume_sets in backup_items:
+            # Split off volume list on ,
+            volume_list = volume_sets.replace("/dev/", "").split(",")
+            # Generate an archive name
+            archive_name = f"{self.archive_basename}-{volume_list[0].replace('/dev/','').replace('/', '_')}"
+
+            # Create the backup archive
+            lvm_archive = LVMBackupArchive(archive_name=archive_name, lvm_volume_list=volume_list, notifiers=self.notifiers)
+            # Add add it to the set
+            backup_set.add_archive(lvm_archive)
+
+        return backup_set