Coverage for bzfs_main/bzfs_jobrunner.py: 99%

1# Copyright 2024 Wolfgang Hoschek AT mac DOT com 

2# 

3# Licensed under the Apache License, Version 2.0 (the "License"); 

4# you may not use this file except in compliance with the License. 

5# You may obtain a copy of the License at 

6# 

7# http://www.apache.org/licenses/LICENSE-2.0 

8# 

9# Unless required by applicable law or agreed to in writing, software 

10# distributed under the License is distributed on an "AS IS" BASIS, 

11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

12# See the License for the specific language governing permissions and 

13# limitations under the License. 

14 

15# 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""" 

34 

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) 

56from logging import ( 

57 Logger, 

58) 

59from subprocess import ( 

60 DEVNULL, 

61 PIPE, 

62) 

63from typing import ( 

64 Any, 

65 Final, 

66 NoReturn, 

67 TypeVar, 

68 Union, 

69 final, 

70) 

71 

72import bzfs_main.argparse_actions 

73from bzfs_main import ( 

74 bzfs, 

75) 

76from bzfs_main.argparse_cli import ( 

77 PROG_AUTHOR, 

78) 

79from bzfs_main.detect import ( 

80 DUMMY_DATASET, 

81) 

82from bzfs_main.loggers import ( 

83 get_simple_logger, 

84 reset_logger, 

85 set_logging_runtime_defaults, 

86) 

87from bzfs_main.util import ( 

88 check_range, 

89 utils, 

90) 

91from bzfs_main.util.parallel_tasktree import ( 

92 BARRIER_CHAR, 

93) 

94from bzfs_main.util.parallel_tasktree_policy import ( 

95 process_datasets_in_parallel_and_fault_tolerant, 

96) 

97from bzfs_main.util.utils import ( 

98 DIE_STATUS, 

99 LOG_TRACE, 

100 UMASK, 

101 UNIX_TIME_INFINITY_SECS, 

102 JobStats, 

103 Subprocesses, 

104 TaskTiming, 

105 dry, 

106 format_dict, 

107 format_obj, 

108 getenv_bool, 

109 human_readable_duration, 

110 percent, 

111 shuffle_dict, 

112 terminate_process_subtree, 

113 termination_signal_handler, 

114 validate_dataset_name, 

115) 

116from bzfs_main.util.utils import PROG_NAME as BZFS_PROG_NAME 

117 

118# constants: 

119PROG_NAME: Final[str] = "bzfs_jobrunner" 

120SRC_MAGIC_SUBSTITUTION_TOKEN: Final[str] = "^SRC_HOST" # noqa: S105 

121DST_MAGIC_SUBSTITUTION_TOKEN: Final[str] = "^DST_HOST" # noqa: S105 

122SEP: Final[str] = "," 

123POSIX_END_OF_OPTIONS_MARKER: Final[str] = "--" # args following -- are treated as operands, even if they begin with a hyphen 

124 

125 

126def argument_parser() -> argparse.ArgumentParser: 

127 """Returns the CLI parser used by bzfs_jobrunner.""" 

128 # fmt: off 

