Coverage for bzfs_main/detect.py: 98%
228 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-19 18:44 +0000
1# Copyright 2024 Wolfgang Hoschek AT mac DOT com
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7# http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
14#
15"""Detection of ZFS features and system capabilities on local and remote hosts."""
17from __future__ import (
18 annotations,
19)
20import re
21import subprocess
22import sys
23import threading
24import time
25from dataclasses import (
26 dataclass,
27 field,
28)
29from subprocess import (
30 DEVNULL,
31 PIPE,
32)
33from typing import (
34 TYPE_CHECKING,
35 Final,
36 final,
37)
39from bzfs_main.util.connection import (
40 DEDICATED,
41 SHARED,
42 ConnectionPools,
43)
44from bzfs_main.util.utils import (
45 LOG_TRACE,
46 PROG_NAME,
47 SynchronousExecutor,
48 die,
49 drain,
50 list_formatter,
51 stderr_to_str,
52 xfinally,
53 xprint,
54)
56if TYPE_CHECKING: # pragma: no cover - for type hints only
57 from bzfs_main.bzfs import (
58 Job,
59 )
60 from bzfs_main.configuration import (
61 Params,
62 Remote,
63 )
65# constants:
66DISABLE_PRG: Final[str] = "-"
67DUMMY_DATASET: Final[str] = "dummy"
68POOL_GUID: Final[str] = "guid"
69ZFS_VERSION_IS_AT_LEAST_2_1_0: Final[str] = "zfs>=2.1.0"
70ZFS_VERSION_IS_AT_LEAST_2_2_0: Final[str] = "zfs>=2.2.0"
73#############################################################################
74@dataclass(frozen=True)
75@final
76class RemoteConfCacheItem:
77 """Caches detected programs, zpool features and connection pools, per remote."""
79 connection_pools: ConnectionPools
80 available_programs: dict[str, str]
81 zpool_features: dict[str, dict[str, str]]
82 timestamp_nanos: int = field(default_factory=time.monotonic_ns)
85def detect_available_programs(job: Job) -> None:
86 """Detects programs, zpool features and connection pools for local and remote hosts."""
87 p = params = job.params
88 log = p.log
89 available_programs: dict[str, dict[str, str]] = params.available_programs
90 if "local" not in available_programs:
91 cmd: list[str] = [p.shell_program_local, "-c", _find_available_programs(p)]
92 sp = job.subprocesses
93 proc = sp.subprocess_run(cmd, stdin=DEVNULL, stdout=PIPE, stderr=PIPE, text=True, log=log)
94 xprint(log=log, value=stderr_to_str(proc.stderr), file=sys.stderr, end="")
95 stdout: str = proc.stdout
96 available_programs["local"] = dict.fromkeys(stdout.splitlines(), "")
97 cmd = [p.shell_program_local, "-c", "exit"]
98 proc = sp.subprocess_run(cmd, stdin=DEVNULL, stdout=PIPE, stderr=PIPE, text=True, log=log)
99 xprint(log=log, value=stderr_to_str(proc.stderr), file=sys.stderr, end="")
100 if proc.returncode != 0:
101 _disable_program(p, "sh", ["local"])
103 todo: list[Remote] = []
104 for r in [p.dst, p.src]:
105 loc: str = r.location
106 remote_conf_cache_key: tuple = r.cache_key()
107 cache_item: RemoteConfCacheItem | None = job.remote_conf_cache.get(remote_conf_cache_key)
108 if cache_item is not None:
109 # startup perf: cache avoids ssh connect setup and feature detection roundtrips on revisits to same site
110 p.connection_pools[loc] = cache_item.connection_pools
111 p.available_programs[loc] = cache_item.available_programs
112 p.zpool_features[loc] = cache_item.zpool_features
113 if time.monotonic_ns() - cache_item.timestamp_nanos < p.remote_conf_cache_ttl_nanos:
114 if r.pool in cache_item.zpool_features:
115 continue # cache hit, skip remote detection
116 else:
117 p.zpool_features[loc] = {} # cache miss, invalidate features of zpools before refetching from remote
118 else:
119 p.connection_pools[loc] = ConnectionPools(
120 remote=r, capacities={SHARED: r.max_concurrent_ssh_sessions_per_tcp_connection, DEDICATED: 1}
121 )
122 todo.append(r)
124 lock: threading.Lock = threading.Lock()
126 def run_detect(r: Remote) -> None: # thread-safe
127 loc: str = r.location
128 remote_conf_cache_key: tuple = r.cache_key()
129 connection_pools: ConnectionPools = p.connection_pools[loc]
131 def _cleanup_uncached_pools() -> None:
132 if remote_conf_cache_key not in job.remote_conf_cache:
133 connection_pools.shutdown()
135 with xfinally(_cleanup_uncached_pools):
136 available_programs: dict[str, str] = _detect_available_programs_remote(job, r, r.ssh_user_host)
137 zpool_features: dict[str, str] = _detect_zpool_features(job, r, available_programs)
138 with lock:
139 r.params.available_programs[loc] = available_programs
140 r.params.zpool_features[loc][r.pool] = zpool_features
141 job.remote_conf_cache[remote_conf_cache_key] = RemoteConfCacheItem(
142 connection_pools, available_programs, r.params.zpool_features[loc]
143 )
144 if r.use_zfs_delegation and zpool_features.get("delegation") == "off":
145 die(
146 f"Permission denied as ZFS delegation is disabled for {r.location} "
147 f"dataset: {r.basis_root_dataset}. Manually enable it via 'sudo zpool set delegation=on {r.pool}'"
148 )
150 with SynchronousExecutor.executor_for(max_workers=max(1, len(todo))) as executor:
151 drain(executor.map(run_detect, todo)) # detect ZFS features + system capabilities on src+dst in parallel
153 locations = ["src", "dst", "local"]
154 if params.compression_program == DISABLE_PRG:
155 _disable_program(p, "zstd", locations)
156 if params.mbuffer_program == DISABLE_PRG:
157 _disable_program(p, "mbuffer", locations)
158 if params.ps_program == DISABLE_PRG:
159 _disable_program(p, "ps", locations)
160 if params.pv_program == DISABLE_PRG:
161 _disable_program(p, "pv", locations)
162 if params.shell_program == DISABLE_PRG:
163 _disable_program(p, "sh", locations)
164 if params.sudo_program == DISABLE_PRG:
165 _disable_program(p, "sudo", locations)
166 if params.zpool_program == DISABLE_PRG:
167 _disable_program(p, "zpool", locations)
169 for key, programs in available_programs.items():
170 for program in list(programs.keys()):
171 if program.startswith("uname-"):
172 # uname-Linux foo 5.15.0-69-generic #76-Ubuntu SMP Fri Mar 17 17:19:29 UTC 2023 x86_64 x86_64 x86_64 GNU/Linux
173 # uname-FreeBSD freebsd 14.1-RELEASE FreeBSD 14.1-RELEASE releng/14.1-n267679-10e31f0946d8 GENERIC amd64
174 # uname-Darwin foo 23.6.0 Darwin Kernel Version 23.6.0: Mon Jul 29 21:13:04 PDT 2024; root:xnu-10063.141.2~1/RELEASE_ARM64_T6020 arm64
175 programs.pop(program)
176 uname: str = program[len("uname-") :]
177 programs["uname"] = uname
178 log.log(LOG_TRACE, f"available_programs[{key}][uname]: %s", uname)
179 programs["os"] = uname.split(" ", maxsplit=1)[0] # Linux|FreeBSD|Darwin
180 log.log(LOG_TRACE, f"available_programs[{key}][os]: %s", programs["os"])
181 elif program.startswith("default_shell-"):
182 programs.pop(program)
183 default_shell: str = program[len("default_shell-") :]
184 programs["default_shell"] = default_shell
185 log.log(LOG_TRACE, f"available_programs[{key}][default_shell]: %s", default_shell)
186 ssh_user_host = p.src.ssh_user_host if key == "src" else p.dst.ssh_user_host if key == "dst" else ""
187 if ssh_user_host:
188 _validate_default_shell(default_shell, key, ssh_user_host)
189 elif program.startswith("getconf_cpu_count-"):
190 programs.pop(program)
191 getconf_cpu_count: str = program[len("getconf_cpu_count-") :]
192 programs["getconf_cpu_count"] = getconf_cpu_count
193 log.log(LOG_TRACE, f"available_programs[{key}][getconf_cpu_count]: %s", getconf_cpu_count)
195 for key, programs in available_programs.items():
196 log.debug(f"available_programs[{key}]: %s", list_formatter(programs, separator=", "))
198 for r in [p.dst, p.src]:
199 if is_dummy(r):
200 continue
201 if r.sudo and not p.is_program_available("sudo", r.location):
202 die(f"{p.sudo_program} CLI is not available on {r.location} host: {r.ssh_user_host or 'localhost'}")
204 if (
205 len(p.args.preserve_properties) > 0
206 and any(prop in p.zfs_send_program_opts for prop in ["--props", "-p"])
207 and not p.skip_replication
208 and not p.is_program_available(ZFS_VERSION_IS_AT_LEAST_2_2_0, p.dst.location)
209 ):
210 die(
211 "Cowardly refusing to proceed as --preserve-properties is unreliable on destination ZFS < 2.2.0 when using "
212 "'zfs send --props'. Either upgrade destination ZFS, or remove '--props' from --zfs-send-program-opt(s)."
213 )
216def _disable_program(p: Params, program: str, locations: list[str]) -> None:
217 """Removes the given program from the available_programs mapping."""
218 for location in locations:
219 p.available_programs[location].pop(program, None)
222def _find_available_programs(p: Params) -> str:
223 """POSIX shell script that checks for the existence of various programs; It uses `if` statements instead of `&&` plus
224 `printf` instead of `echo` to ensure maximum compatibility across shells."""
225 cmds: list[str] = []
226 cmds.append("printf 'default_shell-%s\n' \"$SHELL\"")
227 cmds.append("if command -v echo > /dev/null; then printf 'echo\n'; fi")
228 cmds.append(f"if command -v {p.zpool_program} > /dev/null; then printf 'zpool\n'; fi")
229 cmds.append(f"if command -v {p.ssh_program} > /dev/null; then printf 'ssh\n'; fi")
230 cmds.append(f"if command -v {p.shell_program} > /dev/null; then printf 'sh\n'; fi")
231 cmds.append(f"if command -v {p.sudo_program} > /dev/null; then printf 'sudo\n'; fi")
232 cmds.append(f"if command -v {p.compression_program} > /dev/null; then printf 'zstd\n'; fi")
233 cmds.append(f"if command -v {p.mbuffer_program} > /dev/null; then printf 'mbuffer\n'; fi")
234 cmds.append(f"if command -v {p.pv_program} > /dev/null; then printf 'pv\n'; fi")
235 cmds.append(f"if command -v {p.ps_program} > /dev/null; then printf 'ps\n'; fi")
236 cmds.append(
237 f"if command -v {p.getconf_program} > /dev/null; then "
238 f"printf 'getconf_cpu_count-'; {p.getconf_program} _NPROCESSORS_ONLN; "
239 "fi"
240 )
241 cmds.append(f"if command -v {p.uname_program} > /dev/null; then printf 'uname-'; {p.uname_program} -a || true; fi")
242 return "; ".join(cmds)
245def _detect_available_programs_remote(job: Job, remote: Remote, ssh_user_host: str) -> dict[str, str]:
246 """Detects CLI tools available on ``remote`` and updates mapping correspondingly."""
247 p, log = job.params, job.params.log
248 location = remote.location
249 available_programs_minimum = {"sudo": ""}
250 available_programs: dict[str, str] = {}
251 if is_dummy(remote):
252 return available_programs
253 lines: str | None = None
254 try:
255 # on Linux, 'zfs --version' returns with zero status and prints the correct info
256 # on FreeBSD, 'zfs --version' always prints the same (correct) info as Linux, but nonetheless sometimes
257 # returns with non-zero status (sometimes = if the zfs kernel module is not loaded)
258 lines = job.run_ssh_command_with_retries(remote, LOG_TRACE, print_stderr=False, cmd=[p.zfs_program, "--version"])
259 assert lines
260 except (FileNotFoundError, PermissionError): # location is local and program file was not found
261 die(f"{p.zfs_program} CLI is not available on {location} host: {ssh_user_host or 'localhost'}")
262 except subprocess.CalledProcessError as e:
263 stderr: str = stderr_to_str(e.stderr)
264 stdout: str = stderr_to_str(e.stdout)
265 if "unrecognized command '--version'" in stderr and "run: zfs help" in stderr:
266 die(f"Unsupported ZFS platform: {stderr}") # solaris is unsupported
267 elif stderr.startswith("ssh: "):
268 assert e.returncode == 255, e.returncode # error within SSH itself (not during the remote command)
269 die(f"ssh exit code {e.returncode}: {stderr.rstrip()}")
270 elif not stdout.startswith("zfs"): 270 ↛ 273line 270 didn't jump to line 273 because the condition on line 270 was always true
271 die(f"{p.zfs_program} CLI is not available on {location} host: {ssh_user_host or 'localhost'}")
272 else:
273 lines = stdout # FreeBSD if the zfs kernel module is not loaded
274 assert lines
275 if lines: 275 ↛ 287line 275 didn't jump to line 287 because the condition on line 275 was always true
276 # Examples that should parse: "zfs-2.1.5~rc5-ubuntu3", "zfswin-2.2.3rc5"
277 first_line: str = lines.splitlines()[0] if lines.splitlines() else ""
278 match = re.search(r"(\d+)\.(\d+)\.(\d+)", first_line)
279 if not match:
280 die("Unparsable zfs version string: '" + first_line + "'")
281 version = ".".join(match.groups())
282 available_programs["zfs"] = version
283 if is_version_at_least(version, "2.1.0"):
284 available_programs[ZFS_VERSION_IS_AT_LEAST_2_1_0] = ""
285 if is_version_at_least(version, "2.2.0"):
286 available_programs[ZFS_VERSION_IS_AT_LEAST_2_2_0] = ""
287 log.log(LOG_TRACE, f"available_programs[{location}][zfs]: %s", available_programs["zfs"])
289 if p.shell_program != DISABLE_PRG:
290 try:
291 cmd: list[str] = [p.shell_program, "-c", _find_available_programs(p)]
292 stdout = job.run_ssh_command_with_retries(remote, LOG_TRACE, cmd=cmd)
293 available_programs.update(dict.fromkeys(stdout.splitlines(), ""))
294 return available_programs
295 except (FileNotFoundError, PermissionError) as e: # location is local and shell program file was not found
296 if e.filename != p.shell_program: 296 ↛ 297line 296 didn't jump to line 297 because the condition on line 296 was never true
297 raise
298 except subprocess.CalledProcessError:
299 pass
300 log.warning("%s", f"Failed to find {p.shell_program} on {location}. Continuing with minimal assumptions...")
301 available_programs.update(available_programs_minimum)
302 return available_programs
305def is_dummy(r: Remote) -> bool:
306 """Returns True if ``remote`` refers to the synthetic dummy dataset."""
307 return r.root_dataset == DUMMY_DATASET
310def _detect_zpool_features(job: Job, remote: Remote, available_programs: dict) -> dict[str, str]:
311 """Fills ``job.params.zpool_features`` with detected zpool capabilities."""
312 p = params = job.params
313 r, loc, log = remote, remote.location, p.log
314 lines: list[str] = []
315 features: dict[str, str] = {}
316 if is_dummy(r):
317 return {}
318 if params.zpool_program != DISABLE_PRG and (params.shell_program == DISABLE_PRG or "zpool" in available_programs):
319 cmd: list[str] = params.split_args(f"{params.zpool_program} get -Hp -o property,value all", r.pool)
320 try:
321 lines = job.run_ssh_command_with_retries(remote, LOG_TRACE, cmd=cmd).splitlines()
322 except (FileNotFoundError, PermissionError) as e:
323 if e.filename != params.zpool_program: 323 ↛ 324line 323 didn't jump to line 324 because the condition on line 323 was never true
324 raise
325 log.warning("%s", f"Failed to detect zpool features on {loc}: {r.pool}. Continuing with minimal assumptions ...")
326 except (subprocess.CalledProcessError, UnicodeDecodeError):
327 log.warning("%s", f"Failed to detect zpool features on {loc}: {r.pool}. Continuing with minimal assumptions ...")
328 else:
329 props: dict[str, str] = dict(line.split("\t", 1) for line in lines)
330 features = {k: v for k, v in props.items() if k.startswith("feature@") or k == "delegation" or k == POOL_GUID}
331 if len(lines) == 0:
332 cmd = p.split_args(f"{p.zfs_program} list -t filesystem -Hp -o name -s name", r.pool)
333 if job.try_ssh_command_with_retries(remote, LOG_TRACE, cmd=cmd) is None:
334 die(f"Pool does not exist for {loc} dataset: {r.basis_root_dataset}. Manually create the pool first!")
335 return features
338def is_zpool_feature_enabled_or_active(p: Params, remote: Remote, feature: str) -> bool:
339 """Returns True if the given zpool feature is active or enabled on ``remote``."""
340 return p.zpool_features[remote.location][remote.pool].get(feature) in ("active", "enabled")
343def are_bookmarks_enabled(p: Params, remote: Remote) -> bool:
344 """Checks if bookmark related features are enabled on ``remote``."""
345 return is_zpool_feature_enabled_or_active(p, remote, "feature@bookmark_v2") and is_zpool_feature_enabled_or_active(
346 p, remote, "feature@bookmark_written"
347 )
350def is_caching_snapshots(p: Params, remote: Remote) -> bool:
351 """Returns True if snapshot caching is supported and enabled on ``remote``."""
352 return p.is_caching_snapshots and p.is_program_available(ZFS_VERSION_IS_AT_LEAST_2_2_0, remote.location)
355def is_version_at_least(version_str: str, min_version_str: str) -> bool:
356 """Checks if the version string is at least the minimum version string."""
357 return tuple(map(int, version_str.split("."))) >= tuple(map(int, min_version_str.split(".")))
360def is_version_at_most(version_str: str, max_version_str: str) -> bool:
361 """Checks if the version string is at most the maximum version string."""
362 return tuple(map(int, version_str.split("."))) <= tuple(map(int, max_version_str.split(".")))
365def _validate_default_shell(path_to_default_shell: str, location: str, ssh_user_host: str) -> None:
366 """Fails for default shells that do not honor POSIX shell quoting."""
367 shell_name: str = path_to_default_shell.rsplit("/", maxsplit=1)[-1]
368 if shell_name in ("csh", "tcsh", "elvish", "fish", "nu", "nushell", "xonsh"):
369 # On some old FreeBSD systems the default shell is still csh. Also see https://www.grymoire.com/unix/CshTop10.txt
370 die(
371 f"Cowardly refusing to proceed because {PROG_NAME} requires POSIX shell quoting of special characters, "
372 f"but '{path_to_default_shell}' is incompatible. The safe workaround is to first manually set 'sh' "
373 f"instead of '{path_to_default_shell}' as the default shell of the Unix user on {location} host: "
374 f"{ssh_user_host or 'localhost'}, like so: "
375 "chsh -s /bin/sh <YOURUSERNAME>"
376 )