Coverage for bzfs_main/compare_snapshot_lists.py: 100%
218 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-21 12:39 +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 zfs_list_snapshot_iterator(r: Remote, sorted_datasets: list[str]) -> Iterator[str]:
114 """Lists snapshots sorted by dataset name; All snapshots of a given dataset will be adjacent."""
115 assert (not job.is_test_mode) or sorted_datasets == sorted(sorted_datasets), "List is not sorted"
116 # Also see https://openzfs.github.io/openzfs-docs/man/master/7/zfsprops.7.html#written
117 props: str = job.creation_prefix + "creation,guid,createtxg,written,name"
118 types: str = "snapshot"
119 list_bookmarks: bool = p.use_bookmark and r.location == "src" and are_bookmarks_enabled(p, r)
120 if list_bookmarks:
121 types = "snapshot,bookmark" # output list ordering: intentionally makes bookmarks appear *after* snapshots
122 cmd: list[str] = p.split_args(f"{p.zfs_program} list -t {types} -d 1 -Hp -o {props}") # sorted by dataset, createtxg
123 for lines in zfs_list_snapshots_in_parallel(job, r, cmd, sorted_datasets):
124 yield from lines
126 def snapshot_iterator(r: Remote, root_dataset: str, sorted_itr: Iterator[str]) -> Iterator[_ComparableSnapshot]:
127 """Splits/groups snapshot stream into distinct datasets, sorts by GUID within a dataset such that any two snapshots
128 with the same GUID will lie adjacent to each other during the upcoming phase that merges src snapshots and dst
129 snapshots."""
130 list_bookmarks: bool = p.use_bookmark and r.location == "src" and are_bookmarks_enabled(p, r)
131 # streaming group by dataset name (consumes constant memory only)
132 for dataset, group in itertools.groupby(
133 sorted_itr, key=lambda line: line.rsplit("\t", 1)[1].replace("#", "@", 1).split("@", 1)[0]
134 ):
135 snapshots: list[str] = list(group) # fetch all snapshots of current dataset, e.g. dataset=tank1/src/foo
136 tmp_bookmarks: list[str] = []
137 if list_bookmarks: # temporary bookmarks bypass include/exclude filters, e.g. name and rank filters
138 tmp_bookmarks = [snapshot for snapshot in snapshots if is_tmp_bookmark(snapshot)]
139 snapshots = [snapshot for snapshot in snapshots if not is_tmp_bookmark(snapshot)]
140 non_tmp_guids: set[str] = {snapshot.split("\t", 2)[1] for snapshot in snapshots} # for subsequent dedupe
141 tmp_bookmarks = [bookmark for bookmark in tmp_bookmarks if bookmark.split("\t", 2)[1] not in non_tmp_guids]
142 del non_tmp_guids # help gc
143 snapshots = filter_snapshots(job, snapshots, filter_bookmarks=True) # apply include/exclude policy
144 snapshots += tmp_bookmarks
145 snapshots.sort(key=lambda line: line.split("\t", 2)[1]) # stable sort by GUID (2nd remains createtxg)
146 rel_dataset: str = relativize_dataset(dataset, root_dataset) # rel_dataset=/foo, root_dataset=tank1/src
147 last_guid: str = ""
148 for line in snapshots:
149 cols: list[str] = line.split("\t")
150 _creation, guid, _createtxg, _written, snapshot_name = cols
151 if guid == last_guid:
152 assert "#" in snapshot_name
153 continue # ignore bookmarks whose snapshot still exists. also ignore dupes of bookmarks
154 last_guid = guid
155 key = (rel_dataset, guid) # ensures src snapshots and dst snapshots with the same GUID will be adjacent
156 yield _ComparableSnapshot(key, cols)
158 def print_dataset(rel_dataset: str, entries: Iterable[tuple[str, _ComparableSnapshot]]) -> None:
159 entries = sorted( # fetch all snapshots of current dataset and sort em by creation, createtxg, snapshot_tag
160 entries,
161 key=lambda entry: (
162 int((cols := entry[1].cols)[0]), # creation
163 int(cols[2]), # createtxg
164 cols[-1].replace("#", "@", 1).split("@", 1)[1], # snapshot_tag
165 ),
166 )
168 @dataclass
169 @final
170 class SnapshotStats:
171 snapshot_count: int = 0
172 sum_written: int = 0
173 snapshot_count_since: int = 0
174 sum_written_since: int = 0
175 latest_snapshot_idx: int | None = None
176 latest_snapshot_row_str: str | None = None
177 latest_snapshot_creation: str | None = None
178 oldest_snapshot_row_str: str | None = None
179 oldest_snapshot_creation: str | None = None
181 # print metadata of snapshots of current dataset to TSV file; custom stats can later be computed from there
182 stats: defaultdict[str, SnapshotStats] = defaultdict(SnapshotStats)
183 header: str = "location creation_iso createtxg rel_name guid root_dataset rel_dataset name creation written"
184 nonlocal is_first_row
185 if is_first_row:
186 fd.write(header.replace(" ", "\t") + "\n")
187 is_first_row = False
188 for i, entry in enumerate(entries): # entry is tuple[location:str, ComparableSnapshot]
189 location: str = entry[0] # "src" or "dst" or "all"
190 creation, guid, createtxg, written, name = entry[1].cols
191 root_dataset: str = dst.root_dataset if location == CMP_CHOICES_ITEMS[1] else src.root_dataset
192 rel_name: str = relativize_dataset(name, root_dataset)
193 creation_iso: str = isotime_from_unixtime(int(creation))
194 row = (location, creation_iso, createtxg, rel_name, guid, root_dataset, rel_dataset, name, creation, written)
195 # 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
196 row_str = "\t".join(row)
197 if not p.dry_run:
198 fd.write(row_str + "\n")
199 s = stats[location]
200 s.snapshot_count += 1
201 s.sum_written += int(written) if written != "-" else 0
202 s.latest_snapshot_idx = i
203 s.latest_snapshot_row_str = row_str
204 s.latest_snapshot_creation = creation
205 if not s.oldest_snapshot_row_str:
206 s.oldest_snapshot_row_str = row_str
207 s.oldest_snapshot_creation = creation
209 # for convenience, directly log basic summary stats of current dataset
210 k = stats["all"].latest_snapshot_idx # defaults to None
211 k = k if k is not None else -1
212 for entry in entries[k + 1 :]: # aggregate basic stats since latest common snapshot
213 location = entry[0]
214 creation, guid, createtxg, written, name = entry[1].cols
215 s = stats[location]
216 s.snapshot_count_since += 1
217 s.sum_written_since += int(written) if written != "-" else 0
218 prefix: str = f"Comparing {rel_dataset}~"
219 msgs: list[str] = []
220 msgs.append(f"{prefix} of {task}")
221 msgs.append(
222 f"{prefix} Q: No src snapshots are missing on dst, and no dst snapshots are missing on src, "
223 "and there is a common snapshot? A: "
224 + (
225 "n/a"
226 if not is_src_dst_all
227 else str(
228 stats["src"].snapshot_count == 0 and stats["dst"].snapshot_count == 0 and stats["all"].snapshot_count > 0
229 )
230 )
231 )
232 nonlocal now
233 now = now or int(time.time()) # ZFS truncates fractional secs in 'creation' so here we do, too. And keep now stable
234 latcom = "latest common snapshot"
235 for loc in all_src_dst:
236 s = stats[loc]
237 msgs.append(f"{prefix} Latest snapshot only in {loc}: {s.latest_snapshot_row_str or 'n/a'}")
238 msgs.append(f"{prefix} Oldest snapshot only in {loc}: {s.oldest_snapshot_row_str or 'n/a'}")
239 msgs.append(f"{prefix} Snapshots only in {loc}: {s.snapshot_count}")
240 msgs.append(f"{prefix} Snapshot data written only in {loc}: {human_readable_bytes(s.sum_written)}")
241 if loc != "all":
242 na = None if k >= 0 else "n/a"
243 msgs.append(f"{prefix} Snapshots only in {loc} since {latcom}: {na or s.snapshot_count_since}")
244 msgs.append(
245 f"{prefix} Snapshot data written only in {loc} since {latcom}: "
246 f"{na or human_readable_bytes(s.sum_written_since)}"
247 )
248 all_creation = stats["all"].latest_snapshot_creation
249 latest = ("latest", s.latest_snapshot_creation)
250 oldest = ("oldest", s.oldest_snapshot_creation)
251 for label, s_creation in latest, oldest:
252 if loc != "all":
253 hd = "n/a"
254 if s_creation and k >= 0:
255 assert all_creation is not None
256 hd = human_readable_duration(int(all_creation) - int(s_creation), unit="s")
257 msgs.append(f"{prefix} Time diff between {latcom} and {label} snapshot only in {loc}: {hd}")
258 for label, s_creation in latest, oldest:
259 hd = "n/a" if not s_creation else human_readable_duration(now - int(s_creation), unit="s")
260 msgs.append(f"{prefix} Time diff between now and {label} snapshot only in {loc}: {hd}")
261 log.info("%s", "\n".join(msgs))
263 # setup streaming pipeline
264 src_snapshot_itr: Iterator = snapshot_iterator(src, src.root_dataset, zfs_list_snapshot_iterator(src, src_datasets))
265 dst_snapshot_itr: Iterator = snapshot_iterator(dst, dst.root_dataset, zfs_list_snapshot_iterator(dst, dst_datasets))
266 merge_itr = _merge_sorted_iterators(CMP_CHOICES_ITEMS, p.compare_snapshot_lists, src_snapshot_itr, dst_snapshot_itr)
268 interner: HashedInterner[str] = HashedInterner() # reduces memory footprint
269 rel_datasets: dict[str, set[str]] = defaultdict(set)
270 for datasets, remote in (src_datasets, src), (dst_datasets, dst):
271 for dataset in datasets: # rel_dataset=/foo, root_dataset=tank1/src
272 rel_datasets[remote.location].add(interner.intern(relativize_dataset(dataset, remote.root_dataset)))
273 rel_src_or_dst: list[str] = sorted(rel_datasets["src"].union(rel_datasets["dst"]))
275 log.debug("%s", f"Temporary TSV output file comparing {task} is: {tmp_tsv_file}")
276 with open_nofollow(tmp_tsv_file, "w", encoding="utf-8", perm=FILE_PERMISSIONS) as fd:
277 # streaming group by rel_dataset (consumes constant memory only); entry is a Tuple[str, ComparableSnapshot]
278 groups = itertools.groupby(merge_itr, key=lambda entry: entry[1].key[0])
279 _print_datasets(groups, lambda rel_ds, entries: print_dataset(rel_ds, entries), rel_src_or_dst)
280 os.rename(tmp_tsv_file, tsv_file)
281 log.info("%s", f"Final TSV output file comparing {task} is: {tsv_file}")
283 tsv_file = tsv_file[: tsv_file.rindex(".")] + ".rel_datasets_tsv"
284 tmp_tsv_file = tsv_file + ".tmp"
285 with open_nofollow(tmp_tsv_file, "w", encoding="utf-8", perm=FILE_PERMISSIONS) as fd:
286 header: str = "location rel_dataset src_dataset dst_dataset"
287 fd.write(header.replace(" ", "\t") + "\n")
288 src_only: set[str] = rel_datasets["src"].difference(rel_datasets["dst"])
289 dst_only: set[str] = rel_datasets["dst"].difference(rel_datasets["src"])
290 for rel_dataset in rel_src_or_dst:
291 loc = "src" if rel_dataset in src_only else "dst" if rel_dataset in dst_only else "all"
292 src_dataset = src.root_dataset + rel_dataset if rel_dataset not in dst_only else ""
293 dst_dataset = dst.root_dataset + rel_dataset if rel_dataset not in src_only else ""
294 row = (loc, rel_dataset, src_dataset, dst_dataset) # Example: all /foo/bar tank1/src/foo/bar tank2/dst/foo/bar
295 if not p.dry_run:
296 fd.write("\t".join(row) + "\n")
297 os.rename(tmp_tsv_file, tsv_file)
300def _print_datasets(groups: itertools.groupby, fn: Callable[[str, Iterable], None], rel_datasets: Iterable[str]) -> None:
301 """Iterate over grouped datasets and apply fn, adding gaps for missing ones."""
302 rel_datasets = sorted(rel_datasets)
303 n = len(rel_datasets)
304 i = 0
305 for rel_dataset, entries in groups:
306 while i < n and rel_datasets[i] < rel_dataset:
307 fn(rel_datasets[i], []) # Also print summary stats for datasets whose snapshot stream is empty
308 i += 1
309 assert i >= n or rel_datasets[i] == rel_dataset
310 i += 1
311 fn(rel_dataset, entries)
312 while i < n:
313 fn(rel_datasets[i], []) # Also print summary stats for datasets whose snapshot stream is empty
314 i += 1
317def _merge_sorted_iterators(
318 choices: Sequence[str], # ["src", "dst", "all"]
319 choice: str, # Example: "src+dst+all"
320 src_itr: Iterator[_ComparableSnapshot],
321 dst_itr: Iterator[_ComparableSnapshot],
322) -> Iterator[tuple[str, _ComparableSnapshot] | tuple[str, _ComparableSnapshot, _ComparableSnapshot]]:
323 """The typical pipelined merge algorithm of a merge sort, slightly adapted to our specific use case."""
324 assert len(choices) == 3
325 assert choice
326 flags: int = 0
327 for i, item in enumerate(choices):
328 if item in choice:
329 flags |= 1 << i
330 src_next, dst_next = run_in_parallel(lambda: next(src_itr, None), lambda: next(dst_itr, None))
331 while not (src_next is None and dst_next is None):
332 if src_next == dst_next:
333 n = 2
334 if (flags & (1 << n)) != 0:
335 assert src_next is not None
336 assert dst_next is not None
337 yield choices[n], src_next, dst_next
338 src_next = next(src_itr, None)
339 dst_next = next(dst_itr, None)
340 elif src_next is None or (dst_next is not None and dst_next < src_next):
341 n = 1
342 if (flags & (1 << n)) != 0:
343 assert dst_next is not None
344 yield choices[n], dst_next
345 dst_next = next(dst_itr, None)
346 else:
347 assert src_next is not None
348 n = 0
349 if (flags & (1 << n)) != 0 and not _is_comparable_tmp_bookmark(src_next): # suppress tmp bookmarks
350 yield choices[n], src_next
351 src_next = next(src_itr, None)
354def _is_comparable_tmp_bookmark(src_next: object) -> bool:
355 return isinstance(src_next, _ComparableSnapshot) and is_tmp_bookmark(src_next.cols[-1])