Coverage for bzfs_main/bzfs.py: 99%
1165 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-19 18:44 +0000
1# Copyright 2024 Wolfgang Hoschek AT mac DOT com
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7# http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
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 heapq
48import itertools
49import logging
50import os
51import re
52import subprocess
53import sys
54import threading
55import time
56from collections import (
57 Counter,
58 defaultdict,
59)
60from collections.abc import (
61 Collection,
62 Sequence,
63)
64from datetime import (
65 datetime,
66 timedelta,
67)
68from logging import (
69 Logger,
70)
71from subprocess import (
72 DEVNULL,
73 PIPE,
74 CalledProcessError,
75 TimeoutExpired,
76)
77from typing import (
78 Any,
79 Callable,
80 Final,
81 cast,
82 final,
83)
85import bzfs_main.loggers
86from bzfs_main.argparse_actions import (
87 has_timerange_filter,
88)
89from bzfs_main.argparse_cli import (
90 EXCLUDE_DATASET_REGEXES_DEFAULT,
91)
92from bzfs_main.compare_snapshot_lists import (
93 run_compare_snapshot_lists,
94)
95from bzfs_main.configuration import (
96 AlertConfig,
97 CreateSrcSnapshotConfig,
98 LogParams,
99 MonitorSnapshotAlert,
100 Params,
101 Remote,
102 SnapshotLabel,
103 resolve_r2r_mode,
104)
105from bzfs_main.detect import (
106 DISABLE_PRG,
107 RemoteConfCacheItem,
108 are_bookmarks_enabled,
109 detect_available_programs,
110 is_caching_snapshots,
111 is_dummy,
112 is_zpool_feature_enabled_or_active,
113)
114from bzfs_main.filter import (
115 SNAPSHOT_REGEX_FILTER_NAME,
116 dataset_regexes,
117 filter_datasets,
118 filter_lines,
119 filter_lines_except,
120 filter_snapshots,
121)
122from bzfs_main.loggers import (
123 get_simple_logger,
124 reset_logger,
125 set_logging_runtime_defaults,
126)
127from bzfs_main.parallel_batch_cmd import (
128 run_ssh_cmd_parallel,
129 zfs_list_snapshots_in_parallel,
130)
131from bzfs_main.progress_reporter import (
132 ProgressReporter,
133 count_num_bytes_transferred_by_zfs_send,
134)
135from bzfs_main.replication import (
136 delete_bookmarks,
137 delete_datasets,
138 delete_snapshots,
139 is_inconsistent,
140 is_tmp_bookmark,
141 replicate_dataset,
142)
143from bzfs_main.snapshot_cache import (
144 MATURITY_TIME_THRESHOLD_SECS,
145 MONITOR_CACHE_FILE_PREFIX,
146 REPLICATION_CACHE_FILE_PREFIX,
147 SnapshotCache,
148 set_last_modification_time_safe,
149)
150from bzfs_main.util.connection import (
151 SHARED,
152 ConnectionPool,
153 MiniRemote,
154 timeout,
155)
156from bzfs_main.util.parallel_iterator import (
157 run_in_parallel,
158)
159from bzfs_main.util.parallel_tasktree_policy import (
160 process_datasets_in_parallel_and_fault_tolerant,
161)
162from bzfs_main.util.retry import (
163 Retry,
164 RetryableError,
165 RetryTemplate,
166 RetryTerminationError,
167 RetryTiming,
168)
169from bzfs_main.util.utils import (
170 DESCENDANTS_RE_SUFFIX,
171 DIE_STATUS,
172 DONT_SKIP_DATASET,
173 LOG_DEBUG,
174 LOG_TRACE,
175 PROG_NAME,
176 SHELL_CHARS_AND_SLASH,
177 UMASK,
178 YEAR_WITH_FOUR_DIGITS_REGEX,
179 HashedInterner,
180 SortedInterner,
181 Subprocesses,
182 SynchronizedBool,
183 SynchronizedDict,
184 TaskTiming,
185 append_if_absent,
186 compile_regexes,
187 cut,
188 die,
189 has_duplicates,
190 has_prefix,
191 human_readable_bytes,
192 human_readable_duration,
193 is_descendant,
194 nonblocking_file_lock,
195 percent,
196 pretty_print_formatter,
197 replace_in_lines,
198 replace_prefix,
199 sha256_85_urlsafe_base64,
200 sha256_128_urlsafe_base64,
201 sort_datasets,
202 sort_datasets_key,
203 stderr_to_str,
204 termination_signal_handler,
205 validate_dataset_name,
206 validate_property_name,
207 xappend,
208 xfinally,
209 xprint,
210)
212# constants:
213__version__: Final[str] = bzfs_main.argparse_cli.__version__
214CRITICAL_STATUS: Final[int] = 2
215WARNING_STATUS: Final[int] = 1
216STILL_RUNNING_STATUS: Final[int] = 4
217MIN_PYTHON_VERSION: Final[tuple[int, int]] = (3, 9)
218if sys.version_info < MIN_PYTHON_VERSION:
219 print(f"ERROR: {PROG_NAME} requires Python version >= {'.'.join(map(str, MIN_PYTHON_VERSION))}!")
220 sys.exit(DIE_STATUS)
223#############################################################################
224def argument_parser() -> argparse.ArgumentParser:
225 """Returns the CLI parser used by bzfs."""
226 return bzfs_main.argparse_cli.argument_parser()
229def main() -> None:
230 """API for command line clients."""
231 prev_umask: int = os.umask(UMASK)
232 try:
233 set_logging_runtime_defaults()
234 # On CTRL-C and SIGTERM, send signal to all descendant processes to terminate them
235 termination_event: threading.Event = threading.Event()
236 with termination_signal_handler(termination_events=[termination_event]):
237 run_main(argument_parser().parse_args(), sys.argv, termination_event=termination_event)
238 except subprocess.CalledProcessError as e:
239 sys.exit(normalize_called_process_error(e))
240 finally:
241 os.umask(prev_umask) # restore prior global state
244def run_main(
245 args: argparse.Namespace,
246 sys_argv: list[str] | None = None,
247 log: Logger | None = None,
248 termination_event: threading.Event | None = None,
249) -> None:
250 """API for Python clients; visible for testing; may become a public API eventually."""
251 Job(termination_event=termination_event).run_main(args, sys_argv, log)
254#############################################################################
255@final
256class Job:
257 """Executes one bzfs run, coordinating snapshot replication tasks."""
259 def __init__(self, termination_event: threading.Event | None = None) -> None:
260 self.params: Params
261 self.termination_event: Final[threading.Event] = termination_event or threading.Event()
263 def _on_before_attempt(_retry: Retry) -> None:
264 pass
266 self.retry_timing: Final[RetryTiming] = RetryTiming.make_from(self.termination_event).copy(
267 on_before_attempt=_on_before_attempt
268 )
269 self.task_timing: Final[TaskTiming] = TaskTiming.make_from(self.termination_event)
270 self.subprocesses: Subprocesses = Subprocesses(self.termination_event.is_set)
271 self.all_dst_dataset_exists: Final[dict[str, dict[str, bool]]] = defaultdict(lambda: defaultdict(bool))
272 self.dst_dataset_exists: SynchronizedDict[str, bool] = SynchronizedDict({})
273 self.src_properties: dict[str, DatasetProperties] = {}
274 self.dst_properties: dict[str, DatasetProperties] = {}
275 self.all_exceptions: list[str] = []
276 self.all_exceptions_count: int = 0
277 self.max_exceptions_to_summarize: int = 10000
278 self.first_exception: BaseException | None = None
279 self.worst_exception: BaseException | None = None
280 self.remote_conf_cache: dict[tuple, RemoteConfCacheItem] = {}
281 self.max_datasets_per_minibatch_on_list_snaps: dict[str, int] = {}
282 self.max_workers: dict[str, int] = {}
283 self.progress_reporter: ProgressReporter = cast(ProgressReporter, None)
284 self.is_first_replication_task: Final[SynchronizedBool] = SynchronizedBool(True)
285 self.replication_start_time_nanos: int = time.monotonic_ns()
286 self.timeout_nanos: int | None = None # timestamp aka instant in time
287 self.timeout_duration_nanos: int | None = None # duration (not a timestamp); for logging only
288 self.cache: SnapshotCache = SnapshotCache(self)
289 self.stats_lock: Final[threading.Lock] = threading.Lock()
290 self.num_cache_hits: int = 0
291 self.num_cache_misses: int = 0
292 self.num_snapshots_found: int = 0
293 self.num_snapshots_replicated: int = 0
295 self.is_test_mode: bool = False # for testing only
296 self.creation_prefix: str = "" # for testing only
297 self.use_select: bool = False # for testing only
298 self.progress_update_intervals: tuple[float, float] | None = None # for testing only
299 self.error_injection_triggers: dict[str, Counter[str]] = {} # for testing only
300 self.delete_injection_triggers: dict[str, Counter[str]] = {} # for testing only
301 self.param_injection_triggers: dict[str, dict[str, bool | int]] = {} # for testing only
302 self.inject_params: dict[str, bool | int] = {} # for testing only
303 self.injection_lock: threading.Lock = threading.Lock() # for testing only
304 self.max_command_line_bytes: int | None = None # for testing only
306 def shutdown(self) -> None:
307 """Exits any multiplexed ssh sessions that may be leftover."""
308 cache_items: Collection[RemoteConfCacheItem] = self.remote_conf_cache.values()
309 for i, cache_item in enumerate(cache_items):
310 cache_item.connection_pools.shutdown(f"{i + 1}/{len(cache_items)}")
312 def terminate(self) -> None:
313 """Shuts down gracefully; also terminates descendant processes, if any."""
314 with xfinally(self.subprocesses.terminate_process_subtrees):
315 self.shutdown()
317 def _retry_template(self) -> RetryTemplate:
318 p = self.params
319 return RetryTemplate(policy=p.retry_policy.copy(timing=self.retry_timing), log=p.log)
321 def run_main(self, args: argparse.Namespace, sys_argv: list[str] | None = None, log: Logger | None = None) -> None:
322 """Parses CLI arguments, sets up logging, and executes main job loop."""
323 assert isinstance(self.error_injection_triggers, dict)
324 assert isinstance(self.delete_injection_triggers, dict)
325 assert isinstance(self.inject_params, dict)
326 logger_name_suffix: str = ""
328 def _reset_logger() -> None:
329 if logger_name_suffix and log is not None: # reset Logger unless it's a Logger outside of our control
330 reset_logger(log)
332 with xfinally(_reset_logger): # runs _reset_logger() on exit, without masking error raised in body of `with` block
333 try:
334 log_params: LogParams = LogParams(args)
335 logger_name_suffix = "" if log is not None else log_params.logger_name_suffix
336 log = bzfs_main.loggers.get_logger(
337 log_params=log_params, args=args, log=log, logger_name_suffix=logger_name_suffix
338 )
339 log.info("%s", f"Log file is: {log_params.log_file}")
340 except BaseException as e:
341 simple_log: Logger = get_simple_logger(PROG_NAME, logger_name_suffix=logger_name_suffix)
342 try:
343 simple_log.error("Log init: %s", e, exc_info=not isinstance(e, SystemExit))
344 finally:
345 reset_logger(simple_log)
346 raise
348 aux_args: list[str] = []
349 if getattr(args, "include_snapshot_plan", None):
350 aux_args += args.include_snapshot_plan
351 if getattr(args, "delete_dst_snapshots_except_plan", None):
352 aux_args += args.delete_dst_snapshots_except_plan
353 if len(aux_args) > 0:
354 log.info("Auxiliary CLI arguments: %s", " ".join(aux_args))
355 args = argument_parser().parse_args(xappend(aux_args, "--", args.root_dataset_pairs), namespace=args)
357 def log_error_on_exit(error: object, status_code: object, exc_info: bool = False) -> None:
358 log.error("%s%s", f"Exiting {PROG_NAME} with status code {status_code}. Cause: ", error, exc_info=exc_info)
360 try:
361 log.info("CLI arguments: %s %s", " ".join(sys_argv or []), f"[uid: {os.getuid()}, euid: {os.geteuid()}]")
362 if self.is_test_mode:
363 log.log(LOG_TRACE, "Parsed CLI arguments: %s", args)
364 self.params = p = Params(args, sys_argv or [], log_params, log, self.inject_params)
365 self.timeout_duration_nanos = p.timeout_duration_nanos
366 lock_file: str = p.lock_file_name()
367 with nonblocking_file_lock(lock_file) as acquired:
368 if not acquired:
369 msg = "Exiting as same previous periodic job is still running without completion yet per "
370 die(msg + lock_file, STILL_RUNNING_STATUS)
371 try:
372 self.run_tasks() # do the real work
373 except BaseException:
374 self.terminate()
375 raise
376 self.shutdown()
377 with contextlib.suppress(BrokenPipeError):
378 sys.stderr.flush()
379 sys.stdout.flush()
380 except subprocess.CalledProcessError as e:
381 log_error_on_exit(e, e.returncode)
382 raise
383 except SystemExit as e:
384 log_error_on_exit(e, e.code)
385 raise
386 except (subprocess.TimeoutExpired, UnicodeDecodeError) as e:
387 log_error_on_exit(e, DIE_STATUS)
388 raise SystemExit(DIE_STATUS) from e
389 except re.error as e:
390 log_error_on_exit(f"{e} within regex {e.pattern!r}", DIE_STATUS)
391 raise SystemExit(DIE_STATUS) from e
392 except BaseException as e:
393 log_error_on_exit(e, DIE_STATUS, exc_info=True)
394 raise SystemExit(DIE_STATUS) from e
395 finally:
396 log.info("%s", f"Log file was: {log_params.log_file}")
397 log.info("Success. Goodbye!")
398 with contextlib.suppress(BrokenPipeError):
399 sys.stderr.flush()
400 sys.stdout.flush()
402 def run_tasks(self) -> None:
403 """Executes replication cycles, repeating until daemon lifetime expires."""
404 p, log = self.params, self.params.log
405 self.all_exceptions = []
406 self.all_exceptions_count = 0
407 self.first_exception = None
408 self.worst_exception = None
409 self.remote_conf_cache = {}
410 self.validate_once()
411 self.replication_start_time_nanos = time.monotonic_ns()
412 self.progress_reporter = ProgressReporter(log, p.pv_program_opts, self.use_select, self.progress_update_intervals)
413 with xfinally(lambda: self.progress_reporter.stop()):
414 daemon_stoptime_nanos: int = time.monotonic_ns() + p.daemon_lifetime_nanos
415 while True: # loop for daemon mode
416 self.timeout_nanos = (
417 None if p.timeout_duration_nanos is None else time.monotonic_ns() + p.timeout_duration_nanos
418 )
419 self.all_dst_dataset_exists.clear()
420 self.progress_reporter.reset()
421 src, dst = p.src, p.dst
422 for src_root_dataset, dst_root_dataset in p.root_dataset_pairs:
423 if self.termination_event.is_set():
424 self.terminate()
425 break
426 src.root_dataset = src.basis_root_dataset = src_root_dataset
427 dst.root_dataset = dst.basis_root_dataset = dst_root_dataset
428 p.curr_zfs_send_program_opts = p.zfs_send_program_opts.copy()
429 p.curr_zfs_send_resume_opts = p.zfs_send_resume_opts.copy()
430 if p.daemon_lifetime_nanos > 0:
431 self.timeout_nanos = (
432 None if p.timeout_duration_nanos is None else time.monotonic_ns() + p.timeout_duration_nanos
433 )
434 recurs_sep = " " if p.recursive_flag else ""
435 task_description = f"{src.basis_root_dataset} {p.recursive_flag}{recurs_sep}--> {dst.basis_root_dataset}"
436 if len(p.root_dataset_pairs) > 1:
437 log.info("Starting task: %s", task_description + " ...")
438 try:
439 try:
440 self.maybe_inject_error(cmd=[], error_trigger="retryable_run_tasks")
441 timeout(self)
442 self.validate_task()
443 self.run_task() # do the real work
444 except RetryableError as retryable_error:
445 cause: BaseException | None = retryable_error.__cause__
446 assert cause is not None
447 raise cause.with_traceback(cause.__traceback__) # noqa: B904 re-raise of cause without chaining
448 except (CalledProcessError, TimeoutExpired, SystemExit, UnicodeDecodeError, RetryTerminationError) as e:
449 if p.skip_on_error == "fail" or (
450 isinstance(e, subprocess.TimeoutExpired) and p.daemon_lifetime_nanos == 0
451 ):
452 raise
453 log.error("%s", e)
454 self.append_exception(e, "task", task_description)
455 if not self.sleep_until_next_daemon_iteration(daemon_stoptime_nanos):
456 break
457 if not p.skip_replication:
458 self.print_replication_stats(self.replication_start_time_nanos)
459 error_count = self.all_exceptions_count
460 if error_count > 0 and p.daemon_lifetime_nanos == 0:
461 msgs = "\n".join(f"{i + 1}/{error_count}: {e}" for i, e in enumerate(self.all_exceptions))
462 log.error("%s", f"Tolerated {error_count} errors. Error Summary: \n{msgs}")
463 assert self.worst_exception is not None
464 raise self.worst_exception
466 def append_exception(self, e: BaseException, task_name: str, task_description: str) -> None:
467 """Records and logs an exception that was encountered while running a subtask."""
469 def _exit_code(error: BaseException) -> int:
470 """Classifies failures using CLI exit codes, reserving monitoring statuses for explicit SystemExit alerts."""
471 if isinstance(error, subprocess.CalledProcessError):
472 return normalize_called_process_error(error)
473 if isinstance(error, SystemExit) and isinstance(error.code, int):
474 return error.code
475 return DIE_STATUS
477 self.first_exception = self.first_exception or e
478 existing_code = None if self.worst_exception is None else _exit_code(self.worst_exception)
479 if get_worst_exception(existing_code, _exit_code(e)) != existing_code:
480 self.worst_exception = 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,written,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, written, 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 written=int(written) if written != "-" else 0,
780 )
781 basis_src_datasets.append(src_dataset)
782 assert (not self.is_test_mode) or basis_src_datasets == sorted(basis_src_datasets), "List is not sorted"
783 return basis_src_datasets
785 def list_dst_datasets_task(self) -> list[str]:
786 """Lists datasets on the destination host."""
787 p, log = self.params, self.params.log
788 dst = p.dst
789 is_caching: bool = is_caching_snapshots(p, dst) and p.monitor_snapshots_config.enable_monitor_snapshots
790 props: str = "name"
791 props = "snapshots_changed," + props if is_caching else props
792 cmd: list[str] = p.split_args(
793 f"{p.zfs_program} list -t filesystem,volume -s name -Hp -o {props} {p.recursive_flag}", dst.root_dataset
794 )
795 basis_dst_datasets: list[str] = []
796 basis_dst_datasets_str: str | None = self.try_ssh_command_with_retries(dst, LOG_TRACE, cmd=cmd)
797 if basis_dst_datasets_str is None:
798 log.warning("Destination dataset does not exist: %s", dst.root_dataset)
799 else:
800 for line in basis_dst_datasets_str.splitlines():
801 cols: list[str] = line.split("\t")
802 snapshots_changed, dst_dataset = cols if is_caching else ["-"] + cols
803 self.dst_properties[dst_dataset] = DatasetProperties(
804 recordsize=0,
805 snapshots_changed=int(snapshots_changed) if snapshots_changed and snapshots_changed != "-" else 0,
806 written=0,
807 )
808 basis_dst_datasets.append(dst_dataset)
809 assert (not self.is_test_mode) or basis_dst_datasets == sorted(basis_dst_datasets), "List is not sorted"
810 return basis_dst_datasets
812 def create_src_snapshots_task(self, basis_src_datasets: list[str], src_datasets: list[str]) -> None:
813 """Atomically creates a new snapshot of the src datasets selected by --{include|exclude}-dataset* policy; implements
814 --create-src-snapshots.
816 The implementation attempts to fit as many datasets as possible into a single (atomic) 'zfs snapshot' command line,
817 using lexicographical sort order, and using 'zfs snapshot -r' to the extent that this is compatible with the
818 --{include|exclude}-dataset* pruning policy. The snapshots of all datasets that fit within the same single 'zfs
819 snapshot' CLI invocation will be taken within the same ZFS transaction group, and correspondingly have identical
820 'createtxg' ZFS property (but not necessarily identical 'creation' ZFS time property as ZFS actually provides no such
821 guarantee), and thus be consistent. Dataset names that can't fit into a single command line are spread over multiple
822 command line invocations, respecting the limits that the operating system places on the maximum length of a single
823 command line, per `getconf ARG_MAX`.
825 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
826 per dataset. Space complexity is O(max(N, M)).
827 """
828 p, log = self.params, self.params.log
829 src = p.src
830 if len(basis_src_datasets) == 0:
831 die(f"Source dataset does not exist: {src.basis_root_dataset}")
832 datasets_to_snapshot: dict[SnapshotLabel, list[str]] = self.find_datasets_to_snapshot(src_datasets)
833 datasets_to_snapshot = {label: datasets for label, datasets in datasets_to_snapshot.items() if len(datasets) > 0}
834 basis_datasets_to_snapshot: dict[SnapshotLabel, list[str]] = datasets_to_snapshot.copy() # shallow copy
835 commands: dict[SnapshotLabel, list[str]] = {}
836 for label, datasets in datasets_to_snapshot.items():
837 cmd: list[str] = p.split_args(f"{src.sudo} {p.zfs_program} snapshot")
838 if p.recursive:
839 # Run 'zfs snapshot -r' on the roots of subtrees if possible, else fallback to non-recursive CLI flavor
840 root_datasets = self.root_datasets_if_recursive_zfs_snapshot_is_possible(datasets, basis_src_datasets)
841 if root_datasets is not None:
842 cmd.append("-r") # recursive; takes a snapshot of all datasets in the subtree(s)
843 datasets_to_snapshot[label] = root_datasets
844 commands[label] = cmd
845 creation_msg = f"Creating {sum(len(datasets) for datasets in basis_datasets_to_snapshot.values())} snapshots"
846 log.info(p.dry("--create-src-snapshots: %s"), f"{creation_msg} within {len(src_datasets)} datasets ...")
847 # create snapshots in large (parallel) batches, without using a command line that's too big for the OS to handle
848 run_ssh_cmd_parallel(
849 self,
850 src,
851 ((commands[lbl], (f"{ds}@{lbl}" for ds in datasets)) for lbl, datasets in datasets_to_snapshot.items()),
852 fn=lambda cmd, batch: self.run_ssh_command_with_retries(
853 src, is_dry=p.dry_run, print_stdout=True, cmd=cmd + batch, retry_on_generic_ssh_error=False
854 ), # retry_on_generic_ssh_error=False means only retry on SSH connect errors b/c `zfs snapshot` isn't idempotent
855 max_batch_items=2**29,
856 )
857 for label, datasets in basis_datasets_to_snapshot.items():
858 for dataset in datasets:
859 log.debug(p.dry("--create-src-snapshots: %s%s%s%s"), "Created snapshot ", dataset, "@", label)
860 if is_caching_snapshots(p, src):
861 # perf: copy lastmodified time of source dataset into local cache to reduce future 'zfs list -t snapshot' calls
862 self.cache.update_last_modified_cache(basis_datasets_to_snapshot)
864 def delete_destination_snapshots_task(
865 self, basis_src_datasets: list[str], dst_datasets: list[str], max_workers: int, task_description: str
866 ) -> bool:
867 """Deletes existing destination snapshots that do not exist within the source dataset if they are included by the
868 --{include|exclude}-snapshot-* policy, and the destination dataset is included via --{include|exclude}-dataset*
869 policy; implements --delete-dst-snapshots. Does not attempt to delete snapshots that carry a `zfs hold`."""
870 p, log = self.params, self.params.log
871 src, dst = p.src, p.dst
872 kind: str = "bookmark" if p.delete_dst_bookmarks else "snapshot"
873 filter_needs_creation_time: bool = has_timerange_filter(p.snapshot_filters)
874 props: str = "guid,name,userrefs"
875 props = self.creation_prefix + "creation," + props if filter_needs_creation_time else props
876 basis_src_datasets_set: set[str] = set(basis_src_datasets)
877 num_snapshots_found, num_snapshots_deleted = 0, 0
879 def delete_destination_snapshots(dst_dataset: str, tid: str, retry: Retry) -> bool: # thread-safe
880 src_dataset: str = replace_prefix(dst_dataset, old_prefix=dst.root_dataset, new_prefix=src.root_dataset)
881 if src_dataset in basis_src_datasets_set and (are_bookmarks_enabled(p, src) or not p.delete_dst_bookmarks):
882 src_kind: str = kind
883 if not p.delete_dst_snapshots_no_crosscheck:
884 src_kind = "snapshot,bookmark" if are_bookmarks_enabled(p, src) else "snapshot"
885 src_cmd = p.split_args(f"{p.zfs_program} list -t {src_kind} -d 1 -s name -Hp -o guid", src_dataset)
886 else:
887 src_cmd = None
888 dst_cmd = p.split_args(f"{p.zfs_program} list -t {kind} -d 1 -s createtxg -Hp -o {props}", dst_dataset)
889 self.maybe_inject_delete(dst, dataset=dst_dataset, delete_trigger="zfs_list_delete_dst_snapshots")
890 src_snaps_with_guids, dst_snaps_with_guids_str = run_in_parallel( # list src+dst snapshots in parallel
891 lambda: set(self.run_ssh_command(src, LOG_TRACE, cmd=src_cmd).splitlines() if src_cmd else []),
892 lambda: self.try_ssh_command(dst, LOG_TRACE, cmd=dst_cmd),
893 )
894 if dst_snaps_with_guids_str is None:
895 log.warning("Third party deleted destination: %s", dst_dataset)
896 return False
897 held_dst_snapshots: set[str] = set()
898 dst_snaps_with_guids: list[str] = []
899 for line in dst_snaps_with_guids_str.splitlines():
900 dst_snaps_with_guids.append(line[: line.rindex("\t")]) # strip off trailing userrefs column
901 _, name, userrefs = line.rsplit("\t", 2)
902 if userrefs not in ("", "-", "0"): # ZFS snapshot property userrefs > 0 indicates a zfs hold
903 tag: str = name[name.index("@") + 1 :]
904 held_dst_snapshots.add(tag) # don't attempt to delete snapshots that carry a `zfs hold`
905 if p.delete_dst_bookmarks:
906 dst_snaps_with_guids = [
907 line for line in dst_snaps_with_guids if is_tmp_bookmark(line) == p.force_delete_dst_tmp_bookmarks
908 ]
909 num_dst_snaps_with_guids = len(dst_snaps_with_guids)
910 basis_dst_snaps_with_guids: list[str] = dst_snaps_with_guids.copy()
911 if p.delete_dst_bookmarks:
912 replace_in_lines(dst_snaps_with_guids, old="#", new="@", count=1) # treat bookmarks as snapshots
913 # The check against the source dataset happens *after* filtering the dst snapshots with filter_snapshots().
914 # `p.delete_dst_snapshots_except` means the user wants to specify snapshots to *retain* aka *keep*
915 all_except: bool = p.delete_dst_snapshots_except
916 if p.delete_dst_snapshots_except and not is_dummy(src):
917 # However, as here we are in "except" mode AND the source is NOT a dummy, we first filter to get what
918 # the policy says to *keep* (so all_except=False for the filter_snapshots() call), then from that "keep"
919 # list, we later further refine by checking what's on the source dataset.
920 all_except = False
921 dst_snaps_with_guids = filter_snapshots(self, dst_snaps_with_guids, all_except=all_except)
922 if p.delete_dst_bookmarks:
923 replace_in_lines(dst_snaps_with_guids, old="@", new="#", count=1) # restore pre-filtering bookmark state
924 if filter_needs_creation_time:
925 dst_snaps_with_guids = cut(field=2, lines=dst_snaps_with_guids)
926 basis_dst_snaps_with_guids = cut(field=2, lines=basis_dst_snaps_with_guids)
927 if p.delete_dst_snapshots_except and not is_dummy(src): # Non-dummy Source + "Except" (Keep) Mode
928 # Retain dst snapshots that match snapshot filter policy AND are on src dataset, aka
929 # Delete dst snapshots except snapshots that match snapshot filter policy AND are on src dataset.
930 # Concretely, `dst_snaps_with_guids` contains GUIDs of DST snapshots that the filter policy says to KEEP.
931 # We only actually keep them if they are ALSO on the SRC.
932 # So, snapshots to DELETE (`dst_tags_to_delete`) are ALL snapshots on DST (`basis_dst_snaps_with_guids`)
933 # EXCEPT those whose GUIDs are in `dst_snaps_with_guids` AND ALSO in `src_snaps_with_guids`.
934 except_dst_guids: set[str] = set(cut(field=1, lines=dst_snaps_with_guids)).intersection(src_snaps_with_guids)
935 dst_tags_to_delete: list[str] = filter_lines_except(basis_dst_snaps_with_guids, except_dst_guids)
936 else: # Standard Delete Mode OR Dummy Source + "Except" (Keep) Mode
937 # In standard delete mode:
938 # `dst_snaps_with_guids` contains GUIDs of policy-selected snapshots on DST.
939 # We delete those that are NOT on SRC.
940 # `dst_tags_to_delete` = `dst_snaps_with_guids` - `src_snaps_with_guids`.
941 # In dummy source + "except" (keep) mode:
942 # `all_except` was True.
943 # `dst_snaps_with_guids` contains snaps NOT matching the "keep" policy -- these are the ones to delete.
944 # `src_snaps_with_guids` is empty.
945 # `dst_tags_to_delete` = `dst_snaps_with_guids` - {} = `dst_snaps_with_guids`.
946 dst_guids_to_delete = set(cut(field=1, lines=dst_snaps_with_guids)).difference(src_snaps_with_guids)
947 dst_tags_to_delete = filter_lines(dst_snaps_with_guids, dst_guids_to_delete)
948 separator: str = "#" if p.delete_dst_bookmarks else "@"
949 dst_tags_to_delete = cut(field=2, separator=separator, lines=dst_tags_to_delete)
950 if p.delete_dst_bookmarks:
951 delete_bookmarks(self, dst, dst_dataset, snapshot_tags=dst_tags_to_delete)
952 else:
953 dst_tags_to_delete = [tag for tag in dst_tags_to_delete if tag not in held_dst_snapshots]
954 delete_snapshots(self, dst, dst_dataset, snapshot_tags=dst_tags_to_delete)
955 with self.stats_lock:
956 nonlocal num_snapshots_found
957 num_snapshots_found += num_dst_snaps_with_guids
958 nonlocal num_snapshots_deleted
959 num_snapshots_deleted += len(dst_tags_to_delete)
960 if len(dst_tags_to_delete) > 0 and not p.delete_dst_bookmarks:
961 self.dst_properties[dst_dataset].snapshots_changed = 0 # invalidate cache
962 return True
964 # Run delete_destination_snapshots(dataset) for each dataset, while handling errors, retries + parallel exec
965 failed: bool = False
966 if are_bookmarks_enabled(p, dst) or not p.delete_dst_bookmarks:
967 start_time_nanos = time.monotonic_ns()
968 failed = process_datasets_in_parallel_and_fault_tolerant(
969 log=log,
970 datasets=dst_datasets,
971 process_dataset=delete_destination_snapshots, # lambda
972 skip_tree_on_error=lambda dataset: False,
973 skip_on_error=p.skip_on_error,
974 max_workers=max_workers,
975 timing=self.task_timing,
976 termination_handler=self.terminate,
977 enable_barriers=False,
978 task_name="--delete-dst-snapshots",
979 append_exception=self.append_exception,
980 retry_template=self._retry_template(),
981 dry_run=p.dry_run,
982 is_test_mode=self.is_test_mode,
983 )
984 elapsed_nanos = time.monotonic_ns() - start_time_nanos
985 log.info(
986 p.dry("--delete-dst-snapshots: %s"),
987 task_description + f" [Deleted {num_snapshots_deleted} out of {num_snapshots_found} {kind}s "
988 f"within {len(dst_datasets)} datasets; took {human_readable_duration(elapsed_nanos)}]",
989 )
990 return failed
992 def delete_dst_datasets_task(
993 self, basis_src_datasets: list[str], basis_dst_datasets: list[str], sorted_dst_datasets: list[str]
994 ) -> tuple[list[str], list[str]]:
995 """Deletes existing destination datasets that do not exist within the source dataset if they are included via
996 --{include|exclude}-dataset* policy; implements --delete-dst-datasets.
998 Do not recurse without --recursive. With --recursive, never delete non-selected dataset subtrees or their ancestors.
999 """
1000 p = self.params
1001 src, dst = p.src, p.dst
1002 children: dict[str, set[str]] = defaultdict(set)
1003 for dst_dataset in basis_dst_datasets: # Compute the direct children of each NON-FILTERED dataset
1004 parent: str = os.path.dirname(dst_dataset)
1005 children[parent].add(dst_dataset)
1006 to_delete: set[str] = set()
1007 for dst_dataset in reversed(sorted_dst_datasets): # Reverse order facilitates efficient O(N) time algorithm
1008 if children[dst_dataset].issubset(to_delete):
1009 to_delete.add(dst_dataset) # all children are deletable, thus the dataset itself is deletable too
1010 to_delete = to_delete.difference(
1011 replace_prefix(src_dataset, src.root_dataset, dst.root_dataset) for src_dataset in basis_src_datasets
1012 )
1013 delete_datasets(self, dst, to_delete)
1014 sorted_dst_datasets = sorted(set(sorted_dst_datasets).difference(to_delete))
1015 basis_dst_datasets = sorted(set(basis_dst_datasets).difference(to_delete))
1016 return basis_dst_datasets, sorted_dst_datasets
1018 def delete_empty_dst_datasets_task(
1019 self, basis_dst_datasets: list[str], sorted_dst_datasets: list[str]
1020 ) -> tuple[list[str], list[str]]:
1021 """Deletes any existing destination dataset that has no snapshot and no bookmark if all descendants of that dataset
1022 do not have a snapshot or bookmark either; implements --delete-empty-dst-datasets.
1024 To do so, we walk the dataset list (conceptually, a tree) depth-first (i.e. sorted descending). If a dst dataset has
1025 zero snapshots and zero bookmarks and all its children are already marked as orphans, then it is itself an orphan,
1026 and we mark it as such. Walking in a reverse sorted way means that we efficiently check for zero snapshots/bookmarks
1027 not just over the direct children but the entire tree. Finally, delete all orphan datasets in an efficient batched
1028 way.
1029 """
1030 p = self.params
1031 dst = p.dst
1033 # Compute the direct children of each NON-FILTERED dataset. Thus, no non-selected dataset and no ancestor of a
1034 # non-selected dataset will ever be added to the "orphan" set. In other words, this treats non-selected dataset
1035 # subtrees as if they all had snapshots, so non-selected dataset subtrees and their ancestors are guaranteed
1036 # to not get deleted.
1037 children: dict[str, set[str]] = defaultdict(set)
1038 for dst_dataset in basis_dst_datasets:
1039 parent: str = os.path.dirname(dst_dataset)
1040 children[parent].add(dst_dataset)
1042 def compute_orphans(datasets_having_snapshots: set[str]) -> set[str]:
1043 """Returns destination datasets having zero snapshots whose children are all orphans."""
1044 orphans: set[str] = set()
1045 for dst_dataset in reversed(sorted_dst_datasets): # Reverse order facilitates efficient O(N) time algorithm
1046 if (dst_dataset not in datasets_having_snapshots) and children[dst_dataset].issubset(orphans):
1047 orphans.add(dst_dataset)
1048 return orphans
1050 # Compute candidate orphan datasets, which reduces the list of datasets for which we list snapshots via
1051 # 'zfs list -t snapshot ...' from dst_datasets to a subset of dst_datasets, which in turn reduces I/O and improves
1052 # perf. Essentially, this eliminates the I/O to list snapshots for ancestors of excluded datasets.
1053 candidate_orphans: set[str] = compute_orphans(set())
1055 # Compute destination datasets having more than zero snapshots
1056 dst_datasets_having_snapshots: set[str] = set()
1057 with_bookmarks: bool = p.delete_empty_dst_datasets_if_no_bookmarks_and_no_snapshots and are_bookmarks_enabled(p, dst)
1058 btype: str = "bookmark,snapshot" if with_bookmarks else "snapshot"
1059 cmd: list[str] = p.split_args(f"{p.zfs_program} list -t {btype} -d 1 -S name -Hp -o name")
1060 for snapshots in zfs_list_snapshots_in_parallel(self, dst, cmd, sorted(candidate_orphans), ordered=False):
1061 if with_bookmarks:
1062 replace_in_lines(snapshots, old="#", new="@", count=1) # treat bookmarks as snapshots
1063 dst_datasets_having_snapshots.update(snap[: snap.index("@")] for snap in snapshots) # union
1065 orphans: set[str] = compute_orphans(dst_datasets_having_snapshots) # compute the real orphans
1066 delete_datasets(self, dst, orphans) # finally, delete the orphan datasets in an efficient way
1067 sorted_dst_datasets = sorted(set(sorted_dst_datasets).difference(orphans))
1068 basis_dst_datasets = sorted(set(basis_dst_datasets).difference(orphans))
1069 return basis_dst_datasets, sorted_dst_datasets
1071 def monitor_snapshots_task(
1072 self, sorted_src_datasets: list[str], sorted_dst_datasets: list[str], task_description: str
1073 ) -> None:
1074 """Monitors src and dst snapshots; implements --monitor-snapshots."""
1075 p, log = self.params, self.params.log
1076 src, dst = p.src, p.dst
1077 num_cache_hits: int = self.num_cache_hits
1078 num_cache_misses: int = self.num_cache_misses
1079 start_time_nanos: int = time.monotonic_ns()
1080 dst_alert, src_alert = run_in_parallel(
1081 lambda: self.monitor_snapshots(dst, sorted_dst_datasets),
1082 lambda: self.monitor_snapshots(src, sorted_src_datasets),
1083 )
1084 exit_code, _exit_kind, exit_msg = min(dst_alert, src_alert)
1085 if exit_code != 0:
1086 die(exit_msg, -exit_code)
1087 elapsed: str = human_readable_duration(time.monotonic_ns() - start_time_nanos)
1088 num_cache_hits = self.num_cache_hits - num_cache_hits
1089 num_cache_misses = self.num_cache_misses - num_cache_misses
1090 if num_cache_hits > 0 or num_cache_misses > 0:
1091 msg = self._cache_hits_msg(hits=num_cache_hits, misses=num_cache_misses)
1092 else:
1093 msg = ""
1094 log.info(
1095 "--monitor-snapshots done: %s",
1096 f"{task_description} [{len(sorted_src_datasets) + len(sorted_dst_datasets)} datasets; took {elapsed}{msg}]",
1097 )
1099 def monitor_snapshots(self, remote: Remote, sorted_datasets: list[str]) -> tuple[int, str, str]:
1100 """Checks snapshot freshness and warns or errors out when limits are exceeded.
1102 Alerts the user if the ZFS 'creation' time property of the latest or oldest snapshot for any specified snapshot name
1103 pattern within the selected datasets is too old wrt. the specified age limit. The purpose is to check if snapshots
1104 are successfully taken on schedule, successfully replicated on schedule, and successfully pruned on schedule. Process
1105 exit code is 0, 1, 2 on OK, WARNING, CRITICAL, respectively.
1107 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
1108 per dataset. Space complexity is O(max(N, M)).
1109 """
1110 p, log = self.params, self.params.log
1111 alerts: list[MonitorSnapshotAlert] = p.monitor_snapshots_config.alerts
1112 labels: list[SnapshotLabel] = [alert.label for alert in alerts]
1113 oldest_skip_holds: list[bool] = [alert.oldest_skip_holds for alert in alerts]
1114 current_unixtime_millis: float = p.create_src_snapshots_config.current_datetime.timestamp() * 1000
1115 is_debug: bool = log.isEnabledFor(LOG_DEBUG)
1116 if is_caching_snapshots(p, remote):
1117 props: dict[str, DatasetProperties] = self.dst_properties if remote is p.dst else self.src_properties
1118 snapshots_changed_dict: dict[str, int] = {dataset: vals.snapshots_changed for dataset, vals in props.items()}
1119 alerts_hash: str = sha256_128_urlsafe_base64(str(tuple(alerts)))
1120 label_hashes: dict[SnapshotLabel, str] = {
1121 label: sha256_128_urlsafe_base64(label.notimestamp_str()) for label in labels
1122 }
1123 is_caching: bool = False
1124 worst_alert: tuple[int, str, str] = (0, "", "") # -exit_code, exit_kind, exit_msg
1126 def record_alert(exit_code: int, exit_kind: str, exit_msg: str) -> None:
1127 nonlocal worst_alert
1128 worst_alert = min(worst_alert, (-exit_code, exit_kind, exit_msg)) # min() sorts "Latest" before "Oldest" on tie
1130 def monitor_last_modified_cache_file(r: Remote, dataset: str, label: SnapshotLabel, alert_cfg: AlertConfig) -> str:
1131 cache_label: str = os.path.join(MONITOR_CACHE_FILE_PREFIX, alert_cfg.kind[0], label_hashes[label], alerts_hash)
1132 return self.cache.last_modified_cache_file(r, dataset, cache_label)
1134 def alert_msg(
1135 kind: str, dataset: str, snapshot: str, label: SnapshotLabel, snapshot_age_millis: float, delta_millis: int
1136 ) -> str:
1137 assert kind == "Latest" or kind == "Oldest"
1138 lbl = f"{label.prefix}{label.infix}<timestamp>{label.suffix}"
1139 if snapshot_age_millis >= current_unixtime_millis:
1140 return f"No snapshot exists for {dataset}@{lbl}"
1141 msg = f"{kind} snapshot for {dataset}@{lbl} is {human_readable_duration(snapshot_age_millis, unit='ms')} old"
1142 s = f": @{snapshot}" if snapshot else ""
1143 if delta_millis == -1:
1144 return f"{msg}{s}"
1145 return f"{msg} but should be at most {human_readable_duration(delta_millis, unit='ms')} old{s}"
1147 def check_alert(
1148 label: SnapshotLabel, alert_cfg: AlertConfig | None, creation_unixtime_secs: int, dataset: str, snapshot: str
1149 ) -> None: # thread-safe
1150 if alert_cfg is None:
1151 return
1152 if is_caching and not p.dry_run: # update cache with latest state from 'zfs list -t snapshot'
1153 snapshots_changed: int = snapshots_changed_dict.get(dataset, 0)
1154 cache_file: str = monitor_last_modified_cache_file(remote, dataset, label, alert_cfg)
1155 set_last_modification_time_safe(
1156 cache_file, unixtime_in_secs=(creation_unixtime_secs, snapshots_changed), if_more_recent=True
1157 )
1158 warning_millis: int = alert_cfg.warning_millis
1159 critical_millis: int = alert_cfg.critical_millis
1160 alert_kind = alert_cfg.kind
1161 snapshot_age_millis: float = current_unixtime_millis - creation_unixtime_secs * 1000
1162 m = "--monitor_snapshots: "
1163 if snapshot_age_millis > critical_millis:
1164 msg = m + alert_msg(alert_kind, dataset, snapshot, label, snapshot_age_millis, critical_millis)
1165 log.critical("%s", msg)
1166 if not p.monitor_snapshots_config.dont_crit:
1167 record_alert(CRITICAL_STATUS, alert_kind, msg)
1168 elif snapshot_age_millis > warning_millis:
1169 msg = m + alert_msg(alert_kind, dataset, snapshot, label, snapshot_age_millis, warning_millis)
1170 log.warning("%s", msg)
1171 if not p.monitor_snapshots_config.dont_warn:
1172 record_alert(WARNING_STATUS, alert_kind, msg)
1173 elif is_debug:
1174 msg = m + "OK. " + alert_msg(alert_kind, dataset, snapshot, label, snapshot_age_millis, delta_millis=-1)
1175 log.debug("%s", msg)
1177 def alert_latest_snapshot(i: int, creation_unixtime_secs: int, dataset: str, snapshot: str) -> None:
1178 alert: MonitorSnapshotAlert = alerts[i]
1179 check_alert(alert.label, alert.latest, creation_unixtime_secs, dataset, snapshot)
1181 def alert_oldest_snapshot(i: int, creation_unixtime_secs: int, dataset: str, snapshot: str) -> None:
1182 alert: MonitorSnapshotAlert = alerts[i]
1183 check_alert(alert.label, alert.oldest, creation_unixtime_secs, dataset, snapshot)
1185 def find_stale_datasets_and_check_alerts() -> list[str]:
1186 """If the cache is enabled, check which datasets have changed to determine which datasets can be skipped cheaply,
1187 that is, without incurring 'zfs list -t snapshots'.
1189 This is done by comparing the "snapshots_changed" ZFS dataset property with the local cache. See
1190 https://openzfs.github.io/openzfs-docs/man/7/zfsprops.7.html#snapshots_changed
1191 """
1192 stale_datasets: list[str] = []
1193 time_threshold: float = time.time() - MATURITY_TIME_THRESHOLD_SECS
1194 for dataset in sorted_datasets:
1195 is_stale_dataset: bool = False
1196 snapshots_changed: int = snapshots_changed_dict.get(dataset, 0)
1197 for alert in alerts:
1198 for cfg in (alert.latest, alert.oldest):
1199 if cfg is None:
1200 continue
1201 if cfg is alert.oldest and alert.oldest_skip_holds:
1202 is_stale_dataset = True # `zfs hold` and `zfs release` do not update snapshots_changed
1203 continue
1204 if (
1205 snapshots_changed != 0
1206 and snapshots_changed < time_threshold
1207 and ( # always True
1208 cached_unix_times := self.cache.get_snapshots_changed2(
1209 monitor_last_modified_cache_file(remote, dataset, alert.label, cfg)
1210 )
1211 )
1212 and snapshots_changed == cached_unix_times[1] # cached snapshots_changed aka last modified time
1213 and snapshots_changed >= cached_unix_times[0] # creation time of minmax snapshot aka access time
1214 ): # cached state is still valid; emit an alert if the latest/oldest snapshot is too old
1215 lbl = alert.label
1216 check_alert(lbl, cfg, creation_unixtime_secs=cached_unix_times[0], dataset=dataset, snapshot="")
1217 else: # cached state is no longer valid; fallback to 'zfs list -t snapshot'
1218 is_stale_dataset = True
1219 if is_stale_dataset:
1220 stale_datasets.append(dataset)
1221 return stale_datasets
1223 # satisfy request from local cache as much as possible
1224 if is_caching_snapshots(p, remote):
1225 stale_datasets: list[str] = find_stale_datasets_and_check_alerts()
1226 with self.stats_lock:
1227 self.num_cache_misses += len(stale_datasets)
1228 self.num_cache_hits += len(sorted_datasets) - len(stale_datasets)
1229 else:
1230 stale_datasets = sorted_datasets
1232 # fallback to 'zfs list -t snapshot' for any remaining datasets, as these couldn't be satisfied from local cache
1233 is_caching = is_caching_snapshots(p, remote)
1234 datasets_without_snapshots: list[str] = self.handle_minmax_snapshots(
1235 remote,
1236 stale_datasets,
1237 labels,
1238 fn_latest=alert_latest_snapshot,
1239 fn_oldest=alert_oldest_snapshot,
1240 fn_oldest_skip_holds=oldest_skip_holds,
1241 )
1242 for dataset in datasets_without_snapshots:
1243 for i in range(len(alerts)):
1244 alert_latest_snapshot(i, creation_unixtime_secs=0, dataset=dataset, snapshot="")
1245 alert_oldest_snapshot(i, creation_unixtime_secs=0, dataset=dataset, snapshot="")
1246 return worst_alert
1248 def replicate_datasets(self, src_datasets: list[str], task_description: str, max_workers: int) -> bool:
1249 """Replicates a list of datasets."""
1250 assert (not self.is_test_mode) or src_datasets == sorted(src_datasets), "List is not sorted"
1251 p, log = self.params, self.params.log
1252 src, dst = p.src, p.dst
1253 self.num_snapshots_found = 0
1254 self.num_snapshots_replicated = 0
1255 log.info("Starting replication task: %s", task_description + f" [{len(src_datasets)} datasets]")
1256 start_time_nanos: int = time.monotonic_ns()
1258 def src2dst(src_dataset: str) -> str:
1259 return replace_prefix(src_dataset, old_prefix=src.root_dataset, new_prefix=dst.root_dataset)
1261 def dst2src(dst_dataset: str) -> str:
1262 return replace_prefix(dst_dataset, old_prefix=dst.root_dataset, new_prefix=src.root_dataset)
1264 def find_stale_datasets() -> tuple[list[str], dict[str, str]]:
1265 """If the cache is enabled on replication, check which src datasets or dst datasets have changed to determine
1266 which datasets can be skipped cheaply, i.e. without incurring 'zfs list -t snapshots'.
1268 This is done by comparing the "snapshots_changed" ZFS dataset property with the local cache. See
1269 https://openzfs.github.io/openzfs-docs/man/7/zfsprops.7.html#snapshots_changed
1270 """
1271 # First, check which src datasets have changed since the last replication to that destination
1272 cache_files: dict[str, str] = {}
1273 stale_src_datasets1: list[str] = []
1274 maybe_stale_dst_datasets: list[str] = []
1275 userhost_dir: str = sha256_85_urlsafe_base64(p.dst.cache_namespace())
1276 # filter_key: cache is only valid if same --include/excl-snapshot*, --skip-missing-snapshots, --no-use-bookmark
1277 filter_key: tuple = (p.snapshot_filters, p.skip_missing_snapshots, p.use_bookmark)
1278 filter_hash_code: str = sha256_85_urlsafe_base64(str(filter_key))
1279 for src_dataset in src_datasets:
1280 dst_dataset: str = src2dst(src_dataset) # cache is only valid for identical destination dataset
1281 dst_dataset_dir: str = sha256_85_urlsafe_base64(dst_dataset)
1282 cache_label: str = os.path.join(
1283 REPLICATION_CACHE_FILE_PREFIX, userhost_dir, dst_dataset_dir, filter_hash_code
1284 )
1285 cache_file: str = self.cache.last_modified_cache_file(src, src_dataset, cache_label)
1286 cache_files[src_dataset] = cache_file
1287 snapshots_changed: int = self.src_properties[src_dataset].snapshots_changed # get prop "for free"
1288 if (
1289 snapshots_changed != 0
1290 and time.time() > snapshots_changed + MATURITY_TIME_THRESHOLD_SECS
1291 and snapshots_changed == self.cache.get_snapshots_changed(cache_file)
1292 ):
1293 maybe_stale_dst_datasets.append(dst_dataset)
1294 else:
1295 stale_src_datasets1.append(src_dataset)
1297 # For each src dataset that hasn't changed, check if the corresponding dst dataset has changed
1298 stale_src_datasets2: list[str] = []
1299 dst_snapshots_changed_dict: dict[str, int] = self.cache.zfs_get_snapshots_changed(dst, maybe_stale_dst_datasets)
1300 for dst_dataset in maybe_stale_dst_datasets:
1301 snapshots_changed = dst_snapshots_changed_dict.get(dst_dataset, 0)
1302 if (
1303 snapshots_changed != 0
1304 and time.time() > snapshots_changed + MATURITY_TIME_THRESHOLD_SECS
1305 and snapshots_changed
1306 == self.cache.get_snapshots_changed(self.cache.last_modified_cache_file(dst, dst_dataset))
1307 ):
1308 log.info("Already up-to-date [cached]: %s", dst_dataset)
1309 else:
1310 stale_src_datasets2.append(dst2src(dst_dataset))
1311 assert (not self.is_test_mode) or stale_src_datasets1 == sorted(stale_src_datasets1), "List is not sorted"
1312 assert (not self.is_test_mode) or stale_src_datasets2 == sorted(stale_src_datasets2), "List is not sorted"
1313 stale_src_datasets = list(heapq.merge(stale_src_datasets1, stale_src_datasets2)) # merge two sorted lists
1314 assert (not self.is_test_mode) or not has_duplicates(stale_src_datasets), "List contains duplicates"
1315 return stale_src_datasets, cache_files
1317 cache_files: dict[str, str]
1318 if is_caching_snapshots(p, src):
1319 stale_src_datasets, cache_files = find_stale_datasets()
1320 num_cache_misses = len(stale_src_datasets)
1321 num_cache_hits = len(src_datasets) - len(stale_src_datasets)
1322 self.num_cache_misses += num_cache_misses
1323 self.num_cache_hits += num_cache_hits
1324 cmsg = self._cache_hits_msg(hits=num_cache_hits, misses=num_cache_misses)
1325 else:
1326 stale_src_datasets = src_datasets
1327 cache_files = {}
1328 cmsg = ""
1330 done_src_datasets: list[str] = []
1331 done_src_datasets_lock: threading.Lock = threading.Lock()
1333 def _has_descendant_dataset(src_dataset: str) -> bool:
1334 return has_prefix(stale_src_datasets, prefix=src_dataset + "/", is_test_mode=self.is_test_mode)
1336 def _process_dataset_fn(src_dataset: str, tid: str, retry: Retry) -> bool:
1337 has_descendant_dataset: bool = _has_descendant_dataset(src_dataset)
1338 result: bool = replicate_dataset(
1339 job=self, src_dataset=src_dataset, tid=tid, retry=retry, has_descendant_dataset=has_descendant_dataset
1340 )
1341 with done_src_datasets_lock:
1342 done_src_datasets.append(src_dataset) # record datasets that were actually replicated (not skipped)
1343 return result
1345 def _skip_tree_on_error(src_dataset: str) -> bool:
1346 has_descendant_dataset: bool = _has_descendant_dataset(src_dataset)
1347 dst_dataset: str = src2dst(src_dataset)
1348 if (not has_descendant_dataset) or not self.dst_dataset_exists[dst_dataset]:
1349 return True
1350 try:
1351 return is_inconsistent(self, dst, dst_dataset)
1352 except Exception:
1353 return True
1355 # Run replicate_dataset(dataset) for each dataset, while taking care of errors, retries + parallel execution
1356 failed: bool = process_datasets_in_parallel_and_fault_tolerant(
1357 log=log,
1358 datasets=stale_src_datasets,
1359 process_dataset=_process_dataset_fn,
1360 skip_tree_on_error=_skip_tree_on_error,
1361 skip_on_error=p.skip_on_error,
1362 max_workers=max_workers,
1363 timing=self.task_timing,
1364 termination_handler=self.terminate,
1365 enable_barriers=False,
1366 task_name="Replication",
1367 append_exception=self.append_exception,
1368 retry_template=self._retry_template(),
1369 dry_run=p.dry_run,
1370 is_test_mode=self.is_test_mode,
1371 )
1373 if is_caching_snapshots(p, src) and len(done_src_datasets) > 0:
1374 # refresh "snapshots_changed" ZFS dataset property from dst
1375 stale_dst_datasets: list[str] = [src2dst(src_dataset) for src_dataset in sorted(done_src_datasets)]
1376 dst_snapshots_changed_dict: dict[str, int] = self.cache.zfs_get_snapshots_changed(dst, stale_dst_datasets)
1377 for dst_dataset in stale_dst_datasets: # update local cache
1378 dst_snapshots_changed: int = dst_snapshots_changed_dict.get(dst_dataset, 0)
1379 dst_cache_file: str = self.cache.last_modified_cache_file(dst, dst_dataset)
1380 src_dataset: str = dst2src(dst_dataset)
1381 src_snapshots_changed: int = self.src_properties[src_dataset].snapshots_changed
1382 if not p.dry_run:
1383 set_last_modification_time_safe(
1384 cache_files[src_dataset], unixtime_in_secs=src_snapshots_changed, if_more_recent=True
1385 )
1386 set_last_modification_time_safe(
1387 dst_cache_file, unixtime_in_secs=dst_snapshots_changed, if_more_recent=True
1388 )
1390 elapsed_nanos: int = time.monotonic_ns() - start_time_nanos
1391 log.info(
1392 p.dry("Replication done: %s"),
1393 f"{task_description} [Replicated {self.num_snapshots_replicated} out of {self.num_snapshots_found} snapshots"
1394 f" within {len(src_datasets)} datasets; took {human_readable_duration(elapsed_nanos)}{cmsg}]",
1395 )
1396 return failed
1398 def maybe_inject_delete(self, remote: Remote, dataset: str, delete_trigger: str) -> None:
1399 """For testing only; for unit tests to delete datasets during replication and test correct handling of that."""
1400 assert delete_trigger
1401 counter = self.delete_injection_triggers.get("before")
1402 if counter and self.decrement_injection_counter(counter, delete_trigger):
1403 p = self.params
1404 cmd = p.split_args(f"{remote.sudo} {p.zfs_program} destroy -r", p.force_unmount, p.force_hard, dataset or "")
1405 self.run_ssh_command(remote, LOG_DEBUG, print_stdout=True, cmd=cmd)
1407 def maybe_inject_params(self, error_trigger: str) -> None:
1408 """For testing only; for unit tests to simulate errors during replication and test correct handling of them."""
1409 assert error_trigger
1410 counter = self.error_injection_triggers.get("before")
1411 if counter and self.decrement_injection_counter(counter, error_trigger):
1412 self.inject_params = self.param_injection_triggers[error_trigger]
1413 elif error_trigger in self.param_injection_triggers:
1414 self.inject_params = {}
1416 @staticmethod
1417 def recv_option_property_names(recv_opts: list[str]) -> set[str]:
1418 """Extracts -o and -x property names that are already specified on the command line; This can be used to check for
1419 dupes because 'zfs receive' does not accept multiple -o or -x options with the same property name."""
1420 propnames: set[str] = set()
1421 i = 0
1422 n = len(recv_opts)
1423 while i < n:
1424 stripped: str = recv_opts[i].strip()
1425 if stripped in ("-o", "-x"):
1426 i += 1
1427 if i == n or recv_opts[i].strip() in ("-o", "-x"):
1428 die(f"Missing value for {stripped} option in --zfs-recv-program-opt(s): {' '.join(recv_opts)}")
1429 if stripped == "-o" and "=" not in recv_opts[i]:
1430 die(f"Missing value for {stripped} name=value pair in --zfs-recv-program-opt(s): {' '.join(recv_opts)}")
1431 propname: str = recv_opts[i] if stripped == "-x" else recv_opts[i].split("=", 1)[0]
1432 validate_property_name(propname, "--zfs-recv-program-opt(s)")
1433 propnames.add(propname)
1434 i += 1
1435 return propnames
1437 def root_datasets_if_recursive_zfs_snapshot_is_possible(
1438 self, datasets: list[str], basis_datasets: list[str]
1439 ) -> list[str] | None:
1440 """Returns the root datasets within the (filtered) `datasets` list if no incompatible pruning is detected. A dataset
1441 within `datasets` is considered a root dataset if it has no parent, i.e. it is not a descendant of any dataset in
1442 `datasets`. Returns `None` if any (unfiltered) dataset in `basis_dataset` that is a descendant of at least one of the
1443 root datasets is missing in `datasets`, indicating that --include/exclude-dataset* or the snapshot schedule have
1444 pruned a dataset in a way that is incompatible with 'zfs snapshot -r' CLI semantics, thus requiring a switch to the
1445 non-recursive 'zfs snapshot snapshot1 .. snapshot N' CLI flavor.
1447 Assumes that set(datasets).issubset(set(basis_datasets)). Also assumes that datasets and basis_datasets are both
1448 sorted (and thus the output root_datasets list is sorted too), which is why this algorithm is efficient - O(N) time
1449 complexity. The impl is akin to the merge algorithm of a merge sort, adapted to our specific use case.
1450 See root_datasets_if_recursive_zfs_snapshot_is_possible_slow_but_correct() in the unit test suite for an alternative
1451 impl that's easier to grok.
1452 """
1453 assert (not self.is_test_mode) or datasets == sorted(datasets), "List is not sorted"
1454 assert (not self.is_test_mode) or not has_duplicates(datasets), "List contains duplicates"
1455 assert (not self.is_test_mode) or basis_datasets == sorted(basis_datasets), "List is not sorted"
1456 assert (not self.is_test_mode) or not has_duplicates(basis_datasets), "List contains duplicates"
1457 assert (not self.is_test_mode) or set(datasets).issubset(set(basis_datasets)), "Not a subset"
1458 datasets = sort_datasets(datasets)
1459 basis_datasets = sort_datasets(basis_datasets)
1460 root_datasets: list[str] = self.find_root_datasets(datasets)
1461 i = 0
1462 j = 0
1463 k = 0
1464 len_root_datasets = len(root_datasets)
1465 len_basis_datasets = len(basis_datasets)
1466 len_datasets = len(datasets)
1467 while i < len_root_datasets and j < len_basis_datasets: # walk and "merge" the sorted lists, in sync
1468 if sort_datasets_key(basis_datasets[j]) < sort_datasets_key(root_datasets[i]): # irrelevant subtree?
1469 j += 1 # move to next basis_datasets[j]
1470 elif is_descendant(basis_datasets[j], of_root_dataset=root_datasets[i]): # relevant subtree?
1471 while k < len_datasets and sort_datasets_key(datasets[k]) < sort_datasets_key(basis_datasets[j]):
1472 k += 1 # move to next datasets[k]
1473 if k == len_datasets or datasets[k] != basis_datasets[j]: # dataset chopped off by schedule or --incl/excl*?
1474 return None # detected filter pruning that is incompatible with 'zfs snapshot -r'
1475 j += 1 # move to next basis_datasets[j]
1476 else:
1477 i += 1 # move to next root_dataset[i]; no need to check root_datasets that are no longer (or not yet) reachable
1478 return sorted(root_datasets)
1480 @staticmethod
1481 def find_root_datasets(sorted_datasets: list[str]) -> list[str]:
1482 """Returns the roots of the subtrees in the (sorted) input datasets; The output root dataset list is sorted, too; A
1483 dataset is a root dataset if it has no parent, i.e. it is not a descendant of any dataset in the input datasets."""
1484 root_datasets: list[str] = []
1485 skip_dataset: str = DONT_SKIP_DATASET
1486 for dataset in sorted_datasets:
1487 if is_descendant(dataset, of_root_dataset=skip_dataset):
1488 continue
1489 skip_dataset = dataset
1490 root_datasets.append(dataset)
1491 return root_datasets
1493 def find_datasets_to_snapshot(self, sorted_datasets: list[str]) -> dict[SnapshotLabel, list[str]]:
1494 """Given a (sorted) list of source datasets, returns a dict where the key is a snapshot name (aka SnapshotLabel, e.g.
1495 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
1496 created with that name, because these datasets are due per the schedule, either because the 'creation' time of their
1497 most recent snapshot with that name pattern is now too old, or such a snapshot does not even exist.
1499 The baseline implementation uses the 'zfs list -t snapshot' CLI to find the most recent snapshots, which is simple
1500 but doesn't scale well with the number of snapshots, at least if the goal is to take snapshots every second. An
1501 alternative, much more scalable, implementation queries the standard ZFS "snapshots_changed" dataset property
1502 (requires zfs >= 2.2.0), in combination with a local cache that stores this property, as well as the creation time of
1503 the most recent snapshot, for each SnapshotLabel and each dataset.
1504 """
1505 p, log = self.params, self.params.log
1506 src = p.src
1507 config: CreateSrcSnapshotConfig = p.create_src_snapshots_config
1508 datasets_to_snapshot: dict[SnapshotLabel, list[str]] = defaultdict(list)
1509 is_caching: bool = False
1510 interner: HashedInterner[datetime] = HashedInterner() # reduces memory footprint
1511 msgs: list[tuple[datetime, str, SnapshotLabel, str]] = []
1513 def create_snapshot_if_latest_is_too_old(
1514 datasets_to_snapshot: dict[SnapshotLabel, list[str]], dataset: str, label: SnapshotLabel, creation_unixtime: int
1515 ) -> None: # thread-safe
1516 """Schedules creation of a snapshot for the given label if the label's existing latest snapshot is too old."""
1517 creation_dt: datetime = datetime.fromtimestamp(creation_unixtime, tz=config.tz)
1518 log.log(LOG_TRACE, "Latest snapshot creation: %s for %s", creation_dt, label)
1519 duration_amount, duration_unit = config.suffix_durations[label.suffix]
1520 next_event_dt: datetime = config.anchors.round_datetime_up_to_duration_multiple(
1521 creation_dt + timedelta(microseconds=1), duration_amount, duration_unit
1522 )
1523 msg: str = ""
1524 if config.current_datetime >= next_event_dt:
1525 written_threshold: int = config.snapshot_label_attributes[label.notimestamp_str()].written_threshold
1526 if self.src_properties[dataset].written >= written_threshold:
1527 datasets_to_snapshot[label].append(dataset) # mark it as scheduled for snapshot creation
1528 msg = " has passed"
1529 next_event_dt = interner.intern(next_event_dt)
1530 msgs.append((next_event_dt, dataset, label, msg))
1531 if is_caching and not p.dry_run: # update cache with latest state from 'zfs list -t snapshot'
1532 # Per-label cache stores (atime=creation, mtime=snapshots_changed) so later runs can safely trust creation
1533 # only when the label's mtime matches the current dataset-level '=' cache value. Excludes timestamp of label.
1534 cache_file: str = self.cache.last_modified_cache_file(src, dataset, label_hashes[label])
1535 unixtimes: tuple[int, int] = (creation_unixtime, self.src_properties[dataset].snapshots_changed)
1536 set_last_modification_time_safe(cache_file, unixtime_in_secs=unixtimes, if_more_recent=True)
1538 labels: list[SnapshotLabel] = []
1539 config_labels: list[SnapshotLabel] = config.snapshot_labels()
1540 for label in config_labels:
1541 duration_amount_, _duration_unit = config.suffix_durations[label.suffix]
1542 if duration_amount_ == 0 or config.create_src_snapshots_even_if_not_due:
1543 datasets_to_snapshot[label] = sorted_datasets # take snapshot regardless of creation time of existing snaps
1544 else:
1545 labels.append(label)
1546 if len(labels) == 0:
1547 return datasets_to_snapshot # nothing more TBD
1548 label_hashes: dict[SnapshotLabel, str] = {
1549 label: sha256_128_urlsafe_base64(label.notimestamp_str()) for label in labels
1550 }
1552 # satisfy request from local cache as much as possible
1553 cached_datasets_to_snapshot: dict[SnapshotLabel, list[str]] = defaultdict(list)
1554 if is_caching_snapshots(p, src):
1555 sorted_datasets_todo: list[str] = []
1556 time_threshold: float = time.time() - MATURITY_TIME_THRESHOLD_SECS
1557 for dataset in sorted_datasets:
1558 cache: SnapshotCache = self.cache
1559 cached_snapshots_changed: int = cache.get_snapshots_changed(cache.last_modified_cache_file(src, dataset))
1560 if cached_snapshots_changed == 0:
1561 sorted_datasets_todo.append(dataset) # request cannot be answered from cache
1562 continue
1563 if cached_snapshots_changed != self.src_properties[dataset].snapshots_changed: # get that prop "for free"
1564 cache.invalidate_last_modified_cache_dataset(dataset)
1565 sorted_datasets_todo.append(dataset) # request cannot be answered from cache
1566 continue
1567 if cached_snapshots_changed >= time_threshold: # Avoid equal-second races: only trust matured cache entries
1568 sorted_datasets_todo.append(dataset) # cache entry isn't mature enough to be trusted; skip cache
1569 continue
1570 creation_unixtimes: list[int] = []
1571 for label_hash in label_hashes.values():
1572 # For per-label files, atime stores the latest matching snapshot's creation time, while mtime stores
1573 # the dataset-level snapshots_changed observed when this label file was written.
1574 atime, mtime = cache.get_snapshots_changed2(cache.last_modified_cache_file(src, dataset, label_hash))
1575 # Sanity check: trust per-label cache only when:
1576 # - mtime equals the dataset-level '=' cache (same snapshots_changed), and
1577 # - atime is plausible and not later than mtime (creation <= snapshots_changed), and
1578 # - neither atime nor mtime is zero (unknown provenance).
1579 # Otherwise fall back to 'zfs list -t snapshot' to avoid stale creation times after newer changes.
1580 if atime == 0 or mtime == 0 or mtime != cached_snapshots_changed or atime > mtime:
1581 sorted_datasets_todo.append(dataset) # request cannot be answered from cache
1582 break
1583 creation_unixtimes.append(atime)
1584 if len(creation_unixtimes) == len(labels):
1585 for j, label in enumerate(labels):
1586 create_snapshot_if_latest_is_too_old(
1587 cached_datasets_to_snapshot, dataset, label, creation_unixtimes[j]
1588 )
1589 sorted_datasets = sorted_datasets_todo
1591 def create_snapshot_fn(i: int, creation_unixtime_secs: int, dataset: str, snapshot: str) -> None:
1592 create_snapshot_if_latest_is_too_old(datasets_to_snapshot, dataset, labels[i], creation_unixtime_secs)
1594 def on_finish_dataset(dataset: str) -> None:
1595 if is_caching_snapshots(p, src) and not p.dry_run:
1596 set_last_modification_time_safe(
1597 self.cache.last_modified_cache_file(src, dataset),
1598 unixtime_in_secs=self.src_properties[dataset].snapshots_changed,
1599 if_more_recent=True,
1600 )
1602 # fallback to 'zfs list -t snapshot' for any remaining datasets, as these couldn't be satisfied from local cache
1603 is_caching = is_caching_snapshots(p, src)
1604 datasets_without_snapshots: list[str] = self.handle_minmax_snapshots(
1605 src, sorted_datasets, labels, fn_latest=create_snapshot_fn, fn_on_finish_dataset=on_finish_dataset
1606 )
1607 for lbl in labels: # merge (sorted) results from local cache + 'zfs list -t snapshot' into (sorted) combined result
1608 datasets_to_snapshot[lbl].sort()
1609 if datasets_without_snapshots or (lbl in cached_datasets_to_snapshot): # +take snaps for snapshot-less datasets
1610 datasets_to_snapshot[lbl] = list( # inputs to merge() are sorted, and outputs are sorted too
1611 heapq.merge(datasets_to_snapshot[lbl], cached_datasets_to_snapshot[lbl], datasets_without_snapshots)
1612 )
1613 for label, datasets in datasets_to_snapshot.items():
1614 assert (not self.is_test_mode) or datasets == sorted(datasets), "List is not sorted"
1615 assert (not self.is_test_mode) or not has_duplicates(datasets), "List contains duplicates"
1616 assert label
1618 msgs.sort() # sort by time, dataset, label
1619 for i in range(0, len(msgs), 10_000): # reduce logging overhead via mini-batching
1620 text = "".join(
1621 f"\nNext scheduled snapshot time: {next_event_dt} for {dataset}@{label}{msg}"
1622 for next_event_dt, dataset, label, msg in msgs[i : i + 10_000]
1623 )
1624 log.info("Next scheduled snapshot times ...%s", text)
1626 # sort keys to ensure that we take snapshots for dailies before hourlies, and so on
1627 datasets_to_snapshot = {lbl: datasets_to_snapshot[lbl] for lbl in config_labels if lbl in datasets_to_snapshot}
1628 return datasets_to_snapshot
1630 def handle_minmax_snapshots(
1631 self,
1632 remote: Remote,
1633 sorted_datasets: list[str],
1634 labels: list[SnapshotLabel],
1635 *,
1636 fn_latest: Callable[[int, int, str, str], None], # callback function for latest snapshot
1637 fn_oldest: Callable[[int, int, str, str], None] | None = None, # callback function for oldest snapshot
1638 fn_oldest_skip_holds: Sequence[bool] = (),
1639 fn_on_finish_dataset: Callable[[str], None] = lambda dataset: None,
1640 ) -> list[str]: # thread-safe
1641 """For each dataset in `sorted_datasets`, for each label in `labels`, finds the latest and oldest snapshot, and runs
1642 the callback functions on them; Ignores the timestamp of the input labels and the timestamp of the snapshot names.
1644 If the (optional) fn_oldest_skip_holds=True for a given label, then snapshots for that label that carry a 'zfs hold'
1645 are skipped (ignored) when finding the oldest snapshot for fn_oldest. This can be useful for monitor_snapshots(),
1646 given that users often intentionally retain holds for longer than the "normal" snapshot retention period, and this
1647 shouldn't necessarily cause monitoring to emit alerts.
1648 """
1649 if fn_oldest is not None:
1650 assert len(labels) == len(fn_oldest_skip_holds)
1651 assert (not self.is_test_mode) or sorted_datasets == sorted(sorted_datasets), "List is not sorted"
1653 def extract_fields(line: str) -> tuple[int, int, str, bool]:
1654 fields: list[str] = line.split("\t")
1655 if len(fields) == 3:
1656 name, createtxg, creation_unixtime_secs = fields
1657 userrefs = ""
1658 else:
1659 name, createtxg, creation_unixtime_secs, userrefs = fields
1660 return (
1661 int(createtxg),
1662 int(creation_unixtime_secs),
1663 name.split("@", 1)[1],
1664 userrefs in ("", "-", "0"), # ZFS snapshot property userrefs > 0 indicates a zfs hold
1665 )
1667 p = self.params
1668 props: str = "name,createtxg,creation"
1669 props = props if fn_oldest is None or not any(fn_oldest_skip_holds) else props + ",userrefs"
1670 cmd = p.split_args(f"{p.zfs_program} list -t snapshot -d 1 -Hp -o {props}") # sorts by dataset,creation
1671 datasets_with_snapshots: set[str] = set()
1672 interner: SortedInterner[str] = SortedInterner(sorted_datasets) # reduces memory footprint
1673 for lines in zfs_list_snapshots_in_parallel(self, remote, cmd, sorted_datasets, ordered=False):
1674 # streaming group by dataset name (consumes constant memory only)
1675 for dataset, group in itertools.groupby(lines, key=lambda line: line.split("\t", 1)[0].split("@", 1)[0]):
1676 dataset = interner.interned(dataset)
1677 snapshots = sorted( # fetch all snapshots of current dataset and sort by createtxg,creation,name
1678 extract_fields(line) for line in group
1679 ) # perf: sorted() is fast because Powersort is close to O(N) for nearly sorted input, which is our case
1680 assert len(snapshots) > 0
1681 datasets_with_snapshots.add(dataset)
1682 snapshot_names: tuple[str, ...] = tuple(snapshot[2] for snapshot in snapshots)
1683 year_with_4_digits_regex: re.Pattern[str] = YEAR_WITH_FOUR_DIGITS_REGEX
1684 year_with_4_digits_regex_fullmatch = year_with_4_digits_regex.fullmatch
1685 startswith = str.startswith
1686 endswith = str.endswith
1687 fns = ((fn_latest, True),) if fn_oldest is None else ((fn_latest, True), (fn_oldest, False))
1688 for i, label in enumerate(labels):
1689 infix: str = label.infix
1690 start: str = label.prefix + infix
1691 end: str = label.suffix
1692 startlen: int = len(start)
1693 endlen: int = len(end)
1694 minlen: int = startlen + endlen if infix else 4 + startlen + endlen # year_with_four_digits_regex
1695 startlen_4: int = startlen + 4 # [startlen:startlen+4] # year_with_four_digits_regex
1696 has_infix: bool = bool(infix)
1697 for fn, is_reverse in fns:
1698 creation_unixtime_secs: int = 0 # find creation time of latest or oldest snapshot matching the label
1699 minmax_snapshot: str = ""
1700 no_skip_holds: bool = is_reverse or not fn_oldest_skip_holds[i]
1701 has_match: bool = False
1702 for j in range(len(snapshot_names) - 1, -1, -1) if is_reverse else range(len(snapshot_names)):
1703 snapshot_name: str = snapshot_names[j]
1704 if (
1705 endswith(snapshot_name, end) # aka snapshot_name.endswith(end)
1706 and startswith(snapshot_name, start) # aka snapshot_name.startswith(start)
1707 and len(snapshot_name) >= minlen
1708 and (has_infix or year_with_4_digits_regex_fullmatch(snapshot_name, startlen, startlen_4))
1709 ):
1710 has_match = True
1711 if no_skip_holds or snapshots[j][3]:
1712 creation_unixtime_secs = snapshots[j][1]
1713 minmax_snapshot = snapshot_name
1714 break
1715 if (not no_skip_holds) and (not minmax_snapshot) and has_match:
1716 continue # skip fn_oldest monitoring alert when all matching snapshots are held and skipped
1717 fn(i, creation_unixtime_secs, dataset, minmax_snapshot)
1718 fn_on_finish_dataset(dataset)
1719 datasets_without_snapshots = [dataset for dataset in sorted_datasets if dataset not in datasets_with_snapshots]
1720 return datasets_without_snapshots
1722 @staticmethod
1723 def _cache_hits_msg(hits: int, misses: int) -> str:
1724 total = hits + misses
1725 return f", cache hits: {percent(hits, total, print_total=True)}, misses: {percent(misses, total, print_total=True)}"
1727 def run_ssh_command(
1728 self,
1729 remote: MiniRemote,
1730 loglevel: int = logging.INFO,
1731 is_dry: bool = False,
1732 check: bool = True,
1733 print_stdout: bool = False,
1734 print_stderr: bool = True,
1735 cmd: list[str] | None = None,
1736 retry_on_generic_ssh_error: bool = True,
1737 ) -> str:
1738 """Runs the given CLI cmd via ssh on the given remote, and returns stdout."""
1739 assert cmd is not None and isinstance(cmd, list) and len(cmd) > 0
1740 conn_pool: ConnectionPool = self.params.connection_pools[remote.location].pool(SHARED)
1741 with conn_pool.connection() as conn:
1742 log: logging.Logger = self.params.log
1743 try:
1744 process: subprocess.CompletedProcess[str] = conn.run_ssh_command(
1745 cmd=cmd,
1746 job=self,
1747 loglevel=loglevel,
1748 is_dry=is_dry,
1749 check=check,
1750 stdin=DEVNULL,
1751 stdout=PIPE,
1752 stderr=PIPE,
1753 text=True,
1754 )
1755 except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as e:
1756 xprint(log, stderr_to_str(e.stdout) if print_stdout else e.stdout, run=print_stdout, file=sys.stdout, end="")
1757 xprint(log, stderr_to_str(e.stderr) if print_stderr else e.stderr, run=print_stderr, file=sys.stderr, end="")
1758 if retry_on_generic_ssh_error and isinstance(e, subprocess.CalledProcessError):
1759 stderr: str = stderr_to_str(e.stderr)
1760 if stderr.startswith("ssh: "):
1761 assert e.returncode == 255, e.returncode # error within SSH itself (not during the remote command)
1762 raise RetryableError(display_msg="ssh") from e
1763 raise
1764 else:
1765 if is_dry:
1766 return ""
1767 xprint(log, process.stdout, run=print_stdout, file=sys.stdout, end="")
1768 xprint(log, process.stderr, run=print_stderr, file=sys.stderr, end="")
1769 return process.stdout
1771 def try_ssh_command(
1772 self,
1773 remote: MiniRemote,
1774 loglevel: int,
1775 is_dry: bool = False,
1776 print_stdout: bool = False,
1777 cmd: list[str] | None = None,
1778 exists: bool = True,
1779 error_trigger: str | None = None,
1780 ) -> str | None:
1781 """Convenience method that helps retry/react to a dataset or pool that potentially doesn't exist anymore."""
1782 assert cmd is not None and isinstance(cmd, list) and len(cmd) > 0
1783 log = self.params.log
1784 try:
1785 self.maybe_inject_error(cmd=cmd, error_trigger=error_trigger)
1786 return self.run_ssh_command(remote=remote, loglevel=loglevel, is_dry=is_dry, print_stdout=print_stdout, cmd=cmd)
1787 except (subprocess.CalledProcessError, UnicodeDecodeError) as e:
1788 if not isinstance(e, UnicodeDecodeError):
1789 stderr: str = stderr_to_str(e.stderr)
1790 if exists and (
1791 ": dataset does not exist" in stderr
1792 or ": filesystem does not exist" in stderr # solaris 11.4.0
1793 or ": no such pool" in stderr
1794 or "does not have any resumable receive state to abort" in stderr # harmless `zfs receive -A` race
1795 or re.search(r"bookmark '.*' does not exist", stderr)
1796 ):
1797 return None
1798 log.warning("%s", stderr.rstrip())
1799 raise RetryableError("Subprocess failed") from e
1801 def try_ssh_command_with_retries(self, *args: Any, **kwargs: Any) -> str | None:
1802 """Convenience method that auto-retries try_ssh_command() on failure."""
1803 return self._retry_template().call_with_retries(fn=lambda retry: self.try_ssh_command(*args, **kwargs))
1805 def run_ssh_command_with_retries(self, *args: Any, **kwargs: Any) -> str:
1806 """Convenience method that auto-retries run_ssh_command() on transport failure (not on remote command failure)."""
1807 return self._retry_template().call_with_retries(fn=lambda retry: self.run_ssh_command(*args, **kwargs))
1809 def maybe_inject_error(self, cmd: list[str], error_trigger: str | None = None) -> None:
1810 """For testing only; for unit tests to simulate errors during replication and test correct handling of them."""
1811 if error_trigger:
1812 counter = self.error_injection_triggers.get("before")
1813 if counter and self.decrement_injection_counter(counter, error_trigger):
1814 try:
1815 raise CalledProcessError(returncode=1, cmd=" ".join(cmd), stderr=error_trigger + ":dataset is busy")
1816 except subprocess.CalledProcessError as e:
1817 if error_trigger.startswith("retryable_"):
1818 raise RetryableError("Subprocess failed") from e
1819 else:
1820 raise
1822 def decrement_injection_counter(self, counter: Counter[str], trigger: str) -> bool:
1823 """For testing only."""
1824 with self.injection_lock:
1825 if counter[trigger] <= 0:
1826 return False
1827 counter[trigger] -= 1
1828 return True
1831#############################################################################
1832@final
1833class DatasetProperties:
1834 """Properties of a ZFS dataset."""
1836 __slots__ = ("recordsize", "snapshots_changed", "written") # uses more compact memory layout than __dict__
1838 def __init__(self, recordsize: int, snapshots_changed: int, written: int) -> None:
1839 # immutable variables:
1840 self.recordsize: Final[int] = recordsize
1841 self.written: Final[int] = written
1843 # mutable variables:
1844 self.snapshots_changed: int = snapshots_changed
1847#############################################################################
1848# Input format is [[user@]host:]dataset
1849# 1234 5 6
1850_DATASET_LOCATOR_REGEX: Final[re.Pattern[str]] = re.compile(r"(((([^@]*)@)?([^:]+)):)?(.*)", flags=re.DOTALL)
1853def parse_dataset_locator(
1854 input_text: str, validate: bool = True, user: str | None = None, host: str | None = None, port: int | None = None
1855) -> tuple[str, str, str, str, str]:
1856 """Splits user@host:dataset into its components with optional checks."""
1858 def convert_ipv6(hostname: str) -> str: # support IPv6 without getting confused by host:dataset colon separator ...
1859 return hostname.replace("|", ":") # ... and any colons that may be part of a (valid) ZFS dataset name
1861 user_undefined: bool = user is None
1862 if user is None:
1863 user = ""
1864 host_undefined: bool = host is None
1865 if host is None:
1866 host = ""
1867 host = convert_ipv6(host)
1868 user_host, dataset, pool = "", "", ""
1870 if match := _DATASET_LOCATOR_REGEX.fullmatch(input_text): 1870 ↛ 1887line 1870 didn't jump to line 1887 because the condition on line 1870 was always true
1871 if user_undefined:
1872 user = match.group(4) or ""
1873 if host_undefined:
1874 host = match.group(5) or ""
1875 host = convert_ipv6(host)
1876 if host == "-":
1877 host = ""
1878 dataset = match.group(6) or ""
1879 i = dataset.find("/")
1880 pool = dataset[0:i] if i >= 0 else dataset
1882 if user and host:
1883 user_host = f"{user}@{host}"
1884 elif host:
1885 user_host = host
1887 if validate:
1888 validate_user_name(user, input_text)
1889 validate_host_name(host, input_text)
1890 if port is not None:
1891 validate_port(port, f"Invalid port number: '{port}' for: '{input_text}' - ")
1892 validate_dataset_name(dataset, input_text)
1894 return user, host, user_host, pool, dataset
1897def validate_user_name(user: str, input_text: str) -> None:
1898 """Checks that the username is safe for ssh or local usage."""
1899 invalid_chars: str = SHELL_CHARS_AND_SLASH
1900 if user and (user.startswith("-") or ".." in user or any(c.isspace() or c in invalid_chars for c in user)):
1901 die(f"Invalid user name: '{user}' for: '{input_text}'")
1904def validate_host_name(host: str, input_text: str) -> None:
1905 """Checks hostname for forbidden characters or patterns."""
1906 invalid_chars: str = SHELL_CHARS_AND_SLASH
1907 if host and (host.startswith("-") or ".." in host or any(c.isspace() or c in invalid_chars for c in host)):
1908 die(f"Invalid host name: '{host}' for: '{input_text}'")
1911def validate_port(port: str | int | None, message: str) -> None:
1912 """Checks that port specification is a valid integer."""
1913 if isinstance(port, int):
1914 port = str(port)
1915 if port and not port.isdigit():
1916 die(message + f"must be empty or a positive integer: '{port}'")
1919def normalize_called_process_error(error: subprocess.CalledProcessError) -> int:
1920 """Normalizes `CalledProcessError.returncode` to avoid reserved exit codes so callers don't misclassify them."""
1921 ret: int = error.returncode
1922 ret = DIE_STATUS if isinstance(ret, int) and 1 <= ret <= STILL_RUNNING_STATUS else ret
1923 return ret
1926def get_worst_exception(existing_code: int | None, new_code: int | None) -> int:
1927 """Process exit code precedence: fatal/non-monitor failure > monitor CRITICAL/WARNING > STILL_RUNNING > success."""
1928 new_code = DIE_STATUS if new_code is None else new_code
1929 if existing_code is None:
1930 return new_code
1931 assert existing_code is not None
1932 assert new_code is not None
1934 nonfatal: tuple[int, ...] = (0, WARNING_STATUS, CRITICAL_STATUS, STILL_RUNNING_STATUS)
1935 existing_is_fatal = existing_code not in nonfatal
1936 new_is_fatal = new_code not in nonfatal
1937 if existing_is_fatal and new_is_fatal:
1938 return max(existing_code, new_code)
1939 if existing_is_fatal or new_is_fatal:
1940 return existing_code if existing_is_fatal else new_code
1942 monitor: tuple[int, ...] = (WARNING_STATUS, CRITICAL_STATUS)
1943 existing_is_monitor = existing_code in monitor
1944 new_is_monitor = new_code in monitor
1945 if existing_is_monitor and new_is_monitor:
1946 return max(existing_code, new_code)
1947 if existing_is_monitor or new_is_monitor:
1948 return existing_code if existing_is_monitor else new_code
1950 assert existing_code in (0, STILL_RUNNING_STATUS), existing_code
1951 assert new_code in (0, STILL_RUNNING_STATUS), new_code
1952 return max(existing_code, new_code)
1955#############################################################################
1956if __name__ == "__main__": 1956 ↛ 1957line 1956 didn't jump to line 1957 because the condition on line 1956 was never true
1957 main()