Coverage for bzfs_main/filter.py: 100%
226 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"""The filter algorithms that apply include/exclude policies are in filter_datasets() and filter_snapshots()."""
17from __future__ import (
18 annotations,
19)
20import math
21import os
22import re
23from collections.abc import (
24 Iterable,
25)
26from datetime import (
27 timedelta,
28)
29from typing import (
30 TYPE_CHECKING,
31 Final,
32 Optional,
33 Union,
34 cast,
35)
37from bzfs_main.util.utils import (
38 DONT_SKIP_DATASET,
39 LOG_DEBUG,
40 LOG_TRACE,
41 UNIX_TIME_INFINITY_SECS,
42 RegexList,
43 is_descendant,
44 is_included,
45 relativize_dataset,
46 sort_datasets,
47)
49if TYPE_CHECKING: # pragma: no cover - for type hints only
50 from bzfs_main.bzfs import (
51 Job,
52 )
53 from bzfs_main.configuration import (
54 Params,
55 Remote,
56 )
58# constants:
59SNAPSHOT_REGEX_FILTER_NAME: Final[str] = "snapshot_regex"
60SNAPSHOT_REGEX_FILTER_NAMES: Final[frozenset[str]] = frozenset({"include_snapshot_regex", "exclude_snapshot_regex"})
61SNAPSHOT_FILTERS_VAR: Final[str] = "snapshot_filters_var"
62_NEVER_MATCHES_REGEX: Final[str] = r"(?!)" # empty negative lookahead always fails, even on an empty string
65UnixTimeRange = Optional[tuple[Union[timedelta, int], Union[timedelta, int]]] # Type alias
66RankRange = tuple[tuple[str, int, bool], tuple[str, int, bool]] # Type alias
69def filter_datasets(job: Job, remote: Remote, sorted_datasets: list[str]) -> list[str]:
70 """Returns all datasets (and their descendants) that match at least one of the include regexes but none of the exclude
71 regexes.
73 Assumes the list of input datasets is sorted. The list of output datasets will be sorted too.
74 """
75 assert (not job.is_test_mode) or sorted_datasets == sorted(sorted_datasets), "List is not sorted"
76 p, log = job.params, job.params.log
77 results: list[str] = []
78 for i, dataset in enumerate(sorted_datasets):
79 if i == 0 and p.skip_parent:
80 continue
81 rel_dataset: str = relativize_dataset(dataset, remote.root_dataset)
82 if rel_dataset.startswith("/"):
83 rel_dataset = rel_dataset[1:] # strip leading '/' char if any
84 if is_included(rel_dataset, p.include_dataset_regexes, p.exclude_dataset_regexes):
85 results.append(dataset)
86 log.debug("Including b/c dataset regex: %s", dataset)
87 else:
88 log.debug("Excluding b/c dataset regex: %s", dataset)
89 if p.exclude_dataset_property:
90 results = _filter_datasets_by_exclude_property(job, remote, results)
91 is_debug: bool = p.log.isEnabledFor(LOG_DEBUG)
92 for dataset in results:
93 if is_debug:
94 log.debug(f"Finally included {remote.location} dataset: %s", dataset)
95 if job.is_test_mode:
96 assert results == sorted(results), "List is not sorted"
97 # Asserts the following: If a dataset is excluded its descendants are automatically excluded too, and this
98 # decision is never reconsidered even for the descendants because exclude takes precedence over include.
99 resultset: set[str] = set(results)
100 root_datasets: list[str] = [dataset for dataset in results if os.path.dirname(dataset) not in resultset] # no parent
101 for root in root_datasets: # each root is not a descendant of another dataset
102 assert not any(is_descendant(root, of_root_dataset=dataset) for dataset in results if dataset != root)
103 for dataset in results: # each dataset belongs to a subtree rooted at one of the roots
104 assert any(is_descendant(dataset, of_root_dataset=root) for root in root_datasets)
105 return results
108def _filter_datasets_by_exclude_property(job: Job, remote: Remote, sorted_datasets: list[str]) -> list[str]:
109 """Excludes datasets that are marked with a ZFS user property value that, in effect, says 'skip me'."""
110 p, log = job.params, job.params.log
111 results: list[str] = []
112 localhostname: str | None = None
113 skip_dataset: str = DONT_SKIP_DATASET
114 for dataset in sort_datasets(sorted_datasets):
115 if is_descendant(dataset, of_root_dataset=skip_dataset):
116 # skip_dataset shall be ignored or has been deleted by some third party while we're running
117 continue # nothing to do anymore for this dataset subtree (note that datasets is sorted)
118 skip_dataset = DONT_SKIP_DATASET
119 # TODO perf: on zfs >= 2.3 use json via zfs list -j to safely merge all zfs list's into one 'zfs list' call
120 cmd = p.split_args(f"{p.zfs_program} list -t filesystem,volume -Hp -o {p.exclude_dataset_property}", dataset)
121 job.maybe_inject_delete(remote, dataset=dataset, delete_trigger="zfs_list_exclude_property")
122 property_value: str | None = job.try_ssh_command_with_retries(remote, LOG_TRACE, cmd=cmd)
123 if property_value is None:
124 log.warning(f"Third party deleted {remote.location}: %s", dataset)
125 skip_dataset = dataset
126 else:
127 reason: str = ""
128 property_value = property_value.strip()
129 sync: bool
130 if not property_value or property_value == "-" or property_value.lower() == "true":
131 sync = True
132 elif property_value.lower() == "false":
133 sync = False
134 else:
135 import socket # lazy import for startup perf
137 localhostname = localhostname or socket.gethostname()
138 sync = any(localhostname == hostname.strip() for hostname in property_value.split(","))
139 reason = f", localhostname: {localhostname}, hostnames: {property_value}"
141 if sync:
142 results.append(dataset)
143 log.debug("Including b/c dataset prop: %s%s", dataset, reason)
144 else:
145 skip_dataset = dataset
146 log.debug("Excluding b/c dataset prop: %s%s", dataset, reason)
147 return sorted(results)
150def filter_snapshots(
151 job: Job, basis_snapshots: list[str], all_except: bool = False, filter_bookmarks: bool = False
152) -> list[str]:
153 """Returns all snapshots that pass all include/exclude policies.
155 Semantics: Within a single snapshot-filter group, filters are applied sequentially (logical AND). Across groups,
156 results are union-ized (logical OR). Set `all_except=True` to invert the final selection (retain-selected vs
157 delete-selected modes). Bookmarks: when `filter_bookmarks=False`, bookmark entries (with '#') are always retained to
158 assist common-snapshot detection; when `True`, bookmarks are subject to the same filters as snapshots.
159 """
161 def resolve_timerange(timerange: UnixTimeRange) -> UnixTimeRange:
162 """Converts relative timerange values to UTC Unix time in integer seconds."""
163 assert timerange is not None
164 lo, hi = timerange
165 if isinstance(lo, timedelta):
166 lo = math.ceil(current_unixtime_in_secs - lo.total_seconds())
167 if isinstance(hi, timedelta):
168 hi = math.ceil(current_unixtime_in_secs - hi.total_seconds())
169 assert isinstance(lo, int)
170 assert isinstance(hi, int)
171 return (lo, hi) if lo <= hi else (hi, lo)
173 p, log = job.params, job.params.log
174 current_unixtime_in_secs: float = p.create_src_snapshots_config.current_datetime.timestamp()
175 resultset: set[str] = set()
176 for snapshot_filter in p.snapshot_filters:
177 snapshots: list[str] = basis_snapshots
178 for _filter in snapshot_filter:
179 name: str = _filter.name
180 if name == SNAPSHOT_REGEX_FILTER_NAME:
181 snapshots = _filter_snapshots_by_regex(
182 job,
183 snapshots,
184 regexes=cast(tuple[RegexList, RegexList], _filter.options),
185 filter_bookmarks=filter_bookmarks,
186 )
187 elif name == "include_snapshot_times":
188 timerange = resolve_timerange(_filter.timerange) if _filter.timerange is not None else _filter.timerange
189 snapshots = _filter_snapshots_by_creation_time(
190 job, snapshots, include_snapshot_times=timerange, filter_bookmarks=filter_bookmarks
191 )
192 else:
193 assert name == "include_snapshot_times_and_ranks"
194 timerange = resolve_timerange(_filter.timerange) if _filter.timerange is not None else _filter.timerange
195 snapshots = _filter_snapshots_by_creation_time_and_rank(
196 job,
197 snapshots,
198 include_snapshot_times=timerange,
199 include_snapshot_ranks=cast(list[RankRange], _filter.options),
200 filter_bookmarks=filter_bookmarks,
201 )
202 resultset.update(snapshots) # union
204 no_f_bookmarks: bool = not filter_bookmarks
205 snapshots = [line for line in basis_snapshots if (no_f_bookmarks and "#" in line) or ((line in resultset) != all_except)]
206 is_debug: bool = log.isEnabledFor(LOG_DEBUG)
207 for snapshot in snapshots:
208 if is_debug:
209 log.debug("Finally included snapshot: %s", snapshot[snapshot.rindex("\t") + 1 :])
210 return snapshots
213def _filter_snapshots_by_regex(
214 job: Job, snapshots: list[str], regexes: tuple[RegexList, RegexList], filter_bookmarks: bool = False
215) -> list[str]:
216 """Returns all snapshots that match at least one of the include regexes but none of the exclude regexes.
218 Precondition: Each line is TSV of the form ...guid\tname. Regexes are applied to the snapshot or bookmark tag portion
219 of `name` (after '@' or, if `filter_bookmarks=True`, after '#').
220 """
221 exclude_snapshot_regexes, include_snapshot_regexes = regexes
222 log = job.params.log
223 is_debug: bool = log.isEnabledFor(LOG_DEBUG)
224 results: list[str] = []
225 for snapshot in snapshots:
226 i = snapshot.find("@") # snapshot separator
227 if i < 0 and filter_bookmarks:
228 i = snapshot.index("#") # bookmark separator
229 if i < 0:
230 continue # retain bookmarks to help find common snapshots, apply filter only to snapshots
231 elif is_included(snapshot[i + 1 :], include_snapshot_regexes, exclude_snapshot_regexes):
232 results.append(snapshot)
233 if is_debug:
234 log.debug("Including b/c snapshot regex: %s", snapshot[snapshot.rindex("\t") + 1 :])
235 else:
236 if is_debug:
237 log.debug("Excluding b/c snapshot regex: %s", snapshot[snapshot.rindex("\t") + 1 :])
238 return results
241def _filter_snapshots_by_creation_time(
242 job: Job, snapshots: list[str], include_snapshot_times: UnixTimeRange, filter_bookmarks: bool = False
243) -> list[str]:
244 """Filters snapshots to those created within the specified time window.
246 Precondition: Each line is TSV of the form creation\t...\tname. The creation column (first field) is compared against
247 [lo, hi). Bookmarks are skipped unless `filter_bookmarks=True`.
248 """
249 log = job.params.log
250 is_debug: bool = log.isEnabledFor(LOG_DEBUG)
251 lo_snaptime, hi_snaptime = include_snapshot_times or (0, UNIX_TIME_INFINITY_SECS)
252 assert isinstance(lo_snaptime, int)
253 assert isinstance(hi_snaptime, int)
254 results: list[str] = []
255 for snapshot in snapshots:
256 if (not filter_bookmarks) and "@" not in snapshot:
257 continue # retain bookmarks to help find common snapshots, apply filter only to snapshots
258 elif lo_snaptime <= int(snapshot[: snapshot.index("\t")]) < hi_snaptime:
259 results.append(snapshot)
260 if is_debug:
261 log.debug("Including b/c creation time: %s", snapshot[snapshot.rindex("\t") + 1 :])
262 else:
263 if is_debug:
264 log.debug("Excluding b/c creation time: %s", snapshot[snapshot.rindex("\t") + 1 :])
265 return results
268def _filter_snapshots_by_creation_time_and_rank(
269 job: Job,
270 snapshots: list[str],
271 include_snapshot_times: UnixTimeRange,
272 include_snapshot_ranks: list[RankRange],
273 filter_bookmarks: bool = False,
274) -> list[str]:
275 """Filters by creation time and rank within the snapshot list.
277 Precondition: Each line is TSV of the form creation\t...\tname. The creation column (first field) is compared against
278 [lo, hi). Bookmarks are skipped unless `filter_bookmarks=True`.
279 """
281 def get_idx(rank: tuple[str, int, bool], n: int) -> int:
282 """Returns index for rank tuple (kind, value, percent)."""
283 kind, num, is_percent = rank
284 m = round(n * num / 100) if is_percent else min(n, num)
285 assert kind == "latest" or kind == "oldest"
286 return m if kind == "oldest" else n - m
288 assert isinstance(include_snapshot_ranks, list)
289 assert len(include_snapshot_ranks) > 0
290 log = job.params.log
291 is_debug: bool = log.isEnabledFor(LOG_DEBUG)
292 lo_time, hi_time = include_snapshot_times or (0, UNIX_TIME_INFINITY_SECS)
293 assert isinstance(lo_time, int)
294 assert isinstance(hi_time, int)
295 n: int = sum(1 for snapshot in snapshots if "@" in snapshot)
296 for rank_range in include_snapshot_ranks:
297 lo_rank, hi_rank = rank_range
298 lo: int = get_idx(lo_rank, n)
299 hi: int = get_idx(hi_rank, n)
300 lo, hi = (lo, hi) if lo <= hi else (hi, lo)
301 i: int = 0
302 k: int = 0
303 results: list[str] = []
304 for snapshot in snapshots:
305 is_snapshot = "@" in snapshot
306 if (not filter_bookmarks) and not is_snapshot:
307 continue # retain bookmarks to help find common snapshots, apply filter only to snapshots
308 else:
309 msg = ""
310 if is_snapshot and lo <= i < hi:
311 msg = "Including b/c snapshot rank: %s"
312 elif lo_time <= int(snapshot[: snapshot.index("\t")]) < hi_time:
313 msg = "Including b/c creation time: %s"
314 if msg:
315 results.append(snapshot)
316 if is_snapshot:
317 k += 1
318 else:
319 msg = "Excluding b/c snapshot rank: %s"
320 if is_debug:
321 log.debug(msg, snapshot[snapshot.rindex("\t") + 1 :])
322 if is_snapshot:
323 i += 1
324 snapshots = results
325 n = k
326 return snapshots
329def filter_properties(
330 p: Params, props: dict[str, str | None], include_regexes: RegexList, exclude_regexes: RegexList
331) -> dict[str, str | None]:
332 """Returns ZFS props whose name matches at least one of the include regexes but none of the exclude regexes."""
333 log = p.log
334 is_debug: bool = log.isEnabledFor(LOG_DEBUG)
335 results: dict[str, str | None] = {}
336 for propname, propvalue in props.items():
337 if is_included(propname, include_regexes, exclude_regexes):
338 results[propname] = propvalue
339 if is_debug:
340 log.debug("Including b/c property regex: %s", propname)
341 else:
342 if is_debug:
343 log.debug("Excluding b/c property regex: %s", propname)
344 return results
347def filter_lines(input_list: Iterable[str], input_set: set[str]) -> list[str]:
348 """For each line in input_list, includes the line if input_set contains the first column field of that line."""
349 if len(input_set) == 0:
350 return []
351 return [line for line in input_list if line[: line.index("\t")] in input_set]
354def filter_lines_except(input_list: list[str], input_set: set[str]) -> list[str]:
355 """For each line in input_list, includes the line if input_set does not contain the first column field of that line."""
356 if len(input_set) == 0:
357 return input_list
358 return [line for line in input_list if line[: line.index("\t")] not in input_set]
361def dataset_regexes(src: Remote, dst: Remote, datasets: list[str]) -> list[str]:
362 """Converts dataset paths to regex strings relative to src or dst roots."""
363 results: list[str] = []
364 for dataset in datasets:
365 if dataset.startswith("/"):
366 # it's an absolute dataset - convert it to a relative dataset
367 dataset = dataset[1:]
368 if is_descendant(dataset, of_root_dataset=src.root_dataset):
369 dataset = relativize_dataset(dataset, src.root_dataset)
370 elif is_descendant(dataset, of_root_dataset=dst.root_dataset):
371 dataset = relativize_dataset(dataset, dst.root_dataset)
372 else:
373 results.append(_NEVER_MATCHES_REGEX)
374 continue
375 if dataset.startswith("/"):
376 dataset = dataset[1:]
377 if dataset.endswith("/"):
378 dataset = dataset[0:-1]
379 regex: str
380 if dataset:
381 regex = re.escape(dataset)
382 else:
383 regex = ".*"
384 results.append(regex)
385 return results