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 

32update_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) 

70 

71import bzfs_main.argparse_actions 

72import bzfs_main.check_range 

73import bzfs_main.utils 

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.parallel_tasktree import ( 

89 BARRIER_CHAR, 

90) 

91from bzfs_main.parallel_tasktree_policy import ( 

92 process_datasets_in_parallel_and_fault_tolerant, 

93) 

94from bzfs_main.utils import ( 

95 DIE_STATUS, 

96 LOG_TRACE, 

97 UMASK, 

98 JobStats, 

99 Subprocesses, 

100 format_dict, 

101 format_obj, 

102 getenv_bool, 

103 human_readable_duration, 

104 percent, 

105 shuffle_dict, 

106 terminate_process_subtree, 

107 termination_signal_handler, 

108) 

109from bzfs_main.utils import PROG_NAME as BZFS_PROG_NAME 

110 

111# constants: 

112PROG_NAME: Final[str] = "bzfs_jobrunner" 

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

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

115SEP: Final[str] = "," 

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

117 

118 

119def argument_parser() -> argparse.ArgumentParser: 

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

121 # fmt: off 

122 parser = argparse.ArgumentParser( 

123 prog=PROG_NAME, 

124 allow_abbrev=False, 

125 formatter_class=argparse.RawTextHelpFormatter, 

126 description=f""" 

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

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

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

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

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

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

133 

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

135 

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

137 

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

139 

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

141 

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

143geo-replication factors) 

144 

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

146convenient for basic use cases and for testing. 

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

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

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

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

151to the destination (via --replicate). 

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

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

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

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

156 

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

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

159lines: 

160 

161* crontab on source hosts: 

162 

163`* * * * * testuser /etc/bzfs/bzfs_job_example.py --src-host="$(hostname)" --create-src-snapshots --prune-src-snapshots --prune-src-bookmarks` 

164 

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

166 

167 

168* crontab on destination hosts: 

169 

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

171 

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

173 

174### High Frequency Replication (Experimental Feature) 

175 

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

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

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

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

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

181 

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

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

184along these lines: 

185 

186* crontab on source hosts: 

187 

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

189 

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

191 

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

193 

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

195 

196 

197* crontab on destination hosts: 

198 

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

200 

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

202 

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

204 

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

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

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

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

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

210""") 

211 

212 # commands: 

