Coverage for bzfs_main/detect.py: 96%

209 statements  

« prev     ^ index     » next       coverage.py v7.11.0, created at 2025-11-07 04: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.""" 

16 

17from __future__ import ( 

18 annotations, 

19) 

20import re 

21import subprocess 

22import threading 

23import time 

24from dataclasses import ( 

25 dataclass, 

26 field, 

27) 

28from subprocess import ( 

29 DEVNULL, 

30 PIPE, 

31) 

32from typing import ( 

33 TYPE_CHECKING, 

34 Final, 

35) 

36 

37from bzfs_main.connection import ( 

38 DEDICATED, 

39 SHARED, 

40 ConnectionPools, 

41 run_ssh_command, 

42 try_ssh_command, 

43) 

44from bzfs_main.utils import ( 

45 LOG_TRACE, 

46 PROG_NAME, 

47 SynchronousExecutor, 

48 die, 

49 drain, 

50 list_formatter, 

51 stderr_to_str, 

52 xprint, 

53) 

54 

55if TYPE_CHECKING: # pragma: no cover - for type hints only 

56 from bzfs_main.bzfs import ( 

57 Job, 

58 ) 

59 from bzfs_main.configuration import ( 

60 Params, 

61 Remote, 

62 ) 

63 

64# constants: 

65DISABLE_PRG: Final[str] = "-" 

66DUMMY_DATASET: Final[str] = "dummy" 

67ZFS_VERSION_IS_AT_LEAST_2_1_0: Final[str] = "zfs>=2.1.0" 

68ZFS_VERSION_IS_AT_LEAST_2_2_0: Final[str] = "zfs>=2.2.0" 

69 

70 

71############################################################################# 

72@dataclass(frozen=True) 

73class RemoteConfCacheItem: 

74 """Caches detected programs, zpool features and connection pools, per remote.""" 

75 

76 connection_pools: ConnectionPools 

77 available_programs: dict[str, str] 

78 zpool_features: dict[str, dict[str, str]] 

79 timestamp_nanos: int = field(default_factory=time.monotonic_ns) 

80 

81 

82def detect_available_programs(job: Job) -> None: 

83 """Detects programs, zpool features and connection pools for local and remote hosts.""" 

84 p = params = job.params 

85 log = p.log 

86 available_programs: dict[str, dict[str, str]] = params.available_programs 

87 if "local" not in available_programs: 

88 cmd: list[str] = [p.shell_program_local, "-c", _find_available_programs(p)] 

89 sp = job.subprocesses 

90 proc = sp.subprocess_run(cmd, stdin=DEVNULL, stdout=PIPE, stderr=PIPE, text=True) 

91 xprint(log=log, value=stderr_to_str(proc.stderr), end="") 

92 stdout: str = proc.stdout 

93 available_programs["local"] = dict.fromkeys(stdout.splitlines(), "") 

94 cmd = [p.shell_program_local, "-c", "exit"] 

95 proc = sp.subprocess_run(cmd, stdin=DEVNULL, stdout=PIPE, stderr=PIPE, text=True) 

96 xprint(log=log, value=stderr_to_str(proc.stderr), end="") 

97 if proc.returncode != 0: 

98 _disable_program(p, "sh", ["local"]) 

99 

100 todo: list[Remote] = [] 

101 for r in [p.dst, p.src]: 

102 loc: str = r.location 

103 remote_conf_cache_key: tuple = r.cache_key() 

104 cache_item: RemoteConfCacheItem | None = job.remote_conf_cache.get(remote_conf_cache_key) 

105 if cache_item is not None: 

106 # startup perf: cache avoids ssh connect setup and feature detection roundtrips on revisits to same site 

107 p.connection_pools[loc] = cache_item.connection_pools 

108 p.available_programs[loc] = cache_item.available_programs 

109 p.zpool_features[loc] = cache_item.zpool_features 

110 if time.monotonic_ns() - cache_item.timestamp_nanos < p.remote_conf_cache_ttl_nanos: 

111 if r.pool in cache_item.zpool_features: 111 ↛ 119line 111 didn't jump to line 119 because the condition on line 111 was always true

112 continue # cache hit, skip remote detection 

113 else: 

114 p.zpool_features[loc] = {} # cache miss, invalidate features of zpools before refetching from remote 

115 else: 

116 p.connection_pools[loc] = ConnectionPools( 

117 r, {SHARED: r.max_concurrent_ssh_sessions_per_tcp_connection, DEDICATED: 1} 

118 ) 

119 todo.append(r) 

120 

121 lock: threading.Lock = threading.Lock() 

122 

123 def run_detect(r: Remote) -> None: # thread-safe 

124 loc: str = r.location 

125 remote_conf_cache_key: tuple = r.cache_key() 

126 available_programs: dict[str, str] = _detect_available_programs_remote(job, r, r.ssh_user_host) 

127 zpool_features: dict[str, str] = _detect_zpool_features(job, r, available_programs) 

128 with lock: 

129 r.params.available_programs[loc] = available_programs 

130 r.params.zpool_features[loc][r.pool] = zpool_features 

131 job.remote_conf_cache[remote_conf_cache_key] = RemoteConfCacheItem( 

132 p.connection_pools[loc], available_programs, r.params.zpool_features[loc] 

133 ) 

134 if r.use_zfs_delegation and zpool_features.get("delegation") == "off": 

135 die( 

136 f"Permission denied as ZFS delegation is disabled for {r.location} " 

137 f"dataset: {r.basis_root_dataset}. Manually enable it via 'sudo zpool set delegation=on {r.pool}'" 

138 ) 

139 

140 with SynchronousExecutor.executor_for(max_workers=max(1, len(todo))) as executor: 

141 drain(executor.map(run_detect, todo)) # detect ZFS features + system capabilities on src+dst in parallel 

142 

143 locations = ["src", "dst", "local"] 

144 if params.compression_program == DISABLE_PRG: 

145 _disable_program(p, "zstd", locations) 

146 if params.mbuffer_program == DISABLE_PRG: 

147 _disable_program(p, "mbuffer", locations) 

148 if params.ps_program == DISABLE_PRG: 

149 _disable_program(p, "ps", locations) 

150 if params.pv_program == DISABLE_PRG: 

151 _disable_program(p, "pv", locations) 

152 if params.shell_program == DISABLE_PRG: 

153 _disable_program(p, "sh", locations) 

154 if params.sudo_program == DISABLE_PRG: 

155 _disable_program(p, "sudo", locations) 

156 if params.zpool_program == DISABLE_PRG: 

157 _disable_program(p, "zpool", locations) 

158 

159 for key, programs in available_programs.items(): 

160 for program in list(programs.keys()): 

161 if program.startswith("uname-"): 

162 # 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 

163 # uname-FreeBSD freebsd 14.1-RELEASE FreeBSD 14.1-RELEASE releng/14.1-n267679-10e31f0946d8 GENERIC amd64 

164 # 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 

165 programs.pop(program) 

166 uname: str = program[len("uname-") :] 

167 programs["uname"] = uname 

168 log.log(LOG_TRACE, f"available_programs[{key}][uname]: %s", uname) 

169 programs["os"] = uname.split(" ", maxsplit=1)[0] # Linux|FreeBSD|Darwin 

170 log.log(LOG_TRACE, f"available_programs[{key}][os]: %s", programs["os"]) 

171 elif program.startswith("default_shell-"): 

172 programs.pop(program) 

173 default_shell: str = program[len("default_shell-") :] 

174 programs["default_shell"] = default_shell 

175 log.log(LOG_TRACE, f"available_programs[{key}][default_shell]: %s", default_shell) 

176 ssh_user_host = p.src.ssh_user_host if key == "src" else p.dst.ssh_user_host if key == "dst" else "" 

177 _validate_default_shell(default_shell, key, ssh_user_host) 

178 elif program.startswith("getconf_cpu_count-"): 

179 programs.pop(program) 

180 getconf_cpu_count: str = program[len("getconf_cpu_count-") :] 

181 programs["getconf_cpu_count"] = getconf_cpu_count 

182 log.log(LOG_TRACE, f"available_programs[{key}][getconf_cpu_count]: %s", getconf_cpu_count) 

183 

184 for key, programs in available_programs.items(): 

185 log.debug(f"available_programs[{key}]: %s", list_formatter(programs, separator=", ")) 

186 

187 for r in [p.dst, p.src]: 

188 if is_dummy(r): 

189 continue 

190 if r.sudo and not p.is_program_available("sudo", r.location): 

191 die(f"{p.sudo_program} CLI is not available on {r.location} host: {r.ssh_user_host or 'localhost'}") 

192 

193 if ( 

194 len(p.args.preserve_properties) > 0 

195 and any(prop in p.zfs_send_program_opts for prop in ["--props", "-p"]) 

196 and not p.is_program_available(ZFS_VERSION_IS_AT_LEAST_2_2_0, p.dst.location) 

197 ): 

198 die( 

199 "Cowardly refusing to proceed as --preserve-properties is unreliable on destination ZFS < 2.2.0 when using " 

200 "'zfs send --props'. Either upgrade destination ZFS, or remove '--props' from --zfs-send-program-opt(s)." 

201 ) 

202 

203 

204def _disable_program(p: Params, program: str, locations: list[str]) -> None: 

205 """Removes the given program from the available_programs mapping.""" 

206 for location in locations: 

207 p.available_programs[location].pop(program, None) 

208 

209 

210def _find_available_programs(p: Params) -> str: 

211 """POSIX shell script that checks for the existence of various programs; It uses `if` statements instead of `&&` plus 

