Coverage for bzfs_main/bzfs_jobrunner.py: 99%
661 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* High-level orchestrator that calls `bzfs` as part of complex, periodic workflows to manage backup, replication, and
23 pruning jobs across a fleet of multiple source and destination hosts; driven by a fleet-wide job config file (e.g.,
24 `bzfs_job_testbed.py`).
25* Overview of the bzfs_jobrunner.py codebase:
26* The codebase starts with docs, definition of input data and associated argument parsing of CLI options/parameters.
27* Control flow starts in main(), far below ..., which kicks off a "Job".
28* A Job creates zero or more "subjobs" for each local or remote host, via run_main().
29* It executes the subjobs, serially or in parallel, via run_subjobs(), which in turn delegates parallel job coordination to
30 bzfs.process_datasets_in_parallel_and_fault_tolerant().
31* README_bzfs_jobrunner.md is mostly auto-generated from the ArgumentParser help texts as the source of "truth", via
32 update_readme.sh. Simply run that script whenever you change or add ArgumentParser help text.
33"""
35from __future__ import (
36 annotations,
37)
38import argparse
39import contextlib
40import os
41import platform
42import pwd
43import random
44import socket
45import subprocess
46import sys
47import threading
48import time
49import uuid
50from ast import (
51 literal_eval,
52)
53from collections.abc import (
54 Iterable,
55 Mapping,
56)
57from logging import (
58 Logger,
59)
60from subprocess import (
61 DEVNULL,
62 PIPE,
63)
64from typing import (
65 Any,
66 Final,
67 NoReturn,
68 TypeVar,
69 Union,
70 final,
71)
73import bzfs_main.argparse_actions
74from bzfs_main import (
75 bzfs,
76)
77from bzfs_main.argparse_cli import (
78 PROG_AUTHOR,
79)
80from bzfs_main.detect import (
81 DUMMY_DATASET,
82)
83from bzfs_main.loggers import (
84 get_simple_logger,
85 reset_logger,
86 set_logging_runtime_defaults,
87)
88from bzfs_main.util import (
89 check_range,
90 utils,
91)
92from bzfs_main.util.parallel_tasktree import (
93 BARRIER_CHAR,
94)
95from bzfs_main.util.parallel_tasktree_policy import (
96 process_datasets_in_parallel_and_fault_tolerant,
97)
98from bzfs_main.util.utils import (
99 DIE_STATUS,
100 LOG_TRACE,
101 UMASK,
102 UNIX_TIME_INFINITY_SECS,
103 JobStats,
104 Subprocesses,
105 TaskTiming,
106 dry,
107 format_dict,
108 format_obj,
109 getenv_bool,
110 human_readable_duration,
111 percent,
112 shuffle_dict,
113 terminate_process_subtree,
114 termination_signal_handler,
115 validate_dataset_name,
116)
117from bzfs_main.util.utils import PROG_NAME as BZFS_PROG_NAME
119# constants:
120PROG_NAME: Final[str] = "bzfs_jobrunner"
121SRC_MAGIC_SUBSTITUTION_TOKEN: Final[str] = "^SRC_HOST" # noqa: S105
122DST_MAGIC_SUBSTITUTION_TOKEN: Final[str] = "^DST_HOST" # noqa: S105
123SEP: Final[str] = ","
124POSIX_END_OF_OPTIONS_MARKER: Final[str] = "--" # args following -- are treated as operands, even if they begin with a hyphen
127def argument_parser() -> argparse.ArgumentParser:
128 """Returns the CLI parser used by bzfs_jobrunner."""
129 # fmt: off
130 parser = argparse.ArgumentParser(
131 prog=PROG_NAME,
132 allow_abbrev=False,
133 formatter_class=argparse.RawTextHelpFormatter,
134 description=f"""
135This companion program wraps [bzfs](README.md) for periodic snapshot creation, replication, pruning, and monitoring across
136N source hosts and M destination hosts, using one shared fleet-wide [jobconfig](bzfs_testbed/bzfs_job_testbed.py) script.
138Typical use cases include geo-replicated backup where each destination host is in a different region and receives replicas
139from the same set of source hosts, low-latency replication from a primary to a secondary or to M read replicas, and backups
140to removable drives.
142This program can be used to efficiently replicate ...
144a) within a single machine (local mode), or
146b) from a single source host to one or more destination hosts (pull or push or pull-push mode), or
148c) from multiple source hosts to a single destination host (pull or push or pull-push mode), or
150d) from N source hosts to M destination hosts (pull or push or pull-push mode, N and M can be large, M=2 or M=3 are typical
151geo-replication factors)
153You can run this program on a single third-party host and have it talk to all source and destination hosts. That setup is
154convenient for basic use cases and testing, and efficient with `--r2r=pull` or `--r2r=push`.
155In many deployments, a cron job on each source host runs `{PROG_NAME}` periodically to create new snapshots (via
156--create-src-snapshots) and prune outdated snapshots and bookmarks on the source (via --prune-src-snapshots and
157--prune-src-bookmarks), whereas another cron job on each destination host runs `{PROG_NAME}` periodically to prune
158outdated destination snapshots (via --prune-dst-snapshots), and to replicate the recently created snapshots from the source
159to the destination (via --replicate).
160A separate cron job on each source host and each destination host runs `{PROG_NAME}` periodically to alert the user if the
161latest or oldest snapshot is somehow too old (via --monitor-src-snapshots and --monitor-dst-snapshots).
162The frequency of each activity can differ. Typical intervals range from N milliseconds to years.
164Edit the jobconfig script in a central place (e.g. versioned in a git repo), then copy the (very same) shared file onto all
165source hosts and all destination hosts, and add crontab entries (or systemd timers or Monit entries or similar), along these
166lines:
168* crontab on source hosts:
170`* * * * * testuser /bzfs/bzfs_testbed/bzfs_job_testbed.py --src-host="$(hostname)" --create-src-snapshots --prune-src-snapshots --prune-src-bookmarks`
172`* * * * * testuser /bzfs/bzfs_testbed/bzfs_job_testbed.py --src-host="$(hostname)" --monitor-src-snapshots`
174* crontab on destination hosts:
176`* * * * * testuser /bzfs/bzfs_testbed/bzfs_job_testbed.py --dst-host="$(hostname)" --replicate --prune-dst-snapshots`
178`* * * * * testuser /bzfs/bzfs_testbed/bzfs_job_testbed.py --dst-host="$(hostname)" --monitor-dst-snapshots`
180Some deployments choose to move monitoring to one centralized management host, like so:
182`* * * * * testuser /bzfs/bzfs_testbed/bzfs_job_testbed.py --monitor-src-snapshots --monitor-dst-snapshots`
184### Applying Actions to a Subset of Hosts
186`--src-host` and `--dst-host` let you run actions on only a subset of source and destination hosts. For example, you can
187replicate from selected source hosts to selected destination hosts.
188Each `{PROG_NAME}` invocation runs all enabled actions for the final effective values of `--src-hosts` and `--dst-hosts`,
189after applying the `--src-host` and `--dst-host` filters.
192### High Frequency Replication (Experimental Feature)
194Taking snapshots and/or replicating every N milliseconds to roughly every 10 seconds is considered high frequency. Consider
195that `zfs list -t snapshot` performance degrades as snapshot counts grow within the selected datasets. Keep the active
196snapshot count small, and prune at a cadence that matches your snapshot creation rate. Consider using `--skip-parent` and
197`--exclude-dataset*` filters so only datasets that need this frequency are selected.
199To reduce startup overhead, forward `--daemon-lifetime` to `bzfs`, use the `--daemon-*` options, and split the
200crontab entry (or, preferably, a high-frequency systemd timer) into multiple processes, from one source host to one
201destination host, along these lines:
203* crontab on source hosts:
205`* * * * * testuser /bzfs/bzfs_testbed/bzfs_job_testbed.py --src-host="$(hostname)" --dst-host="foo" --create-src-snapshots`
207`* * * * * testuser /bzfs/bzfs_testbed/bzfs_job_testbed.py --src-host="$(hostname)" --dst-host="foo" --prune-src-snapshots`
209`* * * * * testuser /bzfs/bzfs_testbed/bzfs_job_testbed.py --src-host="$(hostname)" --dst-host="foo" --prune-src-bookmarks`
211`* * * * * testuser /bzfs/bzfs_testbed/bzfs_job_testbed.py --src-host="$(hostname)" --dst-host="foo" --monitor-src-snapshots`
214* crontab on destination hosts:
216`* * * * * testuser /bzfs/bzfs_testbed/bzfs_job_testbed.py --src-host="bar" --dst-host="$(hostname)" --replicate`
218`* * * * * testuser /bzfs/bzfs_testbed/bzfs_job_testbed.py --src-host="bar" --dst-host="$(hostname)" --prune-dst-snapshots`
220`* * * * * testuser /bzfs/bzfs_testbed/bzfs_job_testbed.py --src-host="bar" --dst-host="$(hostname)" --monitor-dst-snapshots`
222The daemon processes work like non-daemon processes except that they loop, handle time events, and sleep between events.
223They exit after the interval specified by `--daemon-lifetime` (for example, 86400 seconds). They are then restarted by
224`cron`, or earlier if they fail. While an existing daemon is still running, `cron` may attempt to start another one.
225This is harmless because that extra process exits immediately with a message like this:
226"Exiting as same previous periodic job is still running without completion yet"
227""")
229 # commands:
230 parser.add_argument(
231 "--create-src-snapshots", action="store_true",
232 help="Take snapshots on the selected source hosts as necessary. Typically, this command should be called by a "
233 "program (or cron job) running on each src host.\n\n")
234 parser.add_argument(
235 "--replicate", action="store_true",
236 help="Replicate snapshots from the selected source hosts to the selected destinations hosts as necessary. For pull "
237 "mode (recommended), this command should be called by a program (or cron job) running on each dst "
238 "host; for push mode, on the src host; for pull-push mode on a third-party host, maybe with `--r2r=pull` "
239 "or `--r2r=push`.\n\n")
240 parser.add_argument(
241 "--prune-src-snapshots", action="store_true",
242 help="Prune snapshots on the selected source hosts as necessary. Typically, this command should be called by a "
243 "program (or cron job) running on each src host.\n\n")
244 parser.add_argument(
245 "--prune-src-bookmarks", action="store_true",
246 help="Prune bookmarks on the selected source hosts as necessary. Typically, this command should be called by a "
247 "program (or cron job) running on each src host.\n\n")
248 parser.add_argument(
249 "--prune-dst-snapshots", action="store_true",
250 help="Prune snapshots on the selected destination hosts as necessary. Typically, this command should be called by a "
251 "program (or cron job) running on each dst host.\n\n")
252 parser.add_argument(
253 "--monitor-src-snapshots", action="store_true",
254 help="Alert the user if snapshots on the selected source hosts are too old, using --monitor-snapshot-plan (see "
255 "below). Typically, this command should be called by a program (or cron job) running on each src host.\n\n")
256 parser.add_argument(
257 "--monitor-dst-snapshots", action="store_true",
258 help="Alert the user if snapshots on the selected destination hosts are too old, using --monitor-snapshot-plan (see "
259 "below). Typically, this command should be called by a program (or cron job) running on each dst host.\n\n")
261 # options:
262 parser.add_argument(
263 "--localhost", default=None, action=bzfs_main.argparse_actions.NonEmptyStringAction, metavar="STRING",
264 help="Hostname of localhost. Default is the hostname without the domain name, querying the Operating System.\n\n")
265 parser.add_argument(
266 "--src-hosts", default=None, metavar="LIST_STRING",
267 help="Hostnames of the sources to operate on. Specify a Python list literal such as "
268 "`\"['src1', 'src2']\"`. If omitted, reads the same list literal from stdin.\n\n")
269 parser.add_argument(
270 "--src-host", default=None, action="append", metavar="STRING",
271 help="For subsetting --src-hosts; Can be specified multiple times; Indicates to only use the --src-hosts that are "
272 "contained in the specified --src-host values (optional).\n\n"
273 "Example: `--src-host=src1 --src-host=src2 --src-hosts=\"['src1', 'src2', 'src3', 'src4']\"` indicates to "
274 "effectively only use `--src-hosts=\"['src1', 'src2']\"`.\n\n")
275 dst_hosts_example = {"nas": ["onsite"], "bak-us-west": ["us-west"],
276 "bak-eu-west": ["eu-west"], "archive": ["offsite"]}
277 parser.add_argument(
278 "--dst-hosts", default="{}", metavar="DICT_STRING",
279 help="Dictionary that maps each destination hostname to a list of zero or more logical replication target names "
280 "(the infix portion of snapshot name). As hostname use the real output of the `hostname` CLI. "
281 "The target is an arbitrary user-defined name that serves as an abstraction of the destination hostnames for "
282 "a group of snapshots, like target 'onsite', 'offsite', 'hotspare', a geographically independent datacenter like "
283 "'us-west', or similar. Rather than the snapshot name embedding (i.e. hardcoding) a list of destination "
284 "hostnames where it should be sent to, the snapshot name embeds the user-defined target name, which is later "
285 "mapped by this jobconfig to a list of destination hostnames.\n\n"
286 f"Example: `{format_dict(dst_hosts_example)}`.\n\n"
287 "With this, given a snapshot name, we can find the destination hostnames to which the snapshot shall be "
288 "replicated. Also, given a snapshot name and its own name, a destination host can determine if it shall "
289 "replicate the given snapshot from the source host, or if the snapshot is intended for another destination "
290 "host, in which case it skips the snapshot. A destination host will receive replicas of snapshots for all "
291 "targets that map to that destination host.\n\n"
292 "Removing a mapping can be used to temporarily suspend replication to a given destination host.\n\n")
293 parser.add_argument(
294 "--dst-host", default=None, action="append", metavar="STRING",
295 help="For subsetting --dst-hosts; Can be specified multiple times; Indicates to only use the --dst-hosts keys that "
296 "are contained in the specified --dst-host values (optional).\n\n"
297 "Example: "
298 "`--dst-host=dst1 --dst-host=dst2 --dst-hosts=\"{'dst1': ..., 'dst2': ..., 'dst3': ..., 'dst4': ...}\"` "
299 "indicates to effectively only use `--dst-hosts=\"{'dst1': ..., 'dst2': ...}\"`.\n\n")
300 parser.add_argument(
301 "--retain-dst-targets", default="{}", metavar="DICT_STRING",
302 help="Dictionary that maps each destination hostname to a list of zero or more logical replication target names "
303 "(the infix portion of snapshot name).\n\n"
304 f"Example: `{format_dict(dst_hosts_example)}`. Has same format as --dst-hosts.\n\n"
305 "As part of --prune-dst-snapshots, a destination host will delete any snapshot it has stored whose target has "
306 "no mapping to that destination host in this dictionary. Do not remove a mapping here unless you are sure it's "
307 "ok to delete all those snapshots on that destination host! If in doubt, use --dryrun mode first.\n\n")
308 dst_root_datasets_example = {
309 "nas": "tank2/bak",
310 "bak-us-west": "backups/bak001",
311 "bak-eu-west": "backups/bak999",
312 "archive": f"archives/zoo/{SRC_MAGIC_SUBSTITUTION_TOKEN}",
313 "hotspare": "",
314 }
315 parser.add_argument(
316 "--dst-root-datasets", default="{}", metavar="DICT_STRING",
317 help="Dictionary that maps each destination hostname to a root dataset located on that destination host. The root "
318 "dataset name is an (optional) prefix that will be prepended to each dataset that is replicated to that "
319 "destination host. For backup use cases, this is the backup ZFS pool or a ZFS dataset path within that pool, "
320 "whereas for cloning, master slave replication, or replication from a primary to a secondary, this can also be "
321 "the empty string.\n\n"
322 f"`{SRC_MAGIC_SUBSTITUTION_TOKEN}` and `{DST_MAGIC_SUBSTITUTION_TOKEN}` are optional magic substitution tokens "
323 "that will be auto-replaced at runtime with the actual hostname. This can be used to force the use of a "
324 "separate destination root dataset per source host or per destination host.\n\n"
325 f"Example: `{format_dict(dst_root_datasets_example)}`\n\n")
326 src_snapshot_plan_example = {
327 "prod": {
328 "onsite": {"secondly": 40, "minutely": 40, "hourly": 36, "daily": 31, "weekly": 12, "monthly": 18, "yearly": 5},
329 "us-west": {"secondly": 0, "minutely": 0, "hourly": 36, "daily": 31, "weekly": 12, "monthly": 18, "yearly": 5},
330 "eu-west": {"secondly": 0, "minutely": 0, "hourly": 36, "daily": 31, "weekly": 12, "monthly": 18, "yearly": 5},
331 },
332 "test": {
333 "offsite": {"12hourly": 42, "weekly": 12},
334 },
335 }
336 parser.add_argument(
337 "--src-snapshot-plan", default="{}", metavar="DICT_STRING",
338 help="Retention periods for snapshots to be used if pruning src, and when creating new snapshots on src. "
339 "Snapshots that do not match a retention period will be deleted. A zero or missing retention period indicates "
340 "that no snapshots shall be retained (or even be created) for the given period.\n\n"
341 f"Example: `{format_dict(src_snapshot_plan_example)}`. This example will, for the organization 'prod' and "
342 "the intended logical target 'onsite', create and then retain secondly snapshots that were created less "
343 "than 40 seconds ago, yet retain the latest 40 secondly snapshots regardless of creation time. Analog for "
344 "the latest 40 minutely snapshots, 36 hourly snapshots, etc. "
345 "It will also create and retain snapshots for the targets 'us-west' and 'eu-west' within the 'prod' "
346 "organization. "
347 "In addition, it will create and retain snapshots every 12 hours and every week for the 'test' organization, "
348 "and name them as being intended for the 'offsite' replication target. "
349 "The example creates snapshots with names like "
350 "`prod_onsite_<timestamp>_secondly`, `prod_onsite_<timestamp>_minutely`, "
351 "`prod_us-west_<timestamp>_hourly`, `prod_us-west_<timestamp>_daily`, "
352 "`prod_eu-west_<timestamp>_hourly`, `prod_eu-west_<timestamp>_daily`, "
353 "`test_offsite_<timestamp>_12hourly`, `test_offsite_<timestamp>_weekly`, and so on.\n\n")
354 parser.add_argument(
355 "--src-bookmark-plan", default="{}", metavar="DICT_STRING",
356 help="Retention periods for bookmarks to be used if pruning src. Has same format as --src-snapshot-plan.\n\n")
357 parser.add_argument(
358 "--dst-snapshot-plan", default="{}", metavar="DICT_STRING",
359 help="Retention periods for snapshots to be used if pruning dst. Has same format as --src-snapshot-plan.\n\n")
360 monitor_snapshot_plan_example = {
361 "prod": {
362 "onsite": {
363 "100millisecondly": {"warning": "650 milliseconds", "critical": "2 seconds"},
364 "secondly": {"warning": "2 seconds", "critical": "14 seconds"},
365 "minutely": {"warning": "30 seconds", "critical": "300 seconds"},
366 "hourly": {"warning": "30 minutes", "critical": "300 minutes"},
367 "daily": {"warning": "4 hours", "critical": "8 hours"},
368 "weekly": {"warning": "2 days", "critical": "8 days"},
369 "monthly": {"warning": "2 days", "critical": "8 days"},
370 "yearly": {"warning": "5 days", "critical": "14 days"},
371 "10minutely": {"warning": "0 minutes", "critical": "0 minutes"},
372 },
373 "": {
374 "daily": {"warning": "4 hours", "critical": "8 hours"},
375 },
376 },
377 }
378 parser.add_argument(
379 "--monitor-snapshot-plan", default="{}", metavar="DICT_STRING",
380 help="Alert the user if the ZFS 'creation' time property of the latest or oldest snapshot for any specified "
381 "snapshot pattern within the selected datasets is too old wrt. the specified age limit. The purpose is to "
382 "check if snapshots are successfully taken on schedule, successfully replicated on schedule, and successfully "
383 "pruned on schedule. "
384 "Process exit code is 0, 1, 2 on OK, WARNING, CRITICAL, respectively.\n\n"
385 f"Example DICT_STRING: `{format_dict(monitor_snapshot_plan_example)}`. "
386 "This example alerts the user if the *latest* src or dst snapshot named `prod_onsite_<timestamp>_hourly` is "
387 "more than 30 minutes late (i.e. more than 30+60=90 minutes old) [warning] or more than 300 minutes late (i.e. "
388 "more than 300+60=360 minutes old) [critical]. In addition, the example alerts the user if the *oldest* src or "
389 "dst snapshot named `prod_onsite_<timestamp>_hourly` is more than 30 + 60x36 minutes old [warning] or more "
390 "than 300 + 60x36 minutes old [critical], where 36 is the number of period cycles specified in "
391 "`src_snapshot_plan` or `dst_snapshot_plan`, respectively. "
392 "Analog for the latest snapshot named `prod_<timestamp>_daily`, and so on.\n\n"
393 "Note: A duration that is missing or zero (e.g. '0 minutes') indicates that no snapshots shall be checked for "
394 "the given snapshot name pattern.\n\n")
395 locations = ["src", "dst"]
396 for loc in locations:
397 parser.add_argument(
398 f"--ssh-{loc}-user", default="", metavar="STRING",
399 help=f"Remote SSH username on {loc} hosts to connect to (optional). Examples: 'root', 'alice'.\n\n")
400 for loc in locations:
401 parser.add_argument(
402 f"--ssh-{loc}-port", type=int, min=1, max=65535, action=check_range.CheckRange, metavar="INT",
403 help=f"Remote SSH port on {loc} host to connect to (optional).\n\n")
404 for loc in locations:
405 parser.add_argument(
406 f"--ssh-{loc}-config-file", type=str, action=bzfs_main.argparse_actions.SSHConfigFileNameAction, metavar="FILE",
407 help=f"Path to SSH ssh_config(5) file to connect to {loc} (optional); will be passed into ssh -F CLI. "
408 "The basename must contain the substring 'bzfs_ssh_config'.\n\n")
409 parser.add_argument(
410 "--job-id", required=True, action=bzfs_main.argparse_actions.NonEmptyStringAction, metavar="STRING",
411 help="The identifier that remains constant across all runs of this particular job; will be included in the log file "
412 "name infix. Example: mytestjob\n\n")
413 parser.add_argument(
414 "--job-run", default="", action=bzfs_main.argparse_actions.NonEmptyStringAction, metavar="STRING",
415 help="The identifier of this particular run of the overall job; will be included in the log file name suffix. "
416 "Default is a hex UUID. Example: 0badc0f003a011f0a94aef02ac16083c\n\n")
417 workers_default = 100 # percent
418 parser.add_argument(
419 "--workers", min=1, default=(workers_default, True), action=bzfs_main.argparse_actions.CheckPercentRange,
420 metavar="INT[%]",
421 help="The maximum number of jobs to run in parallel at any time; can be given as a positive integer, "
422 f"optionally followed by the %% percent character (min: %(min)s, default: {workers_default}%%). Percentages "
423 "are relative to the number of CPU cores on the machine. Example: 200%% uses twice as many parallel jobs as "
424 "there are cores on the machine; 75%% uses num_procs = num_cores * 0.75. Examples: 1, 4, 75%%, 150%%\n\n")
425 parser.add_argument(
426 "--work-period-seconds", type=float, min=0, default=0, action=check_range.CheckRange, metavar="FLOAT",
427 help="Reduces bandwidth spikes by spreading out the start of worker jobs over this much time; "
428 "0 disables this feature (default: %(default)s). Examples: 0, 60, 86400\n\n")
429 parser.add_argument(
430 "--jitter", action="store_true",
431 help="Randomize job start time and host order to avoid potential thundering herd problems in large distributed "
432 "systems (optional). Randomizing job start time is only relevant if --work-period-seconds > 0.\n\n")
433 parser.add_argument(
434 "--worker-timeout-seconds", type=float, min=0.001, default=None, action=check_range.CheckRange, metavar="FLOAT",
435 help="If this much time has passed after a worker process has started executing, kill the straggling worker "
436 "(optional). Other workers remain unaffected. Examples: 60, 3600\n\n")
437 parser.add_argument(
438 "--repeat-if-took-more-than-seconds", type=float, min=0.001, default=UNIX_TIME_INFINITY_SECS,
439 action=check_range.CheckRange, metavar="FLOAT",
440 help="Repeat the entire workflow if it took longer than this much time and was successful. Use this (with the POSIX "
441 "`timeout` CLI) before migrating VM storage to converge replication and reduce cutover downtime. Default is "
442 "infinity, i.e. never repeat the workflow. Examples: 1, 0.1\n\n")
443 parser.add_argument(
444 "--spawn-process-per-job", action="store_true",
445 help="Spawn a Python process per subjob instead of a Python thread per subjob (optional). The former is only "
446 "recommended for a job operating in parallel on a large number of hosts as it helps avoid exceeding "
447 "per-process limits such as the default max number of open file descriptors, at the expense of increased "
448 "startup latency.\n\n")
449 parser.add_argument(
450 "--jobrunner-dryrun", action="store_true",
451 help="Do a dry run (aka 'no-op') to print what operations would happen if the command were to be executed "
452 "for real (optional). This option treats both the ZFS source and destination as read-only. Can also be used to "
453 "check if the configuration options are valid.\n\n")
454 parser.add_argument(
455 "--jobrunner-log-level", choices=["CRITICAL", "ERROR", "WARN", "INFO", "DEBUG", "TRACE"], default="INFO",
456 help="Only emit jobrunner messages with equal or higher priority than this log level. Default is '%(default)s'.\n\n")
457 parser.add_argument(
458 "--daemon-replication-frequency", default="minutely", metavar="STRING",
459 help="Specifies how often the bzfs daemon shall replicate from src to dst if --daemon-lifetime is nonzero.\n\n")
460 parser.add_argument(
461 "--daemon-prune-src-frequency", default="minutely", metavar="STRING",
462 help="Specifies how often the bzfs daemon shall prune src if --daemon-lifetime is nonzero.\n\n")
463 parser.add_argument(
464 "--daemon-prune-dst-frequency", default="minutely", metavar="STRING",
465 help="Specifies how often the bzfs daemon shall prune dst if --daemon-lifetime is nonzero.\n\n")
466 parser.add_argument(
467 "--daemon-monitor-snapshots-frequency", default="minutely", metavar="STRING",
468 help="Specifies how often the bzfs daemon shall monitor snapshot age if --daemon-lifetime is nonzero.\n\n")
469 bad_opts = ["--daemon-frequency", "--include-snapshot-plan", "--create-src-snapshots-plan", "--skip-replication",
470 "--include-snapshot-regex", "--exclude-snapshot-regex", "--include-snapshot-times-and-ranks",
471 "--log-file-prefix", "--log-file-infix", "--log-file-suffix",
472 "--delete-dst-datasets", "--delete-dst-snapshots", "--delete-dst-snapshots-except",
473 "--delete-dst-snapshots-except-plan", "--delete-empty-dst-datasets",
474 "--monitor-snapshots", "--timeout"]
475 for loc in locations:
476 bad_opts += [f"--ssh-{loc}-host"] # reject this arg as jobrunner will auto-generate it
477 for bad_opt in bad_opts:
478 parser.add_argument(bad_opt, action=RejectArgumentAction, nargs=0, help=argparse.SUPPRESS)
479 parser.add_argument(
480 "--version", action="version", version=f"{PROG_NAME}-{bzfs_main.argparse_cli.__version__}, by {PROG_AUTHOR}",
481 help="Display version information and exit.\n\n")
482 parser.add_argument(
483 "--root-dataset-pairs", required=True, nargs="+", action=bzfs_main.argparse_actions.DatasetPairsAction,
484 metavar="SRC_DATASET DST_DATASET",
485 help="Source and destination dataset pairs (excluding usernames and excluding hostnames, which will all be "
486 "auto-appended later).\n\n")
487 return parser
488 # fmt: on
491#############################################################################
492def main() -> None:
493 """API for command line clients."""
494 prev_umask: int = os.umask(UMASK)
495 try:
496 set_logging_runtime_defaults()
497 # On CTRL-C and SIGTERM, send signal to all descendant processes to terminate them
498 termination_event: threading.Event = threading.Event()
499 with termination_signal_handler(termination_events=[termination_event]):
500 Job(log=None, termination_event=termination_event).run_main(sys_argv=sys.argv)
501 finally:
502 os.umask(prev_umask) # restore prior global state
505#############################################################################
506@final
507class Job:
508 """Coordinates subjobs per the CLI flags; Each subjob handles one host pair and may run in its own process or thread."""
510 def __init__(self, log: Logger | None, termination_event: threading.Event) -> None:
511 # immutable variables:
512 self.log_was_None: Final[bool] = log is None
513 self.log: Final[Logger] = get_simple_logger(PROG_NAME) if log is None else log
514 self.termination_event: Final[threading.Event] = termination_event
515 self.timing: Final[TaskTiming] = TaskTiming.make_from(self.termination_event)
516 self.subprocesses: Final[Subprocesses] = Subprocesses(termination_event.is_set)
517 self.jobrunner_dryrun: bool = False
518 self.spawn_process_per_job: bool = False
519 self.loopback_address: Final[str] = _detect_loopback_address()
521 # mutable variables:
522 self.first_exception: int | None = None
523 self.worst_exception: int | None = None
524 self.stats: JobStats = JobStats(jobs_all=0)
525 self.cache_existing_dst_pools: set[str] = set()
526 self.cache_known_dst_pools: set[str] = set()
528 self.is_test_mode: bool = False # for testing only
530 def run_main(self, sys_argv: list[str]) -> None:
531 """API for Python clients; visible for testing; may become a public API eventually."""
532 try:
533 self._run_main(sys_argv)
534 finally:
535 if self.log_was_None: # reset Logger unless it's a Logger outside of our control
536 reset_logger(self.log)
538 def _run_main(self, sys_argv: list[str]) -> None:
539 self.first_exception = None
540 self.worst_exception = None
541 log: Logger = self.log
542 log.info("CLI arguments: %s", " ".join(sys_argv))
543 nsp = argparse.Namespace(no_argument_file=True) # disable --root-dataset-pairs='+file' option in DatasetPairsAction
544 args, unknown_args = argument_parser().parse_known_args(sys_argv[1:], nsp) # forward all unknown args to `bzfs`
545 log.setLevel(args.jobrunner_log_level)
546 self.jobrunner_dryrun = args.jobrunner_dryrun
547 assert len(args.root_dataset_pairs) > 0
548 src_snapshot_plan: dict = self.validate_snapshot_plan(literal_eval(args.src_snapshot_plan), "--src-snapshot-plan")
549 src_bookmark_plan: dict = self.validate_snapshot_plan(literal_eval(args.src_bookmark_plan), "--src-bookmark-plan")
550 dst_snapshot_plan: dict = self.validate_snapshot_plan(literal_eval(args.dst_snapshot_plan), "--dst-snapshot-plan")
551 monitor_snapshot_plan: dict = self.validate_monitor_snapshot_plan(literal_eval(args.monitor_snapshot_plan))
552 localhostname: str = args.localhost if args.localhost else socket.gethostname()
553 self.validate_host_name(localhostname, "--localhost")
554 log.debug("localhostname: %s", localhostname)
555 src_hosts: list[str] = self.validate_src_hosts(self.parse_src_hosts_from_cli_or_stdin(args.src_hosts))
556 basis_src_hosts: list[str] = src_hosts
557 nb_src_hosts: int = len(basis_src_hosts)
558 log.debug("src_hosts before subsetting: %s", src_hosts)
559 if args.src_host is not None: # retain only the src hosts that are also contained in args.src_host
560 assert isinstance(args.src_host, list)
561 retain_src_hosts: set[str] = set(args.src_host)
562 self.validate_is_subset(retain_src_hosts, src_hosts, "--src-host", "--src-hosts")
563 src_hosts = [host for host in src_hosts if host in retain_src_hosts]
564 dst_hosts: dict[str, list[str]] = self.validate_dst_hosts(literal_eval(args.dst_hosts))
565 nb_dst_hosts: int = len(dst_hosts)
566 if args.dst_host is not None: # retain only the dst hosts that are also contained in args.dst_host
567 assert isinstance(args.dst_host, list)
568 retain_dst_hosts: set[str] = set(args.dst_host)
569 self.validate_is_subset(retain_dst_hosts, dst_hosts.keys(), "--dst-host", "--dst-hosts.keys")
570 dst_hosts = {dst_host: lst for dst_host, lst in dst_hosts.items() if dst_host in retain_dst_hosts}
571 retain_dst_targets: dict[str, list[str]] = self.validate_dst_hosts(literal_eval(args.retain_dst_targets))
572 self.validate_is_subset(dst_hosts.keys(), retain_dst_targets.keys(), "--dst-hosts.keys", "--retain-dst-targets.keys")
573 dst_root_datasets: dict[str, str] = self.validate_dst_root_datasets(literal_eval(args.dst_root_datasets))
574 self.validate_is_subset(
575 dst_root_datasets.keys(), retain_dst_targets.keys(), "--dst-root-dataset.keys", "--retain-dst-targets.keys"
576 )
577 self.validate_is_subset(dst_hosts.keys(), dst_root_datasets.keys(), "--dst-hosts.keys", "--dst-root-dataset.keys")
578 bad_root_datasets: dict[str, str] = {
579 dst_host: root_dataset
580 for dst_host in sorted(dst_hosts.keys())
581 if SRC_MAGIC_SUBSTITUTION_TOKEN not in (root_dataset := dst_root_datasets[dst_host])
582 }
583 if len(src_hosts) > 1 and len(bad_root_datasets) > 0:
584 self.die(
585 "Cowardly refusing to proceed as multiple source hosts must not be configured to write to the same "
586 "destination dataset. "
587 f"Problematic subset of --dst-root-datasets: {bad_root_datasets} for src_hosts: {sorted(src_hosts)}"
588 )
589 bad_root_datasets = {
590 dst_host: root_dataset
591 for dst_host, root_dataset in sorted(dst_root_datasets.items())
592 if root_dataset and SRC_MAGIC_SUBSTITUTION_TOKEN not in root_dataset
593 }
594 if len(basis_src_hosts) > 1 and len(bad_root_datasets) > 0:
595 self.die(
596 "Cowardly refusing to proceed as multiple source hosts are defined in the configuration, but "
597 f"not all non-empty root datasets in --dst-root-datasets contain the '{SRC_MAGIC_SUBSTITUTION_TOKEN}' "
598 "substitution token to prevent collisions on writing destination datasets. "
599 f"Problematic subset of --dst-root-datasets: {bad_root_datasets} for src_hosts: {sorted(basis_src_hosts)}"
600 )
601 if args.jitter: # randomize host order to avoid potential thundering herd problems in large distributed systems
602 random.SystemRandom().shuffle(src_hosts)
603 dst_hosts = shuffle_dict(dst_hosts)
604 ssh_src_user: str = args.ssh_src_user
605 ssh_dst_user: str = args.ssh_dst_user
606 ssh_src_port: int | None = args.ssh_src_port
607 ssh_dst_port: int | None = args.ssh_dst_port
608 ssh_src_config_file: str | None = args.ssh_src_config_file
609 ssh_dst_config_file: str | None = args.ssh_dst_config_file
610 job_id: str = _sanitize(args.job_id)
611 job_run: str = _sanitize(args.job_run) if args.job_run else uuid.uuid1().hex
612 workers, workers_is_percent = args.workers
613 max_workers: int = max(1, round((os.cpu_count() or 1) * workers / 100.0) if workers_is_percent else round(workers))
614 worker_timeout_seconds: int = args.worker_timeout_seconds
615 repeat_if_took_more_than_nanos: int = int(args.repeat_if_took_more_than_seconds * 1_000_000_000)
616 self.spawn_process_per_job = args.spawn_process_per_job
617 username: str = pwd.getpwuid(os.getuid()).pw_name
618 assert username
619 loopback_ids: set[str] = {"localhost", "127.0.0.1", "::1", socket.gethostname()} # ::1 is IPv6 loopback address
620 loopback_ids.update(self.get_localhost_ips()) # union
621 loopback_ids.add(localhostname)
622 loopback_ids = set() if getenv_bool("disable_loopback", False) else loopback_ids
623 log.log(LOG_TRACE, "loopback_ids: %s", sorted(loopback_ids))
625 def zero_pad(number: int, width: int = 6) -> str:
626 """Pads number with leading '0' chars to the given width."""
627 return f"{number:0{width}d}"
629 def jpad(jj: int, tag: str) -> str:
630 """Returns ``tag`` prefixed with slash and zero padded index."""
631 return "/" + zero_pad(jj) + tag
633 def runpad() -> str:
634 """Returns standardized subjob count suffix."""
635 return job_run + SEP + zero_pad(len(subjobs))
637 def update_subjob_name(tag: str) -> str:
638 """Derives next subjob name based on ``tag`` and index ``j``."""
639 if j <= 0:
640 return subjob_name
641 elif j == 1:
642 return subjob_name + jpad(j - 1, tag)
643 else:
644 return subjob_name + "/" + BARRIER_CHAR
646 def resolve_dataset(hostname: str, dataset: str, is_src: bool = True) -> str:
647 """Returns host:dataset string resolving IPv6 and localhost cases."""
648 assert hostname
649 assert dataset
650 ssh_user = ssh_src_user if is_src else ssh_dst_user
651 ssh_user = ssh_user if ssh_user else username
652 lb: str = self.loopback_address
653 loopbck_ids: set[str] = loopback_ids
654 hostname = hostname if hostname not in loopbck_ids else (lb if lb else hostname) if username != ssh_user else "-"
655 hostname = convert_ipv6(hostname)
656 return f"{hostname}:{dataset}"
658 def resolve_dst_dataset(dst_hostname: str, dst_dataset: str) -> str:
659 """Expands ``dst_dataset`` relative to ``dst_hostname`` roots."""
660 assert dst_hostname
661 assert dst_dataset
662 root_dataset: str | None = dst_root_datasets.get(dst_hostname)
663 assert root_dataset is not None, dst_hostname # f"Hostname '{dst_hostname}' missing in --dst-root-datasets"
664 root_dataset = root_dataset.replace(SRC_MAGIC_SUBSTITUTION_TOKEN, src_host)
665 root_dataset = root_dataset.replace(DST_MAGIC_SUBSTITUTION_TOKEN, dst_hostname)
666 resolved_dst_dataset: str = f"{root_dataset}/{dst_dataset}" if root_dataset else dst_dataset
667 validate_dataset_name(resolved_dst_dataset, dst_dataset)
668 return resolve_dataset(dst_hostname, resolved_dst_dataset, is_src=False)
670 for src_host in src_hosts:
671 assert src_host
672 for dst_hostname in dst_hosts:
673 assert dst_hostname
674 dummy: Final[str] = DUMMY_DATASET
675 lhn: Final[str] = localhostname
676 bzfs_prog_header: Final[list[str]] = [BZFS_PROG_NAME, "--no-argument-file"] + unknown_args
677 subjobs: dict[str, list[str]] = {}
678 for i, src_host in enumerate(src_hosts):
679 subjob_name: str = zero_pad(i) + "src-host"
680 src_log_suffix: str = _log_suffix(localhostname, src_host, "")
681 j: int = 0
682 opts: list[str]
684 if args.create_src_snapshots:
685 opts = ["--create-src-snapshots", f"--create-src-snapshots-plan={src_snapshot_plan}", "--skip-replication"]
686 self.add_log_file_opts(opts, "create-src-snapshots", job_id, runpad(), src_log_suffix)
687 self.add_ssh_opts(
688 opts, ssh_src_user=ssh_src_user, ssh_src_port=ssh_src_port, ssh_src_config_file=ssh_src_config_file
689 )
690 opts += [POSIX_END_OF_OPTIONS_MARKER]
691 opts += _flatten(_dedupe([(resolve_dataset(src_host, src), dummy) for src, dst in args.root_dataset_pairs]))
692 subjob_name += "/create-src-snapshots"
693 subjobs[subjob_name] = bzfs_prog_header + opts
695 if args.replicate:
696 j = 0
697 marker: str = "replicate"
698 for dst_hostname, targets in dst_hosts.items():
699 opts = self.replication_opts(
700 dst_snapshot_plan, set(targets), lhn, src_host, dst_hostname, marker, job_id, runpad()
701 )
702 if len(opts) > 0:
703 opts += [f"--daemon-frequency={args.daemon_replication_frequency}"]
704 self.add_ssh_opts(
705 opts,
706 ssh_src_user=ssh_src_user,
707 ssh_dst_user=ssh_dst_user,
708 ssh_src_port=ssh_src_port,
709 ssh_dst_port=ssh_dst_port,
710 ssh_src_config_file=ssh_src_config_file,
711 ssh_dst_config_file=ssh_dst_config_file,
712 )
713 opts += [POSIX_END_OF_OPTIONS_MARKER]
714 dataset_pairs: list[tuple[str, str]] = [
715 (resolve_dataset(src_host, src), resolve_dst_dataset(dst_hostname, dst))
716 for src, dst in args.root_dataset_pairs
717 ]
718 dataset_pairs = self.skip_nonexisting_local_dst_pools(dataset_pairs, worker_timeout_seconds)
719 if len(dataset_pairs) > 0:
720 subjobs[subjob_name + jpad(j, marker)] = bzfs_prog_header + opts + _flatten(dataset_pairs)
721 j += 1
722 subjob_name = update_subjob_name(marker)
724 def prune_src(
725 opts: list[str], retention_plan: dict, tag: str, src_host: str = src_host, logsuffix: str = src_log_suffix
726 ) -> None:
727 """Creates prune subjob options for ``tag`` using ``retention_plan``."""
728 opts += ["--skip-replication", f"--delete-dst-snapshots-except-plan={retention_plan}"]
729 opts += [f"--daemon-frequency={args.daemon_prune_src_frequency}"]
730 self.add_log_file_opts(opts, tag, job_id, runpad(), logsuffix)
731 self.add_ssh_opts( # i.e. dst=src, src=dummy
732 opts, ssh_dst_user=ssh_src_user, ssh_dst_port=ssh_src_port, ssh_dst_config_file=ssh_src_config_file
733 )
734 opts += [POSIX_END_OF_OPTIONS_MARKER]
735 opts += _flatten(_dedupe([(dummy, resolve_dataset(src_host, src)) for src, dst in args.root_dataset_pairs]))
736 nonlocal subjob_name
737 subjob_name += f"/{tag}"
738 subjobs[subjob_name] = bzfs_prog_header + opts
740 if args.prune_src_snapshots:
741 prune_src(["--delete-dst-snapshots"], src_snapshot_plan, tag="prune-src-snapshots")
743 if args.prune_src_bookmarks:
744 prune_src(["--delete-dst-snapshots=bookmarks"], src_bookmark_plan, tag="prune-src-bookmarks")
746 if args.prune_dst_snapshots:
747 self.validate_true(
748 retain_dst_targets, "--retain-dst-targets must not be empty. Cowardly refusing to delete all snapshots!"
749 )
750 j = 0
751 marker = "prune-dst-snapshots"
752 for dst_hostname in dst_hosts:
753 curr_retain_targets: set[str] = set(retain_dst_targets[dst_hostname])
754 curr_dst_snapshot_plan = { # only retain targets that belong to the host
755 org: {target: periods for target, periods in target_periods.items() if target in curr_retain_targets}
756 for org, target_periods in dst_snapshot_plan.items()
757 }
758 opts = ["--delete-dst-snapshots", "--skip-replication"]
759 opts += [f"--delete-dst-snapshots-except-plan={curr_dst_snapshot_plan}"]
760 opts += [f"--daemon-frequency={args.daemon_prune_dst_frequency}"]
761 self.add_log_file_opts(opts, marker, job_id, runpad(), _log_suffix(lhn, src_host, dst_hostname))
762 self.add_ssh_opts(
763 opts, ssh_dst_user=ssh_dst_user, ssh_dst_port=ssh_dst_port, ssh_dst_config_file=ssh_dst_config_file
764 )
765 opts += [POSIX_END_OF_OPTIONS_MARKER]
766 dataset_pairs = [(dummy, resolve_dst_dataset(dst_hostname, dst)) for src, dst in args.root_dataset_pairs]
767 dataset_pairs = _dedupe(dataset_pairs)
768 dataset_pairs = self.skip_nonexisting_local_dst_pools(dataset_pairs, worker_timeout_seconds)
769 if len(dataset_pairs) > 0:
770 subjobs[subjob_name + jpad(j, marker)] = bzfs_prog_header + opts + _flatten(dataset_pairs)
771 j += 1
772 subjob_name = update_subjob_name(marker)
774 def monitor_snapshots_opts(tag: str, monitor_plan: dict, logsuffix: str) -> list[str]:
775 """Returns monitor subjob options for ``tag`` and ``monitor_plan``."""
776 opts = [f"--monitor-snapshots={monitor_plan}", "--skip-replication"]
777 opts += [f"--daemon-frequency={args.daemon_monitor_snapshots_frequency}"]
778 self.add_log_file_opts(opts, tag, job_id, runpad(), logsuffix)
779 return opts
781 def build_monitor_plan(monitor_plan: dict, snapshot_plan: dict, cycles_prefix: str) -> dict:
782 """Expands ``monitor_plan`` with cycle defaults from ``snapshot_plan``."""
784 def alert_dicts(alertdict: dict, cycles: int) -> dict:
785 """Returns alert dictionaries with explicit ``cycles`` value."""
786 latest_dict = alertdict.copy()
787 for prefix in ("src_snapshot_", "dst_snapshot_", ""):
788 latest_dict.pop(f"{prefix}cycles", None)
789 oldest_dict = latest_dict.copy()
790 oldest_dict["cycles"] = int(alertdict.get(f"{cycles_prefix}cycles", cycles))
791 latest_dict.pop("oldest_skip_holds", None)
792 return {"latest": latest_dict, "oldest": oldest_dict}
794 return {
795 org: {
796 target: {
797 periodunit: alert_dicts(alertdict, snapshot_plan.get(org, {}).get(target, {}).get(periodunit, 1))
798 for periodunit, alertdict in periods.items()
799 }
800 for target, periods in target_periods.items()
801 }
802 for org, target_periods in monitor_plan.items()
803 }
805 subjob_name += "/" + BARRIER_CHAR + "/monitor-snapshots"
806 if args.monitor_src_snapshots:
807 marker = "monitor-src-snapshots"
808 monitor_plan = build_monitor_plan(monitor_snapshot_plan, src_snapshot_plan, "src_snapshot_")
809 opts = monitor_snapshots_opts(marker, monitor_plan, src_log_suffix)
810 self.add_ssh_opts( # i.e. dst=src, src=dummy
811 opts, ssh_dst_user=ssh_src_user, ssh_dst_port=ssh_src_port, ssh_dst_config_file=ssh_src_config_file
812 )
813 opts += [POSIX_END_OF_OPTIONS_MARKER]
814 opts += _flatten(_dedupe([(dummy, resolve_dataset(src_host, src)) for src, dst in args.root_dataset_pairs]))
815 subjobs[subjob_name + "/." + marker] = bzfs_prog_header + opts
817 if args.monitor_dst_snapshots:
818 j = 0
819 marker = "monitor-dst-snapshots"
820 for dst_hostname, targets in dst_hosts.items():
821 monitor_targets: set[str] = set(targets).intersection(set(retain_dst_targets[dst_hostname]))
822 monitor_plan = { # only retain targets that belong to the host
823 org: {target: periods for target, periods in target_periods.items() if target in monitor_targets}
824 for org, target_periods in monitor_snapshot_plan.items()
825 }
826 monitor_plan = build_monitor_plan(monitor_plan, dst_snapshot_plan, "dst_snapshot_")
827 opts = monitor_snapshots_opts(marker, monitor_plan, _log_suffix(lhn, src_host, dst_hostname))
828 self.add_ssh_opts(
829 opts, ssh_dst_user=ssh_dst_user, ssh_dst_port=ssh_dst_port, ssh_dst_config_file=ssh_dst_config_file
830 )
831 opts += [POSIX_END_OF_OPTIONS_MARKER]
832 dataset_pairs = [(dummy, resolve_dst_dataset(dst_hostname, dst)) for src, dst in args.root_dataset_pairs]
833 dataset_pairs = _dedupe(dataset_pairs)
834 dataset_pairs = self.skip_nonexisting_local_dst_pools(dataset_pairs, worker_timeout_seconds)
835 if len(dataset_pairs) > 0:
836 subjobs[subjob_name + jpad(j, marker)] = bzfs_prog_header + opts + _flatten(dataset_pairs)
837 j += 1
838 subjob_name = update_subjob_name(marker)
840 msg = f"Ready to run {len(subjobs)} subjobs using {len(src_hosts)}/{nb_src_hosts} src hosts: {src_hosts}, "
841 msg += f"{len(dst_hosts)}/{nb_dst_hosts} dst hosts: {list(dst_hosts.keys())}"
842 log.info("%s", dry(msg, is_dry_run=self.jobrunner_dryrun))
843 log.log(LOG_TRACE, "subjobs: \n%s", _pretty_print_formatter(subjobs))
844 while True:
845 start_time_nanos = time.monotonic_ns()
846 self.first_exception = None
847 self.worst_exception = None
848 self.run_subjobs(subjobs, max_workers, worker_timeout_seconds, args.work_period_seconds, args.jitter)
849 ex = self.worst_exception
850 if isinstance(ex, int):
851 assert ex != 0
852 sys.exit(ex)
853 assert ex is None, ex
854 if time.monotonic_ns() - start_time_nanos <= repeat_if_took_more_than_nanos:
855 break
856 log.info("Succeeded. Bye!")
858 def replication_opts(
859 self,
860 dst_snapshot_plan: dict[str, dict[str, dict[str, int]]],
861 targets: set[str],
862 localhostname: str,
863 src_hostname: str,
864 dst_hostname: str,
865 tag: str,
866 job_id: str,
867 job_run: str,
868 ) -> list[str]:
869 """Returns CLI options for one replication subjob."""
870 log = self.log
871 log.debug("%s", f"Replicating targets {sorted(targets)} from {src_hostname} to {dst_hostname} ...")
872 include_snapshot_plan = { # only replicate targets that belong to the destination host and are relevant
873 org: {
874 target: {
875 duration_unit: duration_amount
876 for duration_unit, duration_amount in periods.items()
877 if duration_amount > 0
878 }
879 for target, periods in target_periods.items()
880 if target in targets
881 }
882 for org, target_periods in dst_snapshot_plan.items()
883 }
884 include_snapshot_plan = { # only replicate orgs that have at least one relevant target_period
885 org: target_periods
886 for org, target_periods in include_snapshot_plan.items()
887 if any(len(periods) > 0 for periods in target_periods.values())
888 }
889 opts: list[str] = []
890 if len(include_snapshot_plan) > 0:
891 opts += [f"--include-snapshot-plan={include_snapshot_plan}"]
892 self.add_log_file_opts(opts, tag, job_id, job_run, _log_suffix(localhostname, src_hostname, dst_hostname))
893 return opts
895 def skip_nonexisting_local_dst_pools(
896 self, root_dataset_pairs: list[tuple[str, str]], timeout_secs: float | None = None
897 ) -> list[tuple[str, str]]:
898 """Skip dst datasets that point to removable destination drives that are not currently (locally) attached, if any."""
900 def zpool(dataset: str) -> str:
901 """Returns pool name portion of ``dataset``."""
902 return dataset.split("/", 1)[0]
904 assert len(root_dataset_pairs) > 0
905 if getenv_bool("disable_skip_nonexisting_local_dst_pools", False):
906 return root_dataset_pairs
907 unknown_dst_pools = {zpool(dst) for src, dst in root_dataset_pairs}
908 unknown_dst_pools = unknown_dst_pools.difference(self.cache_known_dst_pools)
910 # Here we treat a zpool as existing if the zpool isn't local, aka if it isn't prefixed with "-:". A remote host
911 # will raise an appropriate error if it turns out that the remote zpool doesn't actually exist.
912 unknown_local_dst_pools = {pool for pool in unknown_dst_pools if pool.startswith("-:")}
913 if len(unknown_local_dst_pools) > 0: # `zpool list` if local
914 existing_pools = {pool[len("-:") :] for pool in unknown_local_dst_pools}
915 cmd = "zpool version".split(" ") # ensure `zpool` CLI works and returns exit code zero
916 sp = subprocess.run(cmd, stdin=DEVNULL, stdout=PIPE, stderr=PIPE, text=True, timeout=timeout_secs, check=False)
917 if sp.returncode != 0:
918 self.die(f"Unexpected error {sp.returncode} on checking for existing local dst zpools: {sp.stderr.strip()}")
919 cmd = "zpool list -Hp -o name".split(" ") + sorted(existing_pools) # don't wake up drives of unrelated pools
920 sp = subprocess.run(cmd, stdin=DEVNULL, stdout=PIPE, stderr=PIPE, text=True, timeout=timeout_secs, check=False)
921 if sp.returncode not in (0, 1): # 1 means pool not found
922 self.die(f"Unexpected error {sp.returncode} on checking for existing local dst pools: {sp.stderr.strip()}")
923 existing_pools = {"-:" + pool for pool in sp.stdout.splitlines() if pool}
924 self.cache_existing_dst_pools.update(existing_pools) # union
925 unknown_remote_dst_pools = unknown_dst_pools.difference(unknown_local_dst_pools)
926 self.cache_existing_dst_pools.update(unknown_remote_dst_pools) # union
927 self.cache_known_dst_pools.update(unknown_dst_pools) # union
928 results: list[tuple[str, str]] = []
929 for src, dst in root_dataset_pairs:
930 if zpool(dst) in self.cache_existing_dst_pools:
931 results.append((src, dst))
932 else:
933 self.log.warning("Skipping dst dataset for which local dst pool does not exist: %s", dst)
934 return results
936 @staticmethod
937 def add_ssh_opts(
938 opts: list[str],
939 *,
940 ssh_src_user: str | None = None,
941 ssh_dst_user: str | None = None,
942 ssh_src_port: int | None = None,
943 ssh_dst_port: int | None = None,
944 ssh_src_config_file: str | None = None,
945 ssh_dst_config_file: str | None = None,
946 ) -> None:
947 """Appends ssh related options to ``opts`` if specified."""
948 assert isinstance(opts, list)
949 opts += [f"--ssh-src-user={ssh_src_user}"] if ssh_src_user else []
950 opts += [f"--ssh-dst-user={ssh_dst_user}"] if ssh_dst_user else []
951 opts += [f"--ssh-src-port={ssh_src_port}"] if ssh_src_port is not None else []
952 opts += [f"--ssh-dst-port={ssh_dst_port}"] if ssh_dst_port is not None else []
953 opts += [f"--ssh-src-config-file={ssh_src_config_file}"] if ssh_src_config_file else []
954 opts += [f"--ssh-dst-config-file={ssh_dst_config_file}"] if ssh_dst_config_file else []
956 @staticmethod
957 def add_log_file_opts(opts: list[str], tag: str, job_id: str, job_run: str, logsuffix: str) -> None:
958 """Appends standard log-file CLI options to ``opts``."""
959 opts += [f"--log-file-prefix={PROG_NAME}{SEP}{tag}{SEP}"]
960 opts += [f"--log-file-infix={SEP}{job_id}"]
961 opts += [f"--log-file-suffix={SEP}{job_run}{logsuffix}{SEP}"]
963 def run_subjobs(
964 self,
965 subjobs: dict[str, list[str]],
966 max_workers: int,
967 timeout_secs: float | None,
968 work_period_seconds: float,
969 jitter: bool,
970 ) -> None:
971 """Executes subjobs sequentially or in parallel, respecting '~' barriers.
973 Design note on subjob failure, isolation and termination:
974 - On subjob failure the subjob's subtree is skipped.
975 - Subjob failures are converted to return codes (not exceptions) so the policy does not invoke a termination handler
976 on such failures and sibling subjobs continue unaffected. This preserves per-subjob isolation, both within a single
977 process (thread-per-subjob mode; default) as well as in process-per-subjob mode (``--spawn-process-per-job``).
978 - The first failure is retained for diagnostics.
979 - Process exit code precedence is: fatal/non-monitor failure > monitor CRITICAL/WARNING > STILL_RUNNING > success.
980 """
981 self.stats = JobStats(len(subjobs))
982 log = self.log
983 num_intervals = 1 + len(subjobs) if jitter else len(subjobs)
984 interval_nanos = 0 if len(subjobs) == 0 else round(1_000_000_000 * max(0.0, work_period_seconds) / num_intervals)
985 assert interval_nanos >= 0
986 if jitter: # randomize job start time to avoid potential thundering herd problems in large distributed systems
987 sleep_nanos = random.SystemRandom().randint(0, interval_nanos)
988 log.info("Jitter: Delaying job start time by sleeping for %s ...", human_readable_duration(sleep_nanos))
989 self.timing.sleep(sleep_nanos) # allows early wakeup on async termination
990 sorted_subjobs: list[str] = sorted(subjobs.keys())
991 spawn_process_per_job: bool = self.spawn_process_per_job
992 log.log(LOG_TRACE, "%s: %s", "spawn_process_per_job", spawn_process_per_job)
993 if process_datasets_in_parallel_and_fault_tolerant(
994 log=log,
995 datasets=sorted_subjobs,
996 process_dataset=lambda subjob, tid, retry: self.run_subjob(
997 subjobs[subjob], name=subjob, timeout_secs=timeout_secs, spawn_process_per_job=spawn_process_per_job
998 )
999 == 0,
1000 priority=lambda dataset: dataset.count("/"), # start runnable jobs sorted by tree depth aka breadth-first order
1001 skip_tree_on_error=lambda subjob: True,
1002 skip_on_error="dataset",
1003 max_workers=max_workers,
1004 interval_nanos=lambda last_update_nanos, dataset, submit_count: interval_nanos,
1005 timing=self.timing,
1006 termination_handler=self.subprocesses.terminate_process_subtrees,
1007 task_name="Subjob",
1008 dry_run=False,
1009 is_test_mode=self.is_test_mode,
1010 ):
1011 self.first_exception = DIE_STATUS if self.first_exception is None else self.first_exception
1012 self.worst_exception = bzfs.get_worst_exception(self.worst_exception, DIE_STATUS)
1013 stats = self.stats
1014 jobs_skipped = stats.jobs_all - stats.jobs_started
1015 msg = f"{stats}, skipped:" + percent(jobs_skipped, total=stats.jobs_all, print_total=True)
1016 log.info("Final Progress: %s", msg)
1017 assert stats.jobs_running == 0, msg
1018 assert stats.jobs_completed == stats.jobs_started, msg
1019 skipped_jobs_dict = {subjob: subjobs[subjob] for subjob in sorted_subjobs if subjob not in stats.started_job_names}
1020 if len(skipped_jobs_dict) > 0:
1021 log.debug("Skipped subjobs: \n%s", _pretty_print_formatter(skipped_jobs_dict))
1022 assert jobs_skipped == len(skipped_jobs_dict), msg
1024 def run_subjob(
1025 self, cmd: list[str], name: str, timeout_secs: float | None, spawn_process_per_job: bool
1026 ) -> int | None: # thread-safe
1027 """Executes one worker job and updates shared Stats."""
1028 start_time_nanos = time.monotonic_ns()
1029 returncode = None
1030 log = self.log
1031 cmd_str = " ".join(cmd)
1032 stats = self.stats
1033 try:
1034 msg: str = stats.submit_job(name)
1035 log.log(LOG_TRACE, "Starting worker job: %s", cmd_str)
1036 log.info("Progress: %s", msg)
1037 start_time_nanos = time.monotonic_ns()
1038 if spawn_process_per_job:
1039 returncode = self.run_worker_job_spawn_process_per_job(cmd, timeout_secs)
1040 else:
1041 returncode = self.run_worker_job_in_current_thread(cmd, timeout_secs)
1042 except BaseException as e:
1043 log.error("Worker job failed with unexpected exception: %s for command: %s", e, cmd_str)
1044 raise
1045 else:
1046 elapsed_human: str = human_readable_duration(time.monotonic_ns() - start_time_nanos)
1047 if returncode != 0:
1048 with stats.lock:
1049 if self.first_exception is None:
1050 self.first_exception = DIE_STATUS if returncode is None else returncode
1051 self.worst_exception = bzfs.get_worst_exception(self.worst_exception, returncode)
1052 log.error("Worker job failed with exit code %s in %s: %s", returncode, elapsed_human, cmd_str)
1053 else:
1054 log.debug("Worker job succeeded in %s: %s", elapsed_human, cmd_str)
1055 return returncode
1056 finally:
1057 msg = stats.complete_job(failed=returncode != 0, elapsed_nanos=time.monotonic_ns() - start_time_nanos)
1058 log.info("Progress: %s", msg)
1060 def run_worker_job_in_current_thread(self, cmd: list[str], timeout_secs: float | None) -> int | None:
1061 """Runs ``bzfs`` in-process and return its exit code."""
1062 log = self.log
1063 if timeout_secs is not None:
1064 cmd = cmd[0:1] + [f"--timeout={round(1000 * timeout_secs)}milliseconds"] + cmd[1:]
1065 try:
1066 if not self.jobrunner_dryrun:
1067 self._bzfs_run_main(cmd)
1068 return 0
1069 except subprocess.CalledProcessError as e:
1070 return bzfs.normalize_called_process_error(e)
1071 except SystemExit as e:
1072 assert e.code is None or isinstance(e.code, int)
1073 return e.code
1074 except BaseException:
1075 log.exception("Worker job failed with unexpected exception for command: %s", " ".join(cmd))
1076 return DIE_STATUS
1078 def _bzfs_run_main(self, cmd: list[str]) -> None:
1079 """Delegates execution to :mod:`bzfs` using parsed arguments."""
1080 bzfs_job = bzfs.Job(termination_event=self.termination_event)
1081 bzfs_job.is_test_mode = self.is_test_mode
1082 bzfs_job.run_main(bzfs.argument_parser().parse_args(cmd[1:]), cmd)
1084 def run_worker_job_spawn_process_per_job(self, cmd: list[str], timeout_secs: float | None) -> int | None:
1085 """Spawns a subprocess for the worker job and waits for completion."""
1086 log = self.log
1087 if len(cmd) > 0 and cmd[0] == BZFS_PROG_NAME:
1088 cmd = [sys.executable, "-m", "bzfs_main." + cmd[0]] + cmd[1:]
1089 if self.jobrunner_dryrun:
1090 return 0
1092 with self.subprocesses.popen_and_track(cmd, stdin=subprocess.DEVNULL, text=True) as proc:
1093 try:
1094 if self.termination_event.is_set():
1095 timeout_secs = 1.0 if timeout_secs is None else timeout_secs
1096 raise subprocess.TimeoutExpired(cmd, timeout_secs) # do not wait for normal completion
1097 proc.communicate(timeout=timeout_secs) # Wait for the subprocess to complete and exit normally
1098 returncode: int | None = proc.returncode
1099 except subprocess.TimeoutExpired:
1100 returncode = proc.returncode
1101 cmd_str = " ".join(cmd)
1102 if self.termination_event.is_set():
1103 log.error("%s", f"Terminating worker job due to async termination request: {cmd_str}")
1104 else:
1105 log.error("%s", f"Terminating worker job as it failed to complete within {timeout_secs}s: {cmd_str}")
1106 proc.terminate() # Sends SIGTERM signal to job subprocess
1107 assert timeout_secs is not None
1108 timeout_secs = min(1.0, timeout_secs)
1109 try:
1110 proc.communicate(timeout=timeout_secs) # Wait for the subprocess to exit
1111 except subprocess.TimeoutExpired:
1112 log.error("%s", f"Killing worker job as it failed to terminate within {timeout_secs}s: {cmd_str}")
1113 terminate_process_subtree(root_pids=[proc.pid]) # Send SIGTERM to process subtree
1114 proc.kill() # Sends SIGKILL signal to job subprocess because SIGTERM wasn't enough
1115 timeout_secs = min(0.025, timeout_secs)
1116 with contextlib.suppress(subprocess.TimeoutExpired):
1117 proc.communicate(timeout=timeout_secs) # Wait for the subprocess to exit
1118 if proc.returncode and not returncode:
1119 returncode = proc.returncode
1120 return returncode
1122 def validate_src_hosts(self, src_hosts: list[str]) -> list[str]:
1123 """Checks ``src_hosts`` contains valid hostnames."""
1124 context = "--src-hosts"
1125 self.validate_type(src_hosts, list, context)
1126 for src_hostname in src_hosts:
1127 self.validate_host_name(src_hostname, context)
1128 return src_hosts
1130 def validate_dst_hosts(self, dst_hosts: dict[str, list[str]]) -> dict[str, list[str]]:
1131 """Checks destination hosts dictionary."""
1132 context = "--dst-hosts"
1133 self.validate_type(dst_hosts, dict, context)
1134 for dst_hostname, targets in dst_hosts.items():
1135 self.validate_host_name(dst_hostname, context)
1136 self.validate_type(targets, list, f"{context} targets")
1137 for target in targets:
1138 self.validate_type(target, str, f"{context} target")
1139 return dst_hosts
1141 def validate_dst_root_datasets(self, dst_root_datasets: dict[str, str]) -> dict[str, str]:
1142 """Checks that each destination root dataset string is valid."""
1143 context = "--dst-root-datasets"
1144 self.validate_type(dst_root_datasets, dict, context)
1145 for dst_hostname, dst_root_dataset in dst_root_datasets.items():
1146 self.validate_host_name(dst_hostname, context)
1147 self.validate_type(dst_root_dataset, str, f"{context} root dataset")
1148 return dst_root_datasets
1150 def validate_snapshot_plan(
1151 self, snapshot_plan: dict[str, dict[str, dict[str, int]]], context: str
1152 ) -> dict[str, dict[str, dict[str, int]]]:
1153 """Checks snapshot plan structure and value types."""
1154 self.validate_type(snapshot_plan, dict, context)
1155 for org, target_periods in snapshot_plan.items():
1156 self.validate_type(org, str, f"{context} org")
1157 self.validate_type(target_periods, dict, f"{context} target_periods")
1158 for target, periods in target_periods.items():
1159 self.validate_type(target, str, f"{context} org/target")
1160 self.validate_type(periods, dict, f"{context} org/periods")
1161 for period_unit, period_amount in periods.items():
1162 self.validate_non_empty_string(period_unit, f"{context} org/target/period_unit")
1163 self.validate_non_negative_int(period_amount, f"{context} org/target/period_amount")
1164 return snapshot_plan
1166 def validate_monitor_snapshot_plan(
1167 self, monitor_snapshot_plan: dict[str, dict[str, dict[str, dict[str, str | int]]]]
1168 ) -> dict[str, dict[str, dict[str, dict[str, str | int]]]]:
1169 """Checks snapshot monitoring plan configuration."""
1170 context = "--monitor-snapshot-plan"
1171 self.validate_type(monitor_snapshot_plan, dict, context)
1172 for org, target_periods in monitor_snapshot_plan.items():
1173 self.validate_type(org, str, f"{context} org")
1174 self.validate_type(target_periods, dict, f"{context} target_periods")
1175 for target, periods in target_periods.items():
1176 self.validate_type(target, str, f"{context} org/target")
1177 self.validate_type(periods, dict, f"{context} org/periods")
1178 for period_unit, alert_dict in periods.items():
1179 self.validate_non_empty_string(period_unit, f"{context} org/target/period_unit")
1180 self.validate_type(alert_dict, dict, f"{context} org/target/alert_dict")
1181 for key, value in alert_dict.items():
1182 self.validate_non_empty_string(key, f"{context} org/target/alert_dict/key")
1183 self.validate_type(value, Union[str, int], f"{context} org/target/alert_dict/value")
1184 return monitor_snapshot_plan
1186 def validate_is_subset(self, x: Iterable[str], y: Iterable[str], x_name: str, y_name: str) -> None:
1187 """Raises error if ``x`` contains an item not present in ``y``."""
1188 if isinstance(x, str) or not isinstance(x, Iterable):
1189 self.die(f"{x_name} must be an Iterable")
1190 if isinstance(y, str) or not isinstance(y, Iterable):
1191 self.die(f"{y_name} must be an Iterable")
1192 if not set(x).issubset(set(y)):
1193 diff = sorted(set(x).difference(set(y)))
1194 self.die(f"{x_name} must be a subset of {y_name}. diff: {diff}, {x_name}: {sorted(x)}, {y_name}: {sorted(y)}")
1196 def validate_host_name(self, hostname: str, context: str) -> None:
1197 """Checks host name string."""
1198 self.validate_non_empty_string(hostname, f"{context} hostname")
1199 bzfs.validate_host_name(hostname, context)
1201 def validate_non_empty_string(self, value: str, name: str) -> None:
1202 """Checks that ``value`` is a non-empty string."""
1203 self.validate_type(value, str, name)
1204 if not value:
1205 self.die(f"{name} must not be empty!")
1207 def validate_non_negative_int(self, value: int, name: str) -> None:
1208 """Checks ``value`` is an int >= 0."""
1209 self.validate_type(value, int, name)
1210 if value < 0:
1211 self.die(f"{name} must be a non-negative integer: {value}")
1213 def validate_true(self, expr: object, msg: str) -> None:
1214 """Raises error if ``expr`` evaluates to ``False``."""
1215 if not bool(expr):
1216 self.die(msg)
1218 def validate_type(self, value: Any, expected_type: Any, name: str) -> None:
1219 """Checks ``value`` is instance of ``expected_type`` or union thereof."""
1220 if hasattr(expected_type, "__origin__") and expected_type.__origin__ is Union: # for compat with python < 3.10
1221 union_types = expected_type.__args__
1222 for t in union_types:
1223 if isinstance(value, t):
1224 return
1225 type_msg = " or ".join([t.__name__ for t in union_types])
1226 self.die(f"{name} must be of type {type_msg} but got {type(value).__name__}: {value}")
1227 elif not isinstance(value, expected_type):
1228 self.die(f"{name} must be of type {expected_type.__name__} but got {type(value).__name__}: {value}")
1230 def parse_src_hosts_from_cli_or_stdin(self, raw_src_hosts: str | None) -> list:
1231 """Resolve --src-hosts from CLI or stdin with robust TTY/empty handling."""
1232 if raw_src_hosts is None:
1233 # If stdin is an interactive TTY, don't block waiting for input; fail clearly instead
1234 try:
1235 is_tty: bool = getattr(sys.stdin, "isatty", lambda: False)()
1236 except Exception:
1237 is_tty = False
1238 if is_tty:
1239 self.die("Missing --src-hosts and stdin is a TTY. Provide --src-hosts or pipe the list on stdin.")
1240 stdin_text: str = sys.stdin.read()
1241 if not stdin_text.strip(): # avoid literal_eval("") SyntaxError and provide a clear message
1242 self.die("Missing --src-hosts and stdin is empty. Provide --src-hosts or pipe a list on stdin.")
1243 raw_src_hosts = stdin_text
1244 try:
1245 value = literal_eval(raw_src_hosts)
1246 except Exception as e:
1247 self.die(f"Invalid --src-hosts format: {e} for input: {raw_src_hosts}")
1248 if not isinstance(value, list):
1249 example: str = format_obj(["hostname1", "hostname2"])
1250 self.die(f"Invalid --src-hosts: expected a Python list literal, e.g. {example} but got: {format_obj(value)}")
1251 return value
1253 def die(self, msg: str) -> NoReturn:
1254 """Log ``msg`` and exit the program."""
1255 self.log.error("%s", msg)
1256 utils.die(msg)
1258 def get_localhost_ips(self) -> set[str]:
1259 """Returns all network addresses of the local host, i.e. all configured addresses on all network interfaces, without
1260 depending on name resolution."""
1261 ips: set[str] = set()
1262 if platform.system() == "Linux":
1263 try:
1264 proc = subprocess.run(["hostname", "-I"], stdin=DEVNULL, stdout=PIPE, text=True, check=True) # noqa: S607
1265 except Exception as e:
1266 self.log.warning("Cannot run 'hostname -I' on localhost: %s", e)
1267 else:
1268 ips = {ip for ip in proc.stdout.strip().split() if ip}
1269 self.log.log(LOG_TRACE, "localhost_ips: %s", sorted(ips))
1270 return ips
1273#############################################################################
1274@final
1275class RejectArgumentAction(argparse.Action):
1276 """An argparse Action that immediately fails if it is ever triggered."""
1278 def __call__(
1279 self, parser: argparse.ArgumentParser, namespace: argparse.Namespace, values: Any, option_string: str | None = None
1280 ) -> None:
1281 """Abort argument parsing if a protected option is seen."""
1282 parser.error(f"Security: Overriding protected argument '{option_string}' is not allowed.")
1285#############################################################################
1286def _dedupe(root_dataset_pairs: list[tuple[str, str]]) -> list[tuple[str, str]]:
1287 """Returns a list with duplicate dataset pairs removed while preserving order."""
1288 return list(dict.fromkeys(root_dataset_pairs))
1291_T = TypeVar("_T")
1294def _flatten(root_dataset_pairs: Iterable[Iterable[_T]]) -> list[_T]:
1295 """Flattens an iterable of pairs into a single list."""
1296 return [item for pair in root_dataset_pairs for item in pair]
1299def _sanitize(filename: str) -> str:
1300 """Replaces potentially problematic characters in ``filename`` with '!'."""
1301 for s in (" ", "..", "/", "\\", SEP):
1302 filename = filename.replace(s, "!")
1303 return filename
1306def _log_suffix(localhostname: str, src_hostname: str, dst_hostname: str) -> str:
1307 """Returns a log file suffix in a format that contains the given hostnames."""
1308 return f"{SEP}{_sanitize(localhostname)}{SEP}{_sanitize(src_hostname)}{SEP}{_sanitize(dst_hostname)}"
1311def _pretty_print_formatter(dictionary: Mapping[str, object]) -> object:
1312 """Lazy JSON formatter used to avoid overhead in disabled log levels."""
1314 @final
1315 class PrettyPrintFormatter:
1316 """Wrapper returning formatted JSON on ``str`` conversion."""
1318 def __str__(self) -> str:
1319 import json
1321 return json.dumps(dictionary, indent=4, sort_keys=True)
1323 return PrettyPrintFormatter()
1326def _detect_loopback_address() -> str:
1327 """Detects if a loopback connection over IPv4 or IPv6 is possible."""
1328 try:
1329 addr = "127.0.0.1"
1330 with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
1331 s.bind((addr, 0))
1332 return addr
1333 except OSError:
1334 pass
1336 try:
1337 addr = "::1"
1338 with socket.socket(socket.AF_INET6, socket.SOCK_STREAM) as s:
1339 s.bind((addr, 0))
1340 return addr
1341 except OSError:
1342 pass
1344 return ""
1347def convert_ipv6(hostname: str) -> str:
1348 """Supports IPv6 without getting confused by host:dataset colon separator and any colons that may be part of a (valid)
1349 ZFS dataset name."""
1350 return hostname.replace(":", "|") # Also see bzfs.convert_ipv6() for the reverse conversion
1353#############################################################################
1354if __name__ == "__main__": 1354 ↛ 1355line 1354 didn't jump to line 1355 because the condition on line 1354 was never true
1355 main()