129 parser = argparse.ArgumentParser( 

130 prog=PROG_NAME, 

131 allow_abbrev=False, 

132 formatter_class=argparse.RawTextHelpFormatter, 

133 description=f""" 

134This companion program wraps [bzfs](README.md) for periodic snapshot creation, replication, pruning, and monitoring across 

135N source hosts and M destination hosts, using one shared fleet-wide [jobconfig](bzfs_testbed/bzfs_job_testbed.py) script. 

136 

137Typical use cases include geo-replicated backup where each destination host is in a different region and receives replicas 

138from the same set of source hosts, low-latency replication from a primary to a secondary or to M read replicas, and backups 

139to removable drives. 

140 

141This program can be used to efficiently replicate ... 

142 

143a) within a single machine (local mode), or 

144 

145b) from a single source host to one or more destination hosts (pull or push or pull-push mode), or 

146 

147c) from multiple source hosts to a single destination host (pull or push or pull-push mode), or 

148 

149d) 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 

150geo-replication factors) 

151 

152You can run this program on a single third-party host and have it talk to all source and destination hosts. That setup is 

153convenient for basic use cases and testing, and efficient with `--r2r=pull` or `--r2r=push`. 

154In many deployments, a cron job on each source host runs `{PROG_NAME}` periodically to create new snapshots (via 

155--create-src-snapshots) and prune outdated snapshots and bookmarks on the source (via --prune-src-snapshots and 

156--prune-src-bookmarks), whereas another cron job on each destination host runs `{PROG_NAME}` periodically to prune 

157outdated destination snapshots (via --prune-dst-snapshots), and to replicate the recently created snapshots from the source 

158to the destination (via --replicate). 

159A separate cron job on each source host and each destination host runs `{PROG_NAME}` periodically to alert the user if the 

160latest or oldest snapshot is somehow too old (via --monitor-src-snapshots and --monitor-dst-snapshots). 

161The frequency of each activity can differ. Typical intervals range from N milliseconds to years. 

162 

163Edit the jobconfig script in a central place (e.g. versioned in a git repo), then copy the (very same) shared file onto all 

164source hosts and all destination hosts, and add crontab entries (or systemd timers or Monit entries or similar), along these 

165lines: 

166 

167* crontab on source hosts: 

168 

169`* * * * * testuser /bzfs/bzfs_testbed/bzfs_job_testbed.py --src-host="$(hostname)" --create-src-snapshots --prune-src-snapshots --prune-src-bookmarks` 

170 

171`* * * * * testuser /bzfs/bzfs_testbed/bzfs_job_testbed.py --src-host="$(hostname)" --monitor-src-snapshots` 

172 

173* crontab on destination hosts: 

174 

175`* * * * * testuser /bzfs/bzfs_testbed/bzfs_job_testbed.py --dst-host="$(hostname)" --replicate --prune-dst-snapshots` 

176 

177`* * * * * testuser /bzfs/bzfs_testbed/bzfs_job_testbed.py --dst-host="$(hostname)" --monitor-dst-snapshots` 

178 

179Some deployments choose to move monitoring to one centralized management host, like so: 

180 

181`* * * * * testuser /bzfs/bzfs_testbed/bzfs_job_testbed.py --monitor-src-snapshots --monitor-dst-snapshots` 

182 

183### Applying Actions to a Subset of Hosts 

184 

185`--src-host` and `--dst-host` let you run actions on only a subset of source and destination hosts. For example, you can 

186replicate from selected source hosts to selected destination hosts. 

187Each `{PROG_NAME}` invocation runs all enabled actions for the final effective values of `--src-hosts` and `--dst-hosts`, 

188after applying the `--src-host` and `--dst-host` filters. 

189 

190 

191### High Frequency Replication (Experimental Feature) 

192 

193Taking snapshots and/or replicating every N milliseconds to roughly every 10 seconds is considered high frequency. Consider 

194that `zfs list -t snapshot` performance degrades as snapshot counts grow within the selected datasets. Keep the active 

195snapshot count small, and prune at a cadence that matches your snapshot creation rate. Consider using `--skip-parent` and 

196`--exclude-dataset*` filters so only datasets that need this frequency are selected. 

197 

198To reduce startup overhead, forward `--daemon-lifetime` to `bzfs`, use the `--daemon-*` options, and split the 

199crontab entry (or, preferably, a high-frequency systemd timer) into multiple processes, from one source host to one 

200destination host, along these lines: 

201 

202* crontab on source hosts: 

203 

204`* * * * * testuser /bzfs/bzfs_testbed/bzfs_job_testbed.py --src-host="$(hostname)" --dst-host="foo" --create-src-snapshots` 

205 

206`* * * * * testuser /bzfs/bzfs_testbed/bzfs_job_testbed.py --src-host="$(hostname)" --dst-host="foo" --prune-src-snapshots` 

207 

208`* * * * * testuser /bzfs/bzfs_testbed/bzfs_job_testbed.py --src-host="$(hostname)" --dst-host="foo" --prune-src-bookmarks` 

209 

210`* * * * * testuser /bzfs/bzfs_testbed/bzfs_job_testbed.py --src-host="$(hostname)" --dst-host="foo" --monitor-src-snapshots` 

211 

212 

213* crontab on destination hosts: 

214 

215`* * * * * testuser /bzfs/bzfs_testbed/bzfs_job_testbed.py --src-host="bar" --dst-host="$(hostname)" --replicate` 

216 

217`* * * * * testuser /bzfs/bzfs_testbed/bzfs_job_testbed.py --src-host="bar" --dst-host="$(hostname)" --prune-dst-snapshots` 

218 

219`* * * * * testuser /bzfs/bzfs_testbed/bzfs_job_testbed.py --src-host="bar" --dst-host="$(hostname)" --monitor-dst-snapshots` 

220 

221The daemon processes work like non-daemon processes except that they loop, handle time events, and sleep between events. 

222They exit after the interval specified by `--daemon-lifetime` (for example, 86400 seconds). They are then restarted by 

223`cron`, or earlier if they fail. While an existing daemon is still running, `cron` may attempt to start another one. 

224This is harmless because that extra process exits immediately with a message like this: 

225"Exiting as same previous periodic job is still running without completion yet" 

226""") 

227 

228 # commands: 

229 parser.add_argument( 

230 "--create-src-snapshots", action="store_true", 

231 help="Take snapshots on the selected source hosts as necessary. Typically, this command should be called by a " 

232 "program (or cron job) running on each src host.\n\n") 

233 parser.add_argument( 

234 "--replicate", action="store_true", 

235 help="Replicate snapshots from the selected source hosts to the selected destinations hosts as necessary. For pull " 

236 "mode (recommended), this command should be called by a program (or cron job) running on each dst " 

237 "host; for push mode, on the src host; for pull-push mode on a third-party host, maybe with `--r2r=pull` " 

238 "or `--r2r=push`.\n\n") 

239 parser.add_argument( 

240 "--prune-src-snapshots", action="store_true", 

241 help="Prune snapshots on the selected source hosts as necessary. Typically, this command should be called by a " 

242 "program (or cron job) running on each src host.\n\n") 

243 parser.add_argument( 

244 "--prune-src-bookmarks", action="store_true", 

245 help="Prune bookmarks on the selected source hosts as necessary. Typically, this command should be called by a " 

246 "program (or cron job) running on each src host.\n\n") 

247 parser.add_argument( 

248 "--prune-dst-snapshots", action="store_true", 

249 help="Prune snapshots on the selected destination hosts as necessary. Typically, this command should be called by a " 

250 "program (or cron job) running on each dst host.\n\n") 

251 parser.add_argument( 

252 "--monitor-src-snapshots", action="store_true", 

253 help="Alert the user if snapshots on the selected source hosts are too old, using --monitor-snapshot-plan (see " 

254 "below). Typically, this command should be called by a program (or cron job) running on each src host.\n\n") 

255 parser.add_argument( 

256 "--monitor-dst-snapshots", action="store_true", 

257 help="Alert the user if snapshots on the selected destination hosts are too old, using --monitor-snapshot-plan (see " 

258 "below). Typically, this command should be called by a program (or cron job) running on each dst host.\n\n") 

259 

260 # options: 

261 parser.add_argument( 

262 "--localhost", default=None, action=bzfs_main.argparse_actions.NonEmptyStringAction, metavar="STRING", 

263 help="Hostname of localhost. Default is the hostname without the domain name, querying the Operating System.\n\n") 

264 parser.add_argument( 

265 "--src-hosts", default=None, metavar="LIST_STRING", 

266 help="Hostnames of the sources to operate on. Specify a Python list literal such as " 

267 "`\"['src1', 'src2']\"`. If omitted, reads the same list literal from stdin.\n\n") 