212 `printf` instead of `echo` to ensure maximum compatibility across shells.""" 

213 cmds: list[str] = [] 

214 cmds.append("printf 'default_shell-%s\n' \"$SHELL\"") 

215 cmds.append("if command -v echo > /dev/null; then printf 'echo\n'; fi") 

216 cmds.append(f"if command -v {p.zpool_program} > /dev/null; then printf 'zpool\n'; fi") 

217 cmds.append(f"if command -v {p.ssh_program} > /dev/null; then printf 'ssh\n'; fi") 

218 cmds.append(f"if command -v {p.shell_program} > /dev/null; then printf 'sh\n'; fi") 

219 cmds.append(f"if command -v {p.sudo_program} > /dev/null; then printf 'sudo\n'; fi") 

220 cmds.append(f"if command -v {p.compression_program} > /dev/null; then printf 'zstd\n'; fi") 

221 cmds.append(f"if command -v {p.mbuffer_program} > /dev/null; then printf 'mbuffer\n'; fi") 

222 cmds.append(f"if command -v {p.pv_program} > /dev/null; then printf 'pv\n'; fi") 

223 cmds.append(f"if command -v {p.ps_program} > /dev/null; then printf 'ps\n'; fi") 

224 cmds.append( 

225 f"if command -v {p.getconf_program} > /dev/null; then " 

226 f"printf 'getconf_cpu_count-'; {p.getconf_program} _NPROCESSORS_ONLN; " 

227 "fi" 

228 ) 

229 cmds.append(f"if command -v {p.uname_program} > /dev/null; then printf 'uname-'; {p.uname_program} -a || true; fi") 

230 return "; ".join(cmds) 

231 

232 

233def _detect_available_programs_remote(job: Job, remote: Remote, ssh_user_host: str) -> dict[str, str]: 

234 """Detects CLI tools available on ``remote`` and updates mapping correspondingly.""" 

235 p, log = job.params, job.params.log 

236 location = remote.location 

237 available_programs_minimum = {"sudo": ""} 

238 available_programs: dict[str, str] = {} 

239 if is_dummy(remote): 

240 return available_programs 

241 lines: str | None = None 

242 try: 

243 # on Linux, 'zfs --version' returns with zero status and prints the correct info 

244 # on FreeBSD, 'zfs --version' always prints the same (correct) info as Linux, but nonetheless sometimes 

245 # returns with non-zero status (sometimes = if the zfs kernel module is not loaded) 

246 lines = run_ssh_command(job, remote, LOG_TRACE, print_stderr=False, cmd=[p.zfs_program, "--version"]) 

247 assert lines 

248 except (FileNotFoundError, PermissionError): # location is local and program file was not found 

249 die(f"{p.zfs_program} CLI is not available on {location} host: {ssh_user_host or 'localhost'}") 

250 except subprocess.CalledProcessError as e: 

251 if "unrecognized command '--version'" in e.stderr and "run: zfs help" in e.stderr: 

252 die(f"Unsupported ZFS platform: {e.stderr}") # solaris is unsupported 

253 elif not e.stdout.startswith("zfs"): 253 ↛ 256line 253 didn't jump to line 256 because the condition on line 253 was always true

254 die(f"{p.zfs_program} CLI is not available on {location} host: {ssh_user_host or 'localhost'}") 

255 else: 

256 lines = e.stdout # FreeBSD if the zfs kernel module is not loaded 

257 assert lines 

258 if lines: 258 ↛ 270line 258 didn't jump to line 270 because the condition on line 258 was always true

259 # Examples that should parse: "zfs-2.1.5~rc5-ubuntu3", "zfswin-2.2.3rc5" 

260 first_line: str = lines.splitlines()[0] if lines.splitlines() else "" 

261 match = re.search(r"(\d+)\.(\d+)\.(\d+)", first_line) 

262 if not match: 

263 die("Unparsable zfs version string: '" + first_line + "'") 

264 version = ".".join(match.groups()) 

265 available_programs["zfs"] = version 

266 if is_version_at_least(version, "2.1.0"): 

267 available_programs[ZFS_VERSION_IS_AT_LEAST_2_1_0] = "" 

268 if is_version_at_least(version, "2.2.0"): 

269 available_programs[ZFS_VERSION_IS_AT_LEAST_2_2_0] = "" 

270 log.log(LOG_TRACE, f"available_programs[{location}][zfs]: %s", available_programs["zfs"]) 

271 

272 if p.shell_program != DISABLE_PRG: 

273 try: 

274 cmd: list[str] = [p.shell_program, "-c", _find_available_programs(p)] 

275 stdout: str = run_ssh_command(job, remote, LOG_TRACE, cmd=cmd) 

276 available_programs.update(dict.fromkeys(stdout.splitlines(), "")) 

277 return available_programs 

278 except (FileNotFoundError, PermissionError) as e: # location is local and shell program file was not found 

279 if e.filename != p.shell_program: 279 ↛ 280line 279 didn't jump to line 280 because the condition on line 279 was never true

280 raise 

281 except subprocess.CalledProcessError: 

282 pass 

283 log.warning("%s", f"Failed to find {p.shell_program} on {location}. Continuing with minimal assumptions...") 

284 available_programs.update(available_programs_minimum) 

285 return available_programs 

286 

287 

288def is_dummy(r: Remote) -> bool: 

289 """Returns True if ``remote`` refers to the synthetic dummy dataset.""" 

290 return r.root_dataset == DUMMY_DATASET 

291 

292 

293def _detect_zpool_features(job: Job, remote: Remote, available_programs: dict) -> dict[str, str]: 

294 """Fills ``job.params.zpool_features`` with detected zpool capabilities.""" 

295 p = params = job.params 

296 r, loc, log = remote, remote.location, p.log 

297 lines: list[str] = [] 

298 features: dict[str, str] = {} 

299 if is_dummy(r): 

300 return {} 

301 if params.zpool_program != DISABLE_PRG and (params.shell_program == DISABLE_PRG or "zpool" in available_programs): 

302 cmd: list[str] = params.split_args(f"{params.zpool_program} get -Hp -o property,value all", r.pool) 

303 try: 

304 lines = run_ssh_command(job, remote, LOG_TRACE, check=False, cmd=cmd).splitlines() 

305 except (FileNotFoundError, PermissionError) as e: 

306 if e.filename != params.zpool_program: 

307 raise 

308 log.warning("%s", f"Failed to detect zpool features on {loc}: {r.pool}. Continuing with minimal assumptions ...") 

309 else: 

310 props: dict[str, str] = dict(line.split("\t", 1) for line in lines) 

311 features = {k: v for k, v in props.items() if k.startswith("feature@") or k == "delegation"} 

312 if len(lines) == 0: 

313 cmd = p.split_args(f"{p.zfs_program} list -t filesystem -Hp -o name -s name", r.pool) 

314 if try_ssh_command(job, remote, LOG_TRACE, cmd=cmd) is None: 

315 die(f"Pool does not exist for {loc} dataset: {r.basis_root_dataset}. Manually create the pool first!") 

316 return features 

317 

318 

319def is_zpool_feature_enabled_or_active(p: Params, remote: Remote, feature: str) -> bool: 

320 """Returns True if the given zpool feature is active or enabled on ``remote``.""" 

321 return p.zpool_features[remote.location][remote.pool].get(feature) in ("active", "enabled") 

322 

323 

324def are_bookmarks_enabled(p: Params, remote: Remote) -> bool: 

325 """Checks if bookmark related features are enabled on ``remote``.""" 

326 return is_zpool_feature_enabled_or_active(p, remote, "feature@bookmark_v2") and is_zpool_feature_enabled_or_active( 

327 p, remote, "feature@bookmark_written" 

328 ) 

329 

330 

331def is_caching_snapshots(p: Params, remote: Remote) -> bool: 

332 """Returns True if snapshot caching is supported and enabled on ``remote``.""" 

333 return p.is_caching_snapshots and p.is_program_available(ZFS_VERSION_IS_AT_LEAST_2_2_0, remote.location) 

334 

335 

336def is_version_at_least(version_str: str, min_version_str: str) -> bool: 

337 """Checks if the version string is at least the minimum version string.""" 

338 return tuple(map(int, version_str.split("."))) >= tuple(map(int, min_version_str.split("."))) 

339 

340 

341def _validate_default_shell(path_to_default_shell: str, location: str, ssh_user_host: str) -> None: 

342 """Fails if the remote user uses csh or tcsh as the default shell.""" 

343 if path_to_default_shell in ("csh", "tcsh") or path_to_default_shell.endswith(("/csh", "/tcsh")): 

344 # On some old FreeBSD systems the default shell is still csh. Also see https://www.grymoire.com/unix/CshTop10.txt 

345 die( 

346 f"Cowardly refusing to proceed because {PROG_NAME} is not compatible with csh-style quoting of special " 

347 f"characters. The safe workaround is to first manually set 'sh' instead of '{path_to_default_shell}' as " 

348 f"the default shell of the Unix user on {location} host: {ssh_user_host or 'localhost'}, like so: " 

349 "chsh -s /bin/sh YOURUSERNAME" 

350 )