Coverage for bzfs_main/configuration.py: 99%
669 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-21 12:39 +0000
1# Copyright 2024 Wolfgang Hoschek AT mac DOT com
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7# http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
14#
15"""Configuration subsystem; All CLI option/parameter values are reachable from the "Params" class."""
17from __future__ import (
18 annotations,
19)
20import argparse
21import ast
22import os
23import platform
24import random
25import re
26import shutil
27import stat
28import sys
29import tempfile
30import threading
31import time
32from collections.abc import (
33 Iterable,
34)
35from dataclasses import (
36 dataclass,
37)
38from datetime import (
39 datetime,
40 tzinfo,
41)
42from logging import (
43 Logger,
44)
45from typing import (
46 Final,
47 Literal,
48 NamedTuple,
49 cast,
50 final,
51)
53from bzfs_main.argparse_actions import (
54 SnapshotFilter,
55 optimize_snapshot_filters,
56)
57from bzfs_main.argparse_cli import (
58 LOG_DIR_DEFAULT,
59 ZFS_RECV_GROUPS,
60 ZFS_RECV_O,
61 ZFS_RECV_O_INCLUDE_REGEX_DEFAULT,
62 __version__,
63)
64from bzfs_main.detect import (
65 DISABLE_PRG,
66)
67from bzfs_main.filter import (
68 SNAPSHOT_FILTERS_VAR,
69)
70from bzfs_main.period_anchors import (
71 PeriodAnchors,
72)
73from bzfs_main.util import (
74 utils,
75)
76from bzfs_main.util.connection import (
77 ConnectionPools,
78 MiniParams,
79 MiniRemote,
80)
81from bzfs_main.util.retry import (
82 RetryPolicy,
83)
84from bzfs_main.util.utils import (
85 DIR_PERMISSIONS,
86 FILE_PERMISSIONS,
87 PROG_NAME,
88 SHELL_CHARS,
89 UNIX_DOMAIN_SOCKET_PATH_MAX_LENGTH,
90 UNIX_TIME_INFINITY_SECS,
91 RegexList,
92 SnapshotPeriods,
93 SynchronizedBool,
94 append_if_absent,
95 compile_regexes,
96 current_datetime,
97 die,
98 get_home_directory,
99 get_timezone,
100 getenv_bool,
101 getenv_int,
102 is_included,
103 ninfix,
104 nprefix,
105 nsuffix,
106 open_nofollow,
107 parse_duration_to_milliseconds,
108 pid_exists,
109 sha256_hex,
110 sha256_urlsafe_base64,
111 urlsafe_base64,
112 validate_dataset_name,
113 validate_file_permissions,
114 validate_is_not_a_symlink,
115 validate_property_name,
116 xappend,
117)
119# constants:
120_UNSET_ENV_VARS_LOCK: Final[threading.Lock] = threading.Lock()
121_UNSET_ENV_VARS_LATCH: Final[SynchronizedBool] = SynchronizedBool(True)
124#############################################################################
125@final
126class LogParams:
127 """Option values for logging."""
129 def __init__(self, args: argparse.Namespace) -> None:
130 """Reads from ArgumentParser via args."""
131 # immutable variables:
132 if args.quiet:
133 log_level: str = "ERROR"
134 elif args.verbose >= 2:
135 log_level = "TRACE"
136 elif args.verbose >= 1:
137 log_level = "DEBUG"
138 else:
139 log_level = "INFO"
140 self.log_level: Final[str] = log_level
141 self.timestamp: Final[str] = datetime.now().isoformat(sep="_", timespec="seconds") # 2024-09-03_12:26:15
142 self.isatty: Final[bool] = getenv_bool("isatty", True)
143 self.quiet: Final[bool] = args.quiet
144 self.terminal_columns: Final[int] = (
145 getenv_int("terminal_columns", shutil.get_terminal_size(fallback=(120, 24)).columns)
146 if self.isatty and args.pv_program != DISABLE_PRG and not self.quiet
147 else 0
148 )
149 self.home_dir: Final[str] = get_home_directory()
150 log_parent_dir: Final[str] = args.log_dir if args.log_dir else os.path.join(self.home_dir, LOG_DIR_DEFAULT)
151 if LOG_DIR_DEFAULT not in os.path.basename(log_parent_dir):
152 die(f"Basename of --log-dir must contain the substring '{LOG_DIR_DEFAULT}', but got: {log_parent_dir}")
153 sep: str = "_" if args.log_subdir == "daily" else ":"
154 timestamp: str = self.timestamp
155 subdir: str = timestamp[: timestamp.rindex(sep) if args.log_subdir == "minutely" else timestamp.index(sep)]
156 # 2024-09-03 (d), 2024-09-03_12 (h), 2024-09-03_12:26 (m)
157 self.log_dir: Final[str] = os.path.join(log_parent_dir, subdir)
158 os.makedirs(log_parent_dir, mode=DIR_PERMISSIONS, exist_ok=True)
159 validate_is_not_a_symlink("--log-dir ", log_parent_dir)
160 validate_file_permissions(log_parent_dir, DIR_PERMISSIONS)
161 os.makedirs(self.log_dir, mode=DIR_PERMISSIONS, exist_ok=True)
162 validate_is_not_a_symlink("--log-dir subdir ", self.log_dir)
163 validate_file_permissions(self.log_dir, DIR_PERMISSIONS)
164 self.log_file_prefix: Final[str] = args.log_file_prefix
165 self.log_file_infix: Final[str] = args.log_file_infix
166 self.log_file_suffix: Final[str] = args.log_file_suffix
167 fd, self.log_file = tempfile.mkstemp(
168 suffix=".log",
169 prefix=f"{self.log_file_prefix}{self.timestamp}{self.log_file_infix}{self.log_file_suffix}-",
170 dir=self.log_dir,
171 )
172 os.fchmod(fd, FILE_PERMISSIONS)
173 os.close(fd)
174 self.pv_log_file: Final[str] = self.log_file[: -len(".log")] + ".pv"
175 log_file_stem: str = os.path.basename(self.log_file)[: -len(".log")]
176 # Python's standard logger naming API interprets chars such as '.', '-', ':', spaces, etc in special ways, e.g.
177 # logging.getLogger("foo.bar") vs logging.getLogger("foo-bar"). Thus, we sanitize the Python logger name via a regex:
178 self.logger_name_suffix: Final[str] = re.sub(r"[^A-Za-z0-9_]", repl="_", string=log_file_stem)
179 cache_root_dir: str = os.path.join(log_parent_dir, ".cache")
180 os.makedirs(cache_root_dir, mode=DIR_PERMISSIONS, exist_ok=True)
181 validate_file_permissions(cache_root_dir, DIR_PERMISSIONS)
182 self.last_modified_cache_dir: Final[str] = os.path.join(cache_root_dir, "mods")
184 # Create/update "current" symlink to current_dir, which is a subdir containing further symlinks to log files.
185 # For parallel usage, ensures there is no time window when the symlinks are inconsistent or do not exist.
186 current: str = "current"
187 dot_current_dir: str = os.path.join(log_parent_dir, f".{current}")
188 current_dir: str = os.path.join(dot_current_dir, log_file_stem)
189 os.makedirs(dot_current_dir, mode=DIR_PERMISSIONS, exist_ok=True)
190 validate_is_not_a_symlink("--log-dir: .current ", dot_current_dir)
191 try:
192 os.makedirs(current_dir, mode=DIR_PERMISSIONS, exist_ok=True)
193 _create_symlink(self.log_file, current_dir, f"{current}.log")
194 _create_symlink(self.pv_log_file, current_dir, f"{current}.pv")
195 _create_symlink(self.log_dir, current_dir, f"{current}.dir")
196 dst_file: str = os.path.join(current_dir, current)
197 os.symlink(os.path.relpath(current_dir, start=log_parent_dir), dst_file)
198 os.replace(dst_file, os.path.join(log_parent_dir, current)) # atomic rename
199 _delete_stale_files(dot_current_dir, prefix="", millis=5000, dirs=True, exclude=os.path.basename(current_dir))
200 except FileNotFoundError:
201 pass # harmless concurrent cleanup
203 def __repr__(self) -> str:
204 return str(self.__dict__)
207#############################################################################
208@final
209class Params(MiniParams):
210 """All parsed CLI options combined into a single bundle; simplifies passing around numerous settings and defaults."""
212 def __init__(
213 self,
214 args: argparse.Namespace,
215 sys_argv: list[str],
216 log_params: LogParams,
217 log: Logger,
218 inject_params: dict[str, bool | int] | None = None, # for testing only
219 ) -> None:
220 """Reads from ArgumentParser via args."""
221 # immutable variables:
222 assert args is not None
223 assert isinstance(sys_argv, list)
224 assert log_params is not None
225 assert log is not None
226 self.args: Final[argparse.Namespace] = args
227 self.sys_argv: Final[list[str]] = sys_argv
228 self.log_params: Final[LogParams] = log_params
229 self.log: Logger = log
230 self.inject_params: Final[dict[str, bool | int]] = inject_params if inject_params is not None else {} # testing-only
231 self.one_or_more_whitespace_regex: Final[re.Pattern[str]] = re.compile(r"\s+")
232 self.two_or_more_spaces_regex: Final[re.Pattern[str]] = re.compile(r" +")
233 self._unset_matching_env_vars(args)
234 self.xperiods: Final[SnapshotPeriods] = SnapshotPeriods()
236 assert len(args.root_dataset_pairs) > 0
237 self.root_dataset_pairs: Final[list[tuple[str, str]]] = args.root_dataset_pairs
238 self.recursive: Final[bool] = args.recursive
239 self.recursive_flag: Final[str] = "-r" if args.recursive else ""
241 self.dry_run: Final[bool] = args.dryrun is not None
242 self.dry_run_recv: Final[str] = "-n" if self.dry_run else ""
243 self.dry_run_destroy: Final[str] = self.dry_run_recv
244 self.dry_run_no_send: Final[bool] = args.dryrun == "send"
245 self.verbose_zfs: Final[bool] = args.verbose >= 2
246 self.verbose_destroy: Final[str] = "" if args.quiet else "-v"
248 self.zfs_send_program_opts: Final[list[str]] = self._fix_send_opts(self.split_args(args.zfs_send_program_opts))
249 self.zfs_send_resume_opts: Final[list[str]] = self._fix_send_resume_opts(self.zfs_send_program_opts)
250 zfs_recv_program_opts: list[str] = self.split_args(args.zfs_recv_program_opts)
251 for extra_opt in args.zfs_recv_program_opt:
252 zfs_recv_program_opts.append(self.validate_arg_str(extra_opt, allow_all=True))
253 preserve_properties = [validate_property_name(name, "--preserve-properties") for name in args.preserve_properties]
254 zfs_recv_program_opts, zfs_recv_x_names = self._fix_recv_opts(zfs_recv_program_opts, frozenset(preserve_properties))
255 self.zfs_recv_program_opts: Final[list[str]] = zfs_recv_program_opts
256 self.zfs_recv_x_names: Final[list[str]] = zfs_recv_x_names
257 if self.verbose_zfs:
258 append_if_absent(self.zfs_send_program_opts, "-v")
259 append_if_absent(self.zfs_recv_program_opts, "-v")
260 # zfs_full_recv_opts: dataset-specific dynamic -o/-x property options are computed later per dataset in
261 # replication._add_recv_property_options():
262 self.zfs_full_recv_opts: Final[list[str]] = self.zfs_recv_program_opts.copy()
263 cpconfigs = [CopyPropertiesConfig(group, flag, args, self) for group, flag in ZFS_RECV_GROUPS.items()]
264 self.zfs_recv_o_config, self.zfs_recv_x_config, self.zfs_set_config = cpconfigs
266 self.force_rollback_to_latest_snapshot: Final[bool] = args.force_rollback_to_latest_snapshot
267 self.force_rollback_to_latest_common_snapshot: Final[SynchronizedBool] = SynchronizedBool(
268 args.force_rollback_to_latest_common_snapshot
269 )
270 self.force: Final[SynchronizedBool] = SynchronizedBool(args.force)
271 self.force_once: Final[bool] = args.force_once
272 self.force_unmount: Final[str] = "-f" if args.force_unmount else ""
273 self.force_hard: Final[str] = "-R" if args.force_destroy_dependents else ""
275 self.skip_parent: Final[bool] = args.skip_parent
276 self.skip_missing_snapshots: Final[str] = args.skip_missing_snapshots
277 self.skip_on_error: Final[str] = args.skip_on_error
278 self.retry_policy: Final[RetryPolicy] = RetryPolicy.from_namespace(args).copy(reraise=True)
279 self.skip_replication: Final[bool] = args.skip_replication
280 self.force_delete_dst_tmp_bookmarks: Final[bool] = args.force_delete_dst_tmp_bookmarks
281 self.delete_dst_snapshots: Final[bool] = args.delete_dst_snapshots is not None or self.force_delete_dst_tmp_bookmarks
282 self.delete_dst_bookmarks: Final[bool] = (
283 args.delete_dst_snapshots == "bookmarks" or self.force_delete_dst_tmp_bookmarks
284 )
285 self.delete_dst_snapshots_no_crosscheck: Final[bool] = args.delete_dst_snapshots_no_crosscheck
286 self.delete_dst_snapshots_except: Final[bool] = args.delete_dst_snapshots_except
287 self.delete_dst_datasets: Final[bool] = args.delete_dst_datasets
288 self.delete_empty_dst_datasets: Final[bool] = args.delete_empty_dst_datasets is not None
289 self.delete_empty_dst_datasets_if_no_bookmarks_and_no_snapshots: Final[bool] = (
290 args.delete_empty_dst_datasets == "snapshots+bookmarks"
291 )
292 self.compare_snapshot_lists: Final[str] = args.compare_snapshot_lists
293 self.daemon_lifetime_nanos: Final[int] = 1_000_000 * parse_duration_to_milliseconds(args.daemon_lifetime)
294 self.daemon_frequency: Final[str] = args.daemon_frequency
295 self.enable_privilege_elevation: Final[bool] = not args.no_privilege_elevation
296 self.no_stream: Final[bool] = args.no_stream
297 self.resume_recv: Final[bool] = not args.no_resume_recv
298 self.create_bookmarks: Final[str] = args.create_bookmarks
299 self.use_bookmark: Final[bool] = not args.no_use_bookmark
300 self.r2r_mode_requested: Final[str] = args.r2r
302 self.src: Final[Remote] = Remote("src", args, self) # src dataset, host and ssh options
303 self.dst: Final[Remote] = Remote("dst", args, self) # dst dataset, host and ssh options
304 self.create_src_snapshots_config: Final[CreateSrcSnapshotConfig] = CreateSrcSnapshotConfig(args, self)
305 self.monitor_snapshots_config: Final[MonitorSnapshotsConfig] = MonitorSnapshotsConfig(args, self)
306 self.is_caching_snapshots: Final[bool] = args.cache_snapshots
308 self.compression_program: Final[str] = self._program_name(args.compression_program)
309 self.compression_program_opts: Final[list[str]] = self.split_args(args.compression_program_opts)
310 for opt in {"-o", "--output-file"}.intersection(self.compression_program_opts):
311 die(f"--compression-program-opts: {opt} is disallowed for security reasons.")
312 self.getconf_program: Final[str] = self._program_name("getconf") # print number of CPUs on POSIX
313 self.mbuffer_program: Final[str] = self._program_name(args.mbuffer_program)
314 self.mbuffer_program_opts: Final[list[str]] = self.split_args(args.mbuffer_program_opts)
315 for opt in {"-i", "-I", "-o", "-O", "-l", "-L", "-t", "-T", "-a", "-A"}.intersection(self.mbuffer_program_opts):
316 die(f"--mbuffer-program-opts: {opt} is disallowed for security reasons.")
317 self.bwlimit: Final[str] = self.validate_arg_str(args.bwlimit) if args.bwlimit else ""
318 self.ps_program: Final[str] = self._program_name(args.ps_program)
319 self.pv_program: Final[str] = self._program_name(args.pv_program)
320 self.pv_program_opts: Final[list[str]] = self.split_args(args.pv_program_opts)
321 bad_pv_opts = {"-o", "--output", "-f", "--log-file", "-S", "--stop-at-size", "-Y", "--sync", "-X", "--discard",
322 "-U", "--store-and-forward", "-d", "--watchfd", "-R", "--remote", "-P", "--pidfile"} # fmt: skip
323 for opt in bad_pv_opts.intersection(self.pv_program_opts):
324 die(f"--pv-program-opts: {opt} is disallowed for security reasons.")
325 if self.bwlimit:
326 self.pv_program_opts.append(f"--rate-limit={self.bwlimit}")
327 self.shell_program_local: Final[str] = "sh"
328 self.shell_program: Final[str] = self._program_name(args.shell_program)
329 self.ssh_program: str = self._program_name(args.ssh_program)
330 self.sudo_program: Final[str] = self._program_name(args.sudo_program)
331 self.uname_program: Final[str] = self._program_name("uname")
332 self.zfs_program: Final[str] = self._program_name("zfs")
333 self.zpool_program: Final[str] = self._program_name(args.zpool_program)
335 # no point creating complex shell pipeline commands for tiny data transfers:
336 self.min_pipe_transfer_size: Final[int] = getenv_int("min_pipe_transfer_size", 1024 * 1024)
337 self.max_datasets_per_batch_on_list_snaps: Final[int] = getenv_int("max_datasets_per_batch_on_list_snaps", 1024)
338 self.max_datasets_per_minibatch_on_list_snaps: int = getenv_int("max_datasets_per_minibatch_on_list_snaps", -1)
339 self.max_snapshots_per_minibatch_on_delete_snaps = getenv_int("max_snapshots_per_minibatch_on_delete_snaps", 2**29)
340 self.dedicated_tcp_connection_per_zfs_send: Final[bool] = getenv_bool("dedicated_tcp_connection_per_zfs_send", True)
341 # threads: with --force-once we intentionally coerce to a single-threaded run to ensure deterministic serial behavior
342 self.threads: Final[tuple[int, bool]] = (1, False) if self.force_once else args.threads
343 timeout_duration_nanos = None if args.timeout is None else 1_000_000 * parse_duration_to_milliseconds(args.timeout)
344 self.timeout_duration_nanos: int | None = timeout_duration_nanos # duration (not a timestamp); for logging only
345 self.no_estimate_send_size: Final[bool] = args.no_estimate_send_size
346 self.remote_conf_cache_ttl_nanos: Final[int] = 1_000_000 * parse_duration_to_milliseconds(
347 args.daemon_remote_conf_cache_ttl
348 )
350 self.os_cpu_count: Final[int | None] = os.cpu_count()
351 self.os_getuid: Final[int] = os.getuid()
352 self.os_geteuid: Final[int] = os.geteuid()
353 self.prog_version: Final[str] = __version__
354 self.python_version: Final[str] = sys.version
355 self.platform_version: Final[str] = platform.version()
356 self.platform_platform: Final[str] = platform.platform()
358 # mutable variables:
359 snapshot_filters = args.snapshot_filters_var if hasattr(args, SNAPSHOT_FILTERS_VAR) else [[]]
360 self.snapshot_filters: list[list[SnapshotFilter]] = [optimize_snapshot_filters(f) for f in snapshot_filters]
361 self.exclude_dataset_property: str | None = args.exclude_dataset_property
362 self.exclude_dataset_regexes: RegexList = [] # deferred to validate_task() phase
363 self.include_dataset_regexes: RegexList = [] # deferred to validate_task() phase
364 self.tmp_exclude_dataset_regexes: RegexList = [] # deferred to validate_task() phase
365 self.tmp_include_dataset_regexes: RegexList = [] # deferred to validate_task() phase
366 self.abs_exclude_datasets: list[str] = [] # deferred to validate_task() phase
367 self.abs_include_datasets: list[str] = [] # deferred to validate_task() phase
368 self.r2r_mode: str = "off" # deferred to validate_task() phase
370 self.curr_zfs_send_program_opts: list[str] = []
371 self.curr_zfs_send_resume_opts: list[str] = []
372 self.zfs_recv_ox_names: set[str] = set()
373 self.available_programs: dict[str, dict[str, str]] = {}
374 self.zpool_features: dict[str, dict[str, dict[str, str]]] = {r.location: {} for r in [self.src, self.dst]}
375 self.connection_pools: dict[str, ConnectionPools] = {}
377 def split_args(self, text: str, *items: str | Iterable[str], allow_all: bool = False) -> list[str]:
378 """Splits option string on runs of one or more whitespace into an option list."""
379 text = text.strip()
380 opts = self.one_or_more_whitespace_regex.split(text) if text else []
381 xappend(opts, items)
382 if not allow_all:
383 self._validate_quoting(opts)
384 return opts
386 def validate_arg(self, opt: str, allow_spaces: bool = False, allow_all: bool = False) -> str | None:
387 """allow_all permits all characters, including whitespace and quotes; See squote() and dquote()."""
388 if allow_all or opt is None:
389 return opt
390 if any(char.isspace() and (char != " " or not allow_spaces) for char in opt):
391 die(f"Option must not contain a whitespace character{' other than space' if allow_spaces else ''}: {opt}")
392 self._validate_quoting([opt])
393 return opt
395 def validate_arg_str(self, opt: str, allow_spaces: bool = False, allow_all: bool = False) -> str:
396 """Returns validated option string, raising if missing or illegal."""
397 if opt is None:
398 die("Option must not be missing")
399 self.validate_arg(opt, allow_spaces=allow_spaces, allow_all=allow_all)
400 return opt
402 @staticmethod
403 def _validate_quoting(opts: list[str]) -> None:
404 """Raises an error if any option contains a quote or shell metacharacter."""
405 for opt in opts:
406 if "'" in opt or '"' in opt or "$" in opt or "`" in opt:
407 die(f"Option must not contain a single quote or double quote or dollar or backtick character: {opt}")
409 @staticmethod
410 def _fix_recv_opts(opts: list[str], preserve_properties: frozenset[str]) -> tuple[list[str], list[str]]:
411 """Returns sanitized ``zfs recv`` options and captured ``-o/-x`` args."""
412 return _fix_send_recv_opts(
413 opts,
414 exclude_long_opts={"--dryrun"},
415 exclude_short_opts="densFA",
416 include_arg_opts={"-o", "-x"},
417 preserve_properties=preserve_properties,
418 )
420 @staticmethod
421 def _fix_send_opts(opts: list[str]) -> list[str]:
422 """Returns sanitized ``zfs send`` options."""
423 return _fix_send_recv_opts(
424 opts,
425 exclude_long_opts={"--dryrun"},
426 exclude_short_opts="den",
427 include_arg_opts={"-X", "--exclude", "--redact"},
428 exclude_arg_opts=frozenset({"-i", "-I", "-t", "--resume"}),
429 )[0]
431 @staticmethod
432 def _fix_send_resume_opts(opts: list[str]) -> list[str]:
433 """Returns sanitized CLI options for `zfs send -t` resume sends."""
434 return _fix_send_recv_opts(
435 opts,
436 exclude_long_opts={"--backup", "--holds", "--props", "--replicate", "--skip-missing"},
437 exclude_short_opts="bhpRsU",
438 include_arg_opts=set(),
439 exclude_arg_opts=frozenset({"-X", "--exclude", "--redact"}),
440 )[0]
442 def _program_name(self, program: str) -> str:
443 """For testing: helps simulate errors caused by external programs."""
444 self.validate_arg_str(program)
445 if not program:
446 die(f"Program name must not be missing: {program}")
447 for char in SHELL_CHARS + ":":
448 if char in program:
449 die(f"Program name must not contain a '{char}' character: {program}")
450 if self.inject_params.get("inject_unavailable_" + program, False):
451 return program + "-xxx" # substitute a program that cannot be found on the PATH
452 if self.inject_params.get("inject_failing_" + program, False):
453 return "false" # substitute a program that will error out with non-zero return code
454 return program
456 def _unset_matching_env_vars(self, args: argparse.Namespace) -> None:
457 """Unset environment variables matching regex filters."""
458 if len(args.exclude_envvar_regex) == 0 and len(args.include_envvar_regex) == 0:
459 return # fast path
460 exclude_envvar_regexes: RegexList = compile_regexes(args.exclude_envvar_regex)
461 include_envvar_regexes: RegexList = compile_regexes(args.include_envvar_regex)
462 # Mutate global state at most once, atomically. First thread wins. The latch isn't strictly necessary for
463 # correctness as all concurrent bzfs.Job instances in bzfs_jobrunner have identical include/exclude_envvar_regex
464 # anyway. It's just for reduced latency.
465 with _UNSET_ENV_VARS_LOCK:
466 if _UNSET_ENV_VARS_LATCH.get_and_set(False):
467 for envvar_name in list(os.environ):
468 # order of include vs exclude is intentionally reversed to correctly implement semantics:
469 # "unset env var iff excluded and not included (include takes precedence)."
470 if is_included(envvar_name, exclude_envvar_regexes, include_envvar_regexes):
471 os.environ.pop(envvar_name, None)
472 self.log.debug("Unsetting b/c envvar regex: %s", envvar_name)
474 def lock_file_name(self) -> str:
475 """Returns unique path used to detect concurrently running jobs.
477 Makes it such that a job that runs periodically declines to start if the same previous periodic job is still running
478 without completion yet. Hashed key avoids overly long filenames while remaining deterministic.
479 """
480 # fmt: off
481 key = (tuple(self.root_dataset_pairs), self.args.recursive, self.args.exclude_dataset_property,
482 tuple(self.args.include_dataset), tuple(self.args.exclude_dataset),
483 tuple(self.args.include_dataset_regex), tuple(self.args.exclude_dataset_regex),
484 tuple(tuple(f) for f in self.snapshot_filters), self.args.skip_replication, self.args.create_src_snapshots,
485 self.args.create_src_snapshots_plan, self.args.create_src_snapshots_timeformat,
486 self.create_src_snapshots_config.anchors,
487 self.args.delete_dst_datasets, self.args.delete_dst_snapshots, self.args.delete_dst_snapshots_except,
488 self.args.force_delete_dst_tmp_bookmarks,
489 self.args.delete_empty_dst_datasets,
490 self.args.compare_snapshot_lists, self.args.monitor_snapshots,
491 self.src.basis_ssh_host, self.dst.basis_ssh_host,
492 self.src.basis_ssh_user, self.dst.basis_ssh_user,
493 self.src.ssh_port, self.dst.ssh_port,
494 os.path.abspath(self.src.ssh_config_file) if self.src.ssh_config_file else "",
495 os.path.abspath(self.dst.ssh_config_file) if self.dst.ssh_config_file else "",
496 )
497 # fmt: on
498 hash_code: str = sha256_hex(str(key))
499 log_parent_dir: str = os.path.dirname(self.log_params.log_dir)
500 locks_dir: str = os.path.join(log_parent_dir, ".locks")
501 os.makedirs(locks_dir, mode=DIR_PERMISSIONS, exist_ok=True)
502 validate_is_not_a_symlink("--locks-dir ", locks_dir)
503 validate_file_permissions(locks_dir, DIR_PERMISSIONS)
504 return os.path.join(locks_dir, f"{PROG_NAME}-lockfile-{hash_code}.lock")
506 def dry(self, msg: str) -> str:
507 """Prefix ``msg`` with 'Dry' when running in dry-run mode."""
508 return utils.dry(msg, self.dry_run)
510 def is_program_available(self, program: str, location: str) -> bool:
511 """Return True if ``program`` was detected on ``location`` host."""
512 return program in self.available_programs.get(location, {})
515#############################################################################
516@final
517class Remote(MiniRemote):
518 """Connection settings for either source or destination host."""
520 def __init__(self, loc: str, args: argparse.Namespace, p: Params) -> None:
521 """Reads from ArgumentParser via args."""
522 # immutable variables:
523 assert loc == "src" or loc == "dst"
524 self.location: str = loc
525 self.params: Params = p
526 self.basis_ssh_user: Final[str] = getattr(args, f"ssh_{loc}_user")
527 self.basis_ssh_host: Final[str] = getattr(args, f"ssh_{loc}_host")
528 self.ssh_port: Final[int | None] = getattr(args, f"ssh_{loc}_port")
529 self.ssh_config_file: Final[str | None] = p.validate_arg(getattr(args, f"ssh_{loc}_config_file"))
530 if self.ssh_config_file and self.ssh_config_file != "none":
531 # `ssh -F none` will not read any config file per https://man7.org/linux/man-pages/man1/ssh.1.html
532 if "bzfs_ssh_config" not in os.path.basename(self.ssh_config_file):
533 die(f"Basename of --ssh-{loc}-config-file must contain substring 'bzfs_ssh_config': {self.ssh_config_file}")
534 with open_nofollow(self.ssh_config_file, "rb"):
535 pass # validate
536 self.ssh_config_file_hash: Final[str] = (
537 sha256_urlsafe_base64(os.path.abspath(self.ssh_config_file), padding=False) if self.ssh_config_file else ""
538 )
539 self.ssh_cipher: Final[str] = p.validate_arg_str(args.ssh_cipher)
540 # disable interactive password prompts and X11 forwarding and pseudo-terminal allocation:
541 ssh_extra_opts: list[str] = ["-oBatchMode=yes", "-oServerAliveInterval=0", "-x", "-T"] + (
542 ["-v"] if args.verbose >= 3 else []
543 )
544 self.ssh_extra_opts: tuple[str, ...] = tuple(ssh_extra_opts)
545 self.max_concurrent_ssh_sessions_per_tcp_connection: Final[int] = args.max_concurrent_ssh_sessions_per_tcp_connection
546 self.ssh_exit_on_shutdown: bool = args.ssh_exit_on_shutdown
547 self.ssh_control_persist_secs: int = args.ssh_control_persist_secs
548 self.ssh_control_persist_margin_secs: int = getenv_int("ssh_control_persist_margin_secs", 2)
549 self.socket_prefix: Final[str] = "s"
550 self.reuse_ssh_connection: bool = getenv_bool("reuse_ssh_connection", True)
551 self.ssh_socket_dir: str = ""
552 if self.reuse_ssh_connection:
553 ssh_home_dir: str = os.path.join(get_home_directory(), ".ssh")
554 os.makedirs(ssh_home_dir, mode=DIR_PERMISSIONS, exist_ok=True)
555 self.ssh_socket_dir = os.path.join(ssh_home_dir, "bzfs")
556 os.makedirs(self.ssh_socket_dir, mode=DIR_PERMISSIONS, exist_ok=True)
557 validate_file_permissions(self.ssh_socket_dir, mode=DIR_PERMISSIONS)
558 self.ssh_exit_on_shutdown_socket_dir: Final[str] = os.path.join(self.ssh_socket_dir, "x")
559 os.makedirs(self.ssh_exit_on_shutdown_socket_dir, mode=DIR_PERMISSIONS, exist_ok=True)
560 validate_file_permissions(self.ssh_exit_on_shutdown_socket_dir, mode=DIR_PERMISSIONS)
561 _delete_stale_files(self.ssh_exit_on_shutdown_socket_dir, prefix=self.socket_prefix, ssh=True)
562 self.sanitize1_regex: Final[re.Pattern[str]] = re.compile(r"[\s\\/@$]") # replace whitespace, /, $, \, @ with ~ char
563 self.sanitize2_regex: Final[re.Pattern[str]] = re.compile(rf"[^a-zA-Z0-9{re.escape('~.:_-')}]") # remove bad chars
565 # mutable variables:
566 self.root_dataset: str = "" # deferred until run_main()
567 self.basis_root_dataset: str = "" # deferred until run_main()
568 self.pool: str = ""
569 self.sudo: str = ""
570 self.use_zfs_delegation: bool = False
571 self.ssh_user: str = ""
572 self.ssh_host: str = ""
573 self.ssh_user_host: str = ""
574 self.is_nonlocal: bool = False
576 def local_ssh_command(self, socket_file: str | None) -> tuple[list[str], str | None]:
577 """Returns the ssh CLI command to run locally in order to talk to the remote host; This excludes the (trailing)
578 command to run on the remote host, which will be appended later; also returns the effective ControlPath used by the
579 ssh CLI command, or ``None`` when SSH multiplexing is not active."""
580 if not self.ssh_user_host:
581 return [], None # dataset is on local host - don't use ssh
583 # dataset is on remote host
584 p: Params = self.params
585 if p.ssh_program == DISABLE_PRG:
586 die("Cannot talk to remote host because ssh CLI is disabled.")
587 ssh_cmd: list[str] = [p.ssh_program] + list(self.ssh_extra_opts)
588 if self.ssh_config_file:
589 ssh_cmd += ["-F", self.ssh_config_file]
590 if self.ssh_cipher: 590 ↛ 592line 590 didn't jump to line 592 because the condition on line 590 was always true
591 ssh_cmd += ["-c", self.ssh_cipher]
592 if self.ssh_port:
593 ssh_cmd += ["-p", str(self.ssh_port)]
595 socket_path: str | None = None
596 if self.reuse_ssh_connection:
597 # Performance: reuse ssh connection for low latency startup of frequent ssh invocations via the 'ssh -S' and
598 # 'ssh -S -M -oControlPersist=60s' options. See https://en.wikibooks.org/wiki/OpenSSH/Cookbook/Multiplexing
599 if socket_file:
600 socket_path = socket_file
601 else:
602 # Generate unique private Unix domain socket file name in user's home dir and pass it to 'ssh -S /path/to/socket'
603 def sanitize(name: str) -> str:
604 name = self.sanitize1_regex.sub("~", name) # replace whitespace, /, $, \, @ with a ~ tilde char
605 name = self.sanitize2_regex.sub("", name) # Remove disallowed chars
606 return name
608 max_rand: int = 999_999_999_999
609 rand_str: str = urlsafe_base64(random.SystemRandom().randint(0, max_rand), max_value=max_rand, padding=False)
610 curr_time: str = urlsafe_base64(time.time_ns(), max_value=2**64 - 1, padding=False)
611 unique: str = f"{os.getpid()}@{curr_time}@{rand_str}"
612 optional: str = f"@{sanitize(self.ssh_host)[:45]}@{sanitize(self.ssh_user)}"
613 socket_name: str = f"{self.socket_prefix}{unique}{optional}"
614 socket_path = os.path.join(self.ssh_exit_on_shutdown_socket_dir, socket_name)
615 socket_path = socket_path[: max(UNIX_DOMAIN_SOCKET_PATH_MAX_LENGTH, len(socket_path) - len(optional))]
616 # `ssh` will error out later if the max OS Unix domain socket path limit cannot be met reasonably as the
617 # home directory path is too long, typically because the Unix user name is unreasonably long.
618 ssh_cmd += ["-S", socket_path]
619 ssh_cmd += [self.ssh_user_host]
620 return ssh_cmd, socket_path
622 def cache_key(self) -> tuple[str, str, int | None, str | None]:
623 """Returns tuple uniquely identifying this Remote for caching."""
624 return self.location, self.ssh_user_host, self.ssh_port, self.ssh_config_file
626 def cache_namespace(self) -> str:
627 """Returns cache namespace string which is a stable, unique directory component for caches that distinguishes
628 endpoints by username+host+port+ssh_config_file where applicable, and uses '-' when no user/host is present (local
629 mode)."""
630 if not self.ssh_user_host:
631 return "-" # local mode
632 return f"{self.ssh_user_host}#{self.ssh_port or ''}#{self.ssh_config_file_hash}"
634 def is_ssh_available(self) -> bool:
635 """Return True if the ssh client program required for this remote is available on the local host."""
636 return self.params.is_program_available("ssh", "local")
638 def __repr__(self) -> str:
639 return str(self.__dict__)
642#############################################################################
643@final
644class CopyPropertiesConfig:
645 """--zfs-recv-o* and --zfs-recv-x* option groups for copying or excluding ZFS properties on receive."""
647 def __init__(self, group: str, flag: str, args: argparse.Namespace, p: Params) -> None:
648 """Reads from ArgumentParser via args."""
649 assert group in ZFS_RECV_GROUPS
650 # immutable variables:
651 grup: str = group
652 self.group: Final[str] = group # one of zfs_recv_o, zfs_recv_x
653 self.flag: Final[str] = flag # one of -o or -x
654 sources: str = p.validate_arg_str(getattr(args, f"{grup}_sources"))
655 self.sources: Final[str] = ",".join(sorted([s.strip() for s in sources.strip().split(",")])) # canonicalize
656 self.targets: Final[str] = p.validate_arg_str(getattr(args, f"{grup}_targets"))
657 include_regexes: list[str] | None = getattr(args, f"{grup}_include_regex")
658 assert ZFS_RECV_O in ZFS_RECV_GROUPS
659 if include_regexes is None:
660 include_regexes = [ZFS_RECV_O_INCLUDE_REGEX_DEFAULT] if grup == ZFS_RECV_O else []
661 self.include_regexes: Final[RegexList] = compile_regexes(include_regexes)
662 self.exclude_regexes: Final[RegexList] = compile_regexes(getattr(args, f"{grup}_exclude_regex"))
664 def __repr__(self) -> str:
665 return str(self.__dict__)
668#############################################################################
669@final
670class SnapshotLabel(NamedTuple):
671 """Contains the individual parts that are concatenated into a ZFS snapshot name."""
673 prefix: str # bzfs_
674 infix: str # us-west_
675 timestamp: str # 2024-11-06_08:30:05
676 suffix: str # _hourly
678 def __str__(self) -> str: # bzfs_us-west_2024-11-06_08:30:05_hourly
679 return f"{self.prefix}{self.infix}{self.timestamp}{self.suffix}"
681 def notimestamp_str(self) -> str: # bzfs_us-west_hourly
682 """Returns the concatenation of all parts except for the timestamp part."""
683 return f"{self.prefix}{self.infix}{self.suffix}"
685 def validate_label(self, input_text: str) -> None:
686 """Validates that the composed snapshot label forms a legal name."""
687 name: str = str(self)
688 validate_dataset_name(name, input_text)
689 if "/" in name:
690 die(f"Invalid ZFS snapshot name: '{name}' for: '{input_text}'")
691 for key, value in {"prefix": self.prefix, "infix": self.infix, "suffix": self.suffix}.items():
692 if key == "prefix":
693 if not value.endswith("_"):
694 die(f"Invalid {input_text}{key}: Must end with an underscore character: '{value}'")
695 if value.count("_") > 1:
696 die(f"Invalid {input_text}{key}: Must not contain multiple underscore characters: '{value}'")
697 elif key == "infix":
698 if value:
699 if not value.endswith("_"):
700 die(f"Invalid {input_text}{key}: Must end with an underscore character: '{value}'")
701 if value.count("_") > 1:
702 die(f"Invalid {input_text}{key}: Must not contain multiple underscore characters: '{value}'")
703 elif value:
704 if not value.startswith("_"):
705 die(f"Invalid {input_text}{key}: Must start with an underscore character: '{value}'")
706 if value.count("_") > 1:
707 die(f"Invalid {input_text}{key}: Must not contain multiple underscore characters: '{value}'")
710#############################################################################
711@final
712class CreateSrcSnapshotConfig:
713 """Option values for --create-src-snapshots, that is, for automatically creating source snapshots."""
715 def __init__(self, args: argparse.Namespace, p: Params) -> None:
716 """Option values for --create-src-snapshots*; reads from ArgumentParser via args."""
717 # immutable variables:
718 self.skip_create_src_snapshots: Final[bool] = not args.create_src_snapshots
719 self.create_src_snapshots_even_if_not_due: Final[bool] = args.create_src_snapshots_even_if_not_due
720 tz_spec: str | None = args.create_src_snapshots_timezone if args.create_src_snapshots_timezone else None
721 self.tz: Final[tzinfo | None] = get_timezone(tz_spec)
722 self.current_datetime: datetime = current_datetime(tz_spec)
723 self.timeformat: Final[str] = args.create_src_snapshots_timeformat
724 self.anchors: Final[PeriodAnchors] = PeriodAnchors.parse(args)
726 # Compute the schedule for upcoming periodic time events (suffix_durations). This event schedule is also used in
727 # daemon mode via sleep_until_next_daemon_iteration()
728 labels: list[SnapshotLabel] = []
729 create_src_snapshots_plan: str = args.create_src_snapshots_plan or str({"bzfs": {"onsite": {"adhoc": 1}}})
730 for org, target_periods in ast.literal_eval(create_src_snapshots_plan).items():
731 for target, periods in target_periods.items():
732 for period_unit, period_amount in periods.items(): # e.g. period_unit can be "10minutely" or "minutely"
733 if not isinstance(period_amount, int) or period_amount < 0:
734 die(f"--create-src-snapshots-plan: Period amount must be a non-negative integer: {period_amount}")
735 if period_amount > 0:
736 suffix: str = nsuffix(period_unit)
737 labels.append(SnapshotLabel(prefix=nprefix(org), infix=ninfix(target), timestamp="", suffix=suffix))
738 suffixes: list[str] = list({label.suffix for label in labels}) # dedupe
739 xperiods: SnapshotPeriods = p.xperiods
740 if self.skip_create_src_snapshots:
741 duration_amount, duration_unit = p.xperiods.suffix_to_duration0(p.daemon_frequency)
742 if duration_amount <= 0 or not duration_unit:
743 die(f"Invalid --daemon-frequency: {p.daemon_frequency}")
744 suffixes = [nsuffix(p.daemon_frequency)]
745 labels = []
746 suffix_durations: dict[str, tuple[int, str]] = {suffix: xperiods.suffix_to_duration1(suffix) for suffix in suffixes}
748 def suffix_key(suffix: str) -> tuple[int, str]:
749 duration_amount, duration_unit = suffix_durations[suffix]
750 duration_milliseconds: int = duration_amount * xperiods.suffix_milliseconds.get(duration_unit, 0)
751 if suffix.endswith(("hourly", "minutely", "secondly")):
752 if duration_milliseconds != 0 and 86400 * 1000 % duration_milliseconds != 0:
753 die(
754 "Invalid --create-src-snapshots-plan: Period duration should be a divisor of 86400 seconds "
755 f"without remainder so that snapshots will be created at the same time of day every day: {suffix}"
756 )
757 if suffix.endswith("monthly"):
758 if duration_amount != 0 and 12 % duration_amount != 0:
759 die(
760 "Invalid --create-src-snapshots-plan: Period duration should be a divisor of 12 months "
761 f"without remainder so that snapshots will be created at the same time every year: {suffix}"
762 )
763 return duration_milliseconds, suffix
765 suffixes.sort(key=suffix_key, reverse=True) # take snapshots for dailies before hourlies, and so on
766 self.suffix_durations: Final[dict[str, tuple[int, str]]] = {sfx: suffix_durations[sfx] for sfx in suffixes} # sort
767 suffix_indexes: dict[str, int] = {suffix: k for k, suffix in enumerate(suffixes)}
768 labels.sort(key=lambda label: (suffix_indexes[label.suffix], label)) # take snapshots for dailies before hourlies
769 self._snapshot_labels: Final[list[SnapshotLabel]] = labels
770 for label in self.snapshot_labels():
771 label.validate_label("--create-src-snapshots-plan ")
773 def snapshot_labels(self) -> list[SnapshotLabel]:
774 """Returns the snapshot name patterns for which snapshots shall be created."""
775 timeformat: str = self.timeformat
776 is_millis: bool = timeformat.endswith("%F") # non-standard hack to append milliseconds
777 if is_millis:
778 timeformat = timeformat[0:-1] + "f" # replace %F with %f (append microseconds)
779 timestamp: str = self.current_datetime.strftime(timeformat)
780 if is_millis:
781 timestamp = timestamp[: -len("000")] # replace microseconds with milliseconds
782 timestamp = timestamp.replace("+", "z") # zfs CLI does not accept the '+' character in snapshot names
783 return [SnapshotLabel(label.prefix, label.infix, timestamp, label.suffix) for label in self._snapshot_labels]
785 def __repr__(self) -> str:
786 return str(self.__dict__)
789#############################################################################
790@dataclass(frozen=True)
791@final
792class AlertConfig:
793 """Thresholds controlling when alerts fire for snapshot age."""
795 kind: Literal["Latest", "Oldest"]
796 warning_millis: int
797 critical_millis: int
800#############################################################################
801@dataclass(frozen=True)
802@final
803class MonitorSnapshotAlert:
804 """Alert configuration for a single monitored snapshot label."""
806 label: SnapshotLabel
807 latest: AlertConfig | None
808 oldest: AlertConfig | None
809 oldest_skip_holds: bool
812#############################################################################
813@final
814class MonitorSnapshotsConfig:
815 """Option values for --monitor-snapshots*, that is, policy describing which snapshots to monitor for staleness."""
817 def __init__(self, args: argparse.Namespace, p: Params) -> None:
818 """Reads from ArgumentParser via args."""
819 # immutable variables:
820 self.monitor_snapshots: Final[dict] = ast.literal_eval(args.monitor_snapshots)
821 self.dont_warn: Final[bool] = args.monitor_snapshots_dont_warn
822 self.dont_crit: Final[bool] = args.monitor_snapshots_dont_crit
823 self.no_latest_check: Final[bool] = args.monitor_snapshots_no_latest_check
824 self.no_oldest_check: Final[bool] = args.monitor_snapshots_no_oldest_check
825 alerts: list[MonitorSnapshotAlert] = []
826 xperiods: SnapshotPeriods = p.xperiods
827 for org, target_periods in self.monitor_snapshots.items():
828 prefix: str = nprefix(org)
829 for target, periods in target_periods.items():
830 for period_unit, alert_dicts in periods.items(): # e.g. period_unit can be "10minutely" or "minutely"
831 label = SnapshotLabel(prefix=prefix, infix=ninfix(target), timestamp="", suffix=nsuffix(period_unit))
832 alert_latest, alert_oldest = None, None
833 oldest_skip_holds: bool = False
834 for alert_type, alert_dict in alert_dicts.items():
835 m = "--monitor-snapshots: "
836 if alert_type not in ["latest", "oldest"]:
837 die(f"{m}'{alert_type}' must be 'latest' or 'oldest' within {args.monitor_snapshots}")
838 warning_millis: int = 0
839 critical_millis: int = 0
840 cycles: int = 1
841 for kind, value in alert_dict.items():
842 context: str = args.monitor_snapshots
843 if kind == "warning":
844 warning_millis = max(0, parse_duration_to_milliseconds(str(value), context=context))
845 elif kind == "critical":
846 critical_millis = max(0, parse_duration_to_milliseconds(str(value), context=context))
847 elif kind == "cycles":
848 cycles = max(0, int(value))
849 elif kind == "oldest_skip_holds" and alert_type == "oldest":
850 if not isinstance(value, bool):
851 die(f"{m}'{kind}' must be a bool within {context}")
852 oldest_skip_holds = value
853 else:
854 die(f"{m}'{kind}' must be 'warning', 'critical' or 'cycles' within {context}")
855 if warning_millis > 0 or critical_millis > 0:
856 duration_amount, duration_unit = xperiods.suffix_to_duration1(label.suffix)
857 duration_milliseconds: int = duration_amount * xperiods.suffix_milliseconds.get(duration_unit, 0)
858 warning_millis += 0 if warning_millis <= 0 else cycles * duration_milliseconds
859 critical_millis += 0 if critical_millis <= 0 else cycles * duration_milliseconds
860 warning_millis = UNIX_TIME_INFINITY_SECS if warning_millis <= 0 else warning_millis
861 critical_millis = UNIX_TIME_INFINITY_SECS if critical_millis <= 0 else critical_millis
862 capitalized_alert_type = cast(Literal["Latest", "Oldest"], sys.intern(alert_type.capitalize()))
863 alert_config = AlertConfig(capitalized_alert_type, warning_millis, critical_millis)
864 if alert_type == "latest":
865 if not self.no_latest_check:
866 alert_latest = alert_config
867 else:
868 assert alert_type == "oldest"
869 if not self.no_oldest_check:
870 alert_oldest = alert_config
871 if alert_latest is not None or alert_oldest is not None:
872 alerts.append(MonitorSnapshotAlert(label, alert_latest, alert_oldest, oldest_skip_holds))
874 def alert_sort_key(alert: MonitorSnapshotAlert) -> tuple[int, SnapshotLabel]:
875 duration_amount, duration_unit = xperiods.suffix_to_duration1(alert.label.suffix)
876 duration_milliseconds: int = duration_amount * xperiods.suffix_milliseconds.get(duration_unit, 0)
877 return duration_milliseconds, alert.label
879 alerts.sort(key=alert_sort_key, reverse=True) # check snapshots for dailies before hourlies, and so on
880 self.alerts: Final[list[MonitorSnapshotAlert]] = alerts
881 self.enable_monitor_snapshots: Final[bool] = len(alerts) > 0
883 def __repr__(self) -> str:
884 return str(self.__dict__)
887#############################################################################
888def _fix_send_recv_opts(
889 opts: list[str],
890 *,
891 exclude_long_opts: set[str],
892 exclude_short_opts: str,
893 include_arg_opts: set[str],
894 exclude_arg_opts: frozenset[str] = frozenset(),
895 preserve_properties: frozenset[str] = frozenset(),
896) -> tuple[list[str], list[str]]:
897 """These opts are instead managed via bzfs CLI args --dryrun, etc."""
898 assert "-" not in exclude_short_opts
899 results: list[str] = []
900 x_names: set[str] = set(preserve_properties)
902 def _append_include_arg(opt: str, arg: str) -> None:
903 if opt == "-o" and "=" in arg and arg.split("=", 1)[0] in preserve_properties:
904 die(f"--preserve-properties: Disallowed ZFS property found in --zfs-recv-program-opt(s): -o {arg}")
905 if opt == "-x":
906 x_names.discard(arg)
907 results.append(opt)
908 results.append(arg)
910 def _is_compact_exclude_arg(opt: str) -> bool:
911 # Compact option forms such as -isnap and --resume=<resume-token> carry the managed argument inline.
912 for exclude_arg_opt in exclude_arg_opts:
913 if len(exclude_arg_opt) == 2 and opt.startswith(exclude_arg_opt) and len(opt) > len(exclude_arg_opt):
914 return True
915 if exclude_arg_opt.startswith("--") and opt.startswith(exclude_arg_opt + "="):
916 return True
917 return False
919 i = 0
920 n = len(opts)
921 while i < n:
922 opt: str = opts[i]
923 i += 1
924 if opt in exclude_arg_opts: # example: {"-X", "--exclude"}
925 i += 1
926 continue
927 elif _is_compact_exclude_arg(opt):
928 continue
929 elif opt in include_arg_opts: # example: {"-o", "-x"}
930 if i < n:
931 _append_include_arg(opt, opts[i])
932 i += 1
933 else:
934 results.append(opt)
935 elif opt not in exclude_long_opts: # example: {"--dryrun", "--verbose"}
936 for include_arg_opt in include_arg_opts:
937 if len(include_arg_opt) == 2 and opt.startswith(include_arg_opt) and len(opt) > len(include_arg_opt):
938 _append_include_arg(include_arg_opt, opt[len(include_arg_opt) :])
939 break
940 else:
941 if opt.startswith("-") and opt != "-" and not opt.startswith("--"):
942 if "=" in opt:
943 die(f"Invalid short opt {opt!r} within {opts}")
944 for char in exclude_short_opts: # example: "den"
945 opt = opt.replace(char, "")
946 if opt == "-":
947 continue
948 results.append(opt)
949 return results, sorted(x_names)
952_SSH_MASTER_DOMAIN_SOCKET_FILE_PID_REGEX: Final[re.Pattern[str]] = re.compile(r"^[0-9]+") # see local_ssh_command()
955def _delete_stale_files(
956 root_dir: str,
957 *,
958 prefix: str,
959 millis: int = 60 * 60 * 1000,
960 dirs: bool = False,
961 exclude: str | None = None,
962 ssh: bool = False,
963) -> None:
964 """Cleans up obsolete files; For example caused by abnormal termination, OS crash."""
965 seconds: float = millis / 1000
966 now: float = time.time()
967 validate_is_not_a_symlink("", root_dir)
968 with os.scandir(root_dir) as iterator:
969 for entry in iterator:
970 if entry.name == exclude or not entry.name.startswith(prefix):
971 continue
972 try:
973 stats = entry.stat(follow_symlinks=False)
974 is_dir = entry.is_dir(follow_symlinks=False)
975 if ((dirs and is_dir) or (not dirs and not is_dir)) and now - stats.st_mtime >= seconds:
976 if dirs:
977 shutil.rmtree(entry.path, ignore_errors=True)
978 elif not (ssh and stat.S_ISSOCK(stats.st_mode)):
979 os.remove(entry.path)
980 elif match := _SSH_MASTER_DOMAIN_SOCKET_FILE_PID_REGEX.match(entry.name[len(prefix) :]): 980 ↛ 969line 980 didn't jump to line 969 because the condition on line 980 was always true
981 pid: int = int(match.group(0))
982 if pid_exists(pid) is False or now - stats.st_mtime >= 31 * 24 * 60 * 60:
983 os.remove(entry.path) # bzfs process is no longer alive; its ssh master process isn't either
984 except FileNotFoundError:
985 pass # harmless
988def _create_symlink(src: str, dst_dir: str, dst: str) -> None:
989 """Creates dst symlink pointing to src using a relative path."""
990 rel_path: str = os.path.relpath(src, start=dst_dir)
991 os.symlink(src=rel_path, dst=os.path.join(dst_dir, dst))
994def resolve_r2r_mode(p: Params) -> str:
995 """Returns the effective r2r mode for the current task and emits fallback warnings."""
996 src, dst = p.src, p.dst
997 log = p.log
998 mode: str = p.r2r_mode_requested
999 assert mode in ("off", "pull", "push"), mode
1001 if p.skip_replication:
1002 return "off"
1004 if mode == "off":
1005 return "off"
1007 if (not src.ssh_user_host) and (not dst.ssh_user_host):
1008 return "off" # we'll do local replication (there's no need for r2r)
1010 if (not src.is_nonlocal) or (not dst.is_nonlocal):
1011 return "off" # at least one of them is local to this host (there's no need for r2r)
1013 r: Remote = dst if mode == "pull" else src
1014 if not p.is_program_available("sh", r.location):
1015 log.warning(
1016 f"--r2r={mode} requires sh on {r.location} host: {r.ssh_user_host or 'localhost'} for remote-to-remote "
1017 "replication; falling back to --r2r=off."
1018 )
1019 return "off"
1021 if is_same_remote(src, dst): # are user@host, port, ssh_config file the same?
1022 return mode # perf: we'll do local replication on that host
1024 if not p.is_program_available("ssh", r.location):
1025 log.warning(
1026 f"--r2r={mode} requires ssh on {r.location} host: {r.ssh_user_host or 'localhost'} "
1027 "for remote-to-remote replication; falling back to --r2r=off."
1028 )
1029 return "off"
1031 if mode == "pull" and src.ssh_config_file and src.ssh_config_file != "none":
1032 log.warning(
1033 "--r2r=pull cannot use --ssh-src-config-file for remote-to-remote ssh; cowardly falling back to --r2r=off."
1034 )
1035 return "off"
1037 if mode == "push" and dst.ssh_config_file and dst.ssh_config_file != "none":
1038 log.warning(
1039 "--r2r=push cannot use --ssh-dst-config-file for remote-to-remote ssh; cowardly falling back to --r2r=off."
1040 )
1041 return "off"
1043 return mode
1046def is_same_remote(src: Remote, dst: Remote) -> bool:
1047 """Returns whether this remote is the same as the other remote."""
1048 if src.ssh_user_host == dst.ssh_user_host and not src.ssh_user_host:
1049 return True # both are local
1050 return (
1051 src.ssh_user_host == dst.ssh_user_host
1052 and src.ssh_port == dst.ssh_port
1053 and src.ssh_config_file == dst.ssh_config_file
1054 )