268 parser.add_argument( 

269 "--src-host", default=None, action="append", metavar="STRING", 

270 help="For subsetting --src-hosts; Can be specified multiple times; Indicates to only use the --src-hosts that are " 

271 "contained in the specified --src-host values (optional).\n\n" 

272 "Example: `--src-host=src1 --src-host=src2 --src-hosts=\"['src1', 'src2', 'src3', 'src4']\"` indicates to " 

273 "effectively only use `--src-hosts=\"['src1', 'src2']\"`.\n\n") 

274 dst_hosts_example = {"nas": ["onsite"], "bak-us-west": ["us-west"], 

275 "bak-eu-west": ["eu-west"], "archive": ["offsite"]} 

276 parser.add_argument( 

277 "--dst-hosts", default="{}", metavar="DICT_STRING", 

278 help="Dictionary that maps each destination hostname to a list of zero or more logical replication target names " 

279 "(the infix portion of snapshot name). As hostname use the real output of the `hostname` CLI. " 

280 "The target is an arbitrary user-defined name that serves as an abstraction of the destination hostnames for " 

281 "a group of snapshots, like target 'onsite', 'offsite', 'hotspare', a geographically independent datacenter like " 

282 "'us-west', or similar. Rather than the snapshot name embedding (i.e. hardcoding) a list of destination " 

283 "hostnames where it should be sent to, the snapshot name embeds the user-defined target name, which is later " 

284 "mapped by this jobconfig to a list of destination hostnames.\n\n" 

285 f"Example: `{format_dict(dst_hosts_example)}`.\n\n" 

286 "With this, given a snapshot name, we can find the destination hostnames to which the snapshot shall be " 

287 "replicated. Also, given a snapshot name and its own name, a destination host can determine if it shall " 

288 "replicate the given snapshot from the source host, or if the snapshot is intended for another destination " 

289 "host, in which case it skips the snapshot. A destination host will receive replicas of snapshots for all " 

290 "targets that map to that destination host.\n\n" 

291 "Removing a mapping can be used to temporarily suspend replication to a given destination host.\n\n") 

292 parser.add_argument( 

293 "--dst-host", default=None, action="append", metavar="STRING", 

294 help="For subsetting --dst-hosts; Can be specified multiple times; Indicates to only use the --dst-hosts keys that " 

295 "are contained in the specified --dst-host values (optional).\n\n" 

296 "Example: " 

297 "`--dst-host=dst1 --dst-host=dst2 --dst-hosts=\"{'dst1': ..., 'dst2': ..., 'dst3': ..., 'dst4': ...}\"` " 

298 "indicates to effectively only use `--dst-hosts=\"{'dst1': ..., 'dst2': ...}\"`.\n\n") 

299 parser.add_argument( 

300 "--retain-dst-targets", default="{}", metavar="DICT_STRING", 

301 help="Dictionary that maps each destination hostname to a list of zero or more logical replication target names " 

302 "(the infix portion of snapshot name).\n\n" 

303 f"Example: `{format_dict(dst_hosts_example)}`. Has same format as --dst-hosts.\n\n" 

304 "As part of --prune-dst-snapshots, a destination host will delete any snapshot it has stored whose target has " 

305 "no mapping to that destination host in this dictionary. Do not remove a mapping here unless you are sure it's " 

306 "ok to delete all those snapshots on that destination host! If in doubt, use --dryrun mode first.\n\n") 

307 dst_root_datasets_example = { 

308 "nas": "tank2/bak", 

309 "bak-us-west": "backups/bak001", 

310 "bak-eu-west": "backups/bak999", 

311 "archive": f"archives/zoo/{SRC_MAGIC_SUBSTITUTION_TOKEN}", 

312 "hotspare": "", 

313 } 

314 parser.add_argument( 

315 "--dst-root-datasets", default="{}", metavar="DICT_STRING", 

316 help="Dictionary that maps each destination hostname to a root dataset located on that destination host. The root " 

317 "dataset name is an (optional) prefix that will be prepended to each dataset that is replicated to that " 

318 "destination host. For backup use cases, this is the backup ZFS pool or a ZFS dataset path within that pool, " 

319 "whereas for cloning, master slave replication, or replication from a primary to a secondary, this can also be " 

320 "the empty string.\n\n" 

321 f"`{SRC_MAGIC_SUBSTITUTION_TOKEN}` and `{DST_MAGIC_SUBSTITUTION_TOKEN}` are optional magic substitution tokens " 

322 "that will be auto-replaced at runtime with the actual hostname. This can be used to force the use of a " 

323 "separate destination root dataset per source host or per destination host.\n\n" 

324 f"Example: `{format_dict(dst_root_datasets_example)}`\n\n") 

325 src_snapshot_plan_example = { 

326 "prod": { 

327 "onsite": {"secondly": 40, "minutely": 40, "hourly": 36, "daily": 31, "weekly": 12, "monthly": 18, "yearly": 5}, 

328 "us-west": {"secondly": 0, "minutely": 0, "hourly": 36, "daily": 31, "weekly": 12, "monthly": 18, "yearly": 5}, 

329 "eu-west": {"secondly": 0, "minutely": 0, "hourly": 36, "daily": 31, "weekly": 12, "monthly": 18, "yearly": 5}, 

330 }, 

331 "test": { 

332 "offsite": {"12hourly": 42, "weekly": 12}, 

333 }, 

334 } 

