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 pruning 

23 jobs across a fleet of multiple source and destination hosts; driven by a fleet-wide job config file 

24 (e.g., `bzfs_job_example.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 JobStats, 

102 Subprocesses, 

103 TaskTiming, 

104 dry, 

105 format_dict, 

106 format_obj, 

107 getenv_bool, 

108 human_readable_duration, 

109 percent, 

110 shuffle_dict, 

111 terminate_process_subtree, 

112 termination_signal_handler, 

113 validate_dataset_name, 

114) 

115from bzfs_main.util.utils import PROG_NAME as BZFS_PROG_NAME 

116 

117# constants: 

118PROG_NAME: Final[str] = "bzfs_jobrunner" 

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

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

121SEP: Final[str] = "," 

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

123 

124 

125def argument_parser() -> argparse.ArgumentParser: 

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

127 # fmt: off 

128 parser = argparse.ArgumentParser( 

129 prog=PROG_NAME, 

130 allow_abbrev=False, 

131 formatter_class=argparse.RawTextHelpFormatter, 

132 description=f""" 

133This program is a convenience wrapper around [bzfs](README.md) that simplifies periodic ZFS snapshot creation, replication, 

134pruning, and monitoring, across a fleet of N source hosts and M destination hosts, using a single fleet-wide shared 

135[jobconfig](bzfs_tests/bzfs_job_example.py) script. 

136For example, this simplifies the deployment of an efficient geo-replicated backup service where each of the M destination 

137hosts is located in a separate geographic region and receives replicas from (the same set of) N source hosts. It also 

138simplifies low latency replication from a primary to a secondary or to M read replicas, or backup to removable drives, etc. 

139 

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

141 

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

143 

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

145 

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

147 

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

149geo-replication factors) 

150 

151You can run this program on a single third-party host and have that talk to all source hosts and destination hosts, which is 

152convenient for basic use cases and for testing. 

153However, typically, a cron job on each source host runs `{PROG_NAME}` periodically to create new snapshots (via 

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

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

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

157to the destination (via --replicate). 

158Yet another cron job on each source and each destination runs `{PROG_NAME}` periodically to alert the user if the latest or 

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

160The frequency of these periodic activities can vary by activity, and is typically every second, minute, hour, day, week, 

161month and/or year (or multiples thereof). 

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 /etc/bzfs/bzfs_job_example.py --src-host="$(hostname)" --create-src-snapshots --prune-src-snapshots --prune-src-bookmarks` 

170 

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

172 

173 

174* crontab on destination hosts: 

175 

176`* * * * * testuser /etc/bzfs/bzfs_job_example.py --dst-host="$(hostname)" --replicate --prune-dst-snapshots` 

177 

178`* * * * * testuser /etc/bzfs/bzfs_job_example.py --dst-host="$(hostname)" --monitor-dst-snapshots` 

179 

180 

181### Applying Actions to a Subset of Hosts 

182 

183`--src-host` and `--dst-host` are ways to tell `{PROG_NAME}` to only perform actions for a specified subset of source hosts 

184and destination hosts, e.g. only replicate from a certain subset of source hosts to a certain subset of destination hosts. 

185Each `{PROG_NAME}` invocation will do all actions that apply to the final effective value of `--src-hosts` and `--dst-hosts`, 

186after the subsetting step has been applied using the `--src-host` and `--dst-host` parameters. 

187 

188 

189### High Frequency Replication (Experimental Feature) 

190 

191Taking snapshots, and/or replicating, from every N milliseconds to every 10 seconds or so is considered high frequency. For 

192such use cases, consider that `zfs list -t snapshot` performance degrades as more and more snapshots currently exist within 

193the selected datasets, so try to keep the number of currently existing snapshots small, and prune them at a frequency that 

194is proportional to the frequency with which snapshots are created. Consider using `--skip-parent` and `--exclude-dataset*` 

195filters to limit the selected datasets only to those that require this level of frequency. 

196 

197In addition, use the `--daemon-*` options to reduce startup overhead, in combination with splitting the crontab entry (or 

198better: high frequency systemd timer) into multiple processes, from a single source host to a single destination host, 

199along these lines: 

200 

201* crontab on source hosts: 

202 

203`* * * * * testuser /etc/bzfs/bzfs_job_example.py --src-host="$(hostname)" --dst-host="foo" --create-src-snapshots` 

204 

205`* * * * * testuser /etc/bzfs/bzfs_job_example.py --src-host="$(hostname)" --dst-host="foo" --prune-src-snapshots` 

206 

207`* * * * * testuser /etc/bzfs/bzfs_job_example.py --src-host="$(hostname)" --dst-host="foo" --prune-src-bookmarks` 

208 

209`* * * * * testuser /etc/bzfs/bzfs_job_example.py --src-host="$(hostname)" --dst-host="foo" --monitor-src-snapshots` 

210 

211 

212* crontab on destination hosts: 

213 

214`* * * * * testuser /etc/bzfs/bzfs_job_example.py --src-host="bar" --dst-host="$(hostname)" --replicate` 

215 

216`* * * * * testuser /etc/bzfs/bzfs_job_example.py --src-host="bar" --dst-host="$(hostname)" --prune-dst-snapshots` 

217 

218`* * * * * testuser /etc/bzfs/bzfs_job_example.py --src-host="bar" --dst-host="$(hostname)" --monitor-dst-snapshots` 

219 

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

221finally exit after, say, 86400 seconds (whatever you specify via `--daemon-lifetime`). The daemons will subsequently be 

222auto-restarted by 'cron', or earlier if they fail. While the daemons are running, 'cron' will attempt to start new 

223(unnecessary) daemons but this is benign as these new processes immediately exit with a message like this: 

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

225""") 

226 

227 # commands: 

228 parser.add_argument( 

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

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

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

232 parser.add_argument( 

233 "--replicate", action="store_true", 

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

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

236 "host; for push mode, on the src host; for pull-push mode on a third-party host.\n\n") 

237 parser.add_argument( 

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

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

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

241 parser.add_argument( 

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

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

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

245 parser.add_argument( 

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

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

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

249 parser.add_argument( 

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

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

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

253 parser.add_argument( 

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

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

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

257 

258 # options: 

259 parser.add_argument( 

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

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

262 parser.add_argument( 

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

264 help="Hostnames of the sources to operate on.\n\n") 

265 parser.add_argument( 

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

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

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

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

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

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

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

273 parser.add_argument( 

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

289 parser.add_argument( 

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

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

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

293 "Example: " 

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

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

296 parser.add_argument( 

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

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

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

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

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

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

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

304 dst_root_datasets_example = { 

305 "nas": "tank2/bak", 

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

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

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

309 "hotspare": "", 

310 } 

311 parser.add_argument( 

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

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

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

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

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

317 "the empty string.\n\n" 

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

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

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

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

322 src_snapshot_plan_example = { 

323 "prod": { 

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

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

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

327 }, 

328 "test": { 

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

330 }, 

331 } 

332 parser.add_argument( 

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

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

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

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

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

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

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

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

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

342 "organization. " 

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

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

345 "The example creates snapshots with names like " 

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

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

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

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

350 parser.add_argument( 

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

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

353 parser.add_argument( 

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

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

356 monitor_snapshot_plan_example = { 

357 "prod": { 

358 "onsite": { 

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

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

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

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

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

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

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

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

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

368 }, 

369 "": { 

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

371 }, 

372 }, 

373 } 

374 parser.add_argument( 

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

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

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

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

379 "pruned on schedule. " 

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

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

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

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

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

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

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

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

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

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

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

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

392 for loc in locations: 

393 parser.add_argument( 

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

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

396 for loc in locations: 

397 parser.add_argument( 

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

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

400 for loc in locations: 

401 parser.add_argument( 

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

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

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

405 parser.add_argument( 

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

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

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

409 parser.add_argument( 

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

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

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

413 workers_default = 100 # percent 

414 parser.add_argument( 

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

416 metavar="INT[%]", 

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

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

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

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

421 parser.add_argument( 

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

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

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

425 parser.add_argument( 

426 "--jitter", action="store_true", 

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

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

429 parser.add_argument( 

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

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

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

433 parser.add_argument( 

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

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

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

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

438 "startup latency.\n\n") 

439 parser.add_argument( 

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

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

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

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

444 parser.add_argument( 

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

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

447 parser.add_argument( 

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

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

450 parser.add_argument( 

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

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

453 parser.add_argument( 

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

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

456 parser.add_argument( 

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

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

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

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

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

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

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

464 "--monitor-snapshots", "--timeout"] 

465 for loc in locations: 

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

467 for bad_opt in bad_opts: 

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

469 parser.add_argument( 

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

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

472 parser.add_argument( 

473 "--help, -h", action="help", # trick to ensure both --help and -h are shown in the help msg 

474 help="Show this help message and exit.\n\n") 

475 parser.add_argument( 

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

477 metavar="SRC_DATASET DST_DATASET", 

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

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

480 return parser 

481 # fmt: on 

482 

483 

484############################################################################# 

485def main() -> None: 

486 """API for command line clients.""" 

487 prev_umask: int = os.umask(UMASK) 

488 try: 

489 set_logging_runtime_defaults() 

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

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

492 with termination_signal_handler(termination_events=[termination_event]): 

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

494 finally: 

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

496 

497 

498############################################################################# 

499@final 

500class Job: 

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

502 

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

504 # immutable variables: 

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

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

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

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

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

510 self.jobrunner_dryrun: bool = False 

511 self.spawn_process_per_job: bool = False 

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

513 

514 # mutable variables: 

515 self.first_exception: int | None = None 

516 self.worst_exception: int | None = None 

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

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

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

520 

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

522 

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

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

525 try: 

526 self._run_main(sys_argv) 

527 finally: 

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

529 reset_logger(self.log) 

530 

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

532 self.first_exception = None 

533 self.worst_exception = None 

534 log: Logger = self.log 

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

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

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

538 log.setLevel(args.jobrunner_log_level) 

539 self.jobrunner_dryrun = args.jobrunner_dryrun 

540 assert len(args.root_dataset_pairs) > 0 

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

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

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

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

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

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

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

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

549 basis_src_hosts: list[str] = src_hosts 

550 nb_src_hosts: int = len(basis_src_hosts) 

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

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

553 assert isinstance(args.src_host, list) 

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

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

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

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

558 nb_dst_hosts: int = len(dst_hosts) 

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

560 assert isinstance(args.dst_host, list) 

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

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

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

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

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

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

567 self.validate_is_subset( 

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

569 ) 

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

571 bad_root_datasets: dict[str, str] = { 

572 dst_host: root_dataset 

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

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

575 } 

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

577 self.die( 

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

579 "destination dataset. " 

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

581 ) 

582 bad_root_datasets = { 

583 dst_host: root_dataset 

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

585 if root_dataset and SRC_MAGIC_SUBSTITUTION_TOKEN not in root_dataset 

586 } 

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

588 self.die( 

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

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

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

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

593 ) 

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

595 random.SystemRandom().shuffle(src_hosts) 

596 dst_hosts = shuffle_dict(dst_hosts) 

597 ssh_src_user: str = args.ssh_src_user 

598 ssh_dst_user: str = args.ssh_dst_user 

599 ssh_src_port: int | None = args.ssh_src_port 

600 ssh_dst_port: int | None = args.ssh_dst_port 

601 ssh_src_config_file: str | None = args.ssh_src_config_file 

602 ssh_dst_config_file: str | None = args.ssh_dst_config_file 

603 job_id: str = _sanitize(args.job_id) 

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

605 workers, workers_is_percent = args.workers 

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

607 worker_timeout_seconds: int = args.worker_timeout_seconds 

608 self.spawn_process_per_job = args.spawn_process_per_job 

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

610 assert username 

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

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

613 loopback_ids.add(localhostname) 

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

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

616 

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

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

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

620 

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

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

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

624 

625 def runpad() -> str: 

626 """Returns standardized subjob count suffix.""" 

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

628 

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

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

631 if j <= 0: 

632 return subjob_name 

633 elif j == 1: 

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

635 else: 

636 return subjob_name + "/" + BARRIER_CHAR 

637 

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

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

640 assert hostname 

641 assert dataset 

642 ssh_user = ssh_src_user if is_src else ssh_dst_user 

643 ssh_user = ssh_user if ssh_user else username 

644 lb: str = self.loopback_address 

645 loopbck_ids: set[str] = loopback_ids 

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

647 hostname = convert_ipv6(hostname) 

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

649 

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

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

652 assert dst_hostname 

653 assert dst_dataset 

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

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

656 root_dataset = root_dataset.replace(SRC_MAGIC_SUBSTITUTION_TOKEN, src_host) 

657 root_dataset = root_dataset.replace(DST_MAGIC_SUBSTITUTION_TOKEN, dst_hostname) 

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

659 validate_dataset_name(resolved_dst_dataset, dst_dataset) 

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

661 

662 for src_host in src_hosts: 

663 assert src_host 

664 for dst_hostname in dst_hosts: 

665 assert dst_hostname 

666 dummy: Final[str] = DUMMY_DATASET 

667 lhn: Final[str] = localhostname 

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

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

670 for i, src_host in enumerate(src_hosts): 

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

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

673 j: int = 0 

674 opts: list[str] 

675 

676 if args.create_src_snapshots: 

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

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

679 self.add_ssh_opts( 

680 opts, ssh_src_user=ssh_src_user, ssh_src_port=ssh_src_port, ssh_src_config_file=ssh_src_config_file 

681 ) 

682 opts += [POSIX_END_OF_OPTIONS_MARKER] 

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

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

685 subjobs[subjob_name] = bzfs_prog_header + opts 

686 

687 if args.replicate: 

688 j = 0 

689 marker: str = "replicate" 

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

691 opts = self.replication_opts( 

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

693 ) 

694 if len(opts) > 0: 

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

696 self.add_ssh_opts( 

697 opts, 

698 ssh_src_user=ssh_src_user, 

699 ssh_dst_user=ssh_dst_user, 

700 ssh_src_port=ssh_src_port, 

701 ssh_dst_port=ssh_dst_port, 

702 ssh_src_config_file=ssh_src_config_file, 

703 ssh_dst_config_file=ssh_dst_config_file, 

704 ) 

705 opts += [POSIX_END_OF_OPTIONS_MARKER] 

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

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

708 for src, dst in args.root_dataset_pairs 

709 ] 

710 dataset_pairs = self.skip_nonexisting_local_dst_pools(dataset_pairs, worker_timeout_seconds) 

711 if len(dataset_pairs) > 0: 

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

713 j += 1 

714 subjob_name = update_subjob_name(marker) 

715 

716 def prune_src( 

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

718 ) -> None: 

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

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

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

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

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

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

725 ) 

726 opts += [POSIX_END_OF_OPTIONS_MARKER] 

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

728 nonlocal subjob_name 

729 subjob_name += f"/{tag}" 

730 subjobs[subjob_name] = bzfs_prog_header + opts 

731 

732 if args.prune_src_snapshots: 

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

734 

735 if args.prune_src_bookmarks: 

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

737 

738 if args.prune_dst_snapshots: 

739 self.validate_true( 

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

741 ) 

742 j = 0 

743 marker = "prune-dst-snapshots" 

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

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

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

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

748 for org, target_periods in dst_snapshot_plan.items() 

749 } 

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

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

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

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

754 self.add_ssh_opts( 

755 opts, ssh_dst_user=ssh_dst_user, ssh_dst_port=ssh_dst_port, ssh_dst_config_file=ssh_dst_config_file 

756 ) 

757 opts += [POSIX_END_OF_OPTIONS_MARKER] 

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

759 dataset_pairs = _dedupe(dataset_pairs) 

760 dataset_pairs = self.skip_nonexisting_local_dst_pools(dataset_pairs, worker_timeout_seconds) 

761 if len(dataset_pairs) > 0: 

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

763 j += 1 

764 subjob_name = update_subjob_name(marker) 

765 

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

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

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

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

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

771 return opts 

772 

773 def build_monitor_plan(monitor_plan: dict, snapshot_plan: dict, cycles_prefix: str) -> dict: