Factor out logic for finding contained source directories in a parent directory (#80).

This commit is contained in:
Dan Helfman 2024-12-03 08:51:10 -08:00
parent 9aaa3c925f
commit 8a6225b7c2
2 changed files with 28 additions and 21 deletions

View File

@ -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(

View File

@ -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
)