335 parser.add_argument( 

336 "--src-snapshot-plan", default="{}", metavar="DICT_STRING", 

337 help="Retention periods for snapshots to be used if pruning src, and when creating new snapshots on src. " 

338 "Snapshots that do not match a retention period will be deleted. A zero or missing retention period indicates " 

339 "that no snapshots shall be retained (or even be created) for the given period.\n\n" 

340 f"Example: `{format_dict(src_snapshot_plan_example)}`. This example will, for the organization 'prod' and " 

341 "the intended logical target 'onsite', create and then retain secondly snapshots that were created less " 

342 "than 40 seconds ago, yet retain the latest 40 secondly snapshots regardless of creation time. Analog for " 

343 "the latest 40 minutely snapshots, 36 hourly snapshots, etc. " 

344 "It will also create and retain snapshots for the targets 'us-west' and 'eu-west' within the 'prod' " 

345 "organization. " 

346 "In addition, it will create and retain snapshots every 12 hours and every week for the 'test' organization, " 

347 "and name them as being intended for the 'offsite' replication target. " 

348 "The example creates snapshots with names like " 

349 "`prod_onsite_<timestamp>_secondly`, `prod_onsite_<timestamp>_minutely`, " 

350 "`prod_us-west_<timestamp>_hourly`, `prod_us-west_<timestamp>_daily`, " 

351 "`prod_eu-west_<timestamp>_hourly`, `prod_eu-west_<timestamp>_daily`, " 

352 "`test_offsite_<timestamp>_12hourly`, `test_offsite_<timestamp>_weekly`, and so on.\n\n") 

353 parser.add_argument( 

354 "--src-bookmark-plan", default="{}", metavar="DICT_STRING", 

355 help="Retention periods for bookmarks to be used if pruning src. Has same format as --src-snapshot-plan.\n\n") 

356 parser.add_argument( 

357 "--dst-snapshot-plan", default="{}", metavar="DICT_STRING", 

358 help="Retention periods for snapshots to be used if pruning dst. Has same format as --src-snapshot-plan.\n\n") 

359 monitor_snapshot_plan_example = { 

360 "prod": { 

361 "onsite": { 

362 "100millisecondly": {"warning": "650 milliseconds", "critical": "2 seconds"}, 

363 "secondly": {"warning": "2 seconds", "critical": "14 seconds"}, 

364 "minutely": {"warning": "30 seconds", "critical": "300 seconds"}, 

365 "hourly": {"warning": "30 minutes", "critical": "300 minutes"}, 

366 "daily": {"warning": "4 hours", "critical": "8 hours"}, 

367 "weekly": {"warning": "2 days", "critical": "8 days"}, 

368 "monthly": {"warning": "2 days", "critical": "8 days"}, 

369 "yearly": {"warning": "5 days", "critical": "14 days"}, 

370 "10minutely": {"warning": "0 minutes", "critical": "0 minutes"}, 

371 }, 

372 "": { 

373 "daily": {"warning": "4 hours", "critical": "8 hours"}, 

374 }, 

375 }, 

376 } 

377 parser.add_argument( 

378 "--monitor-snapshot-plan", default="{}", metavar="DICT_STRING", 

379 help="Alert the user if the ZFS 'creation' time property of the latest or oldest snapshot for any specified " 

380 "snapshot pattern within the selected datasets is too old wrt. the specified age limit. The purpose is to " 

381 "check if snapshots are successfully taken on schedule, successfully replicated on schedule, and successfully " 

382 "pruned on schedule. " 

383 "Process exit code is 0, 1, 2 on OK, WARNING, CRITICAL, respectively.\n\n" 

384 f"Example DICT_STRING: `{format_dict(monitor_snapshot_plan_example)}`. " 

385 "This example alerts the user if the *latest* src or dst snapshot named `prod_onsite_<timestamp>_hourly` is " 

386 "more than 30 minutes late (i.e. more than 30+60=90 minutes old) [warning] or more than 300 minutes late (i.e. " 

387 "more than 300+60=360 minutes old) [critical]. In addition, the example alerts the user if the *oldest* src or " 

388 "dst snapshot named `prod_onsite_<timestamp>_hourly` is more than 30 + 60x36 minutes old [warning] or more " 

389 "than 300 + 60x36 minutes old [critical], where 36 is the number of period cycles specified in " 

390 "`src_snapshot_plan` or `dst_snapshot_plan`, respectively. " 

391 "Analog for the latest snapshot named `prod_<timestamp>_daily`, and so on.\n\n" 

392 "Note: A duration that is missing or zero (e.g. '0 minutes') indicates that no snapshots shall be checked for " 

393 "the given snapshot name pattern.\n\n") 

394 locations = ["src", "dst"] 

395 for loc in locations: 

396 parser.add_argument( 

397 f"--ssh-{loc}-user", default="", metavar="STRING", 

398 help=f"Remote SSH username on {loc} hosts to connect to (optional). Examples: 'root', 'alice'.\n\n") 

399 for loc in locations: 

400 parser.add_argument( 

401 f"--ssh-{loc}-port", type=int, min=1, max=65535, action=check_range.CheckRange, metavar="INT", 

402 help=f"Remote SSH port on {loc} host to connect to (optional).\n\n") 

403 for loc in locations: 

404 parser.add_argument( 

405 f"--ssh-{loc}-config-file", type=str, action=bzfs_main.argparse_actions.SSHConfigFileNameAction, metavar="FILE", 

406 help=f"Path to SSH ssh_config(5) file to connect to {loc} (optional); will be passed into ssh -F CLI. " 

407 "The basename must contain the substring 'bzfs_ssh_config'.\n\n") 

408 parser.add_argument( 

409 "--job-id", required=True, action=bzfs_main.argparse_actions.NonEmptyStringAction, metavar="STRING", 

410 help="The identifier that remains constant across all runs of this particular job; will be included in the log file " 

411 "name infix. Example: mytestjob\n\n") 

412 parser.add_argument( 

413 "--job-run", default="", action=bzfs_main.argparse_actions.NonEmptyStringAction, metavar="STRING", 

414 help="The identifier of this particular run of the overall job; will be included in the log file name suffix. " 

415 "Default is a hex UUID. Example: 0badc0f003a011f0a94aef02ac16083c\n\n") 

416 workers_default = 100 # percent 

417 parser.add_argument( 

418 "--workers", min=1, default=(workers_default, True), action=bzfs_main.argparse_actions.CheckPercentRange, 

419 metavar="INT[%]", 

420 help="The maximum number of jobs to run in parallel at any time; can be given as a positive integer, " 

421 f"optionally followed by the %% percent character (min: %(min)s, default: {workers_default}%%). Percentages " 

422 "are relative to the number of CPU cores on the machine. Example: 200%% uses twice as many parallel jobs as " 

423 "there are cores on the machine; 75%% uses num_procs = num_cores * 0.75. Examples: 1, 4, 75%%, 150%%\n\n") 