213 parser.add_argument( 

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

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

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

217 parser.add_argument( # `choices` are deprecated; use --replicate without argument instead 

218 "--replicate", choices=["pull", "push"], default=None, const="pull", nargs="?", metavar="", 

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

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

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

222 parser.add_argument( 

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

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

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

226 parser.add_argument( 

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

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

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

230 parser.add_argument( 

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

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

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

234 parser.add_argument( 

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

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

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

238 parser.add_argument( 

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

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

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

242 

243 # options: 

244 parser.add_argument( 

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

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

247 parser.add_argument( 

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

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

250 parser.add_argument( 

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

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

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

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

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

256 parser.add_argument( 

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

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

259 "(the infix portion of snapshot name). " 

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

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

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

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

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

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

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

267 parser.add_argument( 

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

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

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

271 parser.add_argument( 

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

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

274 "(the infix portion of snapshot name). " 

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

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

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

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

279 dst_root_datasets_example = { 

280 "nas": "tank2/bak", 

281 "bak-us-west-1": "backups/bak001", 

282 "bak-eu-west-1": "backups/bak999", 

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

284 "hotspare": "", 

285 } 

286 parser.add_argument( 

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

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

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

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

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

292 "the empty string.\n\n" 

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

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

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

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

297 src_snapshot_plan_example = { 

298 "prod": { 

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

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

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

302 }, 

303 "test": { 

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

305 }, 

306 } 

307 parser.add_argument( 

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

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

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

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

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

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

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

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

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

317 "organization. " 

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

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

320 "The example creates snapshots with names like " 

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

322 "`prod_us-west-1_<timestamp>_hourly`, `prod_us-west-1_<timestamp>_daily`, " 

323 "`prod_eu-west-1_<timestamp>_hourly`, `prod_eu-west-1_<timestamp>_daily`, " 

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

325 parser.add_argument( 

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

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

328 parser.add_argument( 

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

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

331 monitor_snapshot_plan_example = { 

332 "prod": { 

333 "onsite": { 

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

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

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

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

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

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

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

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

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

343 }, 

344 "": { 

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

346 }, 

347 }, 

348 } 

349 parser.add_argument( 

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

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

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

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

354 "pruned on schedule. " 

355 "Process exit code is 0, 1, 2 on OK, WARNING, CRITICAL, respectively. " 

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

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

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

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

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

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

362 "or `dst_snapshot_plan`, respectively. " 

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

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

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

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

367 for loc in locations: 

368 parser.add_argument( 

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

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

371 for loc in locations: 

372 parser.add_argument( 

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

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

375 for loc in locations: 

376 parser.add_argument( 

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

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

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

380 parser.add_argument( 

381 "--src-user", default="", metavar="STRING", 

382 help=argparse.SUPPRESS) # deprecated; was renamed to --ssh-src-user 

383 parser.add_argument( 

384 "--dst-user", default="", metavar="STRING", 

385 help=argparse.SUPPRESS) # deprecated; was renamed to --ssh-dst-user 

386 parser.add_argument( 

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

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

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

390 parser.add_argument( 

391 "--jobid", default=None, action=bzfs_main.argparse_actions.NonEmptyStringAction, metavar="STRING", 

392 help=argparse.SUPPRESS) # deprecated; was renamed to --job-run 

393 parser.add_argument( 

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

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

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

397 workers_default = 100 # percent 

398 parser.add_argument( 

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

400 metavar="INT[%]", 

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

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

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

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

405 parser.add_argument( 

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

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

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

409 parser.add_argument( 

410 "--jitter", action="store_true", 

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

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

413 parser.add_argument( 

414 "--worker-timeout-seconds", type=float, min=0.001, default=None, action=bzfs_main.check_range.CheckRange, 

415 metavar="FLOAT", 

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

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

418 parser.add_argument( 

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

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

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

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

423 "startup latency.\n\n") 

424 parser.add_argument( 

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

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

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

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

429 parser.add_argument( 

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

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

432 parser.add_argument( 

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

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

435 parser.add_argument( 

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

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

438 parser.add_argument( 

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

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

441 parser.add_argument( 

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

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

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

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

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

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

448 "--monitor-snapshots", "--timeout"] 

449 for loc in locations: 

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

451 for bad_opt in bad_opts: 

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

453 parser.add_argument( 

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

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

456 parser.add_argument( 

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

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

459 parser.add_argument( 

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

461 metavar="SRC_DATASET DST_DATASET", 

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

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

464 return parser 

465 # fmt: on 

466 

467 

468############################################################################# 

469def main() -> None: 

470 """API for command line clients.""" 

471 prev_umask: int = os.umask(UMASK) 

472 try: 

473 set_logging_runtime_defaults() 

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

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

476 with termination_signal_handler(termination_event=termination_event): 

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

478 finally: 

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

480 

481 

482############################################################################# 

483class Job: 

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

485 

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

487 # immutable variables: 

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

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

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

491 self.subprocesses: Final[Subprocesses] = Subprocesses() 

492 self.jobrunner_dryrun: bool = False 

493 self.spawn_process_per_job: bool = False 

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

495 

496 # mutable variables: 

497 self.first_exception: int | None = None 

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

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

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

501 

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

503 

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

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

506 try: 

507 self._run_main(sys_argv) 

508 finally: 

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

510 reset_logger(self.log) 

511 

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

513 self.first_exception = None 

514 log: Logger = self.log 

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

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

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

518 log.setLevel(args.jobrunner_log_level) 

519 self.jobrunner_dryrun = args.jobrunner_dryrun 

520 assert len(args.root_dataset_pairs) > 0 

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

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

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

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

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

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

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

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

529 basis_src_hosts: list[str] = src_hosts 

530 nb_src_hosts: int = len(basis_src_hosts) 

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

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

533 assert isinstance(args.src_host, list) 

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

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

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

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

538 nb_dst_hosts: int = len(dst_hosts) 

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

540 assert isinstance(args.dst_host, list) 

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

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

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

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

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

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

547 self.validate_is_subset( 

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

549 ) 

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

551 bad_root_datasets: dict[str, str] = { 

552 dst_host: root_dataset 

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

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

555 } 

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

557 self.die( 

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

559 "destination dataset. " 

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

561 ) 

562 bad_root_datasets = { 

563 dst_host: root_dataset 

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

565 if root_dataset and SRC_MAGIC_SUBSTITUTION_TOKEN not in root_dataset 

566 } 

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

568 self.die( 

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

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

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

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

573 ) 

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

575 random.SystemRandom().shuffle(src_hosts) 

576 dst_hosts = shuffle_dict(dst_hosts) 

577 ssh_src_user: str = args.ssh_src_user or args.src_user # --src-user is deprecated 

578 ssh_dst_user: str = args.ssh_dst_user or args.dst_user # --dst-user is deprecated 

579 ssh_src_port: int | None = args.ssh_src_port 

580 ssh_dst_port: int | None = args.ssh_dst_port 

581 ssh_src_config_file: str | None = args.ssh_src_config_file 

582 ssh_dst_config_file: str | None = args.ssh_dst_config_file 

583 job_id: str = _sanitize(args.job_id) 

584 job_run: str = args.job_run if args.job_run is not None else args.jobid # --jobid deprecat; was renamed to --job-run 

585 job_run = _sanitize(job_run) if job_run else uuid.uuid1().hex 

586 workers, workers_is_percent = args.workers 

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

588 worker_timeout_seconds: int = args.worker_timeout_seconds 

589 self.spawn_process_per_job = args.spawn_process_per_job 

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

591 assert username 

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

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

594 loopback_ids.add(localhostname) 

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

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

597 

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

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

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

601 

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

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

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

605 

606 def runpad() -> str: 

607 """Returns standardized subjob count suffix.""" 

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

609 

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

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

612 if j <= 0: 

613 return subjob_name 

614 elif j == 1: 

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

616 else: 

617 return subjob_name + "/" + BARRIER_CHAR 

618 

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

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

621 assert hostname 

622 assert dataset 

623 ssh_user = ssh_src_user if is_src else ssh_dst_user 

624 ssh_user = ssh_user if ssh_user else username 

625 lb: str = self.loopback_address 

626 loopbck_ids: set[str] = loopback_ids 

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

628 hostname = convert_ipv6(hostname) 

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

630 

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

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

633 assert dst_hostname 

634 assert dst_dataset 

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

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

637 root_dataset = root_dataset.replace(SRC_MAGIC_SUBSTITUTION_TOKEN, src_host) 

638 root_dataset = root_dataset.replace(DST_MAGIC_SUBSTITUTION_TOKEN, dst_hostname) 

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

640 bzfs_main.utils.validate_dataset_name(resolved_dst_dataset, dst_dataset) 

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

642 

643 for src_host in src_hosts: 

644 assert src_host 

645 for dst_hostname in dst_hosts: 

646 assert dst_hostname 

647 dummy: Final[str] = DUMMY_DATASET 

648 lhn: Final[str] = localhostname 

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

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

651 for i, src_host in enumerate(src_hosts): 

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

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

654 j: int = 0 

655 opts: list[str] 

656 

657 if args.create_src_snapshots: 

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

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

660 self.add_ssh_opts( 

661 opts, ssh_src_user=ssh_src_user, ssh_src_port=ssh_src_port, ssh_src_config_file=ssh_src_config_file 

662 ) 

663 opts += [POSIX_END_OF_OPTIONS_MARKER] 

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

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

666 subjobs[subjob_name] = bzfs_prog_header + opts 

667 

668 if args.replicate: 

669 j = 0 

670 marker: str = "replicate" 

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

672 opts = self.replication_opts( 

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

674 ) 

675 if len(opts) > 0: 

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

677 self.add_ssh_opts( 

678 opts, 

679 ssh_src_user=ssh_src_user, 

680 ssh_dst_user=ssh_dst_user, 

681 ssh_src_port=ssh_src_port, 

682 ssh_dst_port=ssh_dst_port, 

683 ssh_src_config_file=ssh_src_config_file, 

684 ssh_dst_config_file=ssh_dst_config_file, 

685 ) 

686 opts += [POSIX_END_OF_OPTIONS_MARKER] 

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

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

689 for src, dst in args.root_dataset_pairs 

690 ] 

691 dataset_pairs = self.skip_nonexisting_local_dst_pools(dataset_pairs, worker_timeout_seconds) 

692 if len(dataset_pairs) > 0: 

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

694 j += 1 

695 subjob_name = update_subjob_name(marker) 

696 

697 def prune_src( 

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

699 ) -> None: 

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

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

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

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

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

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

706 ) 

707 opts += [POSIX_END_OF_OPTIONS_MARKER] 

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

709 nonlocal subjob_name 

710 subjob_name += f"/{tag}" 

711 subjobs[subjob_name] = bzfs_prog_header + opts 

712 

713 if args.prune_src_snapshots: 

714 prune_src(["--delete-dst-snapshots"], src_snapshot_plan, "prune-src-snapshots") 

715 

716 if args.prune_src_bookmarks: 

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

718 

719 if args.prune_dst_snapshots: 

720 self.validate_true( 

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

722 ) 

723 j = 0 

724 marker = "prune-dst-snapshots" 

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

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

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

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

729 for org, target_periods in dst_snapshot_plan.items() 

730 } 

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

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

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

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

735 self.add_ssh_opts( 

736 opts, ssh_dst_user=ssh_dst_user, ssh_dst_port=ssh_dst_port, ssh_dst_config_file=ssh_dst_config_file 

737 ) 

738 opts += [POSIX_END_OF_OPTIONS_MARKER] 

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

740 dataset_pairs = _dedupe(dataset_pairs) 

741 dataset_pairs = self.skip_nonexisting_local_dst_pools(dataset_pairs, worker_timeout_seconds) 

742 if len(dataset_pairs) > 0: 

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

744 j += 1 

745 subjob_name = update_subjob_name(marker) 

746 

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

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

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

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

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

752 return opts 

753 

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

755 """Expands ``monitor_plan`` with cycle defaults from ``snapshot_plan``.""" 

756 

757 def alert_dicts(alertdict: dict, cycles: int) -> dict: 

758 """Returns alert dictionaries with explicit ``cycles`` value.""" 

759 latest_dict = alertdict.copy() 

760 for prefix in ("src_snapshot_", "dst_snapshot_", ""):