From 8a6225b7c2f61740a7be8e151804e55a54787041 Mon Sep 17 00:00:00 2001 From: Dan Helfman Date: Tue, 3 Dec 2024 08:51:10 -0800 Subject: [PATCH] Factor out logic for finding contained source directories in a parent directory (#80). --- borgmatic/hooks/data_source/lvm.py | 26 +++++-------------------- borgmatic/hooks/data_source/snapshot.py | 23 ++++++++++++++++++++++ 2 files changed, 28 insertions(+), 21 deletions(-) create mode 100644 borgmatic/hooks/data_source/snapshot.py diff --git a/borgmatic/hooks/data_source/lvm.py b/borgmatic/hooks/data_source/lvm.py index 2948a047..feef0e33 100644 --- a/borgmatic/hooks/data_source/lvm.py +++ b/borgmatic/hooks/data_source/lvm.py @@ -2,11 +2,11 @@ import glob import json import logging import os -import pathlib import shutil import subprocess import borgmatic.config.paths +import borgmatic.hooks.data_source.snapshot import borgmatic.execute logger = logging.getLogger(__name__) @@ -22,24 +22,6 @@ def use_streaming(hook_config, config, log_prefix): # pragma: no cover BORGMATIC_SNAPSHOT_PREFIX = 'borgmatic-' -def get_contained_source_directories(mount_point, source_directories): - ''' - Given a mount point and a sequence of source directories, get the subset of source directories - for which the mount point is the same as that source directory, a parent of it, a grandparent, - etc. The idea is if, say, /var/log and /var/lib are source directories, but there's a logical - volume mount point at /var, then /var is what we want to snapshot. - ''' - if not source_directories: - return () - - return tuple( - source_directory - for source_directory in source_directories - if mount_point == source_directory - or pathlib.PurePosixPath(mount_point) in pathlib.PurePath(source_directory).parents - ) - - def get_logical_volumes(lsblk_command, source_directories=None): ''' Given an lsblk command to run and a sequence of configured source directories, find the @@ -74,7 +56,9 @@ def get_logical_volumes(lsblk_command, source_directories=None): for device in devices_info['blockdevices'] if device['mountpoint'] and device['type'] == 'lvm' for contained_source_directories in ( - get_contained_source_directories(device['mountpoint'], source_directories), + borgmatic.hooks.data_source.snapshot.get_contained_directories( + device['mountpoint'], source_directories + ), ) if not source_directories or contained_source_directories ) @@ -171,7 +155,7 @@ def dump_data_sources( contained_source_directories, ) in requested_logical_volumes: snapshot_name = f'{device_name}_{snapshot_suffix}' - logger.debug(f'{log_prefix}: Creating LVM snapshot {snapshot_name}{dry_run_label}') + logger.debug(f'{log_prefix}: Creating LVM snapshot {snapshot_name} of {mount_point}{dry_run_label}') if not dry_run: snapshot_logical_volume( diff --git a/borgmatic/hooks/data_source/snapshot.py b/borgmatic/hooks/data_source/snapshot.py new file mode 100644 index 00000000..097b51ea --- /dev/null +++ b/borgmatic/hooks/data_source/snapshot.py @@ -0,0 +1,23 @@ +import pathlib + + +IS_A_HOOK = False + + +def get_contained_directories(parent_directory, candidate_contained_directories): + ''' + Given a parent directory and a sequence of candiate directories potentially inside it, get the + subset of contained directories for which the parent directory is actually the parent, a + grandparent, the very same directory, etc. The idea is if, say, /var/log and /var/lib are + candidate contained directories, but there's a parent directory (logical volume, dataset, + subvolume, etc.) at /var, then /var is what we want to snapshot. + ''' + if not candidate_contained_directories: + return () + + return tuple( + candidate + for candidate in candidate_contained_directories + if parent_directory == candidate + or pathlib.PurePosixPath(parent_directory) in pathlib.PurePath(candidate).parents + )