424 parser.add_argument( 

425 "--work-period-seconds", type=float, min=0, default=0, action=check_range.CheckRange, metavar="FLOAT", 

426 help="Reduces bandwidth spikes by spreading out the start of worker jobs over this much time; " 

427 "0 disables this feature (default: %(default)s). Examples: 0, 60, 86400\n\n") 

428 parser.add_argument( 

429 "--jitter", action="store_true", 

430 help="Randomize job start time and host order to avoid potential thundering herd problems in large distributed " 

431 "systems (optional). Randomizing job start time is only relevant if --work-period-seconds > 0.\n\n") 

432 parser.add_argument( 

433 "--worker-timeout-seconds", type=float, min=0.001, default=None, action=check_range.CheckRange, metavar="FLOAT", 

434 help="If this much time has passed after a worker process has started executing, kill the straggling worker " 

435 "(optional). Other workers remain unaffected. Examples: 60, 3600\n\n") 

436 parser.add_argument( 

437 "--repeat-if-took-more-than-seconds", type=float, min=0.001, default=UNIX_TIME_INFINITY_SECS, 

438 action=check_range.CheckRange, metavar="FLOAT", 

439 help="Repeat the entire workflow if it took longer than this much time and was successful. Use this (with the POSIX " 

440 "`timeout` CLI) before migrating VM storage to converge replication and reduce cutover downtime. Default is " 

441 "infinity, i.e. never repeat the workflow. Examples: 1, 0.1\n\n") 

442 parser.add_argument( 

443 "--spawn-process-per-job", action="store_true", 

444 help="Spawn a Python process per subjob instead of a Python thread per subjob (optional). The former is only " 

445 "recommended for a job operating in parallel on a large number of hosts as it helps avoid exceeding " 

446 "per-process limits such as the default max number of open file descriptors, at the expense of increased " 

447 "startup latency.\n\n") 

448 parser.add_argument( 

449 "--jobrunner-dryrun", action="store_true", 

450 help="Do a dry run (aka 'no-op') to print what operations would happen if the command were to be executed " 

451 "for real (optional). This option treats both the ZFS source and destination as read-only. Can also be used to " 

452 "check if the configuration options are valid.\n\n") 

453 parser.add_argument( 

454 "--jobrunner-log-level", choices=["CRITICAL", "ERROR", "WARN", "INFO", "DEBUG", "TRACE"], default="INFO", 

455 help="Only emit jobrunner messages with equal or higher priority than this log level. Default is '%(default)s'.\n\n") 

456 parser.add_argument( 

457 "--daemon-replication-frequency", default="minutely", metavar="STRING", 

458 help="Specifies how often the bzfs daemon shall replicate from src to dst if --daemon-lifetime is nonzero.\n\n") 

459 parser.add_argument( 

460 "--daemon-prune-src-frequency", default="minutely", metavar="STRING", 

461 help="Specifies how often the bzfs daemon shall prune src if --daemon-lifetime is nonzero.\n\n") 

462 parser.add_argument( 

463 "--daemon-prune-dst-frequency", default="minutely", metavar="STRING", 

464 help="Specifies how often the bzfs daemon shall prune dst if --daemon-lifetime is nonzero.\n\n") 

465 parser.add_argument( 

466 "--daemon-monitor-snapshots-frequency", default="minutely", metavar="STRING", 

467 help="Specifies how often the bzfs daemon shall monitor snapshot age if --daemon-lifetime is nonzero.\n\n") 

468 bad_opts = ["--daemon-frequency", "--include-snapshot-plan", "--create-src-snapshots-plan", "--skip-replication", 

469 "--include-snapshot-regex", "--exclude-snapshot-regex", "--include-snapshot-times-and-ranks", 

470 "--log-file-prefix", "--log-file-infix", "--log-file-suffix", 

471 "--delete-dst-datasets", "--delete-dst-snapshots", "--delete-dst-snapshots-except", 

472 "--delete-dst-snapshots-except-plan", "--delete-empty-dst-datasets", 

473 "--monitor-snapshots", "--timeout"] 

474 for loc in locations: 

475 bad_opts += [f"--ssh-{loc}-host"] # reject this arg as jobrunner will auto-generate it 

476 for bad_opt in bad_opts: 

477 parser.add_argument(bad_opt, action=RejectArgumentAction, nargs=0, help=argparse.SUPPRESS) 

478 parser.add_argument( 

479 "--version", action="version", version=f"{PROG_NAME}-{bzfs_main.argparse_cli.__version__}, by {PROG_AUTHOR}", 

480 help="Display version information and exit.\n\n") 

481 parser.add_argument( 

482 "--root-dataset-pairs", required=True, nargs="+", action=bzfs_main.argparse_actions.DatasetPairsAction, 

483 metavar="SRC_DATASET DST_DATASET", 

484 help="Source and destination dataset pairs (excluding usernames and excluding hostnames, which will all be " 

485 "auto-appended later).\n\n") 

486 return parser 

487 # fmt: on 

488 

489 

490############################################################################# 

491def main() -> None: 

492 """API for command line clients.""" 

493 prev_umask: int = os.umask(UMASK) 

494 try: 

495 set_logging_runtime_defaults() 

496 # On CTRL-C and SIGTERM, send signal to all descendant processes to terminate them 

497 termination_event: threading.Event = threading.Event() 

498 with termination_signal_handler(termination_events=[termination_event]): 

499 Job(log=None, termination_event=termination_event).run_main(sys_argv=sys.argv) 

500 finally: 

501 os.umask(prev_umask) # restore prior global state 

502 

503 

504############################################################################# 

505@final 

506class Job: 

507 """Coordinates subjobs per the CLI flags; Each subjob handles one host pair and may run in its own process or thread.""" 

