Coverage for bzfs_main/configuration.py: 99%
700 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-19 18:44 +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 Mapping,
35)
36from dataclasses import (
37 dataclass,
38)
39from datetime import (
40 datetime,
41 tzinfo,
42)
43from logging import (
44 Logger,
45)
46from typing import (
47 Final,
48 Literal,
49 NamedTuple,
50 cast,
51 final,
52)
54from bzfs_main.argparse_actions import (
55 SnapshotFilter,
56 optimize_snapshot_filters,
57)
58from bzfs_main.argparse_cli import (
59 LOG_DIR_DEFAULT,
60 ZFS_RECV_GROUPS,
61 ZFS_RECV_O,
62 ZFS_RECV_O_INCLUDE_REGEX_DEFAULT,
63 __version__,
64)
65from bzfs_main.detect import (
66 DISABLE_PRG,
67)
68from bzfs_main.filter import (
69 SNAPSHOT_FILTERS_VAR,
70)
71from bzfs_main.period_anchors import (
72 PeriodAnchors,
73)
74from bzfs_main.util import (
75 utils,
76)
77from bzfs_main.util.connection import (
78 ConnectionPools,
79 get_ssh_config_file_hash,
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 SynchronizedDict,
95 append_if_absent,
96 compile_regexes,
97 current_datetime,
98 die,
99 get_home_directory,
100 get_timezone,
101 getenv_bool,
102 getenv_int,
103 is_included,
104 ninfix,
105 nprefix,
106 nsuffix,
107 open_nofollow,
108 parse_duration_to_milliseconds,
109 pid_exists,
110 sha256_hex,
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:
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:
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_cache: Final[SynchronizedDict[str, str]] = SynchronizedDict({})
537 self.ssh_cipher: Final[str] = p.validate_arg_str(args.ssh_cipher)
538 # disable interactive password prompts and X11 forwarding and pseudo-terminal allocation:
539 ssh_extra_opts: list[str] = ["-oBatchMode=yes", "-oServerAliveInterval=0", "-x", "-T"] + (
540 ["-v"] if args.verbose >= 3 else []
541 )
542 self.ssh_extra_opts: tuple[str, ...] = tuple(ssh_extra_opts)
543 self.max_concurrent_ssh_sessions_per_tcp_connection: Final[int] = args.max_concurrent_ssh_sessions_per_tcp_connection
544 self.ssh_exit_on_shutdown: bool = args.ssh_exit_on_shutdown
545 self.ssh_control_persist_secs: int = args.ssh_control_persist_secs
546 self.ssh_control_persist_margin_secs: int = getenv_int("ssh_control_persist_margin_secs", 2)
547 self.socket_prefix: Final[str] = "s"
548 self.reuse_ssh_connection: bool = getenv_bool("reuse_ssh_connection", True)
549 self.ssh_socket_dir: str = ""
550 if self.reuse_ssh_connection:
551 ssh_home_dir: str = os.path.join(get_home_directory(), ".ssh")
552 os.makedirs(ssh_home_dir, mode=DIR_PERMISSIONS, exist_ok=True)
553 self.ssh_socket_dir = os.path.join(ssh_home_dir, "bzfs")
554 os.makedirs(self.ssh_socket_dir, mode=DIR_PERMISSIONS, exist_ok=True)
555 validate_file_permissions(self.ssh_socket_dir, mode=DIR_PERMISSIONS)
556 self.ssh_exit_on_shutdown_socket_dir: Final[str] = os.path.join(self.ssh_socket_dir, "x")
557 os.makedirs(self.ssh_exit_on_shutdown_socket_dir, mode=DIR_PERMISSIONS, exist_ok=True)
558 validate_file_permissions(self.ssh_exit_on_shutdown_socket_dir, mode=DIR_PERMISSIONS)
559 _delete_stale_files(self.ssh_exit_on_shutdown_socket_dir, prefix=self.socket_prefix, ssh=True)
560 self.sanitize1_regex: Final[re.Pattern[str]] = re.compile(r"[\s\\/@$]") # replace whitespace, /, $, \, @ with ~ char
561 self.sanitize2_regex: Final[re.Pattern[str]] = re.compile(rf"[^a-zA-Z0-9{re.escape('~.:_-')}]") # remove bad chars
563 # mutable variables:
564 self.root_dataset: str = "" # deferred until run_main()
565 self.basis_root_dataset: str = "" # deferred until run_main()
566 self.pool: str = ""
567 self.sudo: str = ""
568 self.use_zfs_delegation: bool = False
569 self.ssh_user: str = ""
570 self.ssh_host: str = ""
571 self.ssh_user_host: str = ""
572 self.is_nonlocal: bool = False
574 def local_ssh_command(self, socket_file: str | None) -> tuple[tuple[str, ...], str | None]:
575 """Returns the ssh CLI command to run locally in order to talk to the remote host; This excludes the (trailing)
576 command to run on the remote host, which will be appended later; also returns the effective ControlPath used by the
577 ssh CLI command, or ``None`` when SSH multiplexing is not active."""
578 if not self.ssh_user_host:
579 return (), None # dataset is on local host - don't use ssh
581 # dataset is on remote host
582 p: Params = self.params
583 if p.ssh_program == DISABLE_PRG:
584 die("Cannot talk to remote host because ssh CLI is disabled.")
585 ssh_cmd: list[str] = [p.ssh_program]
586 ssh_cmd += self.ssh_extra_opts
587 if self.ssh_config_file:
588 ssh_cmd += ["-F", self.ssh_config_file]
589 if self.ssh_cipher: 589 ↛ 591line 589 didn't jump to line 591 because the condition on line 589 was always true
590 ssh_cmd += ["-c", self.ssh_cipher]
591 if self.ssh_port:
592 ssh_cmd += ["-p", str(self.ssh_port)]
594 socket_path: str | None = None
595 if self.reuse_ssh_connection:
596 # Performance: reuse ssh connection for low latency startup of frequent ssh invocations via the 'ssh -S' and
597 # 'ssh -S -M -oControlPersist=60s' options. See https://en.wikibooks.org/wiki/OpenSSH/Cookbook/Multiplexing
598 if socket_file:
599 socket_path = socket_file
600 else:
601 # Generate unique private Unix domain socket file name in user's home dir and pass it to 'ssh -S /path/to/socket'
602 def sanitize(name: str) -> str:
603 name = self.sanitize1_regex.sub("~", name) # replace whitespace, /, $, \, @ with a ~ tilde char
604 name = self.sanitize2_regex.sub("", name) # Remove disallowed chars
605 return name
607 max_rand: int = 999_999_999_999
608 rand_str: str = urlsafe_base64(random.SystemRandom().randint(0, max_rand), max_value=max_rand, padding=False)
609 curr_time: str = urlsafe_base64(time.time_ns(), max_value=2**64 - 1, padding=False)
610 unique: str = f"{os.getpid()}@{curr_time}@{rand_str}"
611 optional: str = f"@{sanitize(self.ssh_host)[:45]}@{sanitize(self.ssh_user)}"
612 socket_name: str = f"{self.socket_prefix}{unique}{optional}"
613 socket_path = os.path.join(self.ssh_exit_on_shutdown_socket_dir, socket_name)
614 socket_path = socket_path[: max(UNIX_DOMAIN_SOCKET_PATH_MAX_LENGTH, len(socket_path) - len(optional))]
615 # `ssh` will error out later if the max OS Unix domain socket path limit cannot be met reasonably as the
616 # home directory path is too long, typically because the Unix user name is unreasonably long.
617 ssh_cmd += ["-S", socket_path]
618 ssh_cmd += [self.ssh_user_host]
619 return tuple(ssh_cmd), socket_path
621 def cache_key(self) -> tuple[str, str, int | None, str | None]:
622 """Returns tuple uniquely identifying this Remote for caching."""
623 return self.location, self.ssh_user_host, self.ssh_port, self.ssh_config_file
625 def cache_namespace(self) -> str:
626 """Returns cache namespace string which is a stable, unique directory component for caches that distinguishes
627 endpoints by username+host+port+ssh_config_file where applicable, and uses '-' when no user/host is present (local
628 mode)."""
629 if not self.ssh_user_host:
630 return "-" # local mode
631 config_hash: str | None = self._ssh_config_file_hash_cache.get(self.ssh_user_host)
632 if config_hash is None:
633 config_hash = get_ssh_config_file_hash(self, self.ssh_config_file)
634 self._ssh_config_file_hash_cache[self.ssh_user_host] = config_hash
635 return f"{self.ssh_user_host}#{self.ssh_port or ''}#{config_hash}"
637 def is_ssh_available(self) -> bool:
638 """Return True if the ssh client program required for this remote is available on the local host."""
639 return self.params.is_program_available("ssh", "local")
641 def __repr__(self) -> str:
642 return str(self.__dict__)
645#############################################################################
646@final
647class CopyPropertiesConfig:
648 """--zfs-recv-o* and --zfs-recv-x* option groups for copying or excluding ZFS properties on receive."""
650 def __init__(self, group: str, flag: str, args: argparse.Namespace, p: Params) -> None:
651 """Reads from ArgumentParser via args."""
652 assert group in ZFS_RECV_GROUPS
653 # immutable variables:
654 grup: str = group
655 self.group: Final[str] = group # one of zfs_recv_o, zfs_recv_x
656 self.flag: Final[str] = flag # one of -o or -x
657 sources: str = p.validate_arg_str(getattr(args, f"{grup}_sources"))
658 self.sources: Final[str] = ",".join(sorted([s.strip() for s in sources.strip().split(",")])) # canonicalize
659 self.targets: Final[str] = p.validate_arg_str(getattr(args, f"{grup}_targets"))
660 include_regexes: list[str] | None = getattr(args, f"{grup}_include_regex")
661 assert ZFS_RECV_O in ZFS_RECV_GROUPS
662 if include_regexes is None:
663 include_regexes = [ZFS_RECV_O_INCLUDE_REGEX_DEFAULT] if grup == ZFS_RECV_O else []
664 self.include_regexes: Final[RegexList] = compile_regexes(include_regexes)
665 self.exclude_regexes: Final[RegexList] = compile_regexes(getattr(args, f"{grup}_exclude_regex"))
667 def __repr__(self) -> str:
668 return str(self.__dict__)
671#############################################################################
672@final
673class SnapshotLabel(NamedTuple):
674 """Contains the individual parts that are concatenated into a ZFS snapshot name."""
676 prefix: str # bzfs_
677 infix: str # us-west_
678 timestamp: str # 2024-11-06_08:30:05
679 suffix: str # _hourly
681 def __str__(self) -> str: # bzfs_us-west_2024-11-06_08:30:05_hourly
682 return f"{self.prefix}{self.infix}{self.timestamp}{self.suffix}"
684 def notimestamp_str(self) -> str: # bzfs_,us-west_,_hourly
685 """Returns the concatenation of all parts except for the timestamp part, with parts separated by a comma."""
686 return f"{self.prefix},{self.infix},{self.suffix}"
688 def validate_label(self, input_text: str) -> None:
689 """Validates that the composed snapshot label forms a legal name."""
690 name: str = str(self)
691 validate_dataset_name(name, input_text)
692 if "/" in name:
693 die(f"Invalid ZFS snapshot name: '{name}' for: '{input_text}'")
694 for key, value in {"prefix": self.prefix, "infix": self.infix, "suffix": self.suffix}.items():
695 if key == "prefix":
696 if not value.endswith("_"):
697 die(f"Invalid {input_text}{key}: Must end with an underscore character: '{value}'")
698 if value.count("_") > 1:
699 die(f"Invalid {input_text}{key}: Must not contain multiple underscore characters: '{value}'")
700 elif key == "infix":
701 if value:
702 if not value.endswith("_"):
703 die(f"Invalid {input_text}{key}: Must end with an underscore character: '{value}'")
704 if value.count("_") > 1:
705 die(f"Invalid {input_text}{key}: Must not contain multiple underscore characters: '{value}'")
706 elif value:
707 if not value.startswith("_"):
708 die(f"Invalid {input_text}{key}: Must start with an underscore character: '{value}'")
709 if value.count("_") > 1:
710 die(f"Invalid {input_text}{key}: Must not contain multiple underscore characters: '{value}'")
713#############################################################################
714WRITTEN_THRESHOLD: Final[str] = "written_threshold"
717@dataclass(frozen=True)
718@final
719class SnapshotLabelAttributes:
720 """Configuration parameters for a SnapshotLabel, with shared attribute names excluded from dataclass fields."""
722 written_threshold: int
724 @classmethod
725 def parse(cls, attributes: dict) -> SnapshotLabelAttributes:
726 """Parses the given configuration dict into a typed object."""
727 written_threshold: int = 0
728 for name, value in attributes.items():
729 if name == WRITTEN_THRESHOLD:
730 if not isinstance(value, int) or value < 0:
731 die(f"--create-src-snapshots-plan-attributes: {WRITTEN_THRESHOLD} must be a non-negative int: {value}")
732 written_threshold = value
733 else:
734 die(f"--create-src-snapshots-plan-attributes: Invalid attribute name: '{name}'")
735 return SnapshotLabelAttributes(written_threshold=written_threshold)
738#############################################################################
739@final
740class CreateSrcSnapshotConfig:
741 """Option values for --create-src-snapshots, that is, for automatically creating source snapshots."""
743 def __init__(self, args: argparse.Namespace, p: Params) -> None:
744 """Option values for --create-src-snapshots*; reads from ArgumentParser via args."""
745 # immutable variables:
746 self.skip_create_src_snapshots: Final[bool] = not args.create_src_snapshots
747 self.create_src_snapshots_even_if_not_due: Final[bool] = args.create_src_snapshots_even_if_not_due
748 tz_spec: str | None = args.create_src_snapshots_timezone if args.create_src_snapshots_timezone else None
749 self.tz: Final[tzinfo | None] = get_timezone(tz_spec)
750 self.current_datetime: datetime = current_datetime(tz_spec)
751 self.timeformat: Final[str] = args.create_src_snapshots_timeformat
752 self.anchors: Final[PeriodAnchors] = PeriodAnchors.parse(args)
754 # Compute the schedule for upcoming periodic time events (suffix_durations). This event schedule is also used in
755 # daemon mode via sleep_until_next_daemon_iteration()
756 labels: list[SnapshotLabel] = []
757 snapshot_label_attributes: dict[str, SnapshotLabelAttributes] = {}
758 create_src_snapshots_plan: str = args.create_src_snapshots_plan or str({"bzfs": {"onsite": {"adhoc": 1}}})
759 for org, target_periods in ast.literal_eval(create_src_snapshots_plan).items():
760 for target, periods in target_periods.items():
761 for period_unit, period_amount in periods.items(): # e.g. period_unit can be "10minutely" or "minutely"
762 if not isinstance(period_amount, int) or period_amount < 0:
763 die(f"--create-src-snapshots-plan: Period amount must be a non-negative integer: {period_amount}")
764 lbl = SnapshotLabel(prefix=nprefix(org), infix=ninfix(target), timestamp="", suffix=nsuffix(period_unit))
765 if period_amount > 0:
766 labels.append(lbl)
767 snapshot_label_attributes[lbl.notimestamp_str()] = SnapshotLabelAttributes.parse({})
768 create_src_snapshots_plan_attributes: str = args.create_src_snapshots_plan_attributes or str({})
769 for org, target_periods in ast.literal_eval(create_src_snapshots_plan_attributes).items():
770 for target, periods in target_periods.items():
771 for period_unit, period_attributes in periods.items(): # e.g. period_unit can be "10minutely" or "minutely"
772 lbl = SnapshotLabel(prefix=nprefix(org), infix=ninfix(target), timestamp="", suffix=nsuffix(period_unit))
773 snapshot_label_attributes[lbl.notimestamp_str()] = SnapshotLabelAttributes.parse(period_attributes)
774 self.snapshot_label_attributes: Final[Mapping[str, SnapshotLabelAttributes]] = snapshot_label_attributes
775 suffixes: list[str] = list({label.suffix for label in labels}) # dedupe
776 xperiods: SnapshotPeriods = p.xperiods
777 if self.skip_create_src_snapshots:
778 duration_amount, duration_unit = xperiods.suffix_to_duration0(p.daemon_frequency)
779 if duration_amount <= 0 or not duration_unit:
780 die(f"Invalid --daemon-frequency: {p.daemon_frequency}")
781 suffixes = [nsuffix(p.daemon_frequency)]
782 labels = []
783 suffix_durations: dict[str, tuple[int, str]] = {suffix: xperiods.suffix_to_duration1(suffix) for suffix in suffixes}
785 def suffix_key(suffix: str) -> tuple[int, str]:
786 duration_amount, duration_unit = suffix_durations[suffix]
787 duration_milliseconds: int = duration_amount * xperiods.suffix_milliseconds.get(duration_unit, 0)
788 if suffix.endswith(("hourly", "minutely", "secondly")):
789 if duration_milliseconds != 0 and 86400 * 1000 % duration_milliseconds != 0:
790 die(
791 "Invalid --create-src-snapshots-plan: Period duration should be a divisor of 86400 seconds "
792 f"without remainder so that snapshots will be created at the same time of day every day: {suffix}"
793 )
794 if suffix.endswith("monthly"):
795 if duration_amount != 0 and 12 % duration_amount != 0:
796 die(
797 "Invalid --create-src-snapshots-plan: Period duration should be a divisor of 12 months "
798 f"without remainder so that snapshots will be created at the same time every year: {suffix}"
799 )
800 return duration_milliseconds, suffix
802 suffixes.sort(key=suffix_key, reverse=True) # take snapshots for dailies before hourlies, and so on
803 self.suffix_durations: Final[dict[str, tuple[int, str]]] = {sfx: suffix_durations[sfx] for sfx in suffixes} # sort
804 suffix_indexes: dict[str, int] = {suffix: k for k, suffix in enumerate(suffixes)}
805 labels.sort(key=lambda label: (suffix_indexes[label.suffix], label)) # take snapshots for dailies before hourlies
806 self._snapshot_labels: Final[list[SnapshotLabel]] = labels
807 for label in self.snapshot_labels():
808 label.validate_label("--create-src-snapshots-plan ")
809 if (not label.infix) and not self.timeformat.startswith("%Y"):
810 die(f"--create-src-snapshots-timeformat: Must start with '%Y' (four-digit year): {self.timeformat!r}")
812 def snapshot_labels(self) -> list[SnapshotLabel]:
813 """Returns the snapshot name patterns for which snapshots shall be created."""
814 timeformat: str = self.timeformat
815 is_millis: bool = timeformat.endswith("%F") # non-standard hack to append milliseconds
816 if is_millis:
817 timeformat = timeformat[0:-1] + "f" # replace %F with %f (append microseconds)
818 timestamp: str = self.current_datetime.strftime(timeformat)
819 if is_millis:
820 timestamp = timestamp[: -len("000")] # replace microseconds with milliseconds
821 timestamp = timestamp.replace("+", "z") # zfs CLI does not accept the '+' character in snapshot names
822 return [SnapshotLabel(label.prefix, label.infix, timestamp, label.suffix) for label in self._snapshot_labels]
824 def __repr__(self) -> str:
825 return str(self.__dict__)
828#############################################################################
829@dataclass(frozen=True)
830@final
831class AlertConfig:
832 """Thresholds controlling when alerts fire for snapshot age."""
834 kind: Literal["Latest", "Oldest"]
835 warning_millis: int
836 critical_millis: int
839#############################################################################
840@dataclass(frozen=True)
841@final
842class MonitorSnapshotAlert:
843 """Alert configuration for a single monitored snapshot label."""
845 label: SnapshotLabel
846 latest: AlertConfig | None
847 oldest: AlertConfig | None
848 oldest_skip_holds: bool
851#############################################################################
852@final
853class MonitorSnapshotsConfig:
854 """Option values for --monitor-snapshots*, that is, policy describing which snapshots to monitor for staleness."""
856 def __init__(self, args: argparse.Namespace, p: Params) -> None:
857 """Reads from ArgumentParser via args."""
858 # immutable variables:
859 self.monitor_snapshots: Final[dict] = ast.literal_eval(args.monitor_snapshots)
860 self.dont_warn: Final[bool] = args.monitor_snapshots_dont_warn
861 self.dont_crit: Final[bool] = args.monitor_snapshots_dont_crit
862 self.no_latest_check: Final[bool] = args.monitor_snapshots_no_latest_check
863 self.no_oldest_check: Final[bool] = args.monitor_snapshots_no_oldest_check
864 alerts: list[MonitorSnapshotAlert] = []
865 xperiods: SnapshotPeriods = p.xperiods
866 for org, target_periods in self.monitor_snapshots.items():
867 prefix: str = nprefix(org)
868 for target, periods in target_periods.items():
869 for period_unit, alert_dicts in periods.items(): # e.g. period_unit can be "10minutely" or "minutely"
870 label = SnapshotLabel(prefix=prefix, infix=ninfix(target), timestamp="", suffix=nsuffix(period_unit))
871 alert_latest, alert_oldest = None, None
872 oldest_skip_holds: bool = False
873 for alert_type, alert_dict in alert_dicts.items():
874 m = "--monitor-snapshots: "
875 if alert_type not in ["latest", "oldest"]:
876 die(f"{m}'{alert_type}' must be 'latest' or 'oldest' within {args.monitor_snapshots}")
877 warning_millis: int = 0
878 critical_millis: int = 0
879 cycles: int = 1
880 for kind, value in alert_dict.items():
881 context: str = args.monitor_snapshots
882 if kind == "warning":
883 warning_millis = max(0, parse_duration_to_milliseconds(str(value), context=context))
884 elif kind == "critical":
885 critical_millis = max(0, parse_duration_to_milliseconds(str(value), context=context))
886 elif kind == "cycles":
887 cycles = max(0, int(value))
888 elif kind == "oldest_skip_holds" and alert_type == "oldest":
889 if not isinstance(value, bool):
890 die(f"{m}'{kind}' must be a bool within {context}")
891 oldest_skip_holds = value
892 else:
893 die(f"{m}'{kind}' must be 'warning', 'critical' or 'cycles' within {context}")
894 if warning_millis > 0 or critical_millis > 0:
895 duration_amount, duration_unit = xperiods.suffix_to_duration1(label.suffix)
896 duration_milliseconds: int = duration_amount * xperiods.suffix_milliseconds.get(duration_unit, 0)
897 warning_millis += 0 if warning_millis <= 0 else cycles * duration_milliseconds
898 critical_millis += 0 if critical_millis <= 0 else cycles * duration_milliseconds
899 warning_millis = UNIX_TIME_INFINITY_SECS if warning_millis <= 0 else warning_millis
900 critical_millis = UNIX_TIME_INFINITY_SECS if critical_millis <= 0 else critical_millis
901 capitalized_alert_type = cast(Literal["Latest", "Oldest"], sys.intern(alert_type.capitalize()))
902 alert_config = AlertConfig(capitalized_alert_type, warning_millis, critical_millis)
903 if alert_type == "latest":
904 if not self.no_latest_check:
905 alert_latest = alert_config
906 else:
907 assert alert_type == "oldest"
908 if not self.no_oldest_check:
909 alert_oldest = alert_config
910 if alert_latest is not None or alert_oldest is not None:
911 alerts.append(MonitorSnapshotAlert(label, alert_latest, alert_oldest, oldest_skip_holds))
913 def alert_sort_key(alert: MonitorSnapshotAlert) -> tuple[int, SnapshotLabel]:
914 duration_amount, duration_unit = xperiods.suffix_to_duration1(alert.label.suffix)
915 duration_milliseconds: int = duration_amount * xperiods.suffix_milliseconds.get(duration_unit, 0)
916 return duration_milliseconds, alert.label
918 alerts.sort(key=alert_sort_key, reverse=True) # check snapshots for dailies before hourlies, and so on
919 self.alerts: Final[list[MonitorSnapshotAlert]] = alerts
920 self.enable_monitor_snapshots: Final[bool] = len(alerts) > 0
922 def __repr__(self) -> str:
923 return str(self.__dict__)
926#############################################################################
927def _fix_send_recv_opts(
928 opts: list[str],
929 *,
930 exclude_long_opts: set[str],
931 exclude_short_opts: str,
932 include_arg_opts: set[str],
933 exclude_arg_opts: frozenset[str] = frozenset(),
934 preserve_properties: frozenset[str] = frozenset(),
935) -> tuple[list[str], list[str]]:
936 """These opts are instead managed via bzfs CLI args --dryrun, etc."""
937 assert "-" not in exclude_short_opts
938 results: list[str] = []
939 x_names: set[str] = set(preserve_properties)
941 def _append_include_arg(opt: str, arg: str) -> None:
942 if opt == "-o" and "=" in arg and arg.split("=", 1)[0] in preserve_properties:
943 die(f"--preserve-properties: Disallowed ZFS property found in --zfs-recv-program-opt(s): -o {arg}")
944 if opt == "-x":
945 x_names.discard(arg)
946 results.append(opt)
947 results.append(arg)
949 def _is_compact_exclude_arg(opt: str) -> bool:
950 # Compact option forms such as -isnap and --resume=<resume-token> carry the managed argument inline.
951 for exclude_arg_opt in exclude_arg_opts:
952 if len(exclude_arg_opt) == 2 and opt.startswith(exclude_arg_opt) and len(opt) > len(exclude_arg_opt):
953 return True
954 if exclude_arg_opt.startswith("--") and opt.startswith(exclude_arg_opt + "="):
955 return True
956 return False
958 i = 0
959 n = len(opts)
960 while i < n:
961 opt: str = opts[i]
962 i += 1
963 if opt in exclude_arg_opts: # example: {"-X", "--exclude"}
964 i += 1
965 continue
966 elif _is_compact_exclude_arg(opt):
967 continue
968 elif opt in include_arg_opts: # example: {"-o", "-x"}
969 if i < n:
970 _append_include_arg(opt, opts[i])
971 i += 1
972 else:
973 results.append(opt)
974 elif opt not in exclude_long_opts: # example: {"--dryrun", "--verbose"}
975 for include_arg_opt in include_arg_opts:
976 if len(include_arg_opt) == 2 and opt.startswith(include_arg_opt) and len(opt) > len(include_arg_opt):
977 _append_include_arg(include_arg_opt, opt[len(include_arg_opt) :])
978 break
979 else:
980 if opt.startswith("-") and opt != "-" and not opt.startswith("--"):
981 if "=" in opt:
982 die(f"Invalid short opt {opt!r} within {opts}")
983 for char in exclude_short_opts: # example: "den"
984 opt = opt.replace(char, "")
985 if opt == "-":
986 continue
987 results.append(opt)
988 return results, sorted(x_names)
991_SSH_MASTER_DOMAIN_SOCKET_FILE_PID_REGEX: Final[re.Pattern[str]] = re.compile(r"^[0-9]+") # see local_ssh_command()
994def _delete_stale_files(
995 root_dir: str,
996 *,
997 prefix: str,
998 millis: int = 60 * 60 * 1000,
999 dirs: bool = False,
1000 exclude: str | None = None,
1001 ssh: bool = False,
1002) -> None:
1003 """Cleans up obsolete files; For example caused by abnormal termination, OS crash."""
1004 seconds: float = millis / 1000
1005 now: float = time.time()
1006 validate_is_not_a_symlink("", root_dir)
1007 with os.scandir(root_dir) as iterator:
1008 for entry in iterator:
1009 if entry.name == exclude or not entry.name.startswith(prefix):
1010 continue
1011 try:
1012 stats = entry.stat(follow_symlinks=False)
1013 is_dir = entry.is_dir(follow_symlinks=False)
1014 if ((dirs and is_dir) or (not dirs and not is_dir)) and now - stats.st_mtime >= seconds:
1015 if dirs:
1016 shutil.rmtree(entry.path, ignore_errors=True)
1017 elif not (ssh and stat.S_ISSOCK(stats.st_mode)):
1018 os.remove(entry.path)
1019 elif match := _SSH_MASTER_DOMAIN_SOCKET_FILE_PID_REGEX.match(entry.name[len(prefix) :]): 1019 ↛ 1008line 1019 didn't jump to line 1008 because the condition on line 1019 was always true
1020 pid: int = int(match.group(0))
1021 if pid_exists(pid) is False or now - stats.st_mtime >= 31 * 24 * 60 * 60:
1022 os.remove(entry.path) # bzfs process is no longer alive; its ssh master process isn't either
1023 except FileNotFoundError:
1024 pass # harmless
1027def _create_symlink(src: str, dst_dir: str, dst: str) -> None:
1028 """Creates dst symlink pointing to src using a relative path."""
1029 rel_path: str = os.path.relpath(src, start=dst_dir)
1030 os.symlink(src=rel_path, dst=os.path.join(dst_dir, dst))
1033def resolve_r2r_mode(p: Params) -> str:
1034 """Returns the effective r2r mode for the current task and emits fallback warnings."""
1035 src, dst = p.src, p.dst
1036 log = p.log
1037 mode: str = p.r2r_mode_requested
1038 assert mode in ("off", "pull", "push"), mode
1040 if p.skip_replication:
1041 return "off"
1043 if mode == "off":
1044 return "off"
1046 if (not src.ssh_user_host) and (not dst.ssh_user_host):
1047 return "off" # we'll do local replication (there's no need for r2r)
1049 if (not src.is_nonlocal) or (not dst.is_nonlocal):
1050 return "off" # at least one of them is local to this host (there's no need for r2r)
1052 r: Remote = dst if mode == "pull" else src
1053 if not p.is_program_available("sh", r.location):
1054 log.warning(
1055 f"--r2r={mode} requires sh on {r.location} host: {r.ssh_user_host or 'localhost'} for remote-to-remote "
1056 "replication; falling back to --r2r=off."
1057 )
1058 return "off"
1060 if is_same_remote(src, dst): # are user@host, port, ssh_config file the same?
1061 return mode # perf: we'll do local replication on that host
1063 if not p.is_program_available("ssh", r.location):
1064 log.warning(
1065 f"--r2r={mode} requires ssh on {r.location} host: {r.ssh_user_host or 'localhost'} "
1066 "for remote-to-remote replication; falling back to --r2r=off."
1067 )
1068 return "off"
1070 if mode == "pull" and src.ssh_config_file and src.ssh_config_file != "none":
1071 log.warning(
1072 "--r2r=pull cannot use --ssh-src-config-file for remote-to-remote ssh; cowardly falling back to --r2r=off."
1073 )
1074 return "off"
1076 if mode == "push" and dst.ssh_config_file and dst.ssh_config_file != "none":
1077 log.warning(
1078 "--r2r=push cannot use --ssh-dst-config-file for remote-to-remote ssh; cowardly falling back to --r2r=off."
1079 )
1080 return "off"
1082 return mode
1085def is_same_remote(src: Remote, dst: Remote) -> bool:
1086 """Returns whether this remote is the same as the other remote."""
1087 if src.ssh_user_host == dst.ssh_user_host and not src.ssh_user_host:
1088 return True # both are local
1089 return (
1090 src.ssh_user_host == dst.ssh_user_host
1091 and src.ssh_port == dst.ssh_port
1092 and src.ssh_config_file == dst.ssh_config_file
1093 )