Coverage for bzfs_main/bzfs.py: 99%
1110 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.
15# Inline script metadata conforming to https://packaging.python.org/specifications/inline-script-metadata
16# /// script
17# requires-python = ">=3.9"
18# dependencies = []
19# ///
20#
21"""
22* Main CLI entry point for replicating and managing ZFS snapshots. It handles the low-level mechanics of
23 `zfs send/receive`, data transfer, and snapshot management between two hosts.
24* Overview of the bzfs.py codebase:
25* The codebase starts with docs, definition of input data and associated argument parsing into a "Params" class.
26* All CLI option/parameter values are reachable from the "Params" class.
27* Control flow starts in main(), which kicks off a "Job".
28* A Job runs one or more "tasks" via run_tasks(), each task replicating a separate dataset tree.
29* The core replication algorithm is in run_task() and especially in replicate_datasets() and replicate_dataset().
30* The filter algorithms that apply include/exclude policies are in filter_datasets() and filter_snapshots().
31* The --create-src-snapshots-* and --delete-* and --compare-* and --monitor-* algorithms also start in run_task().
32* The main retry logic is in call_with_retries() and clear_resumable_recv_state_if_necessary().
33* Progress reporting for use during `zfs send/recv` data transfers is in class ProgressReporter.
34* Executing a CLI command on a local or remote host is in run_ssh_command().
35* Network connection management is in refresh_ssh_connection_if_necessary() and class ConnectionPool.
36* Cache functionality can be found by searching for this regex: .*cach.*
37* The parallel processing engine is in itr_ssh_cmd_parallel() and process_datasets_in_parallel_and_fault_tolerant().
38* README.md is mostly auto-generated from the ArgumentParser help texts as the source of "truth", via update_readme.sh.
39 Simply run that script whenever you change or add ArgumentParser help text.
40"""
42from __future__ import (
43 annotations,
44)
45import argparse
46import contextlib
47import fcntl
48import heapq
49import itertools
50import logging
51import os
52import re
53import subprocess
54import sys
55import threading
56import time
57from collections import (
58 Counter,
59 defaultdict,
60)
61from collections.abc import (
62 Collection,
63 Sequence,
64)
65from datetime import (
66 datetime,
67 timedelta,
68)
69from logging import (
70 Logger,
71)
72from pathlib import (
73 Path,
74)
75from subprocess import (
76 DEVNULL,
77 PIPE,
78 CalledProcessError,
79 TimeoutExpired,
80)
81from typing import (
82 Any,
83 Callable,
84 Final,
85 cast,
86 final,
87)
89import bzfs_main.loggers
90from bzfs_main.argparse_actions import (
91 has_timerange_filter,
92)
93from bzfs_main.argparse_cli import (
94 EXCLUDE_DATASET_REGEXES_DEFAULT,
95)
96from bzfs_main.compare_snapshot_lists import (
97 run_compare_snapshot_lists,
98)
99from bzfs_main.configuration import (
100 AlertConfig,
101 CreateSrcSnapshotConfig,
102 LogParams,
103 MonitorSnapshotAlert,
104 Params,
105 Remote,
106 SnapshotLabel,
107 resolve_r2r_mode,
108)
109from bzfs_main.detect import (
110 DISABLE_PRG,
111 RemoteConfCacheItem,
112 are_bookmarks_enabled,
113 detect_available_programs,
114 is_caching_snapshots,
115 is_dummy,
116 is_zpool_feature_enabled_or_active,
117)
118from bzfs_main.filter import (
119 SNAPSHOT_REGEX_FILTER_NAME,
120 dataset_regexes,
121 filter_datasets,
122 filter_lines,
123 filter_lines_except,
124 filter_snapshots,
125)
126from bzfs_main.loggers import (
127 get_simple_logger,
128 reset_logger,
129 set_logging_runtime_defaults,
130)
131from bzfs_main.parallel_batch_cmd import (
132 run_ssh_cmd_parallel,
133 zfs_list_snapshots_in_parallel,
134)
135from bzfs_main.progress_reporter import (
136 ProgressReporter,
137 count_num_bytes_transferred_by_zfs_send,
138)
139from bzfs_main.replication import (
140 delete_bookmarks,
141 delete_datasets,
142 delete_snapshots,
143 is_tmp_bookmark,
144 replicate_dataset,
145)
146from bzfs_main.snapshot_cache import (
147 MATURITY_TIME_THRESHOLD_SECS,
148 MONITOR_CACHE_FILE_PREFIX,
149 REPLICATION_CACHE_FILE_PREFIX,
150 SnapshotCache,
151 set_last_modification_time_safe,
152)
153from bzfs_main.util.connection import (
154 SHARED,
155 ConnectionPool,
156 MiniJob,
157 MiniRemote,
158 timeout,
159)
160from bzfs_main.util.parallel_iterator import (
161 run_in_parallel,
162)
163from bzfs_main.util.parallel_tasktree_policy import (
164 process_datasets_in_parallel_and_fault_tolerant,
165)
166from bzfs_main.util.retry import (
167 Retry,
168 RetryableError,
169 RetryTemplate,
170 RetryTerminationError,
171 RetryTiming,
172)
173from bzfs_main.util.utils import (
174 DESCENDANTS_RE_SUFFIX,
175 DIE_STATUS,
176 DONT_SKIP_DATASET,
177 FILE_PERMISSIONS,
178 LOG_DEBUG,
179 LOG_TRACE,
180 PROG_NAME,
181 SHELL_CHARS_AND_SLASH,
182 UMASK,
183 YEAR_WITH_FOUR_DIGITS_REGEX,
184 HashedInterner,
185 SortedInterner,
186 Subprocesses,
187 SynchronizedBool,
188 SynchronizedDict,
189 TaskTiming,
190 append_if_absent,
191 compile_regexes,
192 cut,
193 die,
194 has_duplicates,
195 human_readable_bytes,
196 human_readable_duration,
197 is_descendant,
198 percent,
199 pretty_print_formatter,
200 replace_in_lines,
201 replace_prefix,
202 sha256_85_urlsafe_base64,
203 sha256_128_urlsafe_base64,
204 stderr_to_str,
205 termination_signal_handler,
206 validate_dataset_name,
207 validate_property_name,
208 xappend,
209 xfinally,
210 xprint,
211)
213# constants:
214__version__: Final[str] = bzfs_main.argparse_cli.__version__
215CRITICAL_STATUS: Final[int] = 2
216WARNING_STATUS: Final[int] = 1
217STILL_RUNNING_STATUS: Final[int] = 4
218MIN_PYTHON_VERSION: Final[tuple[int, int]] = (3, 9)
219if sys.version_info < MIN_PYTHON_VERSION:
220 print(f"ERROR: {PROG_NAME} requires Python version >= {'.'.join(map(str, MIN_PYTHON_VERSION))}!")
221 sys.exit(DIE_STATUS)
224#############################################################################
225def argument_parser() -> argparse.ArgumentParser:
226 """Returns the CLI parser used by bzfs."""
227 return bzfs_main.argparse_cli.argument_parser()
230def main() -> None:
231 """API for command line clients."""
232 prev_umask: int = os.umask(UMASK)
233 try:
234 set_logging_runtime_defaults()
235 # On CTRL-C and SIGTERM, send signal to all descendant processes to terminate them
236 termination_event: threading.Event = threading.Event()
237 with termination_signal_handler(termination_events=[termination_event]):
238 run_main(argument_parser().parse_args(), sys.argv, termination_event=termination_event)
239 except subprocess.CalledProcessError as e:
240 sys.exit(normalize_called_process_error(e))
241 finally:
242 os.umask(prev_umask) # restore prior global state
245def run_main(
246 args: argparse.Namespace,
247 sys_argv: list[str] | None = None,
248 log: Logger | None = None,
249 termination_event: threading.Event | None = None,
250) -> None:
251 """API for Python clients; visible for testing; may become a public API eventually."""
252 Job(termination_event=termination_event).run_main(args, sys_argv, log)
255#############################################################################
256@final
257class Job(MiniJob):
258 """Executes one bzfs run, coordinating snapshot replication tasks."""
260 def __init__(self, termination_event: threading.Event | None = None) -> None:
261 self.params: Params
262 self.termination_event: Final[threading.Event] = termination_event or threading.Event()
263 self.retry_timing: Final[RetryTiming] = RetryTiming.make_from(self.termination_event).copy(
264 on_before_attempt=lambda retry: None
265 )
266 self.task_timing: Final[TaskTiming] = TaskTiming.make_from(self.termination_event)
267 self.subprocesses: Subprocesses = Subprocesses(self.termination_event.is_set)
268 self.all_dst_dataset_exists: Final[dict[str, dict[str, bool]]] = defaultdict(lambda: defaultdict(bool))
269 self.dst_dataset_exists: SynchronizedDict[str, bool] = SynchronizedDict({})
270 self.src_properties: dict[str, DatasetProperties] = {}
271 self.dst_properties: dict[str, DatasetProperties] = {}
272 self.all_exceptions: list[str] = []
273 self.all_exceptions_count: int = 0
274 self.max_exceptions_to_summarize: int = 10000
275 self.first_exception: BaseException | None = None
276 self.remote_conf_cache: dict[tuple, RemoteConfCacheItem] = {}
277 self.max_datasets_per_minibatch_on_list_snaps: dict[str, int] = {}
278 self.max_workers: dict[str, int] = {}
279 self.progress_reporter: ProgressReporter = cast(ProgressReporter, None)
280 self.is_first_replication_task: Final[SynchronizedBool] = SynchronizedBool(True)
281 self.replication_start_time_nanos: int = time.monotonic_ns()
282 self.timeout_nanos: int | None = None # timestamp aka instant in time
283 self.timeout_duration_nanos: int | None = None # duration (not a timestamp); for logging only
284 self.cache: SnapshotCache = SnapshotCache(self)
285 self.stats_lock: Final[threading.Lock] = threading.Lock()
286 self.num_cache_hits: int = 0
287 self.num_cache_misses: int = 0
288 self.num_snapshots_found: int = 0
289 self.num_snapshots_replicated: int = 0
291 self.is_test_mode: bool = False # for testing only
292 self.creation_prefix: str = "" # for testing only
293 self.use_select: bool = False # for testing only
294 self.progress_update_intervals: tuple[float, float] | None = None # for testing only
295 self.error_injection_triggers: dict[str, Counter[str]] = {} # for testing only
296 self.delete_injection_triggers: dict[str, Counter[str]] = {} # for testing only
297 self.param_injection_triggers: dict[str, dict[str, bool | int]] = {} # for testing only
298 self.inject_params: dict[str, bool | int] = {} # for testing only
299 self.injection_lock: threading.Lock = threading.Lock() # for testing only
300 self.max_command_line_bytes: int | None = None # for testing only
302 def shutdown(self) -> None:
303 """Exits any multiplexed ssh sessions that may be leftover."""
304 cache_items: Collection[RemoteConfCacheItem] = self.remote_conf_cache.values()
305 for i, cache_item in enumerate(cache_items):
306 cache_item.connection_pools.shutdown(f"{i + 1}/{len(cache_items)}")
308 def terminate(self) -> None:
309 """Shuts down gracefully; also terminates descendant processes, if any."""
310 with xfinally(self.subprocesses.terminate_process_subtrees):
311 self.shutdown()
313 def _retry_template(self) -> RetryTemplate:
314 p = self.params
315 return RetryTemplate(policy=p.retry_policy.copy(timing=self.retry_timing), log=p.log)
317 def run_main(self, args: argparse.Namespace, sys_argv: list[str] | None = None, log: Logger | None = None) -> None:
318 """Parses CLI arguments, sets up logging, and executes main job loop."""
319 assert isinstance(self.error_injection_triggers, dict)
320 assert isinstance(self.delete_injection_triggers, dict)
321 assert isinstance(self.inject_params, dict)
322 logger_name_suffix: str = ""
324 def _reset_logger() -> None:
325 if logger_name_suffix and log is not None: # reset Logger unless it's a Logger outside of our control
326 reset_logger(log)
328 with xfinally(_reset_logger): # runs _reset_logger() on exit, without masking error raised in body of `with` block
329 try:
330 log_params: LogParams = LogParams(args)
331 logger_name_suffix = "" if log is not None else log_params.logger_name_suffix
332 log = bzfs_main.loggers.get_logger(
333 log_params=log_params, args=args, log=log, logger_name_suffix=logger_name_suffix
334 )
335 log.info("%s", f"Log file is: {log_params.log_file}")
336 except BaseException as e:
337 simple_log: Logger = get_simple_logger(PROG_NAME, logger_name_suffix=logger_name_suffix)
338 try:
339 simple_log.error("Log init: %s", e, exc_info=not isinstance(e, SystemExit))
340 finally:
341 reset_logger(simple_log)
342 raise
344 aux_args: list[str] = []
345 if getattr(args, "include_snapshot_plan", None):
346 aux_args += args.include_snapshot_plan
347 if getattr(args, "delete_dst_snapshots_except_plan", None):
348 aux_args += args.delete_dst_snapshots_except_plan
349 if len(aux_args) > 0:
350 log.info("Auxiliary CLI arguments: %s", " ".join(aux_args))
351 args = argument_parser().parse_args(xappend(aux_args, "--", args.root_dataset_pairs), namespace=args)
353 def log_error_on_exit(error: Any, status_code: Any, exc_info: bool = False) -> None:
354 log.error("%s%s", f"Exiting {PROG_NAME} with status code {status_code}. Cause: ", error, exc_info=exc_info)
356 try:
357 log.info("CLI arguments: %s %s", " ".join(sys_argv or []), f"[uid: {os.getuid()}, euid: {os.geteuid()}]")
358 if self.is_test_mode:
359 log.log(LOG_TRACE, "Parsed CLI arguments: %s", args)
360 self.params = p = Params(args, sys_argv or [], log_params, log, self.inject_params)
361 self.timeout_duration_nanos = p.timeout_duration_nanos
362 lock_file: str = p.lock_file_name()
363 lock_fd = os.open(
364 lock_file, os.O_WRONLY | os.O_TRUNC | os.O_CREAT | os.O_NOFOLLOW | os.O_CLOEXEC, FILE_PERMISSIONS
365 )
366 with xfinally(lambda: os.close(lock_fd)):
367 try:
368 # Acquire an exclusive lock; will raise a BlockingIOError if lock is already held by this process or
369 # another process. The (advisory) lock is auto-released when the process terminates or the fd is
370 # closed.
371 fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB) # LOCK_NB ... non-blocking
372 except BlockingIOError:
373 msg = "Exiting as same previous periodic job is still running without completion yet per "
374 die(msg + lock_file, STILL_RUNNING_STATUS)
376 # xfinally: unlink the lock_file while still holding the flock on its fd - it's a correct and safe
377 # standard POSIX pattern:
378 # - Performing unlink() before close(fd) avoids a race where a subsequent bzfs process could recreate and
379 # lock a fresh inode for the same path between our close() and a later unlink(). In that case, a late
380 # unlink would delete the newer process's lock_file path.
381 # - At this point, critical work is complete; the remaining steps are shutdown mechanics that have no
382 # side effect, so this pattern is correct, safe, and simple.
383 with xfinally(lambda: Path(lock_file).unlink(missing_ok=True)): # don't accumulate stale files
384 try:
385 self.run_tasks() # do the real work
386 except BaseException:
387 self.terminate()
388 raise
389 self.shutdown()
390 with contextlib.suppress(BrokenPipeError):
391 sys.stderr.flush()
392 sys.stdout.flush()
393 except subprocess.CalledProcessError as e:
394 log_error_on_exit(e, e.returncode)
395 raise
396 except SystemExit as e:
397 log_error_on_exit(e, e.code)
398 raise
399 except (subprocess.TimeoutExpired, UnicodeDecodeError) as e:
400 log_error_on_exit(e, DIE_STATUS)
401 raise SystemExit(DIE_STATUS) from e
402 except re.error as e:
403 log_error_on_exit(f"{e} within regex {e.pattern!r}", DIE_STATUS)
404 raise SystemExit(DIE_STATUS) from e
405 except BaseException as e:
406 log_error_on_exit(e, DIE_STATUS, exc_info=True)
407 raise SystemExit(DIE_STATUS) from e
408 finally:
409 log.info("%s", f"Log file was: {log_params.log_file}")
410 log.info("Success. Goodbye!")
411 with contextlib.suppress(BrokenPipeError):
412 sys.stderr.flush()
413 sys.stdout.flush()
415 def run_tasks(self) -> None:
416 """Executes replication cycles, repeating until daemon lifetime expires."""
417 p, log = self.params, self.params.log
418 self.all_exceptions = []
419 self.all_exceptions_count = 0
420 self.first_exception = None
421 self.remote_conf_cache = {}
422 self.validate_once()
423 self.replication_start_time_nanos = time.monotonic_ns()
424 self.progress_reporter = ProgressReporter(log, p.pv_program_opts, self.use_select, self.progress_update_intervals)
425 with xfinally(lambda: self.progress_reporter.stop()):
426 daemon_stoptime_nanos: int = time.monotonic_ns() + p.daemon_lifetime_nanos
427 while True: # loop for daemon mode
428 self.timeout_nanos = (
429 None if p.timeout_duration_nanos is None else time.monotonic_ns() + p.timeout_duration_nanos
430 )
431 self.all_dst_dataset_exists.clear()
432 self.progress_reporter.reset()
433 src, dst = p.src, p.dst
434 for src_root_dataset, dst_root_dataset in p.root_dataset_pairs:
435 if self.termination_event.is_set():
436 self.terminate()
437 break
438 src.root_dataset = src.basis_root_dataset = src_root_dataset
439 dst.root_dataset = dst.basis_root_dataset = dst_root_dataset
440 p.curr_zfs_send_program_opts = p.zfs_send_program_opts.copy()
441 p.curr_zfs_send_resume_opts = p.zfs_send_resume_opts.copy()
442 if p.daemon_lifetime_nanos > 0:
443 self.timeout_nanos = (
444 None if p.timeout_duration_nanos is None else time.monotonic_ns() + p.timeout_duration_nanos
445 )
446 recurs_sep = " " if p.recursive_flag else ""
447 task_description = f"{src.basis_root_dataset} {p.recursive_flag}{recurs_sep}--> {dst.basis_root_dataset}"
448 if len(p.root_dataset_pairs) > 1:
449 log.info("Starting task: %s", task_description + " ...")
450 try:
451 try:
452 self.maybe_inject_error(cmd=[], error_trigger="retryable_run_tasks")
453 timeout(self)
454 self.validate_task()
455 self.run_task() # do the real work
456 except RetryableError as retryable_error:
457 cause: BaseException | None = retryable_error.__cause__
458 assert cause is not None
459 raise cause.with_traceback(cause.__traceback__) # noqa: B904 re-raise of cause without chaining
460 except (CalledProcessError, TimeoutExpired, SystemExit, UnicodeDecodeError, RetryTerminationError) as e:
461 if p.skip_on_error == "fail" or (
462 isinstance(e, subprocess.TimeoutExpired) and p.daemon_lifetime_nanos == 0
463 ):
464 raise
465 log.error("%s", e)
466 self.append_exception(e, "task", task_description)
467 if not self.sleep_until_next_daemon_iteration(daemon_stoptime_nanos):
468 break
469 if not p.skip_replication:
470 self.print_replication_stats(self.replication_start_time_nanos)
471 error_count = self.all_exceptions_count
472 if error_count > 0 and p.daemon_lifetime_nanos == 0:
473 msgs = "\n".join(f"{i + 1}/{error_count}: {e}" for i, e in enumerate(self.all_exceptions))
474 log.error("%s", f"Tolerated {error_count} errors. Error Summary: \n{msgs}")
475 assert self.first_exception is not None
476 raise self.first_exception
478 def append_exception(self, e: BaseException, task_name: str, task_description: str) -> None:
479 """Records and logs an exception that was encountered while running a subtask."""
480 self.first_exception = self.first_exception or e
481 if len(self.all_exceptions) < self.max_exceptions_to_summarize: # cap max memory consumption
482 self.all_exceptions.append(str(e))
483 self.all_exceptions_count += 1
484 self.params.log.error(f"#{self.all_exceptions_count}: Done with %s: %s", task_name, task_description)
486 def sleep_until_next_daemon_iteration(self, daemon_stoptime_nanos: int) -> bool:
487 """Pauses until next scheduled snapshot time or daemon stop; Returns True to continue daemon loop; False to stop."""
488 sleep_nanos: int = daemon_stoptime_nanos - time.monotonic_ns()
489 if sleep_nanos <= 0:
490 return False
491 self.progress_reporter.pause()
492 p, log = self.params, self.params.log
493 config: CreateSrcSnapshotConfig = p.create_src_snapshots_config
494 curr_datetime: datetime = config.current_datetime + timedelta(microseconds=1)
495 next_snapshotting_event_dt: datetime = min(
496 (
497 config.anchors.round_datetime_up_to_duration_multiple(curr_datetime, duration_amount, duration_unit)
498 for duration_amount, duration_unit in config.suffix_durations.values()
499 ),
500 default=curr_datetime + timedelta(days=10 * 365), # infinity
501 )
502 offset: timedelta = next_snapshotting_event_dt - datetime.now(config.tz)
503 offset_nanos: int = (offset.days * 86400 + offset.seconds) * 1_000_000_000 + offset.microseconds * 1_000
504 sleep_nanos = min(sleep_nanos, max(0, offset_nanos))
505 log.info("Daemon sleeping for: %s%s", human_readable_duration(sleep_nanos), f" ... [Log {p.log_params.log_file}]")
506 self.termination_event.wait(sleep_nanos / 1_000_000_000) # allow early wakeup on async termination
507 config.current_datetime = datetime.now(config.tz)
508 return time.monotonic_ns() < daemon_stoptime_nanos and not self.termination_event.is_set()
510 def print_replication_stats(self, start_time_nanos: int) -> None:
511 """Logs overall replication statistics after a job cycle completes."""
512 p, log = self.params, self.params.log
513 elapsed_nanos: int = time.monotonic_ns() - start_time_nanos
514 msg = p.dry(f"zfs sent {self.num_snapshots_replicated} snapshots in {human_readable_duration(elapsed_nanos)}.")
515 if p.is_program_available("pv", "local"):
516 sent_bytes: int = count_num_bytes_transferred_by_zfs_send(p.log_params.pv_log_file)
517 sent_bytes_per_sec: int = round(1_000_000_000 * sent_bytes / (elapsed_nanos or 1))
518 msg += f" zfs sent {human_readable_bytes(sent_bytes)} [{human_readable_bytes(sent_bytes_per_sec)}/s]."
519 log.info("%s", msg.ljust(p.log_params.terminal_columns - len("2024-01-01 23:58:45 [I] ")))
521 def validate_once(self) -> None:
522 """Validates CLI parameters and compiles regex lists one time only, which will later be reused many times."""
523 p = self.params
524 p.zfs_recv_ox_names = self.recv_option_property_names(p.zfs_recv_program_opts)
525 for snapshot_filter in p.snapshot_filters:
526 for _filter in snapshot_filter:
527 if _filter.name == SNAPSHOT_REGEX_FILTER_NAME:
528 exclude_snapshot_regexes_strings, include_snapshot_regexes_strings = cast(
529 tuple[list[str], list[str]], _filter.options
530 )
531 exclude_snapshot_regexes = compile_regexes(exclude_snapshot_regexes_strings)
532 include_snapshot_regexes = compile_regexes(include_snapshot_regexes_strings or [".*"])
533 _filter.options = (exclude_snapshot_regexes, include_snapshot_regexes)
535 exclude_regexes: list[str] = [EXCLUDE_DATASET_REGEXES_DEFAULT]
536 if len(p.args.exclude_dataset_regex) > 0: # some patterns don't exclude anything
537 exclude_regexes = [regex for regex in p.args.exclude_dataset_regex if regex != "" and regex != "!.*"]
538 include_regexes: list[str] = p.args.include_dataset_regex
540 # relative datasets need not be compiled more than once as they don't change between tasks
541 def separate_abs_vs_rel_datasets(datasets: list[str]) -> tuple[list[str], list[str]]:
542 abs_datasets: list[str] = []
543 rel_datasets: list[str] = []
544 for dataset in datasets:
545 (abs_datasets if dataset.startswith("/") else rel_datasets).append(dataset)
546 return abs_datasets, rel_datasets
548 p.abs_exclude_datasets, rel_exclude_datasets = separate_abs_vs_rel_datasets(p.args.exclude_dataset)
549 p.abs_include_datasets, rel_include_datasets = separate_abs_vs_rel_datasets(p.args.include_dataset)
550 suffix = DESCENDANTS_RE_SUFFIX
551 p.tmp_exclude_dataset_regexes, p.tmp_include_dataset_regexes = (
552 compile_regexes(exclude_regexes + dataset_regexes(p.src, p.dst, rel_exclude_datasets), suffix=suffix),
553 compile_regexes(include_regexes + dataset_regexes(p.src, p.dst, rel_include_datasets), suffix=suffix),
554 )
556 if p.pv_program != DISABLE_PRG:
557 pv_program_opts_set = set(p.pv_program_opts)
558 if pv_program_opts_set.isdisjoint({"--bytes", "-b", "--bits", "-8"}):
559 die("--pv-program-opts must contain one of --bytes or --bits for progress metrics to function.")
560 if not p.log_params.quiet:
561 for opts in [["--eta", "-e"], ["--fineta", "-I"], ["--average-rate", "-a"]]:
562 if pv_program_opts_set.isdisjoint(opts):
563 die(f"--pv-program-opts must contain one of {', '.join(opts)} for progress report line to function.")
565 src, dst = p.src, p.dst
566 for remote in [src, dst]:
567 r, loc = remote, remote.location
568 validate_user_name(r.basis_ssh_user, f"--ssh-{loc}-user")
569 validate_host_name(r.basis_ssh_host, f"--ssh-{loc}-host")
570 validate_port(r.ssh_port, f"--ssh-{loc}-port ")
572 args, log = p.args, p.log
573 if args.force_delete_dst_tmp_bookmarks:
574 if not args.skip_replication:
575 die("--force-delete-dst-tmp-bookmarks requires --skip-replication")
576 incompatible_option: str = ""
577 if args.delete_dst_snapshots is not None:
578 incompatible_option = "--delete-dst-snapshots"
579 elif args.delete_dst_snapshots_except_plan is not None:
580 incompatible_option = "--delete-dst-snapshots-except-plan"
581 elif args.include_snapshot_plan is not None:
582 incompatible_option = "--include-snapshot-plan"
583 if incompatible_option:
584 die(f"--force-delete-dst-tmp-bookmarks cannot be combined with {incompatible_option}")
585 log.warning(
586 "DANGER: --force-delete-dst-tmp-bookmarks can delete the only temporary bookmark preserving incremental "
587 "replication continuity after an interrupted replication. This option is intended only for rare "
588 "administrative cleanup in pathologic situations, not normal periodic operation."
589 )
591 def validate_task(self) -> None:
592 """Validates a single replication task before execution."""
593 p, log = self.params, self.params.log
594 src, dst = p.src, p.dst
595 for remote in [src, dst]:
596 r = remote
597 r.ssh_user, r.ssh_host, r.ssh_user_host, r.pool, r.root_dataset = parse_dataset_locator(
598 r.basis_root_dataset, user=r.basis_ssh_user, host=r.basis_ssh_host, port=r.ssh_port
599 )
600 r.sudo, r.use_zfs_delegation = self.sudo_cmd(r.ssh_user_host, r.ssh_user)
601 local_addrs = ("",) if self.is_test_mode else ("", "127.0.0.1", "::1") # ::1 is IPv6 version of loopback address
602 remote.is_nonlocal = r.ssh_host not in local_addrs
603 self.dst_dataset_exists = SynchronizedDict(self.all_dst_dataset_exists[dst.ssh_user_host])
605 if src.ssh_host == dst.ssh_host:
606 msg = f"src: {src.basis_root_dataset}, dst: {dst.basis_root_dataset}"
607 if src.root_dataset == dst.root_dataset:
608 die(f"Source and destination dataset must not be the same! {msg}")
609 if p.recursive and (
610 is_descendant(src.root_dataset, of_root_dataset=dst.root_dataset)
611 or is_descendant(dst.root_dataset, of_root_dataset=src.root_dataset)
612 ):
613 die(f"Source and destination dataset trees must not overlap! {msg}")
615 suffx: str = DESCENDANTS_RE_SUFFIX # also match descendants of a matching dataset
616 p.exclude_dataset_regexes, p.include_dataset_regexes = (
617 p.tmp_exclude_dataset_regexes + compile_regexes(dataset_regexes(src, dst, p.abs_exclude_datasets), suffix=suffx),
618 p.tmp_include_dataset_regexes + compile_regexes(dataset_regexes(src, dst, p.abs_include_datasets), suffix=suffx),
619 )
620 if len(p.include_dataset_regexes) == 0:
621 p.include_dataset_regexes = [(re.compile(r".*"), False)]
623 detect_available_programs(self)
624 p.r2r_mode = resolve_r2r_mode(p)
626 if is_zpool_feature_enabled_or_active(p, dst, "feature@large_blocks"):
627 append_if_absent(p.curr_zfs_send_program_opts, "--large-block")
628 append_if_absent(p.curr_zfs_send_resume_opts, "--large-block")
630 self.max_workers = {}
631 self.max_datasets_per_minibatch_on_list_snaps = {}
632 for r in [src, dst]:
633 cpus: int = int(p.available_programs[r.location].get("getconf_cpu_count", 8))
634 threads, is_percent = p.threads
635 cpus = max(1, round(cpus * threads / 100.0) if is_percent else round(threads))
636 self.max_workers[r.location] = cpus
637 bs: int = max(1, p.max_datasets_per_batch_on_list_snaps) # 1024 by default
638 max_datasets_per_minibatch: int = p.max_datasets_per_minibatch_on_list_snaps
639 if max_datasets_per_minibatch <= 0:
640 max_datasets_per_minibatch = max(1, bs // cpus)
641 max_datasets_per_minibatch = min(bs, max_datasets_per_minibatch)
642 self.max_datasets_per_minibatch_on_list_snaps[r.location] = max_datasets_per_minibatch
643 log.log(
644 LOG_TRACE,
645 "%s",
646 f"max_datasets_per_batch_on_list_snaps: {p.max_datasets_per_batch_on_list_snaps}, "
647 f"max_datasets_per_minibatch_on_list_snaps: {max_datasets_per_minibatch}, "
648 f"max_workers: {self.max_workers[r.location]}, "
649 f"location: {r.location}",
650 )
651 if self.is_test_mode:
652 log.log(LOG_TRACE, "Validated Param values: %s", pretty_print_formatter(self.params))
654 def sudo_cmd(self, ssh_user_host: str, ssh_user: str) -> tuple[str, bool]:
655 """Returns sudo command prefix and whether root privileges are required."""
656 p: Params = self.params
657 assert isinstance(ssh_user_host, str)
658 assert isinstance(ssh_user, str)
659 assert isinstance(p.sudo_program, str)
660 assert isinstance(p.enable_privilege_elevation, bool)
662 is_root: bool = True
663 if ssh_user_host != "":
664 if ssh_user == "":
665 if os.getuid() != 0:
666 is_root = False
667 elif ssh_user != "root":
668 is_root = False
669 elif os.getuid() != 0:
670 is_root = False
672 if is_root:
673 sudo = "" # using sudo in an attempt to make ZFS operations work even if we are not root user?
674 use_zfs_delegation = False # or instead using 'zfs allow' delegation?
675 return sudo, use_zfs_delegation
676 elif p.enable_privilege_elevation:
677 if p.sudo_program == DISABLE_PRG:
678 die(f"sudo CLI is not available on host: {ssh_user_host or 'localhost'}")
679 # The '-n' option makes 'sudo' safer and more fail-fast. It avoids having sudo prompt the user for input of any
680 # kind. If a password is required for the sudo command to run, sudo will display an error message and exit.
681 return p.sudo_program + " -n", False
682 else:
683 return "", True
685 def run_task(self) -> None:
686 """Replicates all snapshots for the current root dataset pair."""
688 def filter_src_datasets() -> list[str]: # apply --{include|exclude}-dataset policy
689 return filter_datasets(self, src, basis_src_datasets) if src_datasets is None else src_datasets
691 p, log = self.params, self.params.log
692 src, dst = p.src, p.dst
693 max_workers: int = min(self.max_workers[src.location], self.max_workers[dst.location])
694 recursive_sep: str = " " if p.recursive_flag else ""
695 task_description: str = f"{src.basis_root_dataset} {p.recursive_flag}{recursive_sep}--> {dst.basis_root_dataset} ..."
696 failed: bool = False
697 src_datasets: list[str] | None = None
698 basis_src_datasets: list[str] = []
699 self.src_properties = {}
700 self.dst_properties = {}
701 if not is_dummy(src): # find src dataset or all datasets in src dataset tree (with --recursive)
702 basis_src_datasets = self.list_src_datasets_task()
704 if not p.create_src_snapshots_config.skip_create_src_snapshots:
705 log.info(p.dry("--create-src-snapshots: %s"), f"{src.basis_root_dataset} {p.recursive_flag}{recursive_sep}...")
706 src_datasets = filter_src_datasets() # apply include/exclude policy
707 self.create_src_snapshots_task(basis_src_datasets, src_datasets)
709 # Optionally, replicate src.root_dataset (optionally including its descendants) to dst.root_dataset
710 if not p.skip_replication:
711 if len(basis_src_datasets) == 0:
712 die(f"Replication: Source dataset does not exist: {src.basis_root_dataset}")
713 if is_dummy(dst):
714 die("Replication: Destination may be a dummy dataset only if exclusively creating snapshots on the source!")
715 src_datasets = filter_src_datasets() # apply include/exclude policy
716 failed = self.replicate_datasets(src_datasets, task_description, max_workers)
718 if failed or not (
719 p.delete_dst_datasets
720 or p.delete_dst_snapshots
721 or p.delete_empty_dst_datasets
722 or p.compare_snapshot_lists
723 or p.monitor_snapshots_config.enable_monitor_snapshots
724 ):
725 return
726 log.info("Listing dst datasets: %s", task_description)
727 if is_dummy(dst):
728 die("Destination may be a dummy dataset only if exclusively creating snapshots on the source!")
729 basis_dst_datasets: list[str] = self.list_dst_datasets_task()
730 dst_datasets: list[str] = filter_datasets(self, dst, basis_dst_datasets) # apply include/exclude policy
732 if p.delete_dst_datasets and not failed:
733 log.info(p.dry("--delete-dst-datasets: %s"), task_description)
734 basis_dst_datasets, dst_datasets = self.delete_dst_datasets_task(
735 basis_src_datasets, basis_dst_datasets, dst_datasets
736 )
738 if p.delete_dst_snapshots and not failed:
739 log.info(p.dry("--delete-dst-snapshots: %s"), task_description + f" [{len(dst_datasets)} datasets]")
740 failed = self.delete_destination_snapshots_task(basis_src_datasets, dst_datasets, max_workers, task_description)
742 if p.delete_empty_dst_datasets and p.recursive and not failed:
743 log.info(p.dry("--delete-empty-dst-datasets: %s"), task_description)
744 basis_dst_datasets, dst_datasets = self.delete_empty_dst_datasets_task(basis_dst_datasets, dst_datasets)
746 if p.compare_snapshot_lists and not failed:
747 log.info("--compare-snapshot-lists: %s", task_description)
748 if len(basis_src_datasets) == 0 and not is_dummy(src):
749 die(f"Source dataset does not exist: {src.basis_root_dataset}")
750 src_datasets = filter_src_datasets() # apply include/exclude policy
751 run_compare_snapshot_lists(self, src_datasets, dst_datasets)
753 if p.monitor_snapshots_config.enable_monitor_snapshots and not failed:
754 log.info("--monitor-snapshots: %s", task_description)
755 if len(basis_src_datasets) == 0 and not is_dummy(src):
756 die(f"Source dataset does not exist: {src.basis_root_dataset}")
757 if len(basis_dst_datasets) == 0:
758 die(f"Destination dataset does not exist: {dst.basis_root_dataset}")
759 src_datasets = filter_src_datasets() # apply include/exclude policy
760 self.monitor_snapshots_task(src_datasets, dst_datasets, task_description)
762 def list_src_datasets_task(self) -> list[str]:
763 """Lists datasets on the source host."""
764 p = self.params
765 src = p.src
766 basis_src_datasets: list[str] = []
767 is_caching: bool = is_caching_snapshots(p, src)
768 props: str = "volblocksize,recordsize,name"
769 props = "snapshots_changed," + props if is_caching else props
770 cmd: list[str] = p.split_args(
771 f"{p.zfs_program} list -t filesystem,volume -s name -Hp -o {props} {p.recursive_flag}", src.root_dataset
772 )
773 for line in (self.try_ssh_command_with_retries(src, LOG_DEBUG, cmd=cmd) or "").splitlines():
774 cols: list[str] = line.split("\t")
775 snapshots_changed, volblocksize, recordsize, src_dataset = cols if is_caching else ["-"] + cols
776 self.src_properties[src_dataset] = DatasetProperties(
777 recordsize=int(recordsize) if recordsize != "-" else -int(volblocksize),
778 snapshots_changed=int(snapshots_changed) if snapshots_changed and snapshots_changed != "-" else 0,
779 )
780 basis_src_datasets.append(src_dataset)
781 assert (not self.is_test_mode) or basis_src_datasets == sorted(basis_src_datasets), "List is not sorted"
782 return basis_src_datasets
784 def list_dst_datasets_task(self) -> list[str]:
785 """Lists datasets on the destination host."""
786 p, log = self.params, self.params.log
787 dst = p.dst
788 is_caching: bool = is_caching_snapshots(p, dst) and p.monitor_snapshots_config.enable_monitor_snapshots
789 props: str = "name"
790 props = "snapshots_changed," + props if is_caching else props
791 cmd: list[str] = p.split_args(
792 f"{p.zfs_program} list -t filesystem,volume -s name -Hp -o {props} {p.recursive_flag}", dst.root_dataset
793 )
794 basis_dst_datasets: list[str] = []
795 basis_dst_datasets_str: str | None = self.try_ssh_command_with_retries(dst, LOG_TRACE, cmd=cmd)
796 if basis_dst_datasets_str is None:
797 log.warning("Destination dataset does not exist: %s", dst.root_dataset)
798 else:
799 for line in basis_dst_datasets_str.splitlines():
800 cols: list[str] = line.split("\t")
801 snapshots_changed, dst_dataset = cols if is_caching else ["-"] + cols
802 self.dst_properties[dst_dataset] = DatasetProperties(
803 recordsize=0,
804 snapshots_changed=int(snapshots_changed) if snapshots_changed and snapshots_changed != "-" else 0,
805 )
806 basis_dst_datasets.append(dst_dataset)
807 assert (not self.is_test_mode) or basis_dst_datasets == sorted(basis_dst_datasets), "List is not sorted"
808 return basis_dst_datasets
810 def create_src_snapshots_task(self, basis_src_datasets: list[str], src_datasets: list[str]) -> None:
811 """Atomically creates a new snapshot of the src datasets selected by --{include|exclude}-dataset* policy; implements
812 --create-src-snapshots.
814 The implementation attempts to fit as many datasets as possible into a single (atomic) 'zfs snapshot' command line,
815 using lexicographical sort order, and using 'zfs snapshot -r' to the extent that this is compatible with the
816 --{include|exclude}-dataset* pruning policy. The snapshots of all datasets that fit within the same single 'zfs
817 snapshot' CLI invocation will be taken within the same ZFS transaction group, and correspondingly have identical
818 'createtxg' ZFS property (but not necessarily identical 'creation' ZFS time property as ZFS actually provides no such
819 guarantee), and thus be consistent. Dataset names that can't fit into a single command line are spread over multiple
820 command line invocations, respecting the limits that the operating system places on the maximum length of a single
821 command line, per `getconf ARG_MAX`.
823 Time complexity is O((N log N) + (N * M log M)) where N is the number of datasets and M is the number of snapshots
824 per dataset. Space complexity is O(max(N, M)).
825 """
826 p, log = self.params, self.params.log
827 src = p.src
828 if len(basis_src_datasets) == 0:
829 die(f"Source dataset does not exist: {src.basis_root_dataset}")
830 datasets_to_snapshot: dict[SnapshotLabel, list[str]] = self.find_datasets_to_snapshot(src_datasets)
831 datasets_to_snapshot = {label: datasets for label, datasets in datasets_to_snapshot.items() if len(datasets) > 0}
832 basis_datasets_to_snapshot: dict[SnapshotLabel, list[str]] = datasets_to_snapshot.copy() # shallow copy
833 commands: dict[SnapshotLabel, list[str]] = {}
834 for label, datasets in datasets_to_snapshot.items():
835 cmd: list[str] = p.split_args(f"{src.sudo} {p.zfs_program} snapshot")
836 if p.recursive:
837 # Run 'zfs snapshot -r' on the roots of subtrees if possible, else fallback to non-recursive CLI flavor
838 root_datasets = self.root_datasets_if_recursive_zfs_snapshot_is_possible(datasets, basis_src_datasets)
839 if root_datasets is not None:
840 cmd.append("-r") # recursive; takes a snapshot of all datasets in the subtree(s)
841 datasets_to_snapshot[label] = root_datasets
842 commands[label] = cmd
843 creation_msg = f"Creating {sum(len(datasets) for datasets in basis_datasets_to_snapshot.values())} snapshots"
844 log.info(p.dry("--create-src-snapshots: %s"), f"{creation_msg} within {len(src_datasets)} datasets ...")
845 # create snapshots in large (parallel) batches, without using a command line that's too big for the OS to handle
846 run_ssh_cmd_parallel(
847 self,
848 src,
849 ((commands[lbl], (f"{ds}@{lbl}" for ds in datasets)) for lbl, datasets in datasets_to_snapshot.items()),
850 fn=lambda cmd, batch: self.run_ssh_command_with_retries(
851 src, is_dry=p.dry_run, print_stdout=True, cmd=cmd + batch, retry_on_generic_ssh_error=False
852 ), # retry_on_generic_ssh_error=False means only retry on SSH connect errors b/c `zfs snapshot` isn't idempotent
853 max_batch_items=2**29,
854 )
855 for label, datasets in basis_datasets_to_snapshot.items():
856 for dataset in datasets:
857 log.debug(p.dry("--create-src-snapshots: %s%s%s%s"), "Created snapshot ", dataset, "@", label)
858 if is_caching_snapshots(p, src):
859 # perf: copy lastmodified time of source dataset into local cache to reduce future 'zfs list -t snapshot' calls
860 self.cache.update_last_modified_cache(basis_datasets_to_snapshot)
862 def delete_destination_snapshots_task(
863 self, basis_src_datasets: list[str], dst_datasets: list[str], max_workers: int, task_description: str
864 ) -> bool:
865 """Deletes existing destination snapshots that do not exist within the source dataset if they are included by the
866 --{include|exclude}-snapshot-* policy, and the destination dataset is included via --{include|exclude}-dataset*
867 policy; implements --delete-dst-snapshots. Does not attempt to delete snapshots that carry a `zfs hold`."""
868 p, log = self.params, self.params.log
869 src, dst = p.src, p.dst
870 kind: str = "bookmark" if p.delete_dst_bookmarks else "snapshot"
871 filter_needs_creation_time: bool = has_timerange_filter(p.snapshot_filters)
872 props: str = "guid,name,userrefs"
873 props = self.creation_prefix + "creation," + props if filter_needs_creation_time else props
874 basis_src_datasets_set: set[str] = set(basis_src_datasets)
875 num_snapshots_found, num_snapshots_deleted = 0, 0
877 def delete_destination_snapshots(dst_dataset: str, tid: str, retry: Retry) -> bool: # thread-safe
878 src_dataset: str = replace_prefix(dst_dataset, old_prefix=dst.root_dataset, new_prefix=src.root_dataset)
879 if src_dataset in basis_src_datasets_set and (are_bookmarks_enabled(p, src) or not p.delete_dst_bookmarks):
880 src_kind: str = kind
881 if not p.delete_dst_snapshots_no_crosscheck:
882 src_kind = "snapshot,bookmark" if are_bookmarks_enabled(p, src) else "snapshot"
883 src_cmd = p.split_args(f"{p.zfs_program} list -t {src_kind} -d 1 -s name -Hp -o guid", src_dataset)
884 else:
885 src_cmd = None
886 dst_cmd = p.split_args(f"{p.zfs_program} list -t {kind} -d 1 -s createtxg -Hp -o {props}", dst_dataset)
887 self.maybe_inject_delete(dst, dataset=dst_dataset, delete_trigger="zfs_list_delete_dst_snapshots")
888 src_snaps_with_guids, dst_snaps_with_guids_str = run_in_parallel( # list src+dst snapshots in parallel
889 lambda: set(self.run_ssh_command(src, LOG_TRACE, cmd=src_cmd).splitlines() if src_cmd else []),
890 lambda: self.try_ssh_command(dst, LOG_TRACE, cmd=dst_cmd),
891 )
892 if dst_snaps_with_guids_str is None:
893 log.warning("Third party deleted destination: %s", dst_dataset)
894 return False
895 held_dst_snapshots: set[str] = set()
896 dst_snaps_with_guids: list[str] = []
897 for line in dst_snaps_with_guids_str.splitlines():
898 dst_snaps_with_guids.append(line[: line.rindex("\t")]) # strip off trailing userrefs column
899 _, name, userrefs = line.rsplit("\t", 2)
900 if userrefs not in ("", "-", "0"): # ZFS snapshot property userrefs > 0 indicates a zfs hold
901 tag: str = name[name.index("@") + 1 :]
902 held_dst_snapshots.add(tag) # don't attempt to delete snapshots that carry a `zfs hold`
903 if p.delete_dst_bookmarks:
904 dst_snaps_with_guids = [
905 line for line in dst_snaps_with_guids if is_tmp_bookmark(line) == p.force_delete_dst_tmp_bookmarks
906 ]
907 num_dst_snaps_with_guids = len(dst_snaps_with_guids)
908 basis_dst_snaps_with_guids: list[str] = dst_snaps_with_guids.copy()
909 if p.delete_dst_bookmarks:
910 replace_in_lines(dst_snaps_with_guids, old="#", new="@", count=1) # treat bookmarks as snapshots
911 # The check against the source dataset happens *after* filtering the dst snapshots with filter_snapshots().
912 # `p.delete_dst_snapshots_except` means the user wants to specify snapshots to *retain* aka *keep*
913 all_except: bool = p.delete_dst_snapshots_except
914 if p.delete_dst_snapshots_except and not is_dummy(src):
915 # However, as here we are in "except" mode AND the source is NOT a dummy, we first filter to get what
916 # the policy says to *keep* (so all_except=False for the filter_snapshots() call), then from that "keep"
917 # list, we later further refine by checking what's on the source dataset.
918 all_except = False
919 dst_snaps_with_guids = filter_snapshots(self, dst_snaps_with_guids, all_except=all_except)
920 if p.delete_dst_bookmarks:
921 replace_in_lines(dst_snaps_with_guids, old="@", new="#", count=1) # restore pre-filtering bookmark state
922 if filter_needs_creation_time:
923 dst_snaps_with_guids = cut(field=2, lines=dst_snaps_with_guids)
924 basis_dst_snaps_with_guids = cut(field=2, lines=basis_dst_snaps_with_guids)
925 if p.delete_dst_snapshots_except and not is_dummy(src): # Non-dummy Source + "Except" (Keep) Mode
926 # Retain dst snapshots that match snapshot filter policy AND are on src dataset, aka
927 # Delete dst snapshots except snapshots that match snapshot filter policy AND are on src dataset.
928 # Concretely, `dst_snaps_with_guids` contains GUIDs of DST snapshots that the filter policy says to KEEP.
929 # We only actually keep them if they are ALSO on the SRC.
930 # So, snapshots to DELETE (`dst_tags_to_delete`) are ALL snapshots on DST (`basis_dst_snaps_with_guids`)
931 # EXCEPT those whose GUIDs are in `dst_snaps_with_guids` AND ALSO in `src_snaps_with_guids`.
932 except_dst_guids: set[str] = set(cut(field=1, lines=dst_snaps_with_guids)).intersection(src_snaps_with_guids)
933 dst_tags_to_delete: list[str] = filter_lines_except(basis_dst_snaps_with_guids, except_dst_guids)
934 else: # Standard Delete Mode OR Dummy Source + "Except" (Keep) Mode
935 # In standard delete mode:
936 # `dst_snaps_with_guids` contains GUIDs of policy-selected snapshots on DST.
937 # We delete those that are NOT on SRC.
938 # `dst_tags_to_delete` = `dst_snaps_with_guids` - `src_snaps_with_guids`.
939 # In dummy source + "except" (keep) mode:
940 # `all_except` was True.
941 # `dst_snaps_with_guids` contains snaps NOT matching the "keep" policy -- these are the ones to delete.
942 # `src_snaps_with_guids` is empty.
943 # `dst_tags_to_delete` = `dst_snaps_with_guids` - {} = `dst_snaps_with_guids`.
944 dst_guids_to_delete = set(cut(field=1, lines=dst_snaps_with_guids)).difference(src_snaps_with_guids)
945 dst_tags_to_delete = filter_lines(dst_snaps_with_guids, dst_guids_to_delete)
946 separator: str = "#" if p.delete_dst_bookmarks else "@"
947 dst_tags_to_delete = cut(field=2, separator=separator, lines=dst_tags_to_delete)
948 if p.delete_dst_bookmarks:
949 delete_bookmarks(self, dst, dst_dataset, snapshot_tags=dst_tags_to_delete)
950 else:
951 dst_tags_to_delete = [tag for tag in dst_tags_to_delete if tag not in held_dst_snapshots]
952 delete_snapshots(self, dst, dst_dataset, snapshot_tags=dst_tags_to_delete)
953 with self.stats_lock:
954 nonlocal num_snapshots_found
955 num_snapshots_found += num_dst_snaps_with_guids
956 nonlocal num_snapshots_deleted
957 num_snapshots_deleted += len(dst_tags_to_delete)
958 if len(dst_tags_to_delete) > 0 and not p.delete_dst_bookmarks:
959 self.dst_properties[dst_dataset].snapshots_changed = 0 # invalidate cache
960 return True
962 # Run delete_destination_snapshots(dataset) for each dataset, while handling errors, retries + parallel exec
963 failed: bool = False
964 if are_bookmarks_enabled(p, dst) or not p.delete_dst_bookmarks:
965 start_time_nanos = time.monotonic_ns()
966 failed = process_datasets_in_parallel_and_fault_tolerant(
967 log=log,
968 datasets=dst_datasets,
969 process_dataset=delete_destination_snapshots, # lambda
970 skip_tree_on_error=lambda dataset: False,
971 skip_on_error=p.skip_on_error,
972 max_workers=max_workers,
973 timing=self.task_timing,
974 termination_handler=self.terminate,
975 enable_barriers=False,
976 task_name="--delete-dst-snapshots",
977 append_exception=self.append_exception,
978 retry_template=self._retry_template(),
979 dry_run=p.dry_run,
980 is_test_mode=self.is_test_mode,
981 )
982 elapsed_nanos = time.monotonic_ns() - start_time_nanos
983 log.info(
984 p.dry("--delete-dst-snapshots: %s"),
985 task_description + f" [Deleted {num_snapshots_deleted} out of {num_snapshots_found} {kind}s "
986 f"within {len(dst_datasets)} datasets; took {human_readable_duration(elapsed_nanos)}]",
987 )
988 return failed
990 def delete_dst_datasets_task(
991 self, basis_src_datasets: list[str], basis_dst_datasets: list[str], sorted_dst_datasets: list[str]
992 ) -> tuple[list[str], list[str]]:
993 """Deletes existing destination datasets that do not exist within the source dataset if they are included via
994 --{include|exclude}-dataset* policy; implements --delete-dst-datasets.
996 Do not recurse without --recursive. With --recursive, never delete non-selected dataset subtrees or their ancestors.
997 """
998 p = self.params
999 src, dst = p.src, p.dst
1000 children: dict[str, set[str]] = defaultdict(set)
1001 for dst_dataset in basis_dst_datasets: # Compute the direct children of each NON-FILTERED dataset
1002 parent: str = os.path.dirname(dst_dataset)
1003 children[parent].add(dst_dataset)
1004 to_delete: set[str] = set()
1005 for dst_dataset in reversed(sorted_dst_datasets): # Reverse order facilitates efficient O(N) time algorithm
1006 if children[dst_dataset].issubset(to_delete):
1007 to_delete.add(dst_dataset) # all children are deletable, thus the dataset itself is deletable too
1008 to_delete = to_delete.difference(
1009 replace_prefix(src_dataset, src.root_dataset, dst.root_dataset) for src_dataset in basis_src_datasets
1010 )
1011 delete_datasets(self, dst, to_delete)
1012 sorted_dst_datasets = sorted(set(sorted_dst_datasets).difference(to_delete))
1013 basis_dst_datasets = sorted(set(basis_dst_datasets).difference(to_delete))
1014 return basis_dst_datasets, sorted_dst_datasets
1016 def delete_empty_dst_datasets_task(
1017 self, basis_dst_datasets: list[str], sorted_dst_datasets: list[str]
1018 ) -> tuple[list[str], list[str]]:
1019 """Deletes any existing destination dataset that has no snapshot and no bookmark if all descendants of that dataset
1020 do not have a snapshot or bookmark either; implements --delete-empty-dst-datasets.
1022 To do so, we walk the dataset list (conceptually, a tree) depth-first (i.e. sorted descending). If a dst dataset has
1023 zero snapshots and zero bookmarks and all its children are already marked as orphans, then it is itself an orphan,
1024 and we mark it as such. Walking in a reverse sorted way means that we efficiently check for zero snapshots/bookmarks
1025 not just over the direct children but the entire tree. Finally, delete all orphan datasets in an efficient batched
1026 way.
1027 """
1028 p = self.params
1029 dst = p.dst
1031 # Compute the direct children of each NON-FILTERED dataset. Thus, no non-selected dataset and no ancestor of a
1032 # non-selected dataset will ever be added to the "orphan" set. In other words, this treats non-selected dataset
1033 # subtrees as if they all had snapshots, so non-selected dataset subtrees and their ancestors are guaranteed
1034 # to not get deleted.
1035 children: dict[str, set[str]] = defaultdict(set)
1036 for dst_dataset in basis_dst_datasets:
1037 parent: str = os.path.dirname(dst_dataset)
1038 children[parent].add(dst_dataset)
1040 def compute_orphans(datasets_having_snapshots: set[str]) -> set[str]:
1041 """Returns destination datasets having zero snapshots whose children are all orphans."""
1042 orphans: set[str] = set()
1043 for dst_dataset in reversed(sorted_dst_datasets): # Reverse order facilitates efficient O(N) time algorithm
1044 if (dst_dataset not in datasets_having_snapshots) and children[dst_dataset].issubset(orphans):
1045 orphans.add(dst_dataset)
1046 return orphans
1048 # Compute candidate orphan datasets, which reduces the list of datasets for which we list snapshots via
1049 # 'zfs list -t snapshot ...' from dst_datasets to a subset of dst_datasets, which in turn reduces I/O and improves
1050 # perf. Essentially, this eliminates the I/O to list snapshots for ancestors of excluded datasets.
1051 candidate_orphans: set[str] = compute_orphans(set())
1053 # Compute destination datasets having more than zero snapshots
1054 dst_datasets_having_snapshots: set[str] = set()
1055 with_bookmarks: bool = p.delete_empty_dst_datasets_if_no_bookmarks_and_no_snapshots and are_bookmarks_enabled(p, dst)
1056 btype: str = "bookmark,snapshot" if with_bookmarks else "snapshot"
1057 cmd: list[str] = p.split_args(f"{p.zfs_program} list -t {btype} -d 1 -S name -Hp -o name")
1058 for snapshots in zfs_list_snapshots_in_parallel(self, dst, cmd, sorted(candidate_orphans), ordered=False):
1059 if with_bookmarks:
1060 replace_in_lines(snapshots, old="#", new="@", count=1) # treat bookmarks as snapshots
1061 dst_datasets_having_snapshots.update(snap[: snap.index("@")] for snap in snapshots) # union
1063 orphans: set[str] = compute_orphans(dst_datasets_having_snapshots) # compute the real orphans
1064 delete_datasets(self, dst, orphans) # finally, delete the orphan datasets in an efficient way
1065 sorted_dst_datasets = sorted(set(sorted_dst_datasets).difference(orphans))
1066 basis_dst_datasets = sorted(set(basis_dst_datasets).difference(orphans))
1067 return basis_dst_datasets, sorted_dst_datasets
1069 def monitor_snapshots_task(
1070 self, sorted_src_datasets: list[str], sorted_dst_datasets: list[str], task_description: str
1071 ) -> None:
1072 """Monitors src and dst snapshots; implements --monitor-snapshots."""
1073 p, log = self.params, self.params.log
1074 src, dst = p.src, p.dst
1075 num_cache_hits: int = self.num_cache_hits
1076 num_cache_misses: int = self.num_cache_misses
1077 start_time_nanos: int = time.monotonic_ns()
1078 dst_alert, src_alert = run_in_parallel(
1079 lambda: self.monitor_snapshots(dst, sorted_dst_datasets),
1080 lambda: self.monitor_snapshots(src, sorted_src_datasets),
1081 )
1082 exit_code, _exit_kind, exit_msg = min(dst_alert, src_alert)
1083 if exit_code != 0:
1084 die(exit_msg, -exit_code)
1085 elapsed: str = human_readable_duration(time.monotonic_ns() - start_time_nanos)
1086 num_cache_hits = self.num_cache_hits - num_cache_hits
1087 num_cache_misses = self.num_cache_misses - num_cache_misses
1088 if num_cache_hits > 0 or num_cache_misses > 0:
1089 msg = self._cache_hits_msg(hits=num_cache_hits, misses=num_cache_misses)
1090 else:
1091 msg = ""
1092 log.info(
1093 "--monitor-snapshots done: %s",
1094 f"{task_description} [{len(sorted_src_datasets) + len(sorted_dst_datasets)} datasets; took {elapsed}{msg}]",
1095 )
1097 def monitor_snapshots(self, remote: Remote, sorted_datasets: list[str]) -> tuple[int, str, str]:
1098 """Checks snapshot freshness and warns or errors out when limits are exceeded.
1100 Alerts the user if the ZFS 'creation' time property of the latest or oldest snapshot for any specified snapshot name
1101 pattern within the selected datasets is too old wrt. the specified age limit. The purpose is to check if snapshots
1102 are successfully taken on schedule, successfully replicated on schedule, and successfully pruned on schedule. Process
1103 exit code is 0, 1, 2 on OK, WARNING, CRITICAL, respectively.
1105 Time complexity is O((N log N) + (N * M log M)) where N is the number of datasets and M is the number of snapshots
1106 per dataset. Space complexity is O(max(N, M)).
1107 """
1108 p, log = self.params, self.params.log
1109 alerts: list[MonitorSnapshotAlert] = p.monitor_snapshots_config.alerts
1110 labels: list[SnapshotLabel] = [alert.label for alert in alerts]
1111 oldest_skip_holds: list[bool] = [alert.oldest_skip_holds for alert in alerts]
1112 current_unixtime_millis: float = p.create_src_snapshots_config.current_datetime.timestamp() * 1000
1113 is_debug: bool = log.isEnabledFor(LOG_DEBUG)
1114 if is_caching_snapshots(p, remote):
1115 props: dict[str, DatasetProperties] = self.dst_properties if remote is p.dst else self.src_properties
1116 snapshots_changed_dict: dict[str, int] = {dataset: vals.snapshots_changed for dataset, vals in props.items()}
1117 alerts_hash: str = sha256_128_urlsafe_base64(str(tuple(alerts)))
1118 label_hashes: dict[SnapshotLabel, str] = {
1119 label: sha256_128_urlsafe_base64(label.notimestamp_str()) for label in labels
1120 }
1121 is_caching: bool = False
1122 worst_alert: tuple[int, str, str] = (0, "", "") # -exit_code, exit_kind, exit_msg
1124 def record_alert(exit_code: int, exit_kind: str, exit_msg: str) -> None:
1125 nonlocal worst_alert
1126 worst_alert = min(worst_alert, (-exit_code, exit_kind, exit_msg)) # min() sorts "Latest" before "Oldest" on tie
1128 def monitor_last_modified_cache_file(r: Remote, dataset: str, label: SnapshotLabel, alert_cfg: AlertConfig) -> str:
1129 cache_label: str = os.path.join(MONITOR_CACHE_FILE_PREFIX, alert_cfg.kind[0], label_hashes[label], alerts_hash)
1130 return self.cache.last_modified_cache_file(r, dataset, cache_label)
1132 def alert_msg(
1133 kind: str, dataset: str, snapshot: str, label: SnapshotLabel, snapshot_age_millis: float, delta_millis: int
1134 ) -> str:
1135 assert kind == "Latest" or kind == "Oldest"
1136 lbl = f"{label.prefix}{label.infix}<timestamp>{label.suffix}"
1137 if snapshot_age_millis >= current_unixtime_millis:
1138 return f"No snapshot exists for {dataset}@{lbl}"
1139 msg = f"{kind} snapshot for {dataset}@{lbl} is {human_readable_duration(snapshot_age_millis, unit='ms')} old"
1140 s = f": @{snapshot}" if snapshot else ""
1141 if delta_millis == -1:
1142 return f"{msg}{s}"
1143 return f"{msg} but should be at most {human_readable_duration(delta_millis, unit='ms')} old{s}"
1145 def check_alert(
1146 label: SnapshotLabel, alert_cfg: AlertConfig | None, creation_unixtime_secs: int, dataset: str, snapshot: str
1147 ) -> None: # thread-safe
1148 if alert_cfg is None:
1149 return
1150 if is_caching and not p.dry_run: # update cache with latest state from 'zfs list -t snapshot'
1151 snapshots_changed: int = snapshots_changed_dict.get(dataset, 0)
1152 cache_file: str = monitor_last_modified_cache_file(remote, dataset, label, alert_cfg)
1153 set_last_modification_time_safe(
1154 cache_file, unixtime_in_secs=(creation_unixtime_secs, snapshots_changed), if_more_recent=True
1155 )
1156 warning_millis: int = alert_cfg.warning_millis
1157 critical_millis: int = alert_cfg.critical_millis
1158 alert_kind = alert_cfg.kind
1159 snapshot_age_millis: float = current_unixtime_millis - creation_unixtime_secs * 1000
1160 m = "--monitor_snapshots: "
1161 if snapshot_age_millis > critical_millis:
1162 msg = m + alert_msg(alert_kind, dataset, snapshot, label, snapshot_age_millis, critical_millis)
1163 log.critical("%s", msg)
1164 if not p.monitor_snapshots_config.dont_crit:
1165 record_alert(CRITICAL_STATUS, alert_kind, msg)
1166 elif snapshot_age_millis > warning_millis:
1167 msg = m + alert_msg(alert_kind, dataset, snapshot, label, snapshot_age_millis, warning_millis)
1168 log.warning("%s", msg)
1169 if not p.monitor_snapshots_config.dont_warn:
1170 record_alert(WARNING_STATUS, alert_kind, msg)
1171 elif is_debug:
1172 msg = m + "OK. " + alert_msg(alert_kind, dataset, snapshot, label, snapshot_age_millis, delta_millis=-1)
1173 log.debug("%s", msg)
1175 def alert_latest_snapshot(i: int, creation_unixtime_secs: int, dataset: str, snapshot: str) -> None:
1176 alert: MonitorSnapshotAlert = alerts[i]
1177 check_alert(alert.label, alert.latest, creation_unixtime_secs, dataset, snapshot)
1179 def alert_oldest_snapshot(i: int, creation_unixtime_secs: int, dataset: str, snapshot: str) -> None:
1180 alert: MonitorSnapshotAlert = alerts[i]
1181 check_alert(alert.label, alert.oldest, creation_unixtime_secs, dataset, snapshot)
1183 def find_stale_datasets_and_check_alerts() -> list[str]:
1184 """If the cache is enabled, check which datasets have changed to determine which datasets can be skipped cheaply,
1185 that is, without incurring 'zfs list -t snapshots'.
1187 This is done by comparing the "snapshots_changed" ZFS dataset property with the local cache. See
1188 https://openzfs.github.io/openzfs-docs/man/7/zfsprops.7.html#snapshots_changed
1189 """
1190 stale_datasets: list[str] = []
1191 time_threshold: float = time.time() - MATURITY_TIME_THRESHOLD_SECS
1192 for dataset in sorted_datasets:
1193 is_stale_dataset: bool = False
1194 snapshots_changed: int = snapshots_changed_dict.get(dataset, 0)
1195 for alert in alerts:
1196 for cfg in (alert.latest, alert.oldest):
1197 if cfg is None:
1198 continue
1199 if (
1200 snapshots_changed != 0
1201 and snapshots_changed < time_threshold
1202 and ( # always True
1203 cached_unix_times := self.cache.get_snapshots_changed2(
1204 monitor_last_modified_cache_file(remote, dataset, alert.label, cfg)
1205 )
1206 )
1207 and snapshots_changed == cached_unix_times[1] # cached snapshots_changed aka last modified time
1208 and snapshots_changed >= cached_unix_times[0] # creation time of minmax snapshot aka access time
1209 ): # cached state is still valid; emit an alert if the latest/oldest snapshot is too old
1210 lbl = alert.label
1211 check_alert(lbl, cfg, creation_unixtime_secs=cached_unix_times[0], dataset=dataset, snapshot="")
1212 else: # cached state is no longer valid; fallback to 'zfs list -t snapshot'
1213 is_stale_dataset = True
1214 if is_stale_dataset:
1215 stale_datasets.append(dataset)
1216 return stale_datasets
1218 # satisfy request from local cache as much as possible
1219 if is_caching_snapshots(p, remote):
1220 stale_datasets: list[str] = find_stale_datasets_and_check_alerts()
1221 with self.stats_lock:
1222 self.num_cache_misses += len(stale_datasets)
1223 self.num_cache_hits += len(sorted_datasets) - len(stale_datasets)
1224 else:
1225 stale_datasets = sorted_datasets
1227 # fallback to 'zfs list -t snapshot' for any remaining datasets, as these couldn't be satisfied from local cache
1228 is_caching = is_caching_snapshots(p, remote)
1229 datasets_without_snapshots: list[str] = self.handle_minmax_snapshots(
1230 remote,
1231 stale_datasets,
1232 labels,
1233 fn_latest=alert_latest_snapshot,
1234 fn_oldest=alert_oldest_snapshot,
1235 fn_oldest_skip_holds=oldest_skip_holds,
1236 )
1237 for dataset in datasets_without_snapshots:
1238 for i in range(len(alerts)):
1239 alert_latest_snapshot(i, creation_unixtime_secs=0, dataset=dataset, snapshot="")
1240 alert_oldest_snapshot(i, creation_unixtime_secs=0, dataset=dataset, snapshot="")
1241 return worst_alert
1243 def replicate_datasets(self, src_datasets: list[str], task_description: str, max_workers: int) -> bool:
1244 """Replicates a list of datasets."""
1245 assert (not self.is_test_mode) or src_datasets == sorted(src_datasets), "List is not sorted"
1246 p, log = self.params, self.params.log
1247 src, dst = p.src, p.dst
1248 self.num_snapshots_found = 0
1249 self.num_snapshots_replicated = 0
1250 log.info("Starting replication task: %s", task_description + f" [{len(src_datasets)} datasets]")
1251 start_time_nanos: int = time.monotonic_ns()
1253 def src2dst(src_dataset: str) -> str:
1254 return replace_prefix(src_dataset, old_prefix=src.root_dataset, new_prefix=dst.root_dataset)
1256 def dst2src(dst_dataset: str) -> str:
1257 return replace_prefix(dst_dataset, old_prefix=dst.root_dataset, new_prefix=src.root_dataset)
1259 def find_stale_datasets() -> tuple[list[str], dict[str, str]]:
1260 """If the cache is enabled on replication, check which src datasets or dst datasets have changed to determine
1261 which datasets can be skipped cheaply, i.e. without incurring 'zfs list -t snapshots'.
1263 This is done by comparing the "snapshots_changed" ZFS dataset property with the local cache. See
1264 https://openzfs.github.io/openzfs-docs/man/7/zfsprops.7.html#snapshots_changed
1265 """
1266 # First, check which src datasets have changed since the last replication to that destination
1267 cache_files: dict[str, str] = {}
1268 stale_src_datasets1: list[str] = []
1269 maybe_stale_dst_datasets: list[str] = []
1270 userhost_dir: str = sha256_85_urlsafe_base64(p.dst.cache_namespace())
1271 # filter_key: cache is only valid if same --include/excl-snapshot*, --skip-missing-snapshots, --no-use-bookmark
1272 filter_key: tuple = (p.snapshot_filters, p.skip_missing_snapshots, p.use_bookmark)
1273 filter_hash_code: str = sha256_85_urlsafe_base64(str(filter_key))
1274 for src_dataset in src_datasets:
1275 dst_dataset: str = src2dst(src_dataset) # cache is only valid for identical destination dataset
1276 dst_dataset_dir: str = sha256_85_urlsafe_base64(dst_dataset)
1277 cache_label: str = os.path.join(
1278 REPLICATION_CACHE_FILE_PREFIX, userhost_dir, dst_dataset_dir, filter_hash_code
1279 )
1280 cache_file: str = self.cache.last_modified_cache_file(src, src_dataset, cache_label)
1281 cache_files[src_dataset] = cache_file
1282 snapshots_changed: int = self.src_properties[src_dataset].snapshots_changed # get prop "for free"
1283 if (
1284 snapshots_changed != 0
1285 and time.time() > snapshots_changed + MATURITY_TIME_THRESHOLD_SECS
1286 and snapshots_changed == self.cache.get_snapshots_changed(cache_file)
1287 ):
1288 maybe_stale_dst_datasets.append(dst_dataset)
1289 else:
1290 stale_src_datasets1.append(src_dataset)
1292 # For each src dataset that hasn't changed, check if the corresponding dst dataset has changed
1293 stale_src_datasets2: list[str] = []
1294 dst_snapshots_changed_dict: dict[str, int] = self.cache.zfs_get_snapshots_changed(dst, maybe_stale_dst_datasets)
1295 for dst_dataset in maybe_stale_dst_datasets:
1296 snapshots_changed = dst_snapshots_changed_dict.get(dst_dataset, 0)
1297 if (
1298 snapshots_changed != 0
1299 and time.time() > snapshots_changed + MATURITY_TIME_THRESHOLD_SECS
1300 and snapshots_changed
1301 == self.cache.get_snapshots_changed(self.cache.last_modified_cache_file(dst, dst_dataset))
1302 ):
1303 log.info("Already up-to-date [cached]: %s", dst_dataset)
1304 else:
1305 stale_src_datasets2.append(dst2src(dst_dataset))
1306 assert (not self.is_test_mode) or stale_src_datasets1 == sorted(stale_src_datasets1), "List is not sorted"
1307 assert (not self.is_test_mode) or stale_src_datasets2 == sorted(stale_src_datasets2), "List is not sorted"
1308 stale_src_datasets = list(heapq.merge(stale_src_datasets1, stale_src_datasets2)) # merge two sorted lists
1309 assert (not self.is_test_mode) or not has_duplicates(stale_src_datasets), "List contains duplicates"
1310 return stale_src_datasets, cache_files
1312 if is_caching_snapshots(p, src):
1313 stale_src_datasets, cache_files = find_stale_datasets()
1314 num_cache_misses = len(stale_src_datasets)
1315 num_cache_hits = len(src_datasets) - len(stale_src_datasets)
1316 self.num_cache_misses += num_cache_misses
1317 self.num_cache_hits += num_cache_hits
1318 cmsg = self._cache_hits_msg(hits=num_cache_hits, misses=num_cache_misses)
1319 else:
1320 stale_src_datasets = src_datasets
1321 cache_files = {}
1322 cmsg = ""
1324 done_src_datasets: list[str] = []
1325 done_src_datasets_lock: threading.Lock = threading.Lock()
1327 def _process_dataset_fn(src_dataset: str, tid: str, retry: Retry) -> bool:
1328 result: bool = replicate_dataset(job=self, src_dataset=src_dataset, tid=tid, retry=retry)
1329 with done_src_datasets_lock:
1330 done_src_datasets.append(src_dataset) # record datasets that were actually replicated (not skipped)
1331 return result
1333 # Run replicate_dataset(dataset) for each dataset, while taking care of errors, retries + parallel execution
1334 failed: bool = process_datasets_in_parallel_and_fault_tolerant(
1335 log=log,
1336 datasets=stale_src_datasets,
1337 process_dataset=_process_dataset_fn,
1338 skip_tree_on_error=lambda dataset: not self.dst_dataset_exists[src2dst(dataset)],
1339 skip_on_error=p.skip_on_error,
1340 max_workers=max_workers,
1341 timing=self.task_timing,
1342 termination_handler=self.terminate,
1343 enable_barriers=False,
1344 task_name="Replication",
1345 append_exception=self.append_exception,
1346 retry_template=self._retry_template(),
1347 dry_run=p.dry_run,
1348 is_test_mode=self.is_test_mode,
1349 )
1351 if is_caching_snapshots(p, src) and len(done_src_datasets) > 0:
1352 # refresh "snapshots_changed" ZFS dataset property from dst
1353 stale_dst_datasets: list[str] = [src2dst(src_dataset) for src_dataset in sorted(done_src_datasets)]
1354 dst_snapshots_changed_dict: dict[str, int] = self.cache.zfs_get_snapshots_changed(dst, stale_dst_datasets)
1355 for dst_dataset in stale_dst_datasets: # update local cache
1356 dst_snapshots_changed: int = dst_snapshots_changed_dict.get(dst_dataset, 0)
1357 dst_cache_file: str = self.cache.last_modified_cache_file(dst, dst_dataset)
1358 src_dataset: str = dst2src(dst_dataset)
1359 src_snapshots_changed: int = self.src_properties[src_dataset].snapshots_changed
1360 if not p.dry_run:
1361 set_last_modification_time_safe(
1362 cache_files[src_dataset], unixtime_in_secs=src_snapshots_changed, if_more_recent=True
1363 )
1364 set_last_modification_time_safe(
1365 dst_cache_file, unixtime_in_secs=dst_snapshots_changed, if_more_recent=True
1366 )
1368 elapsed_nanos: int = time.monotonic_ns() - start_time_nanos
1369 log.info(
1370 p.dry("Replication done: %s"),
1371 f"{task_description} [Replicated {self.num_snapshots_replicated} out of {self.num_snapshots_found} snapshots"
1372 f" within {len(src_datasets)} datasets; took {human_readable_duration(elapsed_nanos)}{cmsg}]",
1373 )
1374 return failed
1376 def maybe_inject_delete(self, remote: Remote, dataset: str, delete_trigger: str) -> None:
1377 """For testing only; for unit tests to delete datasets during replication and test correct handling of that."""
1378 assert delete_trigger
1379 counter = self.delete_injection_triggers.get("before")
1380 if counter and self.decrement_injection_counter(counter, delete_trigger):
1381 p = self.params
1382 cmd = p.split_args(f"{remote.sudo} {p.zfs_program} destroy -r", p.force_unmount, p.force_hard, dataset or "")
1383 self.run_ssh_command(remote, LOG_DEBUG, print_stdout=True, cmd=cmd)
1385 def maybe_inject_params(self, error_trigger: str) -> None:
1386 """For testing only; for unit tests to simulate errors during replication and test correct handling of them."""
1387 assert error_trigger
1388 counter = self.error_injection_triggers.get("before")
1389 if counter and self.decrement_injection_counter(counter, error_trigger):
1390 self.inject_params = self.param_injection_triggers[error_trigger]
1391 elif error_trigger in self.param_injection_triggers:
1392 self.inject_params = {}
1394 @staticmethod
1395 def recv_option_property_names(recv_opts: list[str]) -> set[str]:
1396 """Extracts -o and -x property names that are already specified on the command line; This can be used to check for
1397 dupes because 'zfs receive' does not accept multiple -o or -x options with the same property name."""
1398 propnames: set[str] = set()
1399 i = 0
1400 n = len(recv_opts)
1401 while i < n:
1402 stripped: str = recv_opts[i].strip()
1403 if stripped in ("-o", "-x"):
1404 i += 1
1405 if i == n or recv_opts[i].strip() in ("-o", "-x"):
1406 die(f"Missing value for {stripped} option in --zfs-recv-program-opt(s): {' '.join(recv_opts)}")
1407 if stripped == "-o" and "=" not in recv_opts[i]:
1408 die(f"Missing value for {stripped} name=value pair in --zfs-recv-program-opt(s): {' '.join(recv_opts)}")
1409 propname: str = recv_opts[i] if stripped == "-x" else recv_opts[i].split("=", 1)[0]
1410 validate_property_name(propname, "--zfs-recv-program-opt(s)")
1411 propnames.add(propname)
1412 i += 1
1413 return propnames
1415 def root_datasets_if_recursive_zfs_snapshot_is_possible(
1416 self, datasets: list[str], basis_datasets: list[str]
1417 ) -> list[str] | None:
1418 """Returns the root datasets within the (filtered) `datasets` list if no incompatible pruning is detected. A dataset
1419 within `datasets` is considered a root dataset if it has no parent, i.e. it is not a descendant of any dataset in
1420 `datasets`. Returns `None` if any (unfiltered) dataset in `basis_dataset` that is a descendant of at least one of the
1421 root datasets is missing in `datasets`, indicating that --include/exclude-dataset* or the snapshot schedule have
1422 pruned a dataset in a way that is incompatible with 'zfs snapshot -r' CLI semantics, thus requiring a switch to the
1423 non-recursive 'zfs snapshot snapshot1 .. snapshot N' CLI flavor.
1425 Assumes that set(datasets).issubset(set(basis_datasets)). Also assumes that datasets and basis_datasets are both
1426 sorted (and thus the output root_datasets list is sorted too), which is why this algorithm is efficient - O(N) time
1427 complexity. The impl is akin to the merge algorithm of a merge sort, adapted to our specific use case.
1428 See root_datasets_if_recursive_zfs_snapshot_is_possible_slow_but_correct() in the unit test suite for an alternative
1429 impl that's easier to grok.
1430 """
1431 assert (not self.is_test_mode) or datasets == sorted(datasets), "List is not sorted"
1432 assert (not self.is_test_mode) or not has_duplicates(datasets), "List contains duplicates"
1433 assert (not self.is_test_mode) or basis_datasets == sorted(basis_datasets), "List is not sorted"
1434 assert (not self.is_test_mode) or not has_duplicates(basis_datasets), "List contains duplicates"
1435 assert (not self.is_test_mode) or set(datasets).issubset(set(basis_datasets)), "Not a subset"
1436 root_datasets: list[str] = self.find_root_datasets(datasets)
1437 i = 0
1438 j = 0
1439 k = 0
1440 len_root_datasets = len(root_datasets)
1441 len_basis_datasets = len(basis_datasets)
1442 len_datasets = len(datasets)
1443 while i < len_root_datasets and j < len_basis_datasets: # walk and "merge" the sorted lists, in sync
1444 if basis_datasets[j] < root_datasets[i]: # irrelevant subtree?
1445 j += 1 # move to next basis_datasets[j]
1446 elif is_descendant(basis_datasets[j], of_root_dataset=root_datasets[i]): # relevant subtree?
1447 while k < len_datasets and datasets[k] < basis_datasets[j]:
1448 k += 1 # move to next datasets[k]
1449 if k == len_datasets or datasets[k] != basis_datasets[j]: # dataset chopped off by schedule or --incl/excl*?
1450 return None # detected filter pruning that is incompatible with 'zfs snapshot -r'
1451 j += 1 # move to next basis_datasets[j]
1452 else:
1453 i += 1 # move to next root_dataset[i]; no need to check root_datasets that are no longer (or not yet) reachable
1454 return root_datasets
1456 @staticmethod
1457 def find_root_datasets(sorted_datasets: list[str]) -> list[str]:
1458 """Returns the roots of the subtrees in the (sorted) input datasets; The output root dataset list is sorted, too; A
1459 dataset is a root dataset if it has no parent, i.e. it is not a descendant of any dataset in the input datasets."""
1460 root_datasets: list[str] = []
1461 skip_dataset: str = DONT_SKIP_DATASET
1462 for dataset in sorted_datasets:
1463 if is_descendant(dataset, of_root_dataset=skip_dataset):
1464 continue
1465 skip_dataset = dataset
1466 root_datasets.append(dataset)
1467 return root_datasets
1469 def find_datasets_to_snapshot(self, sorted_datasets: list[str]) -> dict[SnapshotLabel, list[str]]:
1470 """Given a (sorted) list of source datasets, returns a dict where the key is a snapshot name (aka SnapshotLabel, e.g.
1471 bzfs_2024-11-06_08:30:05_hourly) and the value is the (sorted) (sub)list of datasets for which a snapshot needs to be
1472 created with that name, because these datasets are due per the schedule, either because the 'creation' time of their
1473 most recent snapshot with that name pattern is now too old, or such a snapshot does not even exist.
1475 The baseline implementation uses the 'zfs list -t snapshot' CLI to find the most recent snapshots, which is simple
1476 but doesn't scale well with the number of snapshots, at least if the goal is to take snapshots every second. An
1477 alternative, much more scalable, implementation queries the standard ZFS "snapshots_changed" dataset property
1478 (requires zfs >= 2.2.0), in combination with a local cache that stores this property, as well as the creation time of
1479 the most recent snapshot, for each SnapshotLabel and each dataset.
1480 """
1481 p, log = self.params, self.params.log
1482 src = p.src
1483 config: CreateSrcSnapshotConfig = p.create_src_snapshots_config
1484 datasets_to_snapshot: dict[SnapshotLabel, list[str]] = defaultdict(list)
1485 is_caching: bool = False
1486 interner: HashedInterner[datetime] = HashedInterner() # reduces memory footprint
1487 msgs: list[tuple[datetime, str, SnapshotLabel, str]] = []
1489 def create_snapshot_if_latest_is_too_old(
1490 datasets_to_snapshot: dict[SnapshotLabel, list[str]], dataset: str, label: SnapshotLabel, creation_unixtime: int
1491 ) -> None: # thread-safe
1492 """Schedules creation of a snapshot for the given label if the label's existing latest snapshot is too old."""
1493 creation_dt: datetime = datetime.fromtimestamp(creation_unixtime, tz=config.tz)
1494 log.log(LOG_TRACE, "Latest snapshot creation: %s for %s", creation_dt, label)
1495 duration_amount, duration_unit = config.suffix_durations[label.suffix]
1496 next_event_dt: datetime = config.anchors.round_datetime_up_to_duration_multiple(
1497 creation_dt + timedelta(microseconds=1), duration_amount, duration_unit
1498 )
1499 msg: str = ""
1500 if config.current_datetime >= next_event_dt:
1501 datasets_to_snapshot[label].append(dataset) # mark it as scheduled for snapshot creation
1502 msg = " has passed"
1503 next_event_dt = interner.intern(next_event_dt)
1504 msgs.append((next_event_dt, dataset, label, msg))
1505 if is_caching and not p.dry_run: # update cache with latest state from 'zfs list -t snapshot'
1506 # Per-label cache stores (atime=creation, mtime=snapshots_changed) so later runs can safely trust creation
1507 # only when the label's mtime matches the current dataset-level '=' cache value. Excludes timestamp of label.
1508 cache_file: str = self.cache.last_modified_cache_file(src, dataset, label_hashes[label])
1509 unixtimes: tuple[int, int] = (creation_unixtime, self.src_properties[dataset].snapshots_changed)
1510 set_last_modification_time_safe(cache_file, unixtime_in_secs=unixtimes, if_more_recent=True)
1512 labels: list[SnapshotLabel] = []
1513 config_labels: list[SnapshotLabel] = config.snapshot_labels()
1514 for label in config_labels:
1515 duration_amount_, _duration_unit = config.suffix_durations[label.suffix]
1516 if duration_amount_ == 0 or config.create_src_snapshots_even_if_not_due:
1517 datasets_to_snapshot[label] = sorted_datasets # take snapshot regardless of creation time of existing snaps
1518 else:
1519 labels.append(label)
1520 if len(labels) == 0:
1521 return datasets_to_snapshot # nothing more TBD
1522 label_hashes: dict[SnapshotLabel, str] = {
1523 label: sha256_128_urlsafe_base64(label.notimestamp_str()) for label in labels
1524 }
1526 # satisfy request from local cache as much as possible
1527 cached_datasets_to_snapshot: dict[SnapshotLabel, list[str]] = defaultdict(list)
1528 if is_caching_snapshots(p, src):
1529 sorted_datasets_todo: list[str] = []
1530 time_threshold: float = time.time() - MATURITY_TIME_THRESHOLD_SECS
1531 for dataset in sorted_datasets:
1532 cache: SnapshotCache = self.cache
1533 cached_snapshots_changed: int = cache.get_snapshots_changed(cache.last_modified_cache_file(src, dataset))
1534 if cached_snapshots_changed == 0:
1535 sorted_datasets_todo.append(dataset) # request cannot be answered from cache
1536 continue
1537 if cached_snapshots_changed != self.src_properties[dataset].snapshots_changed: # get that prop "for free"
1538 cache.invalidate_last_modified_cache_dataset(dataset)
1539 sorted_datasets_todo.append(dataset) # request cannot be answered from cache
1540 continue
1541 if cached_snapshots_changed >= time_threshold: # Avoid equal-second races: only trust matured cache entries
1542 sorted_datasets_todo.append(dataset) # cache entry isn't mature enough to be trusted; skip cache
1543 continue
1544 creation_unixtimes: list[int] = []
1545 for label_hash in label_hashes.values():
1546 # For per-label files, atime stores the latest matching snapshot's creation time, while mtime stores
1547 # the dataset-level snapshots_changed observed when this label file was written.
1548 atime, mtime = cache.get_snapshots_changed2(cache.last_modified_cache_file(src, dataset, label_hash))
1549 # Sanity check: trust per-label cache only when:
1550 # - mtime equals the dataset-level '=' cache (same snapshots_changed), and
1551 # - atime is plausible and not later than mtime (creation <= snapshots_changed), and
1552 # - neither atime nor mtime is zero (unknown provenance).
1553 # Otherwise fall back to 'zfs list -t snapshot' to avoid stale creation times after newer changes.
1554 if atime == 0 or mtime == 0 or mtime != cached_snapshots_changed or atime > mtime:
1555 sorted_datasets_todo.append(dataset) # request cannot be answered from cache
1556 break
1557 creation_unixtimes.append(atime)
1558 if len(creation_unixtimes) == len(labels):
1559 for j, label in enumerate(labels):
1560 create_snapshot_if_latest_is_too_old(
1561 cached_datasets_to_snapshot, dataset, label, creation_unixtimes[j]
1562 )
1563 sorted_datasets = sorted_datasets_todo
1565 def create_snapshot_fn(i: int, creation_unixtime_secs: int, dataset: str, snapshot: str) -> None:
1566 create_snapshot_if_latest_is_too_old(datasets_to_snapshot, dataset, labels[i], creation_unixtime_secs)
1568 def on_finish_dataset(dataset: str) -> None:
1569 if is_caching_snapshots(p, src) and not p.dry_run:
1570 set_last_modification_time_safe(
1571 self.cache.last_modified_cache_file(src, dataset),
1572 unixtime_in_secs=self.src_properties[dataset].snapshots_changed,
1573 if_more_recent=True,
1574 )
1576 # fallback to 'zfs list -t snapshot' for any remaining datasets, as these couldn't be satisfied from local cache
1577 is_caching = is_caching_snapshots(p, src)
1578 datasets_without_snapshots: list[str] = self.handle_minmax_snapshots(
1579 src, sorted_datasets, labels, fn_latest=create_snapshot_fn, fn_on_finish_dataset=on_finish_dataset
1580 )
1581 for lbl in labels: # merge (sorted) results from local cache + 'zfs list -t snapshot' into (sorted) combined result
1582 datasets_to_snapshot[lbl].sort()
1583 if datasets_without_snapshots or (lbl in cached_datasets_to_snapshot): # +take snaps for snapshot-less datasets
1584 datasets_to_snapshot[lbl] = list( # inputs to merge() are sorted, and outputs are sorted too
1585 heapq.merge(datasets_to_snapshot[lbl], cached_datasets_to_snapshot[lbl], datasets_without_snapshots)
1586 )
1587 for label, datasets in datasets_to_snapshot.items():
1588 assert (not self.is_test_mode) or datasets == sorted(datasets), "List is not sorted"
1589 assert (not self.is_test_mode) or not has_duplicates(datasets), "List contains duplicates"
1590 assert label
1592 msgs.sort() # sort by time, dataset, label
1593 for i in range(0, len(msgs), 10_000): # reduce logging overhead via mini-batching
1594 text = "".join(
1595 f"\nNext scheduled snapshot time: {next_event_dt} for {dataset}@{label}{msg}"
1596 for next_event_dt, dataset, label, msg in msgs[i : i + 10_000]
1597 )
1598 log.info("Next scheduled snapshot times ...%s", text)
1600 # sort keys to ensure that we take snapshots for dailies before hourlies, and so on
1601 datasets_to_snapshot = {lbl: datasets_to_snapshot[lbl] for lbl in config_labels if lbl in datasets_to_snapshot}
1602 return datasets_to_snapshot
1604 def handle_minmax_snapshots(
1605 self,
1606 remote: Remote,
1607 sorted_datasets: list[str],
1608 labels: list[SnapshotLabel],
1609 *,
1610 fn_latest: Callable[[int, int, str, str], None], # callback function for latest snapshot
1611 fn_oldest: Callable[[int, int, str, str], None] | None = None, # callback function for oldest snapshot
1612 fn_oldest_skip_holds: Sequence[bool] = (),
1613 fn_on_finish_dataset: Callable[[str], None] = lambda dataset: None,
1614 ) -> list[str]: # thread-safe
1615 """For each dataset in `sorted_datasets`, for each label in `labels`, finds the latest and oldest snapshot, and runs
1616 the callback functions on them; Ignores the timestamp of the input labels and the timestamp of the snapshot names.
1618 If the (optional) fn_oldest_skip_holds=True for a given label, then snapshots for that label that carry a 'zfs hold'
1619 are skipped (ignored) when finding the oldest snapshot for fn_oldest. This can be useful for monitor_snapshots(),
1620 given that users often intentionally retain holds for longer than the "normal" snapshot retention period, and this
1621 shouldn't necessarily cause monitoring to emit alerts.
1622 """
1623 if fn_oldest is not None:
1624 assert len(labels) == len(fn_oldest_skip_holds)
1625 assert (not self.is_test_mode) or sorted_datasets == sorted(sorted_datasets), "List is not sorted"
1627 def extract_fields(line: str) -> tuple[int, int, str, bool]:
1628 fields: list[str] = line.split("\t")
1629 if len(fields) == 3:
1630 name, createtxg, creation_unixtime_secs = fields
1631 userrefs = ""
1632 else:
1633 name, createtxg, creation_unixtime_secs, userrefs = fields
1634 return (
1635 int(createtxg),
1636 int(creation_unixtime_secs),
1637 name.split("@", 1)[1],
1638 userrefs in ("", "-", "0"), # ZFS snapshot property userrefs > 0 indicates a zfs hold
1639 )
1641 p = self.params
1642 props: str = "name,createtxg,creation"
1643 props = props if fn_oldest is None or not any(fn_oldest_skip_holds) else props + ",userrefs"
1644 cmd = p.split_args(f"{p.zfs_program} list -t snapshot -d 1 -Hp -o {props}") # sorts by dataset,creation
1645 datasets_with_snapshots: set[str] = set()
1646 interner: SortedInterner[str] = SortedInterner(sorted_datasets) # reduces memory footprint
1647 for lines in zfs_list_snapshots_in_parallel(self, remote, cmd, sorted_datasets, ordered=False):
1648 # streaming group by dataset name (consumes constant memory only)
1649 for dataset, group in itertools.groupby(lines, key=lambda line: line.split("\t", 1)[0].split("@", 1)[0]):
1650 dataset = interner.interned(dataset)
1651 snapshots = sorted( # fetch all snapshots of current dataset and sort by createtxg,creation,name
1652 extract_fields(line) for line in group
1653 ) # perf: sorted() is fast because Powersort is close to O(N) for nearly sorted input, which is our case
1654 assert len(snapshots) > 0
1655 datasets_with_snapshots.add(dataset)
1656 snapshot_names: tuple[str, ...] = tuple(snapshot[2] for snapshot in snapshots)
1657 year_with_4_digits_regex: re.Pattern[str] = YEAR_WITH_FOUR_DIGITS_REGEX
1658 year_with_4_digits_regex_fullmatch = year_with_4_digits_regex.fullmatch
1659 startswith = str.startswith
1660 endswith = str.endswith
1661 fns = ((fn_latest, True),) if fn_oldest is None else ((fn_latest, True), (fn_oldest, False))
1662 for i, label in enumerate(labels):
1663 infix: str = label.infix
1664 start: str = label.prefix + infix
1665 end: str = label.suffix
1666 startlen: int = len(start)
1667 endlen: int = len(end)
1668 minlen: int = startlen + endlen if infix else 4 + startlen + endlen # year_with_four_digits_regex
1669 startlen_4: int = startlen + 4 # [startlen:startlen+4] # year_with_four_digits_regex
1670 has_infix: bool = bool(infix)
1671 for fn, is_reverse in fns:
1672 creation_unixtime_secs: int = 0 # find creation time of latest or oldest snapshot matching the label
1673 minmax_snapshot: str = ""
1674 no_skip_holds: bool = is_reverse or not fn_oldest_skip_holds[i]
1675 for j in range(len(snapshot_names) - 1, -1, -1) if is_reverse else range(len(snapshot_names)):
1676 snapshot_name: str = snapshot_names[j]
1677 if (
1678 endswith(snapshot_name, end) # aka snapshot_name.endswith(end)
1679 and startswith(snapshot_name, start) # aka snapshot_name.startswith(start)
1680 and len(snapshot_name) >= minlen
1681 and (has_infix or year_with_4_digits_regex_fullmatch(snapshot_name, startlen, startlen_4))
1682 and (no_skip_holds or snapshots[j][3])
1683 ):
1684 creation_unixtime_secs = snapshots[j][1]
1685 minmax_snapshot = snapshot_name
1686 break
1687 fn(i, creation_unixtime_secs, dataset, minmax_snapshot)
1688 fn_on_finish_dataset(dataset)
1689 datasets_without_snapshots = [dataset for dataset in sorted_datasets if dataset not in datasets_with_snapshots]
1690 return datasets_without_snapshots
1692 @staticmethod
1693 def _cache_hits_msg(hits: int, misses: int) -> str:
1694 total = hits + misses
1695 return f", cache hits: {percent(hits, total, print_total=True)}, misses: {percent(misses, total, print_total=True)}"
1697 def run_ssh_command(
1698 self,
1699 remote: MiniRemote,
1700 loglevel: int = logging.INFO,
1701 is_dry: bool = False,
1702 check: bool = True,
1703 print_stdout: bool = False,
1704 print_stderr: bool = True,
1705 cmd: list[str] | None = None,
1706 retry_on_generic_ssh_error: bool = True,
1707 ) -> str:
1708 """Runs the given CLI cmd via ssh on the given remote, and returns stdout."""
1709 assert cmd is not None and isinstance(cmd, list) and len(cmd) > 0
1710 conn_pool: ConnectionPool = self.params.connection_pools[remote.location].pool(SHARED)
1711 with conn_pool.connection() as conn:
1712 log: logging.Logger = self.params.log
1713 try:
1714 process: subprocess.CompletedProcess[str] = conn.run_ssh_command(
1715 cmd=cmd,
1716 job=self,
1717 loglevel=loglevel,
1718 is_dry=is_dry,
1719 check=check,
1720 stdin=DEVNULL,
1721 stdout=PIPE,
1722 stderr=PIPE,
1723 text=True,
1724 )
1725 except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as e:
1726 xprint(log, stderr_to_str(e.stdout) if print_stdout else e.stdout, run=print_stdout, file=sys.stdout, end="")
1727 xprint(log, stderr_to_str(e.stderr) if print_stderr else e.stderr, run=print_stderr, file=sys.stderr, end="")
1728 if retry_on_generic_ssh_error and isinstance(e, subprocess.CalledProcessError):
1729 stderr: str = stderr_to_str(e.stderr)
1730 if stderr.startswith("ssh: "):
1731 assert e.returncode == 255, e.returncode # error within SSH itself (not during the remote command)
1732 raise RetryableError(display_msg="ssh") from e
1733 raise
1734 else:
1735 if is_dry:
1736 return ""
1737 xprint(log, process.stdout, run=print_stdout, file=sys.stdout, end="")
1738 xprint(log, process.stderr, run=print_stderr, file=sys.stderr, end="")
1739 return process.stdout
1741 def try_ssh_command(
1742 self,
1743 remote: MiniRemote,
1744 loglevel: int,
1745 is_dry: bool = False,
1746 print_stdout: bool = False,
1747 cmd: list[str] | None = None,
1748 exists: bool = True,
1749 error_trigger: str | None = None,
1750 ) -> str | None:
1751 """Convenience method that helps retry/react to a dataset or pool that potentially doesn't exist anymore."""
1752 assert cmd is not None and isinstance(cmd, list) and len(cmd) > 0
1753 log = self.params.log
1754 try:
1755 self.maybe_inject_error(cmd=cmd, error_trigger=error_trigger)
1756 return self.run_ssh_command(remote=remote, loglevel=loglevel, is_dry=is_dry, print_stdout=print_stdout, cmd=cmd)
1757 except (subprocess.CalledProcessError, UnicodeDecodeError) as e:
1758 if not isinstance(e, UnicodeDecodeError):
1759 stderr: str = stderr_to_str(e.stderr)
1760 if exists and (
1761 ": dataset does not exist" in stderr
1762 or ": filesystem does not exist" in stderr # solaris 11.4.0
1763 or ": no such pool" in stderr
1764 or "does not have any resumable receive state to abort" in stderr # harmless `zfs receive -A` race
1765 or re.search(r"bookmark '.*' does not exist", stderr)
1766 ):
1767 return None
1768 log.warning("%s", stderr.rstrip())
1769 raise RetryableError("Subprocess failed") from e
1771 def try_ssh_command_with_retries(self, *args: Any, **kwargs: Any) -> str | None:
1772 """Convenience method that auto-retries try_ssh_command() on failure."""
1773 return self._retry_template().call_with_retries(fn=lambda retry: self.try_ssh_command(*args, **kwargs))
1775 def run_ssh_command_with_retries(self, *args: Any, **kwargs: Any) -> str:
1776 """Convenience method that auto-retries run_ssh_command() on transport failure (not on remote command failure)."""
1777 return self._retry_template().call_with_retries(fn=lambda retry: self.run_ssh_command(*args, **kwargs))
1779 def maybe_inject_error(self, cmd: list[str], error_trigger: str | None = None) -> None:
1780 """For testing only; for unit tests to simulate errors during replication and test correct handling of them."""
1781 if error_trigger:
1782 counter = self.error_injection_triggers.get("before")
1783 if counter and self.decrement_injection_counter(counter, error_trigger):
1784 try:
1785 raise CalledProcessError(returncode=1, cmd=" ".join(cmd), stderr=error_trigger + ":dataset is busy")
1786 except subprocess.CalledProcessError as e:
1787 if error_trigger.startswith("retryable_"):
1788 raise RetryableError("Subprocess failed") from e
1789 else:
1790 raise
1792 def decrement_injection_counter(self, counter: Counter[str], trigger: str) -> bool:
1793 """For testing only."""
1794 with self.injection_lock:
1795 if counter[trigger] <= 0:
1796 return False
1797 counter[trigger] -= 1
1798 return True
1801#############################################################################
1802@final
1803class DatasetProperties:
1804 """Properties of a ZFS dataset."""
1806 __slots__ = ("recordsize", "snapshots_changed") # uses more compact memory layout than __dict__
1808 def __init__(self, recordsize: int, snapshots_changed: int) -> None:
1809 # immutable variables:
1810 self.recordsize: Final[int] = recordsize
1812 # mutable variables:
1813 self.snapshots_changed: int = snapshots_changed
1816#############################################################################
1817# Input format is [[user@]host:]dataset
1818# 1234 5 6
1819_DATASET_LOCATOR_REGEX: Final[re.Pattern[str]] = re.compile(r"(((([^@]*)@)?([^:]+)):)?(.*)", flags=re.DOTALL)
1822def parse_dataset_locator(
1823 input_text: str, validate: bool = True, user: str | None = None, host: str | None = None, port: int | None = None
1824) -> tuple[str, str, str, str, str]:
1825 """Splits user@host:dataset into its components with optional checks."""
1827 def convert_ipv6(hostname: str) -> str: # support IPv6 without getting confused by host:dataset colon separator ...
1828 return hostname.replace("|", ":") # ... and any colons that may be part of a (valid) ZFS dataset name
1830 user_undefined: bool = user is None
1831 if user is None:
1832 user = ""
1833 host_undefined: bool = host is None
1834 if host is None:
1835 host = ""
1836 host = convert_ipv6(host)
1837 user_host, dataset, pool = "", "", ""
1839 if match := _DATASET_LOCATOR_REGEX.fullmatch(input_text): 1839 ↛ 1856line 1839 didn't jump to line 1856 because the condition on line 1839 was always true
1840 if user_undefined:
1841 user = match.group(4) or ""
1842 if host_undefined:
1843 host = match.group(5) or ""
1844 host = convert_ipv6(host)
1845 if host == "-":
1846 host = ""
1847 dataset = match.group(6) or ""
1848 i = dataset.find("/")
1849 pool = dataset[0:i] if i >= 0 else dataset
1851 if user and host:
1852 user_host = f"{user}@{host}"
1853 elif host:
1854 user_host = host
1856 if validate:
1857 validate_user_name(user, input_text)
1858 validate_host_name(host, input_text)
1859 if port is not None:
1860 validate_port(port, f"Invalid port number: '{port}' for: '{input_text}' - ")
1861 validate_dataset_name(dataset, input_text)
1863 return user, host, user_host, pool, dataset
1866def validate_user_name(user: str, input_text: str) -> None:
1867 """Checks that the username is safe for ssh or local usage."""
1868 invalid_chars: str = SHELL_CHARS_AND_SLASH
1869 if user and (user.startswith("-") or ".." in user or any(c.isspace() or c in invalid_chars for c in user)):
1870 die(f"Invalid user name: '{user}' for: '{input_text}'")
1873def validate_host_name(host: str, input_text: str) -> None:
1874 """Checks hostname for forbidden characters or patterns."""
1875 invalid_chars: str = SHELL_CHARS_AND_SLASH
1876 if host and (host.startswith("-") or ".." in host or any(c.isspace() or c in invalid_chars for c in host)):
1877 die(f"Invalid host name: '{host}' for: '{input_text}'")
1880def validate_port(port: str | int | None, message: str) -> None:
1881 """Checks that port specification is a valid integer."""
1882 if isinstance(port, int):
1883 port = str(port)
1884 if port and not port.isdigit():
1885 die(message + f"must be empty or a positive integer: '{port}'")
1888def normalize_called_process_error(error: subprocess.CalledProcessError) -> int:
1889 """Normalizes `CalledProcessError.returncode` to avoid reserved exit codes so callers don't misclassify them."""
1890 ret: int = error.returncode
1891 ret = DIE_STATUS if isinstance(ret, int) and 1 <= ret <= STILL_RUNNING_STATUS else ret
1892 return ret
1895#############################################################################
1896if __name__ == "__main__": 1896 ↛ 1897line 1896 didn't jump to line 1897 because the condition on line 1896 was never true
1897 main()