508 

509 def __init__(self, log: Logger | None, termination_event: threading.Event) -> None: 

510 # immutable variables: 

511 self.log_was_None: Final[bool] = log is None 

512 self.log: Final[Logger] = get_simple_logger(PROG_NAME) if log is None else log 

513 self.termination_event: Final[threading.Event] = termination_event 

514 self.timing: Final[TaskTiming] = TaskTiming.make_from(self.termination_event) 

515 self.subprocesses: Final[Subprocesses] = Subprocesses(termination_event.is_set) 

516 self.jobrunner_dryrun: bool = False 

517 self.spawn_process_per_job: bool = False 

518 self.loopback_address: Final[str] = _detect_loopback_address() 

519 

520 # mutable variables: 

521 self.first_exception: int | None = None 

522 self.worst_exception: int | None = None 

523 self.stats: JobStats = JobStats(jobs_all=0) 

524 self.cache_existing_dst_pools: set[str] = set() 

525 self.cache_known_dst_pools: set[str] = set() 

526 

527 self.is_test_mode: bool = False # for testing only 

528 

529 def run_main(self, sys_argv: list[str]) -> None: 

530 """API for Python clients; visible for testing; may become a public API eventually.""" 

531 try: 

532 self._run_main(sys_argv) 

533 finally: 

534 if self.log_was_None: # reset Logger unless it's a Logger outside of our control 534 ↛ exitline 534 didn't return from function 'run_main' because the condition on line 534 was always true

535 reset_logger(self.log) 

536 

537 def _run_main(self, sys_argv: list[str]) -> None: 

538 self.first_exception = None 

539 self.worst_exception = None 

540 log: Logger = self.log 

541 log.info("CLI arguments: %s", " ".join(sys_argv)) 

542 nsp = argparse.Namespace(no_argument_file=True) # disable --root-dataset-pairs='+file' option in DatasetPairsAction 

543 args, unknown_args = argument_parser().parse_known_args(sys_argv[1:], nsp) # forward all unknown args to `bzfs` 

544 log.setLevel(args.jobrunner_log_level) 

545 self.jobrunner_dryrun = args.jobrunner_dryrun 

546 assert len(args.root_dataset_pairs) > 0 

547 src_snapshot_plan: dict = self.validate_snapshot_plan(literal_eval(args.src_snapshot_plan), "--src-snapshot-plan") 

548 src_bookmark_plan: dict = self.validate_snapshot_plan(literal_eval(args.src_bookmark_plan), "--src-bookmark-plan") 

549 dst_snapshot_plan: dict = self.validate_snapshot_plan(literal_eval(args.dst_snapshot_plan), "--dst-snapshot-plan") 

550 monitor_snapshot_plan: dict = self.validate_monitor_snapshot_plan(literal_eval(args.monitor_snapshot_plan)) 

551 localhostname: str = args.localhost if args.localhost else socket.gethostname() 

552 self.validate_host_name(localhostname, "--localhost") 

553 log.debug("localhostname: %s", localhostname) 

554 src_hosts: list[str] = self.validate_src_hosts(self.parse_src_hosts_from_cli_or_stdin(args.src_hosts)) 

555 basis_src_hosts: list[str] = src_hosts 

556 nb_src_hosts: int = len(basis_src_hosts) 

557 log.debug("src_hosts before subsetting: %s", src_hosts) 

558 if args.src_host is not None: # retain only the src hosts that are also contained in args.src_host 

559 assert isinstance(args.src_host, list) 

560 retain_src_hosts: set[str] = set(args.src_host) 

561 self.validate_is_subset(retain_src_hosts, src_hosts, "--src-host", "--src-hosts") 

562 src_hosts = [host for host in src_hosts if host in retain_src_hosts] 

563 dst_hosts: dict[str, list[str]] = self.validate_dst_hosts(literal_eval(args.dst_hosts)) 

564 nb_dst_hosts: int = len(dst_hosts) 

565 if args.dst_host is not None: # retain only the dst hosts that are also contained in args.dst_host 

566 assert isinstance(args.dst_host, list) 

567 retain_dst_hosts: set[str] = set(args.dst_host) 

568 self.validate_is_subset(retain_dst_hosts, dst_hosts.keys(), "--dst-host", "--dst-hosts.keys") 

569 dst_hosts = {dst_host: lst for dst_host, lst in dst_hosts.items() if dst_host in retain_dst_hosts} 

570 retain_dst_targets: dict[str, list[str]] = self.validate_dst_hosts(literal_eval(args.retain_dst_targets)) 

571 self.validate_is_subset(dst_hosts.keys(), retain_dst_targets.keys(), "--dst-hosts.keys", "--retain-dst-targets.keys") 

572 dst_root_datasets: dict[str, str] = self.validate_dst_root_datasets(literal_eval(args.dst_root_datasets)) 

573 self.validate_is_subset( 

574 dst_root_datasets.keys(), retain_dst_targets.keys(), "--dst-root-dataset.keys", "--retain-dst-targets.keys" 

575 ) 

576 self.validate_is_subset(dst_hosts.keys(), dst_root_datasets.keys(), "--dst-hosts.keys", "--dst-root-dataset.keys") 

577 bad_root_datasets: dict[str, str] = { 

578 dst_host: root_dataset 

579 for dst_host in sorted(dst_hosts.keys()) 

580 if SRC_MAGIC_SUBSTITUTION_TOKEN not in (root_dataset := dst_root_datasets[dst_host]) 

581 } 

582 if len(src_hosts) > 1 and len(bad_root_datasets) > 0: 

583 self.die( 

584 "Cowardly refusing to proceed as multiple source hosts must not be configured to write to the same " 

585 "destination dataset. " 

586 f"Problematic subset of --dst-root-datasets: {bad_root_datasets} for src_hosts: {sorted(src_hosts)}" 

587 ) 

