Coverage for bzfs_main/compare_snapshot_lists.py: 100%
219 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"""Implementation of bzfs --compare-snapshot-lists."""
17from __future__ import (
18 annotations,
19)
20import itertools
21import os
22import time
23from collections import (
24 defaultdict,
25)
26from collections.abc import (
27 Iterable,
28 Iterator,
29 Sequence,
30)
31from dataclasses import (
32 dataclass,
33 field,
34)
35from typing import (
36 TYPE_CHECKING,
37 Callable,
38 final,
39)
41from bzfs_main.argparse_cli import (
42 CMP_CHOICES_ITEMS,
43)
44from bzfs_main.detect import (
45 are_bookmarks_enabled,
46)
47from bzfs_main.filter import (
48 filter_snapshots,
49)
50from bzfs_main.parallel_batch_cmd import (
51 zfs_list_snapshots_in_parallel,
52)
53from bzfs_main.replication import (
54 is_tmp_bookmark,
55)
56from bzfs_main.util.parallel_iterator import (
57 run_in_parallel,
58)
59from bzfs_main.util.utils import (
60 DIR_PERMISSIONS,
61 FILE_PERMISSIONS,
62 HashedInterner,
63 human_readable_bytes,
64 human_readable_duration,
65 isotime_from_unixtime,
66 open_nofollow,
67 relativize_dataset,
68)
70if TYPE_CHECKING: # pragma: no cover - for type hints only
71 from bzfs_main.bzfs import (
72 Job,
73 )
74 from bzfs_main.configuration import (
75 Remote,
76 )
79@dataclass(order=True, frozen=True)
80@final
81class _ComparableSnapshot:
82 """Snapshot entry comparable by rel_dataset and GUID for sorting and merging."""
84 key: tuple[str, str] # rel_dataset, guid
85 cols: list[str] = field(compare=False) # excluded from comparison/equality checks
88def run_compare_snapshot_lists(job: Job, src_datasets: list[str], dst_datasets: list[str]) -> None:
89 """Compares source and destination dataset trees recursively with respect to snapshots, for example to check if all
90 recently taken snapshots have been successfully replicated by a periodic job; implements --compare-snapshot-lists.
92 Lists snapshots only contained in source (tagged with 'src'), only contained in destination (tagged with 'dst'), and
93 contained in both source and destination (tagged with 'all'), in the form of a TSV file, along with other snapshot
94 metadata.
96 Implemented with a time and space efficient streaming algorithm; easily scales to millions of datasets and any number of
97 snapshots. Time complexity is O((N log N) + (N * M log M)) where N is the number of datasets and M is the number of
98 snapshots per dataset. Space complexity is O(max(N, M)). Assumes that both src_datasets and dst_datasets are sorted.
99 """
100 p, log = job.params, job.params.log
101 src, dst = p.src, p.dst
102 task: str = src.root_dataset + " vs. " + dst.root_dataset
103 tsv_dir: str = p.log_params.log_file[: -len(".log")] + ".cmp"
104 os.makedirs(tsv_dir, mode=DIR_PERMISSIONS, exist_ok=True)
105 tsv_file: str = os.path.join(tsv_dir, (src.root_dataset + "%" + dst.root_dataset).replace("/", "~") + ".tsv")
106 tmp_tsv_file: str = tsv_file + ".tmp"
107 compare_snapshot_lists: set[str] = set(p.compare_snapshot_lists.split("+"))
108 is_src_dst_all: bool = all(choice in compare_snapshot_lists for choice in CMP_CHOICES_ITEMS)
109 all_src_dst: list[str] = [loc for loc in ("all", "src", "dst") if loc in compare_snapshot_lists]
110 is_first_row: bool = True
111 now: int | None = None
113 def _dataset_name(line: str) -> str:
114 """Extracts the dataset name from a ZFS output line whose final column is a snapshot or bookmark name."""
115 return line.rsplit("\t", 1)[1].replace("#", "@", 1).split("@", 1)[0]
117 def zfs_list_snapshot_iterator(r: Remote, sorted_datasets: list[str]) -> Iterator[str]:
118 """Lists snapshots sorted by dataset name; All snapshots of a given dataset will be adjacent."""
119 assert (not job.is_test_mode) or sorted_datasets == sorted(sorted_datasets), "List is not sorted"
120 # Also see https://openzfs.github.io/openzfs-docs/man/master/7/zfsprops.7.html#written
121 props: str = job.creation_prefix + "creation,guid,createtxg,written,name"
122 types: str = "snapshot"
123 list_bookmarks: bool = p.use_bookmark and r.location == "src" and are_bookmarks_enabled(p, r)
124 if list_bookmarks:
125 types = "snapshot,bookmark" # output list ordering: intentionally makes bookmarks appear *after* snapshots
126 cmd: list[str] = p.split_args(f"{p.zfs_program} list -t {types} -d 1 -Hp -o {props}") # sorted by dataset, createtxg
127 for lines in zfs_list_snapshots_in_parallel(job, r, cmd, sorted_datasets):
128 if list_bookmarks:
129 # `zfs list` can interleave a space-named sibling dataset between a dataset's snapshots and bookmarks.
130 # e.g. [tank/src@foo, tank/src copy@bar, tank/src#bar] --> [tank/src@foo, tank/src#bar, tank/src copy@bar]
131 lines.sort(key=_dataset_name) # stable sort by dataset name
132 yield from lines
134 def snapshot_iterator(r: Remote, root_dataset: str, sorted_itr: Iterator[str]) -> Iterator[_ComparableSnapshot]:
135 """Splits/groups snapshot stream into distinct datasets, sorts by GUID within a dataset such that any two snapshots
136 with the same GUID will lie adjacent to each other during the upcoming phase that merges src snapshots and dst
137 snapshots."""
138 list_bookmarks: bool = p.use_bookmark and r.location == "src" and are_bookmarks_enabled(p, r)
139 # streaming group by dataset name (consumes constant memory only)
140 for dataset, group in itertools.groupby(sorted_itr, key=_dataset_name):
141 snapshots: list[str] = list(group) # fetch all snapshots of current dataset, e.g. dataset=tank1/src/foo
142 tmp_bookmarks: list[str] = []
143 if list_bookmarks: # temporary bookmarks bypass include/exclude filters, e.g. name and rank filters
144 tmp_bookmarks = [snapshot for snapshot in snapshots if is_tmp_bookmark(snapshot)]
145 snapshots = [snapshot for snapshot in snapshots if not is_tmp_bookmark(snapshot)]
146 non_tmp_guids: set[str] = {snapshot.split("\t", 2)[1] for snapshot in snapshots} # for subsequent dedupe
147 tmp_bookmarks = [bookmark for bookmark in tmp_bookmarks if bookmark.split("\t", 2)[1] not in non_tmp_guids]
148 del non_tmp_guids # help gc
149 snapshots = filter_snapshots(job, snapshots, filter_bookmarks=True) # apply include/exclude policy
150 snapshots += tmp_bookmarks
151 snapshots.sort(key=lambda line: line.split("\t", 2)[1]) # stable sort by GUID (2nd remains createtxg)
152 rel_dataset: str = relativize_dataset(dataset, root_dataset) # rel_dataset=/foo, root_dataset=tank1/src
153 last_guid: str = ""
154 for line in snapshots:
155 cols: list[str] = line.split("\t")
156 _creation, guid, _createtxg, _written, snapshot_name = cols
157 if guid == last_guid:
158 assert "#" in snapshot_name
159 continue # ignore bookmarks whose snapshot still exists. also ignore dupes of bookmarks
160 last_guid = guid
161 key = (rel_dataset, guid) # ensures src snapshots and dst snapshots with the same GUID will be adjacent
162 yield _ComparableSnapshot(key, cols)
164 def print_dataset(rel_dataset: str, entries: Iterable[tuple[str, _ComparableSnapshot]]) -> None:
165 entries = sorted( # fetch all snapshots of current dataset and sort em by creation, createtxg, snapshot_tag
166 entries,
167 key=lambda entry: (
168 int((cols := entry[1].cols)[0]), # creation
169 int(cols[2]), # createtxg
170 cols[-1].replace("#", "@", 1).split("@", 1)[1], # snapshot_tag
171 ),
172 )
174 @dataclass
175 @final
176 class SnapshotStats:
177 snapshot_count: int = 0
178 sum_written: int = 0
179 snapshot_count_since: int = 0
180 sum_written_since: int = 0
181 latest_snapshot_idx: int | None = None
182 latest_snapshot_row_str: str | None = None
183 latest_snapshot_creation: str | None = None
184 oldest_snapshot_row_str: str | None = None
185 oldest_snapshot_creation: str | None = None
187 # print metadata of snapshots of current dataset to TSV file; custom stats can later be computed from there
188 stats: defaultdict[str, SnapshotStats] = defaultdict(SnapshotStats)
189 header: str = "location creation_iso createtxg rel_name guid root_dataset rel_dataset name creation written"
190 nonlocal is_first_row
191 if is_first_row:
192 fd.write(header.replace(" ", "\t") + "\n")
193 is_first_row = False
194 for i, entry in enumerate(entries): # entry is tuple[location:str, ComparableSnapshot]
195 location: str = entry[0] # "src" or "dst" or "all"
196 creation, guid, createtxg, written, name = entry[1].cols
197 root_dataset: str = dst.root_dataset if location == CMP_CHOICES_ITEMS[1] else src.root_dataset
198 rel_name: str = relativize_dataset(name, root_dataset)
199 creation_iso: str = isotime_from_unixtime(int(creation))
200 row = (location, creation_iso, createtxg, rel_name, guid, root_dataset, rel_dataset, name, creation, written)
201 # Example: src 2024-11-06_08:30:05 17435050 /foo@test_2024-11-06_08:30:05_daily 2406491805272097867 tank1/src /foo tank1/src/foo@test_2024-10-06_08:30:04_daily 1730878205 24576
202 row_str = "\t".join(row)
203 if not p.dry_run:
204 fd.write(row_str + "\n")
205 s = stats[location]
206 s.snapshot_count += 1
207 s.sum_written += int(written) if written != "-" else 0
208 s.latest_snapshot_idx = i
209 s.latest_snapshot_row_str = row_str
210 s.latest_snapshot_creation = creation
211 if not s.oldest_snapshot_row_str:
212 s.oldest_snapshot_row_str = row_str
213 s.oldest_snapshot_creation = creation
215 # for convenience, directly log basic summary stats of current dataset
216 k = stats["all"].latest_snapshot_idx # defaults to None
217 k = k if k is not None else -1
218 for entry in entries[k + 1 :]: # aggregate basic stats since latest common snapshot
219 location = entry[0]
220 creation, guid, createtxg, written, name = entry[1].cols
221 s = stats[location]
222 s.snapshot_count_since += 1
223 s.sum_written_since += int(written) if written != "-" else 0
224 prefix: str = f"Comparing {rel_dataset}~"
225 msgs: list[str] = []
226 msgs.append(f"{prefix} of {task}")
227 msgs.append(
228 f"{prefix} Q: No src snapshots are missing on dst, and no dst snapshots are missing on src, "
229 "and there is a common snapshot? A: "
230 + (
231 "n/a"
232 if not is_src_dst_all
233 else str(
234 stats["src"].snapshot_count == 0 and stats["dst"].snapshot_count == 0 and stats["all"].snapshot_count > 0
235 )
236 )
237 )
238 nonlocal now
239 now = now or int(time.time()) # ZFS truncates fractional secs in 'creation' so here we do, too. And keep now stable
240 latcom = "latest common snapshot"
241 for loc in all_src_dst:
242 s = stats[loc]
243 msgs.append(f"{prefix} Latest snapshot only in {loc}: {s.latest_snapshot_row_str or 'n/a'}")
244 msgs.append(f"{prefix} Oldest snapshot only in {loc}: {s.oldest_snapshot_row_str or 'n/a'}")
245 msgs.append(f"{prefix} Snapshots only in {loc}: {s.snapshot_count}")
246 msgs.append(f"{prefix} Snapshot data written only in {loc}: {human_readable_bytes(s.sum_written)}")
247 if loc != "all":
248 na = None if k >= 0 else "n/a"
249 msgs.append(f"{prefix} Snapshots only in {loc} since {latcom}: {na or s.snapshot_count_since}")
250 msgs.append(
251 f"{prefix} Snapshot data written only in {loc} since {latcom}: "
252 f"{na or human_readable_bytes(s.sum_written_since)}"
253 )
254 all_creation = stats["all"].latest_snapshot_creation
255 latest = ("latest", s.latest_snapshot_creation)
256 oldest = ("oldest", s.oldest_snapshot_creation)
257 for label, s_creation in latest, oldest:
258 if loc != "all":
259 hd = "n/a"
260 if s_creation and k >= 0:
261 assert all_creation is not None
262 hd = human_readable_duration(int(all_creation) - int(s_creation), unit="s")
263 msgs.append(f"{prefix} Time diff between {latcom} and {label} snapshot only in {loc}: {hd}")
264 for label, s_creation in latest, oldest:
265 hd = "n/a" if not s_creation else human_readable_duration(now - int(s_creation), unit="s")
266 msgs.append(f"{prefix} Time diff between now and {label} snapshot only in {loc}: {hd}")
267 log.info("%s", "\n".join(msgs))
269 # setup streaming pipeline
270 src_snapshot_itr: Iterator = snapshot_iterator(src, src.root_dataset, zfs_list_snapshot_iterator(src, src_datasets))
271 dst_snapshot_itr: Iterator = snapshot_iterator(dst, dst.root_dataset, zfs_list_snapshot_iterator(dst, dst_datasets))
272 merge_itr = _merge_sorted_iterators(CMP_CHOICES_ITEMS, p.compare_snapshot_lists, src_snapshot_itr, dst_snapshot_itr)
274 interner: HashedInterner[str] = HashedInterner() # reduces memory footprint
275 rel_datasets: dict[str, set[str]] = defaultdict(set)
276 for datasets, remote in (src_datasets, src), (dst_datasets, dst):
277 for dataset in datasets: # rel_dataset=/foo, root_dataset=tank1/src
278 rel_datasets[remote.location].add(interner.intern(relativize_dataset(dataset, remote.root_dataset)))
279 rel_src_or_dst: list[str] = sorted(rel_datasets["src"].union(rel_datasets["dst"]))
281 log.debug("%s", f"Temporary TSV output file comparing {task} is: {tmp_tsv_file}")
282 with open_nofollow(tmp_tsv_file, "w", encoding="utf-8", perm=FILE_PERMISSIONS) as fd:
283 # streaming group by rel_dataset (consumes constant memory only); entry is a Tuple[str, ComparableSnapshot]
284 groups = itertools.groupby(merge_itr, key=lambda entry: entry[1].key[0])
285 _print_datasets(groups, lambda rel_ds, entries: print_dataset(rel_ds, entries), rel_src_or_dst)
286 os.rename(tmp_tsv_file, tsv_file)
287 log.info("%s", f"Final TSV output file comparing {task} is: {tsv_file}")
289 tsv_file = tsv_file[: tsv_file.rindex(".")] + ".rel_datasets_tsv"
290 tmp_tsv_file = tsv_file + ".tmp"
291 with open_nofollow(tmp_tsv_file, "w", encoding="utf-8", perm=FILE_PERMISSIONS) as fd:
292 header: str = "location rel_dataset src_dataset dst_dataset"
293 fd.write(header.replace(" ", "\t") + "\n")
294 src_only: set[str] = rel_datasets["src"].difference(rel_datasets["dst"])
295 dst_only: set[str] = rel_datasets["dst"].difference(rel_datasets["src"])
296 for rel_dataset in rel_src_or_dst:
297 loc = "src" if rel_dataset in src_only else "dst" if rel_dataset in dst_only else "all"
298 src_dataset = src.root_dataset + rel_dataset if rel_dataset not in dst_only else ""
299 dst_dataset = dst.root_dataset + rel_dataset if rel_dataset not in src_only else ""
300 row = (loc, rel_dataset, src_dataset, dst_dataset) # Example: all /foo/bar tank1/src/foo/bar tank2/dst/foo/bar
301 if not p.dry_run:
302 fd.write("\t".join(row) + "\n")
303 os.rename(tmp_tsv_file, tsv_file)
306def _print_datasets(groups: itertools.groupby, fn: Callable[[str, Iterable], None], rel_datasets: Iterable[str]) -> None:
307 """Iterate over grouped datasets and apply fn, adding gaps for missing ones."""
308 rel_datasets = sorted(rel_datasets)
309 n = len(rel_datasets)
310 i = 0
311 for rel_dataset, entries in groups:
312 while i < n and rel_datasets[i] < rel_dataset:
313 fn(rel_datasets[i], []) # Also print summary stats for datasets whose snapshot stream is empty
314 i += 1
315 assert i >= n or rel_datasets[i] == rel_dataset
316 i += 1
317 fn(rel_dataset, entries)
318 while i < n:
319 fn(rel_datasets[i], []) # Also print summary stats for datasets whose snapshot stream is empty
320 i += 1
323def _merge_sorted_iterators(
324 choices: Sequence[str], # ["src", "dst", "all"]
325 choice: str, # Example: "src+dst+all"
326 src_itr: Iterator[_ComparableSnapshot],
327 dst_itr: Iterator[_ComparableSnapshot],
328) -> Iterator[tuple[str, _ComparableSnapshot] | tuple[str, _ComparableSnapshot, _ComparableSnapshot]]:
329 """The typical pipelined merge algorithm of a merge sort, slightly adapted to our specific use case."""
330 assert len(choices) == 3
331 assert choice
332 enabled: tuple[bool, ...] = tuple(item in choice for item in choices)
333 src_next, dst_next = run_in_parallel(lambda: next(src_itr, None), lambda: next(dst_itr, None))
334 while (src_next is not None) or (dst_next is not None):
335 if src_next == dst_next:
336 n = 2
337 if enabled[n]:
338 assert src_next is not None
339 assert dst_next is not None
340 yield choices[n], src_next, dst_next
341 src_next = next(src_itr, None)
342 dst_next = next(dst_itr, None)
343 elif src_next is None or (dst_next is not None and dst_next < src_next):
344 n = 1
345 if enabled[n]:
346 assert dst_next is not None
347 yield choices[n], dst_next
348 dst_next = next(dst_itr, None)
349 else:
350 assert src_next is not None
351 n = 0
352 if enabled[n] and not _is_comparable_tmp_bookmark(src_next): # suppress tmp bookmarks
353 yield choices[n], src_next
354 src_next = next(src_itr, None)
357def _is_comparable_tmp_bookmark(src_next: object) -> bool:
358 return isinstance(src_next, _ComparableSnapshot) and is_tmp_bookmark(src_next.cols[-1])