588 bad_root_datasets = { 

589 dst_host: root_dataset 

590 for dst_host, root_dataset in sorted(dst_root_datasets.items()) 

591 if root_dataset and SRC_MAGIC_SUBSTITUTION_TOKEN not in root_dataset 

592 } 

593 if len(basis_src_hosts) > 1 and len(bad_root_datasets) > 0: 

594 self.die( 

595 "Cowardly refusing to proceed as multiple source hosts are defined in the configuration, but " 

596 f"not all non-empty root datasets in --dst-root-datasets contain the '{SRC_MAGIC_SUBSTITUTION_TOKEN}' " 

597 "substitution token to prevent collisions on writing destination datasets. " 

598 f"Problematic subset of --dst-root-datasets: {bad_root_datasets} for src_hosts: {sorted(basis_src_hosts)}" 

599 ) 

600 if args.jitter: # randomize host order to avoid potential thundering herd problems in large distributed systems 

601 random.SystemRandom().shuffle(src_hosts) 

602 dst_hosts = shuffle_dict(dst_hosts) 

603 ssh_src_user: str = args.ssh_src_user 

604 ssh_dst_user: str = args.ssh_dst_user 

605 ssh_src_port: int | None = args.ssh_src_port 

606 ssh_dst_port: int | None = args.ssh_dst_port 

607 ssh_src_config_file: str | None = args.ssh_src_config_file 

608 ssh_dst_config_file: str | None = args.ssh_dst_config_file 

609 job_id: str = _sanitize(args.job_id) 

610 job_run: str = _sanitize(args.job_run) if args.job_run else uuid.uuid1().hex 

611 workers, workers_is_percent = args.workers 

612 max_workers: int = max(1, round((os.cpu_count() or 1) * workers / 100.0) if workers_is_percent else round(workers)) 

613 worker_timeout_seconds: int = args.worker_timeout_seconds 

614 repeat_if_took_more_than_nanos: int = int(args.repeat_if_took_more_than_seconds * 1_000_000_000) 

615 self.spawn_process_per_job = args.spawn_process_per_job 

616 username: str = pwd.getpwuid(os.getuid()).pw_name 

617 assert username 

618 loopback_ids: set[str] = {"localhost", "127.0.0.1", "::1", socket.gethostname()} # ::1 is IPv6 loopback address 

619 loopback_ids.update(self.get_localhost_ips()) # union 

620 loopback_ids.add(localhostname) 

621 loopback_ids = set() if getenv_bool("disable_loopback", False) else loopback_ids 

622 log.log(LOG_TRACE, "loopback_ids: %s", sorted(loopback_ids)) 

623 

624 def zero_pad(number: int, width: int = 6) -> str: 

625 """Pads number with leading '0' chars to the given width.""" 

626 return f"{number:0{width}d}" 

627 

628 def jpad(jj: int, tag: str) -> str: 

629 """Returns ``tag`` prefixed with slash and zero padded index.""" 

630 return "/" + zero_pad(jj) + tag 

631 

632 def runpad() -> str: 

633 """Returns standardized subjob count suffix.""" 

634 return job_run + SEP + zero_pad(len(subjobs)) 

635 

636 def update_subjob_name(tag: str) -> str: 

637 """Derives next subjob name based on ``tag`` and index ``j``.""" 

638 if j <= 0: 

639 return subjob_name 

640 elif j == 1: 

641 return subjob_name + jpad(j - 1, tag) 

642 else: 

643 return subjob_name + "/" + BARRIER_CHAR 

644 

645 def resolve_dataset(hostname: str, dataset: str, is_src: bool = True) -> str: 

646 """Returns host:dataset string resolving IPv6 and localhost cases.""" 

647 assert hostname 

648 assert dataset 

649 ssh_user = ssh_src_user if is_src else ssh_dst_user 

650 ssh_user = ssh_user if ssh_user else username 

651 lb: str = self.loopback_address 

652 loopbck_ids: set[str] = loopback_ids 

653 hostname = hostname if hostname not in loopbck_ids else (lb if lb else hostname) if username != ssh_user else "-" 

654 hostname = convert_ipv6(hostname) 

655 return f"{hostname}:{dataset}" 

656 

657 def resolve_dst_dataset(dst_hostname: str, dst_dataset: str) -> str: 

658 """Expands ``dst_dataset`` relative to ``dst_hostname`` roots.""" 

659 assert dst_hostname 

660 assert dst_dataset 

661 root_dataset: str | None = dst_root_datasets.get(dst_hostname) 

662 assert root_dataset is not None, dst_hostname # f"Hostname '{dst_hostname}' missing in --dst-root-datasets" 

663 root_dataset = root_dataset.replace(SRC_MAGIC_SUBSTITUTION_TOKEN, src_host) 

664 root_dataset = root_dataset.replace(DST_MAGIC_SUBSTITUTION_TOKEN, dst_hostname) 

665 resolved_dst_dataset: str = f"{root_dataset}/{dst_dataset}" if root_dataset else dst_dataset 

666 validate_dataset_name(resolved_dst_dataset, dst_dataset) 

667 return resolve_dataset(dst_hostname, resolved_dst_dataset, is_src=False) 

668 

669 for src_host in src_hosts: 

670 assert src_host 

671 for dst_hostname in dst_hosts: 

672 assert dst_hostname 

673 dummy: Final[str] = DUMMY_DATASET 

674 lhn: Final[str] = localhostname 

675 bzfs_prog_header: Final[list[str]] = [BZFS_PROG_NAME, "--no-argument-file"] + unknown_args 

676 subjobs: dict[str, list[str]] = {} 

677 for i, src_host in enumerate(src_hosts): 

678 subjob_name: str = zero_pad(i) + "src-host" 

679 src_log_suffix: str = _log_suffix(localhostname, src_host, "") 

680 j: int = 0 

681 opts: list[str] 

682 

683 if args.create_src_snapshots: 

684 opts = ["--create-src-snapshots", f"--create-src-snapshots-plan={src_snapshot_plan}", "--skip-replication"] 

685 self.add_log_file_opts(opts, "create-src-snapshots", job_id, runpad(), src_log_suffix) 

686 self.add_ssh_opts( 

687 opts, ssh_src_user=ssh_src_user, ssh_src_port=ssh_src_port, ssh_src_config_file=ssh_src_config_file 

688 ) 

689 opts += [POSIX_END_OF_OPTIONS_MARKER] 

690 opts += _flatten(_dedupe([(resolve_dataset(src_host, src), dummy) for src, dst in args.root_dataset_pairs])) 

691 subjob_name += "/create-src-snapshots" 

692 subjobs[subjob_name] = bzfs_prog_header + opts 

693 

694 if args.replicate: 

695 j = 0 

696 marker: str = "replicate" 

697 for dst_hostname, targets in dst_hosts.items(): 

698 opts = self.replication_opts( 

699 dst_snapshot_plan, set(targets), lhn, src_host, dst_hostname, marker, job_id, runpad() 

700 ) 

701 if len(opts) > 0: 

702 opts += [f"--daemon-frequency={args.daemon_replication_frequency}"] 

703 self.add_ssh_opts( 

704 opts, 

705 ssh_src_user=ssh_src_user, 

706 ssh_dst_user=ssh_dst_user, 

707 ssh_src_port=ssh_src_port, 

708 ssh_dst_port=ssh_dst_port, 

709 ssh_src_config_file=ssh_src_config_file, 

710 ssh_dst_config_file=ssh_dst_config_file, 

711 ) 

712 opts += [POSIX_END_OF_OPTIONS_MARKER] 

713 dataset_pairs: list[tuple[str, str]] = [ 

714 (resolve_dataset(src_host, src), resolve_dst_dataset(dst_hostname, dst)) 

715 for src, dst in args.root_dataset_pairs 

716 ] 

717 dataset_pairs = self.skip_nonexisting_local_dst_pools(dataset_pairs, worker_timeout_seconds) 

718 if len(dataset_pairs) > 0: 

719 subjobs[subjob_name + jpad(j, marker)] = bzfs_prog_header + opts + _flatten(dataset_pairs) 

720 j += 1 

721 subjob_name = update_subjob_name(marker) 

722 

723 def prune_src( 

724 opts: list[str], retention_plan: dict, tag: str, src_host: str = src_host, logsuffix: str = src_log_suffix 

725 ) -> None: 

726 """Creates prune subjob options for ``tag`` using ``retention_plan``.""" 

727 opts += ["--skip-replication", f"--delete-dst-snapshots-except-plan={retention_plan}"] 

728 opts += [f"--daemon-frequency={args.daemon_prune_src_frequency}"] 

729 self.add_log_file_opts(opts, tag, job_id, runpad(), logsuffix) 

730 self.add_ssh_opts( # i.e. dst=src, src=dummy 

731 opts, ssh_dst_user=ssh_src_user, ssh_dst_port=ssh_src_port, ssh_dst_config_file=ssh_src_config_file 

732 ) 

733 opts += [POSIX_END_OF_OPTIONS_MARKER] 

734 opts += _flatten(_dedupe([(dummy, resolve_dataset(src_host, src)) for src, dst in args.root_dataset_pairs])) 

735 nonlocal subjob_name 

736 subjob_name += f"/{tag}" 

737 subjobs[subjob_name] = bzfs_prog_header + opts 

738 

739 if args.prune_src_snapshots: 

740 prune_src(["--delete-dst-snapshots"], src_snapshot_plan, tag="prune-src-snapshots") 

741 

742 if args.prune_src_bookmarks: 

743 prune_src(["--delete-dst-snapshots=bookmarks"], src_bookmark_plan, tag="prune-src-bookmarks") 

744 

745 if args.prune_dst_snapshots: 

746 self.validate_true( 

747 retain_dst_targets, "--retain-dst-targets must not be empty. Cowardly refusing to delete all snapshots!" 

748 ) 

749 j = 0 

750 marker = "prune-dst-snapshots" 

751 for dst_hostname, _ in dst_hosts.items(): 

752 curr_retain_targets: set[str] = set(retain_dst_targets[dst_hostname]) 

753 curr_dst_snapshot_plan = { # only retain targets that belong to the host 

754 org: {target: periods for target, periods in target_periods.items() if target in curr_retain_targets} 

755 for org, target_periods in dst_snapshot_plan.items() 

756 } 

757 opts = ["--delete-dst-snapshots", "--skip-replication"] 

758 opts += [f"--delete-dst-snapshots-except-plan={curr_dst_snapshot_plan}"] 

759 opts += [f"--daemon-frequency={args.daemon_prune_dst_frequency}"] 

760 self.add_log_file_opts(opts, marker, job_id, runpad(), _log_suffix(lhn, src_host, dst_hostname)) 

761 self.add_ssh_opts( 

762 opts, ssh_dst_user=ssh_dst_user, ssh_dst_port=ssh_dst_port, ssh_dst_config_file=ssh_dst_config_file 

763 ) 

764 opts += [POSIX_END_OF_OPTIONS_MARKER] 

765 dataset_pairs = [(dummy, resolve_dst_dataset(dst_hostname, dst)) for src, dst in args.root_dataset_pairs] 

766 dataset_pairs = _dedupe(dataset_pairs) 

767 dataset_pairs = self.skip_nonexisting_local_dst_pools(dataset_pairs, worker_timeout_seconds) 

768 if len(dataset_pairs) > 0: 

769 subjobs[subjob_name + jpad(j, marker)] = bzfs_prog_header + opts + _flatten(dataset_pairs) 

770 j += 1 

771 subjob_name = update_subjob_name(marker) 

772 

773 def monitor_snapshots_opts(tag: str, monitor_plan: dict, logsuffix: str) -> list[str]: 

774 """Returns monitor subjob options for ``tag`` and ``monitor_plan``.""" 

775 opts = [f"--monitor-snapshots={monitor_plan}", "--skip-replication"] 

776 opts += [f"--daemon-frequency={args.daemon_monitor_snapshots_frequency}"]