diff --git a/AGENT.md b/AGENT.md index 1942923..d1270e2 100644 --- a/AGENT.md +++ b/AGENT.md @@ -232,6 +232,19 @@ Cast direction is values → column (the small array to the big one's type), not `MILLPOND_FILTER_DROP_FIELD_NAME` is reserved at the config layer (mutex with keep, both empty or exactly one set) and explicitly rejected at startup. The denylist implementation lives in a future commit; the namespace is locked today so that change doesn't require operator env-var churn. +**Include-values source** (`include_values.py`) supplies the allowlist `_apply_filter` reads — the values are now an explicit third argument, fetched per batch via `include_source.current()`, not read off cfg. `build(cfg)` returns exactly ONE object (`StaticIncludeValues` | `ShadowIncludeValues` | `HttpIncludeValues`, distinguished by `.mode`); main starts/stops it around the consume loop and `metrics.include_values_mode{mode}` reports which one actually runs (fleet queries must not pass vacuously on replicas that never got the URL). + +Design spine is the consequence asymmetry: erroneous addition = surplus rows downstream ignores; erroneous removal = silent unrecoverable drop. The invariants, all tested in `tests/unit/test_include_values.py`: + +- Additions apply on first sight; removals need `removal_confirm_polls` consecutive successful ACCEPTED polls absent. Failed polls and refused polls advance nothing — refusal must never pre-charge the countdown (N refused-empty polls + one junk value would otherwise mass-remove instantly). +- Refusal guards (whole poll rejected, `include_values_refused_total{reason}`): `empty` (an empty array is never removal evidence and never an acceptable first state), `bulk_removal` (> half of a multi-value set confirmed at once — world-replacement goes through static config, not an unattended poll), `type_flip` (int↔str: one junk element flips `_normalize` to strings and the filter's values→column cast then fails, dropping ENTIRE batches as `filter_field_missing` with offsets still committed — the worst silent-loss path in the module). +- Authoritative `start()` BLOCKS until the first successful poll and raises on timeout — no proceed-on-bootstrap. The damping state is in-memory; a restart that silently ran on a stale static bootstrap would perform de-facto removals with zero polls of evidence. Halting is recoverable from Kafka; a silent drop is not. +- Shadow mode: `current()` serves the static set; the prober is a private attribute precisely so reading the polled set as authority requires reaching into `_private`, not a one-line mistake. +- Thread-safety: `current()` reads an immutable tuple swapped by attribute assignment (single writer = the poll thread); everything else is poll-thread-only. No locks. +- `_fetch` refuses redirects (urllib re-sends the auth header cross-host otherwise), caps the response at 4 MB, and jitters the poll ±10% so replica fleets don't herd. + +Config validation (`_load_include_values_config` in `config.py`) refuses at startup: URL without keep-filter, MODE/auth without URL (an intended dynamic source silently degrading to static is a removal-by-omission), lone auth halves, non-numeric knobs. The URL⇒keep-field⇒non-empty-static-values chain is what makes the no-bootstrap path unreachable through config — `test_keep_filter_chain_guarantees_bootstrap` pins it. + **Sort** (`_apply_sort` in `main.py`) runs inside `_flush()` after `pa.concat_tables` but before `sink.write()`. The sink sees pre-sorted data; sink-side partition columns (year/month/day/hour, computed by the ducklake extension) are not in scope by design — operators specify sort keys against the source schema. Missing-field handling: if any `cfg.sort_by` field is absent from the batch, the whole sort is skipped (rather than partially sorting on available keys, which would silently differ from intent). Records still flow through unsorted. The metric is `sort_skipped_total{reason="field_missing"}`, deliberately distinct from `records_skipped_total` because no data is being dropped — only the layout improvement is. diff --git a/README.md b/README.md index f67f4ee..65ab452 100644 --- a/README.md +++ b/README.md @@ -79,6 +79,30 @@ Two skip reasons are tracked on `millpond_records_skipped_total`: `MILLPOND_FILTER_DROP_FIELD_NAME` is reserved at the config layer (mutex with keep) and currently rejected at startup. It will become a denylist filter in a future release without env-var churn. +### Dynamic allowlist source + +The allowlist can be sourced at runtime from an HTTP endpoint instead of being fixed at startup: `MILLPOND_INCLUDE_VALUES_URL` names a URL returning a JSON array of scalars (ints or strings, matching the static list's type), polled on a background thread (default every 60s, ±10% jitter). Millpond knows nothing about the endpoint's meaning — the URL and an optional auth header (`MILLPOND_INCLUDE_VALUES_AUTH_HEADER_NAME` + `_AUTH_TOKEN`) are plain config. + +How the static list and the polled set interact: + +| `MILLPOND_FILTER_VALUES` (static) | `MILLPOND_INCLUDE_VALUES_URL` | `MILLPOND_INCLUDE_VALUES_MODE` | Effective allowlist | Endpoint's role | +|---|---|---|---|---| +| set | unset | unset | the static list | none — today's behavior, unchanged | +| set | set | `shadow` (default) | the static list | observability only: polled each interval, exports diff-vs-static gauges and staleness; its values are never applied | +| set | set | `authoritative` | the polled set | live: the static list seeds the initial held set; startup **blocks** until the first successful poll (no proceed-on-stale-bootstrap) | +| unset | unset | unset | no filter — all records kept | — | +| unset | set | any | **startup error** | the URL requires an active keep-filter, which requires static values | +| any | unset | set (or auth vars set) | **startup error** | MODE/auth without a URL means a dynamic source was intended; refusing beats silently running static-only | + +In `authoritative` mode the polled set changes under safety rules shaped by a consequence asymmetry — an erroneous addition writes surplus rows, an erroneous removal silently drops records with no recovery: + +- **Additions** apply on the first successful poll that shows them. +- **Removals** require `MILLPOND_INCLUDE_VALUES_REMOVAL_POLLS` (default 5) *consecutive successful* polls with the value absent. Failed polls freeze the countdown; a reappearing value resets it. +- **Poll failures** keep the last-known-good set indefinitely; staleness is observable via `millpond_include_values_last_success_timestamp_seconds`. +- **Refused polls** (counted on `millpond_include_values_refused_total{reason}`) keep the set and advance nothing: empty arrays (`empty` — never removal evidence, never an acceptable first state), removals of more than half of a multi-value set at once (`bulk_removal`), and int↔str type changes (`type_flip` — a type-flipped set would fail the filter's cast against the column and drop whole batches). + +Rollout is designed to be shadow-first: run `shadow`, alert on `millpond_include_values_shadow_only_static` / `_shadow_only_remote` staying nonzero, and flip to `authoritative` once the symmetric difference holds at zero. `millpond_include_values_mode` reports which mode each replica actually runs, so a fleet-level flip gate can't pass vacuously on a replica that never got the URL. + ### Pre-write sort Sorts the consolidated batch by one or more columns ascending, right before `sink.write()`. The sink sees pre-sorted data, which improves Parquet compression (especially for low-cardinality keys like `team_id`) and downstream reader predicate pushdown. @@ -222,6 +246,14 @@ See [Record Handling](#record-handling) for context. All four variables below ar | `MILLPOND_FILTER_KEEP_FIELD_NAME` | no | | Column name to check against the allowlist. Must be set with `MILLPOND_FILTER_VALUES`. Validated as a safe identifier. | | `MILLPOND_FILTER_DROP_FIELD_NAME` | no | | Reserved for a future denylist filter; setting it today raises at startup. Mutually exclusive with `MILLPOND_FILTER_KEEP_FIELD_NAME`. | | `MILLPOND_FILTER_VALUES` | no | | Comma-separated allowed values. Auto-detected as int if every token parses as an integer, string otherwise. Required when either filter field name is set. | +| `MILLPOND_INCLUDE_VALUES_URL` | no | | HTTP endpoint returning a JSON array of allowlist values (see [Dynamic allowlist source](#dynamic-allowlist-source)). Requires the keep-filter to be configured. | +| `MILLPOND_INCLUDE_VALUES_MODE` | no | `shadow` | `shadow` (static authoritative, endpoint observed for diff metrics) or `authoritative` (polled set live). Only valid with the URL set. | +| `MILLPOND_INCLUDE_VALUES_POLL_INTERVAL_S` | no | `60` | Poll cadence, jittered ±10%. | +| `MILLPOND_INCLUDE_VALUES_REMOVAL_POLLS` | no | `5` | Consecutive successful polls a value must be absent before removal. `1` disables damping (warned at startup). | +| `MILLPOND_INCLUDE_VALUES_REQUEST_TIMEOUT_S` | no | `10` | Per-request HTTP timeout. | +| `MILLPOND_INCLUDE_VALUES_STARTUP_TIMEOUT_S` | no | `60` | Authoritative mode: how long startup blocks for the first successful poll before failing the pod. | +| `MILLPOND_INCLUDE_VALUES_AUTH_HEADER_NAME` | no | | Header name sent with each poll (e.g. an internal-secret header). Must be set together with the token. Redirects are refused so the header can't leak cross-host. | +| `MILLPOND_INCLUDE_VALUES_AUTH_TOKEN` | no | | Header value. Must be set together with the header name. | | `MILLPOND_SORT_BY` | no | | Comma-separated column names; the batch is sorted ascending by these in tuple order before each write. Missing fields cause the sort to be skipped (records still flow). | | `MILLPOND_TYPED_COLUMNS` | no | | Comma-separated `column:type` pairs pinning columns to a target type before write (types: `timestamptz`, `bigint`, `double`, `boolean`, `varchar`). Needed when writing into a table whose columns are already typed and JSON inference would diverge (date-times → `VARCHAR` vs `TIMESTAMPTZ`; all-null `project_id` → `VARCHAR` vs `BIGINT`). Column names validated as safe identifiers; types validated against the allowlist. | diff --git a/millpond/config.py b/millpond/config.py index 370c92f..94d0f8d 100644 --- a/millpond/config.py +++ b/millpond/config.py @@ -77,6 +77,22 @@ class Config: filter_drop_field: str | None filter_values: tuple[int, ...] | tuple[str, ...] | None + # Optional dynamic source for the keep-filter's include set (see + # include_values.py). URL unset = today's static behavior. Mode + # "shadow" (default) polls and reports diff metrics while the static + # list stays authoritative; "authoritative" makes the polled set the + # live filter, with the static list as the bootstrap/fallback seed. + # The auth header is generic (name + token) — nothing here knows what + # the endpoint is. + include_values_url: str | None + include_values_mode: str + include_values_poll_interval_s: float + include_values_removal_polls: int + include_values_request_timeout_s: float + include_values_startup_timeout_s: float + include_values_auth_header_name: str | None + include_values_auth_token: str | None + # Optional pre-write sort. Tuple of column names; sort is ascending # in tuple order. Applied to the consolidated batch right before # sink.write(). None disables the sort entirely. @@ -213,6 +229,91 @@ def _load_filter_fields() -> tuple[str | None, str | None, tuple[int, ...] | tup return keep, None, _parse_filter_values(values_raw) +def _load_include_values_config( + filter_keep_field: str | None, + filter_values: tuple[int, ...] | tuple[str, ...] | None, +) -> dict: + """Read the MILLPOND_INCLUDE_VALUES_* group and validate it against the + static filter config. Startup-refusal beats runtime surprise: + + - a URL without an active keep-filter has nothing to feed; + - MODE (or auth) without a URL means the operator INTENDED a dynamic + source and a typo'd/missing URL would silently degrade to static — + refuse rather than run on the wrong set; + - shadow mode without static values has nothing to diff against; + - a lone auth header name or token is always a misconfiguration. + """ + url = os.environ.get("MILLPOND_INCLUDE_VALUES_URL", "").strip() or None + mode_raw = os.environ.get("MILLPOND_INCLUDE_VALUES_MODE", "").strip().lower() + mode = mode_raw or "shadow" + header_name = os.environ.get("MILLPOND_INCLUDE_VALUES_AUTH_HEADER_NAME", "").strip() or None + token = os.environ.get("MILLPOND_INCLUDE_VALUES_AUTH_TOKEN", "").strip() or None + + if url is None: + if mode_raw: + raise RuntimeError( + "MILLPOND_INCLUDE_VALUES_MODE is set but MILLPOND_INCLUDE_VALUES_URL is not — " + "a dynamic source was intended; refusing to silently run static-only" + ) + if header_name or token: + raise RuntimeError("MILLPOND_INCLUDE_VALUES_AUTH_* requires MILLPOND_INCLUDE_VALUES_URL") + return dict( + include_values_url=None, + include_values_mode="static", + include_values_poll_interval_s=60.0, + include_values_removal_polls=5, + include_values_request_timeout_s=10.0, + include_values_startup_timeout_s=60.0, + include_values_auth_header_name=None, + include_values_auth_token=None, + ) + + if mode not in ("shadow", "authoritative"): + raise RuntimeError(f"MILLPOND_INCLUDE_VALUES_MODE must be 'shadow' or 'authoritative', got {mode!r}") + if filter_keep_field is None: + raise RuntimeError("MILLPOND_INCLUDE_VALUES_URL requires MILLPOND_FILTER_KEEP_FIELD_NAME to be set") + if mode == "shadow" and filter_values is None: + raise RuntimeError("MILLPOND_INCLUDE_VALUES_MODE=shadow requires static MILLPOND_FILTER_VALUES to diff against") + if bool(header_name) != bool(token): + raise RuntimeError( + "MILLPOND_INCLUDE_VALUES_AUTH_HEADER_NAME and MILLPOND_INCLUDE_VALUES_AUTH_TOKEN must be set together" + ) + + def _parse_number(env_name: str, default: str, cast): + raw = os.environ.get(env_name, default) + try: + return cast(raw) + except ValueError: + raise RuntimeError(f"{env_name} must be a number, got {raw!r}") from None + + poll_interval = _parse_number("MILLPOND_INCLUDE_VALUES_POLL_INTERVAL_S", "60", float) + removal_polls = _parse_number("MILLPOND_INCLUDE_VALUES_REMOVAL_POLLS", "5", int) + request_timeout = _parse_number("MILLPOND_INCLUDE_VALUES_REQUEST_TIMEOUT_S", "10", float) + startup_timeout = _parse_number("MILLPOND_INCLUDE_VALUES_STARTUP_TIMEOUT_S", "60", float) + if poll_interval <= 0: + raise RuntimeError("MILLPOND_INCLUDE_VALUES_POLL_INTERVAL_S must be positive") + if removal_polls < 1: + raise RuntimeError("MILLPOND_INCLUDE_VALUES_REMOVAL_POLLS must be >= 1") + if removal_polls == 1: + log.warning( + "MILLPOND_INCLUDE_VALUES_REMOVAL_POLLS=1 disables removal damping — a single poll " + "omitting a value removes it immediately" + ) + if request_timeout <= 0 or startup_timeout <= 0: + raise RuntimeError("MILLPOND_INCLUDE_VALUES_*_TIMEOUT_S must be positive") + + return dict( + include_values_url=url, + include_values_mode=mode, + include_values_poll_interval_s=poll_interval, + include_values_removal_polls=removal_polls, + include_values_request_timeout_s=request_timeout, + include_values_startup_timeout_s=startup_timeout, + include_values_auth_header_name=header_name, + include_values_auth_token=token, + ) + + def _load_sort_by() -> tuple[str, ...] | None: """Parse MILLPOND_SORT_BY into a tuple of column names. @@ -417,6 +518,7 @@ def load() -> Config: filter_keep_field=filter_keep_field, filter_drop_field=filter_drop_field, filter_values=filter_values, + **_load_include_values_config(filter_keep_field, filter_values), sort_by=sort_by, typed_columns=typed_columns, kafka_config_overrides=kafka_overrides, diff --git a/millpond/include_values.py b/millpond/include_values.py new file mode 100644 index 0000000..e71d8fd --- /dev/null +++ b/millpond/include_values.py @@ -0,0 +1,380 @@ +"""Sources for the keep-filter's include-value set. + +`build(cfg)` returns ONE source object with `current()/start()/stop()`: + +- static (no URL configured): today's behavior — the set parsed from + MILLPOND_FILTER_VALUES at startup, immutable for the process lifetime. +- shadow (URL + mode=shadow): `current()` still serves the static set; + a PRIVATE background prober polls the URL and exports diff-vs-static + gauges so the authority flip can be gated on the symmetric difference + holding at zero. The prober's set is deliberately not exposed — + reading it as authority in shadow mode should require reaching into + private attributes, not a one-line mistake. +- authoritative (URL + mode=authoritative): `current()` serves the + polled set, seeded from the static values; startup BLOCKS until the + first successful poll (see `start()` — a halt is recoverable from + Kafka, a silent drop is not). + +The consequence asymmetry drives every safety choice: an erroneous +ADDITION writes surplus rows a downstream reader ignores; an erroneous +REMOVAL silently drops records on the floor with no recovery. Hence: + +- additions apply on the first successful poll that shows them; +- removals require `removal_confirm_polls` CONSECUTIVE successful polls + with the value absent; failed polls advance nothing; +- a REFUSED poll (below) advances nothing either — refusal must not + pre-charge the removal countdown; +- a poll failure keeps the last-known-good set, with staleness + observable via `millpond_include_values_last_success_timestamp_seconds`; +- a successful poll is REFUSED (set kept, `millpond_include_values_ + refused_total{reason}`) when it would (a) empty a non-empty set, + (b) seed an EMPTY initial set, (c) confirm removal of more than half + of a multi-value set at once, or (d) flip the set's scalar type + (int↔str) — a type flip would silently exclude every record at the + filter's cast site, which is a mass drop wearing a different hat; +- the damping state and last-known-good set are IN-MEMORY: a restart + falls back to the static bootstrap, which is why authoritative mode + refuses to start unsynced rather than running on a possibly-stale + bootstrap. + +Nothing in this module knows what the endpoint is or what the values +mean — the URL and auth header are plain config. +""" + +from __future__ import annotations + +import json +import logging +import random +import threading +import time +import urllib.request + +from millpond import metrics + +log = logging.getLogger(__name__) + +Values = tuple[int, ...] | tuple[str, ...] + +# Cap on the endpoint response body. The largest plausible legitimate +# payload (thousands of scalar values) is a few hundred KB; anything +# bigger is a misconfigured URL, and an unbounded read() is an OOM. +_MAX_RESPONSE_BYTES = 4 * 1024 * 1024 + +_LOG_CHANGES_INDIVIDUALLY_MAX = 10 + + +class _NoRedirect(urllib.request.HTTPRedirectHandler): + """Refuse redirects: urllib re-sends custom headers (our auth token) + to the redirect target, including cross-host. For a component that + takes an arbitrary URL plus a secret header, a 3xx is an error.""" + + def redirect_request(self, req, fp, code, msg, headers, newurl): + return None + + +_OPENER = urllib.request.build_opener(_NoRedirect()) + + +def _normalize(elements: list) -> Values: + """Normalize a JSON array into the homogeneous, deduplicated tuple + shape the filter expects — same rule as config._parse_filter_values: + an all-int array becomes tuple[int, ...], anything else becomes + tuple[str, ...]. (bool is an int subclass in Python; a JSON `true` + must not silently become 1, so bools force the string path.)""" + if any(isinstance(e, (dict, list)) for e in elements): + raise ValueError("include-values array must contain only scalars") + if all(isinstance(e, int) and not isinstance(e, bool) for e in elements): + return tuple(sorted(set(elements))) + return tuple(sorted({str(e) for e in elements})) + + +def _values_type(values) -> type | None: + return type(values[0]) if values else None + + +class StaticIncludeValues: + """The startup-parsed static set. `current()` never changes.""" + + mode = "static" + + def __init__(self, values: Values | None): + self._values = values + + def current(self) -> Values | None: + return self._values + + def start(self) -> None: # uniform lifecycle across sources + pass + + def stop(self) -> None: + pass + + +class HttpIncludeValues: + """Polls `url` for a JSON array of include values on a background + thread and maintains the current set under the refusal / damping / + keep-last-on-error semantics in the module docstring. + + `current()` is safe from any thread: the set is an immutable tuple + swapped by plain attribute assignment; all other state is touched + only by the poll thread. + """ + + mode = "authoritative" + + def __init__( + self, + *, + url: str, + poll_interval_s: float, + removal_confirm_polls: int, + auth_header: tuple[str, str] | None = None, + bootstrap: Values | None = None, + shadow_reference: Values | None = None, + request_timeout_s: float = 10.0, + startup_timeout_s: float = 60.0, + ): + self._url = url + self._poll_interval_s = poll_interval_s + self._removal_confirm_polls = removal_confirm_polls + self._auth_header = auth_header + self._shadow_reference = shadow_reference + self._request_timeout_s = request_timeout_s + self._startup_timeout_s = startup_timeout_s + + self._values: Values | None = bootstrap + # value -> consecutive successful-AND-accepted polls it has been + # absent from. Failed polls and refused polls advance nothing. + self._absent_polls: dict[int | str, int] = {} + self._synced = threading.Event() + self._stop = threading.Event() + self._thread: threading.Thread | None = None + + def current(self) -> Values | None: + return self._values + + def start(self) -> None: + """Start the poll thread and BLOCK until the first successful + poll, raising after `startup_timeout_s`. + + No proceed-on-bootstrap escape hatch, deliberately: the damping + state is in-memory, so a restart during an endpoint outage that + silently ran on the (possibly stale) bootstrap would perform + de-facto removals with zero polls of evidence. Refusing to start + halts ingestion — recoverable, alertable, and the data waits in + Kafka — where running on a stale set silently drops records. + """ + self._thread = threading.Thread(target=self._run, name="include-values-poll", daemon=True) + self._thread.start() + if not self._synced.wait(self._startup_timeout_s): + self.stop() + raise RuntimeError( + f"include-values source got no successful poll of {self._url} within {self._startup_timeout_s}s" + ) + + def stop(self) -> None: + self._stop.set() + if self._thread is not None: + self._thread.join(timeout=5.0) + + # -- poll loop --------------------------------------------------------- + + def _run(self) -> None: + while not self._stop.is_set(): + try: + self._poll_once() + except Exception as e: + metrics.include_values_poll_failures_total.inc() + log.warning("include-values poll failed (keeping last set): %s", e) + # ±10% jitter so a fleet of replicas deployed in phase doesn't + # poll the endpoint in lockstep. + self._stop.wait(self._poll_interval_s * random.uniform(0.9, 1.1)) + + def _fetch(self) -> list: + req = urllib.request.Request(self._url) + if self._auth_header is not None: + req.add_header(*self._auth_header) + with _OPENER.open(req, timeout=self._request_timeout_s) as resp: + body = resp.read(_MAX_RESPONSE_BYTES + 1) + if len(body) > _MAX_RESPONSE_BYTES: + raise ValueError(f"include-values response exceeds {_MAX_RESPONSE_BYTES} bytes") + parsed = json.loads(body) + if not isinstance(parsed, list): + raise ValueError(f"include-values endpoint returned {type(parsed).__name__}, expected a JSON array") + return parsed + + def _refuse(self, reason: str, detail: str) -> None: + metrics.include_values_refused_total.labels(reason=reason).inc() + log.error("include-values poll refused (%s): %s — keeping current set", reason, detail) + + def _poll_once(self) -> None: + remote = _normalize(self._fetch()) + remote_set = set(remote) + + current = self._values if self._values is not None else () + current_set = set(current) + + # Type-flip refusal. If the endpoint's scalar type disagrees with + # the held set's (ints vs strings), unioning them would produce a + # mixed set and — worse — a full flip makes the filter's value- + # array cast fail against the column and drop ENTIRE batches as + # filter_field_missing. A type change is an endpoint bug or a + # coordinated migration; either way it goes through static config, + # not an unattended poll. (First sight with no held set accepts + # whatever type the endpoint serves.) + held_type = _values_type(current) + remote_type = _values_type(remote) + if held_type is not None and remote_type is not None and held_type is not remote_type: + self._refuse( + "type_flip", + f"held set is {held_type.__name__}, endpoint served {remote_type.__name__}", + ) + self._record_success_metrics(remote_set) + return + + # Empty-remote refusal, unconditional: an empty array is never + # removal evidence and never an acceptable first state. Against a + # held set it must not even ADVANCE the absence countdown (a run + # of empty responses is an endpoint bug, and letting it pre-charge + # the counters would let one later junk value mass-remove); with + # no held set, accepting () would arm the filter with an empty + # include set and drop every record. Legitimately shrinking to + # zero values is an operator action through static config. + if not remote_set: + if current_set or self._values is None: + self._refuse("empty", "endpoint served an empty array") + if current_set: + self._record_success_metrics(remote_set) + return + + additions = remote_set - current_set + + # Damped removals — computed TENTATIVELY. The counters are only + # committed if this poll is accepted; a refused poll must not + # pre-charge the countdown (a run of refused-empty polls followed + # by one junk value would otherwise mass-remove instantly). + tentative_counts: dict[int | str, int] = {} + confirmed_removals: set = set() + for v in current_set - remote_set: + tentative_counts[v] = self._absent_polls.get(v, 0) + 1 + if tentative_counts[v] >= self._removal_confirm_polls: + confirmed_removals.add(v) + + candidate = (current_set | additions) - confirmed_removals + + if not candidate and current_set: + self._refuse("empty", f"result would empty a non-empty set ({len(current_set)} values)") + self._record_success_metrics(remote_set) + return + # Bulk-removal refusal: confirming removal of more than half of a + # multi-value set in one poll is a fleet-wide config wipe, not + # routine churn. (Removing 1 of 2 is allowed; the guard is against + # the endpoint replacing the world.) + if len(current_set) > 1 and len(confirmed_removals) > max(1, len(current_set) // 2): + self._refuse( + "bulk_removal", + f"would remove {len(confirmed_removals)} of {len(current_set)} values in one poll", + ) + self._record_success_metrics(remote_set) + return + + # Accepted: commit the countdown state and the new set. + self._absent_polls = {v: n for v, n in tentative_counts.items() if v not in confirmed_removals} + + if candidate != current_set or self._values is None: + new_values = _normalize(list(candidate)) + self._log_changes(additions, confirmed_removals) + for _ in additions: + metrics.include_values_changes_total.labels(action="add").inc() + for _ in confirmed_removals: + metrics.include_values_changes_total.labels(action="remove").inc() + self._values = new_values + + self._record_success_metrics(remote_set) + self._synced.set() + + def _log_changes(self, additions: set, removals: set) -> None: + # Individually at WARNING while small (rare, operator-relevant); + # summarized when large (e.g. first sync against a full endpoint) + # so one poll can't flood the log. + if len(additions) <= _LOG_CHANGES_INDIVIDUALLY_MAX: + for v in sorted(additions, key=str): + log.warning("include-values: adding %r", v) + elif additions: + log.warning( + "include-values: adding %d values (sample: %s)", + len(additions), + sorted(additions, key=str)[:_LOG_CHANGES_INDIVIDUALLY_MAX], + ) + for v in sorted(removals, key=str): + log.warning( + "include-values: removing %r (absent %d consecutive polls)", + v, + self._removal_confirm_polls, + ) + + def _record_success_metrics(self, remote_set: set) -> None: + metrics.include_values_size.set(len(self._values or ())) + metrics.include_values_pending_removals.set(len(self._absent_polls)) + metrics.include_values_last_success_timestamp_seconds.set(time.time()) + if self._shadow_reference is not None: + ref = set(self._shadow_reference) + metrics.include_values_shadow_only_static.set(len(ref - remote_set)) + metrics.include_values_shadow_only_remote.set(len(remote_set - ref)) + + +class ShadowIncludeValues: + """Shadow mode: `current()` serves the STATIC set; the HTTP prober is + private and exists only for its diff/staleness metrics. Its polled + set is intentionally unreachable through this object's public + surface.""" + + mode = "shadow" + + def __init__(self, static_values: Values | None, prober: HttpIncludeValues): + self._values = static_values + self._prober = prober + + def current(self) -> Values | None: + return self._values + + def start(self) -> None: + # The prober's sync is not load-bearing in shadow mode — the + # authority is static — so start the thread without blocking. + self._prober._thread = threading.Thread(target=self._prober._run, name="include-values-poll", daemon=True) + self._prober._thread.start() + + def stop(self) -> None: + self._prober.stop() + + +def build(cfg) -> StaticIncludeValues | ShadowIncludeValues | HttpIncludeValues: + """Build the include-values source from config. Exactly one object + comes back; the caller starts/stops it around the consume loop and + reads `current()` per batch. `.mode` is exported as + `millpond_include_values_mode` so fleet-level queries (the shadow→ + authoritative flip gate, staleness alerts) can tell which replicas + are actually running which source instead of passing vacuously on + replicas that never got the URL. + """ + if cfg.include_values_url is None: + return StaticIncludeValues(cfg.filter_values) + + auth = None + if cfg.include_values_auth_header_name is not None: + auth = (cfg.include_values_auth_header_name, cfg.include_values_auth_token) + + http = HttpIncludeValues( + url=cfg.include_values_url, + poll_interval_s=cfg.include_values_poll_interval_s, + removal_confirm_polls=cfg.include_values_removal_polls, + auth_header=auth, + bootstrap=cfg.filter_values, + shadow_reference=cfg.filter_values if cfg.include_values_mode == "shadow" else None, + request_timeout_s=cfg.include_values_request_timeout_s, + startup_timeout_s=cfg.include_values_startup_timeout_s, + ) + if cfg.include_values_mode == "shadow": + return ShadowIncludeValues(cfg.filter_values, http) + return http diff --git a/millpond/main.py b/millpond/main.py index 6419c97..25939ba 100644 --- a/millpond/main.py +++ b/millpond/main.py @@ -8,7 +8,17 @@ import pyarrow.compute as pc from confluent_kafka import TopicPartition -from millpond import arrow_converter, backpressure, config, consumer, ducklake, logging_config, metrics, server +from millpond import ( + arrow_converter, + backpressure, + config, + consumer, + ducklake, + include_values, + logging_config, + metrics, + server, +) log = logging.getLogger(__name__) @@ -54,9 +64,12 @@ def _coerce_columns(table: pa.Table, cfg: config.Config) -> pa.Table: return arrow_converter.coerce_typed_columns(table, cfg.typed_columns) -def _apply_filter(table: pa.Table, cfg: config.Config) -> pa.Table: +def _apply_filter(table: pa.Table, cfg: config.Config, values: tuple[int, ...] | tuple[str, ...] | None) -> pa.Table: """Apply the configured keep-filter: drop records whose value in - `filter_keep_field` is not in `filter_values`. + `filter_keep_field` is not in `values` (the CURRENT include set from + the configured IncludeValuesSource — static or polled; passing it + explicitly per batch is what makes the dynamic source take effect + without any state on the config object). Two reason labels on `records_skipped_total`: - `filter_field_missing`: the configured field is absent from this @@ -87,7 +100,7 @@ def _apply_filter(table: pa.Table, cfg: config.Config) -> pa.Table: reserved at the config layer but not implemented here yet — config rejects it explicitly at startup. """ - if cfg.filter_keep_field is None or cfg.filter_values is None: + if cfg.filter_keep_field is None or values is None: return table field = cfg.filter_keep_field @@ -122,11 +135,11 @@ def _apply_filter(table: pa.Table, cfg: config.Config) -> pa.Table: # a config-vs-schema mismatch (e.g. int values exceeding an int32 # column's range, or non-numeric strings against an int column) # from crashing the pod. - value_array = pa.array(cfg.filter_values).cast(column.type) + value_array = pa.array(values).cast(column.type) except (pa.ArrowInvalid, pa.ArrowNotImplementedError, pa.ArrowTypeError) as e: log.warning( "Filter values %r incompatible with column %r type %s (%s); treating batch as field-missing", - cfg.filter_values, + values, field, column.type, e, @@ -331,6 +344,7 @@ def main(): sink = None kafka = None logger_provider = None + include_source = None cfg = config.load() # metrics.init() FIRST so a failure in OTLP setup (DNS lookup at @@ -366,6 +380,21 @@ def main(): log.info("Kafka consumer created, partitions assigned") backpressure.init(cfg.consume_batch_size) + # Include-values source: what the filter reads each batch. In + # authoritative mode start() BLOCKS until the first successful + # poll (a halt is recoverable from Kafka; running on a stale + # bootstrap silently drops records). The mode gauge is what lets + # fleet-level queries (shadow-flip gate, staleness alerts) tell + # which replicas actually run which source. + include_source = include_values.build(cfg) + metrics.include_values_mode.labels(mode=include_source.mode).set(1) + include_source.start() + log.info( + "Include-values source started (mode=%s, url=%s)", + include_source.mode, + cfg.include_values_url, + ) + shutdown = False def on_signal(signum, _frame): @@ -415,7 +444,7 @@ def on_signal(signum, _frame): if table is not None: skipped = len(values) - len(table) table = _coerce_columns(table, cfg) - table = _apply_filter(table, cfg) + table = _apply_filter(table, cfg, include_source.current()) if len(table) > 0: pending.append(table) pending_bytes += table.nbytes @@ -468,6 +497,12 @@ def on_signal(signum, _frame): log.exception("Fatal error in main loop") raise finally: + # Stop the include-values poll thread first — it's daemonized so + # this is cosmetic on crash paths, but a clean shutdown shouldn't + # leave it racing the final flush. + if include_source is not None: + include_source.stop() + # Final flush only makes sense if both sink and kafka were created; # if startup failed earlier, there's no consumed data to flush. if pending_records > 0 and sink is not None and kafka is not None: diff --git a/millpond/metrics.py b/millpond/metrics.py index 20fc679..5c35f62 100644 --- a/millpond/metrics.py +++ b/millpond/metrics.py @@ -64,6 +64,55 @@ def labels(self, **kwargs): buckets=[100, 500, 1000, 5000, 10000, 50000, 100000, 500000, 1000000], ) +# --- Include-values source (see include_values.py for semantics) --- + +_include_values_size = Gauge( + "millpond_include_values_size", + "Current number of values in the keep-filter include set", + ["pipeline", "broker_source"], +) +_include_values_last_success_timestamp_seconds = Gauge( + "millpond_include_values_last_success_timestamp_seconds", + "Unix time of the last successful include-values poll (alert on age)", + ["pipeline", "broker_source"], +) +_include_values_pending_removals = Gauge( + "millpond_include_values_pending_removals", + "Values in the removal-confirmation countdown (absent but not yet removed)", + ["pipeline", "broker_source"], +) +_include_values_poll_failures_total = Counter( + "millpond_include_values_poll_failures_total", + "Failed include-values polls (the set is kept as-is on failure)", + ["pipeline", "broker_source"], +) +_include_values_refused_total = Counter( + "millpond_include_values_refused_total", + "Successful polls refused by a safety guard (reason: empty|bulk_removal|type_flip)", + ["pipeline", "broker_source", "reason"], +) +_include_values_mode = Gauge( + "millpond_include_values_mode", + "1 for the include-values mode this replica actually runs (static|shadow|authoritative)", + ["pipeline", "broker_source", "mode"], +) +_include_values_changes_total = Counter( + "millpond_include_values_changes_total", + "Applied include-set changes by action (add|remove)", + ["pipeline", "broker_source", "action"], +) +_include_values_shadow_only_static = Gauge( + "millpond_include_values_shadow_only_static", + "Shadow mode: values in the static set but not the polled set", + ["pipeline", "broker_source"], +) +_include_values_shadow_only_remote = Gauge( + "millpond_include_values_shadow_only_remote", + "Shadow mode: values in the polled set but not the static set", + ["pipeline", "broker_source"], +) + + _pending_bytes = Gauge( "millpond_pending_bytes", "Current pending Arrow bytes awaiting flush", @@ -168,6 +217,15 @@ def labels(self, **kwargs): schema_columns_widened_total = _schema_columns_widened_total sort_skipped_total = _sort_skipped_total columns_coerced_total = _columns_coerced_total +include_values_size = _include_values_size +include_values_last_success_timestamp_seconds = _include_values_last_success_timestamp_seconds +include_values_pending_removals = _include_values_pending_removals +include_values_poll_failures_total = _include_values_poll_failures_total +include_values_refused_total = _include_values_refused_total +include_values_mode = _include_values_mode +include_values_changes_total = _include_values_changes_total +include_values_shadow_only_static = _include_values_shadow_only_static +include_values_shadow_only_remote = _include_values_shadow_only_remote rdkafka_replyq = _rdkafka_replyq rdkafka_msg_cnt = _rdkafka_msg_cnt rdkafka_msg_size = _rdkafka_msg_size @@ -184,6 +242,10 @@ def init(pipeline: str, broker_source: str = ""): global pending_bytes, buffer_fullness, consume_batch_size_current, consumer_lag, last_committed_offset global schema_columns_added_total, schema_columns_widened_total, sort_skipped_total global columns_coerced_total + global include_values_size, include_values_last_success_timestamp_seconds + global include_values_pending_removals, include_values_poll_failures_total + global include_values_refused_total, include_values_changes_total, include_values_mode + global include_values_shadow_only_static, include_values_shadow_only_remote global rdkafka_replyq, rdkafka_msg_cnt, rdkafka_msg_size global rdkafka_broker_rtt_avg, rdkafka_broker_rtt_p99 @@ -195,6 +257,9 @@ def init(pipeline: str, broker_source: str = ""): records_skipped_total = _AutoCommonLabels(_records_skipped_total, pipeline, bs) sort_skipped_total = _AutoCommonLabels(_sort_skipped_total, pipeline, bs) columns_coerced_total = _AutoCommonLabels(_columns_coerced_total, pipeline, bs) + include_values_changes_total = _AutoCommonLabels(_include_values_changes_total, pipeline, bs) + include_values_refused_total = _AutoCommonLabels(_include_values_refused_total, pipeline, bs) + include_values_mode = _AutoCommonLabels(_include_values_mode, pipeline, bs) errors_total = _AutoCommonLabels(_errors_total, pipeline, bs) consumer_lag = _AutoCommonLabels(_consumer_lag, pipeline, bs) last_committed_offset = _AutoCommonLabels(_last_committed_offset, pipeline, bs) @@ -207,6 +272,14 @@ def init(pipeline: str, broker_source: str = ""): arrow_conversion_seconds = _arrow_conversion_seconds.labels(pipeline=pipeline, broker_source=bs) flush_size_bytes = _flush_size_bytes.labels(pipeline=pipeline, broker_source=bs) flush_size_records = _flush_size_records.labels(pipeline=pipeline, broker_source=bs) + include_values_size = _include_values_size.labels(pipeline=pipeline, broker_source=bs) + include_values_last_success_timestamp_seconds = _include_values_last_success_timestamp_seconds.labels( + pipeline=pipeline, broker_source=bs + ) + include_values_pending_removals = _include_values_pending_removals.labels(pipeline=pipeline, broker_source=bs) + include_values_poll_failures_total = _include_values_poll_failures_total.labels(pipeline=pipeline, broker_source=bs) + include_values_shadow_only_static = _include_values_shadow_only_static.labels(pipeline=pipeline, broker_source=bs) + include_values_shadow_only_remote = _include_values_shadow_only_remote.labels(pipeline=pipeline, broker_source=bs) pending_bytes = _pending_bytes.labels(pipeline=pipeline, broker_source=bs) buffer_fullness = _buffer_fullness.labels(pipeline=pipeline, broker_source=bs) consume_batch_size_current = _consume_batch_size_current.labels(pipeline=pipeline, broker_source=bs) diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py index 042fcd3..91e64c2 100644 --- a/tests/unit/test_config.py +++ b/tests/unit/test_config.py @@ -546,3 +546,101 @@ def test_log_lists_pairs(self, monkeypatch, caplog): load() msgs = [r.message for r in caplog.records] assert any("Coerce typed columns: timestamp:timestamptz, project_id:bigint" in m for m in msgs) + + +class TestIncludeValuesConfig: + """MILLPOND_INCLUDE_VALUES_* — the dynamic include-set source knobs and + their validation against the static filter config.""" + + @pytest.fixture(autouse=True) + def _env(self, monkeypatch): + monkeypatch.setenv("KAFKA_BOOTSTRAP_SERVERS", "localhost:9092") + monkeypatch.setenv("KAFKA_TOPIC", "test-topic") + monkeypatch.setenv("REPLICA_COUNT", "1") + monkeypatch.setenv("POD_NAME", "millpond-events-0") + monkeypatch.setenv("DUCKLAKE_TABLE", "events") + monkeypatch.setenv("DUCKLAKE_DATA_PATH", "s3://bucket/data") + monkeypatch.setenv("DUCKLAKE_RDS_HOST", "host") + monkeypatch.setenv("DUCKLAKE_RDS_PASSWORD", "pass") + monkeypatch.setenv("DUCKLAKE_CONNECTION", ":memory:") + monkeypatch.setenv("MILLPOND_FILTER_KEEP_FIELD_NAME", "team_id") + monkeypatch.setenv("MILLPOND_FILTER_VALUES", "2,50689") + + def test_unset_url_yields_static_only_defaults(self): + cfg = load() + assert cfg.include_values_url is None + assert cfg.include_values_mode == "static" + + def test_mode_without_url_rejected(self, monkeypatch): + # A set MODE means a dynamic source was intended; a typo'd/missing + # URL must not silently degrade to static-only. + monkeypatch.setenv("MILLPOND_INCLUDE_VALUES_MODE", "authoritative") + with pytest.raises(RuntimeError, match="MODE is set but"): + load() + + def test_timeout_knobs(self, monkeypatch): + monkeypatch.setenv("MILLPOND_INCLUDE_VALUES_URL", "http://cp/values") + monkeypatch.setenv("MILLPOND_INCLUDE_VALUES_REQUEST_TIMEOUT_S", "5") + monkeypatch.setenv("MILLPOND_INCLUDE_VALUES_STARTUP_TIMEOUT_S", "120") + cfg = load() + assert cfg.include_values_request_timeout_s == 5.0 + assert cfg.include_values_startup_timeout_s == 120.0 + + def test_non_numeric_knob_rejected(self, monkeypatch): + monkeypatch.setenv("MILLPOND_INCLUDE_VALUES_URL", "http://cp/values") + monkeypatch.setenv("MILLPOND_INCLUDE_VALUES_POLL_INTERVAL_S", "soon") + with pytest.raises(RuntimeError, match="must be a number"): + load() + + def test_keep_filter_chain_guarantees_bootstrap(self, monkeypatch): + # The include-values source's no-bootstrap path must stay + # unreachable through config: URL requires the keep field, and the + # keep field requires non-empty static values. If this chain is + # ever loosened, the empty-seed refusal in include_values.py + # becomes the only guard — see that module's docstring. + monkeypatch.setenv("MILLPOND_INCLUDE_VALUES_URL", "http://cp/values") + cfg = load() + assert cfg.filter_values, "static filter_values must be non-empty whenever a URL is configured" + + def test_url_with_defaults(self, monkeypatch): + monkeypatch.setenv("MILLPOND_INCLUDE_VALUES_URL", "http://cp/values") + cfg = load() + assert cfg.include_values_url == "http://cp/values" + assert cfg.include_values_mode == "shadow" + assert cfg.include_values_poll_interval_s == 60.0 + assert cfg.include_values_removal_polls == 5 + + def test_authoritative_mode(self, monkeypatch): + monkeypatch.setenv("MILLPOND_INCLUDE_VALUES_URL", "http://cp/values") + monkeypatch.setenv("MILLPOND_INCLUDE_VALUES_MODE", "authoritative") + assert load().include_values_mode == "authoritative" + + def test_bad_mode_rejected(self, monkeypatch): + monkeypatch.setenv("MILLPOND_INCLUDE_VALUES_URL", "http://cp/values") + monkeypatch.setenv("MILLPOND_INCLUDE_VALUES_MODE", "yolo") + with pytest.raises(RuntimeError, match="MODE"): + load() + + def test_url_requires_keep_filter(self, monkeypatch): + monkeypatch.delenv("MILLPOND_FILTER_KEEP_FIELD_NAME") + monkeypatch.delenv("MILLPOND_FILTER_VALUES") + monkeypatch.setenv("MILLPOND_INCLUDE_VALUES_URL", "http://cp/values") + with pytest.raises(RuntimeError, match="FILTER_KEEP_FIELD_NAME"): + load() + + def test_auth_header_requires_both_parts(self, monkeypatch): + monkeypatch.setenv("MILLPOND_INCLUDE_VALUES_URL", "http://cp/values") + monkeypatch.setenv("MILLPOND_INCLUDE_VALUES_AUTH_HEADER_NAME", "X-Secret") + with pytest.raises(RuntimeError, match="set together"): + load() + + def test_auth_without_url_rejected(self, monkeypatch): + monkeypatch.setenv("MILLPOND_INCLUDE_VALUES_AUTH_TOKEN", "tok") + with pytest.raises(RuntimeError, match="requires MILLPOND_INCLUDE_VALUES_URL"): + load() + + def test_nonpositive_interval_rejected(self, monkeypatch): + monkeypatch.setenv("MILLPOND_INCLUDE_VALUES_URL", "http://cp/values") + monkeypatch.setenv("MILLPOND_INCLUDE_VALUES_POLL_INTERVAL_S", "0") + with pytest.raises(RuntimeError, match="must be positive"): + load() diff --git a/tests/unit/test_consumer.py b/tests/unit/test_consumer.py index 5697b80..0cc9beb 100644 --- a/tests/unit/test_consumer.py +++ b/tests/unit/test_consumer.py @@ -159,6 +159,14 @@ def _make_cfg(**overrides) -> Config: filter_keep_field=None, filter_drop_field=None, filter_values=None, + include_values_url=None, + include_values_mode="shadow", + include_values_poll_interval_s=60.0, + include_values_removal_polls=5, + include_values_request_timeout_s=10.0, + include_values_startup_timeout_s=60.0, + include_values_auth_header_name=None, + include_values_auth_token=None, sort_by=None, typed_columns=None, kafka_config_overrides=(("security.protocol", "SSL"),), diff --git a/tests/unit/test_include_values.py b/tests/unit/test_include_values.py new file mode 100644 index 0000000..ab8126d --- /dev/null +++ b/tests/unit/test_include_values.py @@ -0,0 +1,301 @@ +"""Include-values source semantics. + +The safety contract under test (see include_values.py): +- additions apply on first sight; removals need M consecutive successful + ACCEPTED polls absent — failed polls AND refused polls advance nothing; +- a poll failure keeps the last-known set; +- refusal guards: empty result vs a non-empty set, empty initial seed, + bulk removal (> half of a multi-value set at once), and int↔str type + flips (which would break the filter's cast and drop whole batches); +- shadow mode reports diffs while `current()` serves only the static set, + and the prober's set is not reachable through the public surface. + +The metrics module is mocked (same approach as test_main's filter tests) +so counter/gauge calls are visible without a Prometheus registry. +""" + +from unittest.mock import MagicMock, patch + +import pytest + +from millpond.include_values import ( + HttpIncludeValues, + ShadowIncludeValues, + StaticIncludeValues, + _normalize, + build, +) + + +class TestNormalize: + def test_all_ints_sorted_deduped(self): + assert _normalize([3, 1, 2, 3]) == (1, 2, 3) + + def test_strings_sorted_deduped(self): + assert _normalize(["b", "a", "b"]) == ("a", "b") + + def test_mixed_becomes_strings(self): + assert _normalize([1, "a"]) == ("1", "a") + + def test_bools_are_not_ints(self): + # JSON `true` must not silently become 1 and match team_id=1. + assert _normalize([True, 2]) == ("2", "True") + + def test_nested_rejected(self): + with pytest.raises(ValueError): + _normalize([[1, 2]]) + with pytest.raises(ValueError): + _normalize([{"team": 1}]) + + def test_empty(self): + assert _normalize([]) == () + + +class TestStaticIncludeValues: + def test_passthrough(self): + src = StaticIncludeValues((1, 2)) + assert src.current() == (1, 2) + assert src.mode == "static" + src.start() + src.stop() + assert src.current() == (1, 2) + + def test_none_passthrough(self): + assert StaticIncludeValues(None).current() is None + + +def _http(**kwargs): + defaults = dict( + url="http://example.invalid/values", + poll_interval_s=60.0, + removal_confirm_polls=3, + bootstrap=(1, 2, 3), + ) + defaults.update(kwargs) + return HttpIncludeValues(**defaults) + + +@pytest.fixture(autouse=True) +def _mock_metrics(): + with patch("millpond.include_values.metrics", MagicMock()) as m: + yield m + + +def _refusal_reasons(mock_metrics): + return [ + c.kwargs.get("reason") or (c.args[0] if c.args else None) + for c in mock_metrics.include_values_refused_total.labels.call_args_list + ] + + +class TestPollSemantics: + def _poll(self, src, remote): + with patch.object(src, "_fetch", return_value=remote): + src._poll_once() + + def test_addition_applies_immediately(self): + src = _http() + self._poll(src, [1, 2, 3, 4]) + assert src.current() == (1, 2, 3, 4) + + def test_removal_needs_consecutive_absent_polls(self): + src = _http(removal_confirm_polls=3) + self._poll(src, [1, 2]) # 3 absent (1/3) + assert src.current() == (1, 2, 3) + self._poll(src, [1, 2]) # 3 absent (2/3) + assert src.current() == (1, 2, 3) + self._poll(src, [1, 2]) # 3 absent (3/3) -> removed + assert src.current() == (1, 2) + + def test_reappearance_resets_the_countdown(self): + src = _http(removal_confirm_polls=3) + self._poll(src, [1, 2]) # 3 absent (1/3) + self._poll(src, [1, 2]) # 3 absent (2/3) + self._poll(src, [1, 2, 3]) # 3 back -> counter reset + self._poll(src, [1, 2]) # 3 absent (1/3 again) + self._poll(src, [1, 2]) # (2/3) + assert src.current() == (1, 2, 3) + + def test_failed_poll_keeps_set_and_freezes_countdown(self): + src = _http(removal_confirm_polls=2) + self._poll(src, [1, 2]) # 3 absent (1/2) + with patch.object(src, "_fetch", side_effect=OSError("boom")): + with pytest.raises(OSError): + src._poll_once() + assert src.current() == (1, 2, 3) + assert src._absent_polls == {3: 1} + # The absence countdown resumes where it left off, not from zero. + self._poll(src, [1, 2]) # (2/2) -> removed + assert src.current() == (1, 2) + + def test_empty_result_refused_when_set_nonempty(self, _mock_metrics): + src = _http(removal_confirm_polls=1) + self._poll(src, []) + assert src.current() == (1, 2, 3) + assert "empty" in _refusal_reasons(_mock_metrics) + self._poll(src, []) + assert src.current() == (1, 2, 3) + + def test_refused_polls_do_not_precharge_removal(self): + # REGRESSION (review finding): refused-empty polls must not advance + # the absence counters. Otherwise N refused polls followed by one + # junk value confirm-removes every real value instantly. + src = _http(removal_confirm_polls=5) + for _ in range(5): + self._poll(src, []) + assert src._absent_polls == {} + self._poll(src, [999]) + # 999 added; real values have only ONE accepted absent poll — far + # from confirmation. Nothing removed. + assert src.current() == (1, 2, 3, 999) + assert all(n == 1 for n in src._absent_polls.values()) + + def test_bulk_removal_refused(self, _mock_metrics): + # Confirming removal of > half of a multi-value set in one poll is + # refused (endpoint replacing the world != routine churn). + src = _http(bootstrap=(1, 2, 3, 4), removal_confirm_polls=1) + self._poll(src, [9]) + # The WHOLE poll is refused — additions too. A world-replacement + # response is an endpoint bug or a migration; either goes through + # static config, not an unattended poll. + assert src.current() == (1, 2, 3, 4) + assert "bulk_removal" in _refusal_reasons(_mock_metrics) + # And nothing was committed to the countdown. + assert src._absent_polls == {} + + def test_removing_one_of_two_is_allowed(self): + src = _http(bootstrap=(1, 2), removal_confirm_polls=1) + self._poll(src, [1]) + assert src.current() == (1,) + + def test_type_flip_refused(self, _mock_metrics): + # REGRESSION (review finding): a junk element flips _normalize to + # strings; accepting that set would make _apply_filter's cast fail + # against an int column and drop ENTIRE batches. Refuse instead. + src = _http(bootstrap=(2, 50689)) + self._poll(src, [2, 50689, "beta-team"]) + assert src.current() == (2, 50689) + assert "type_flip" in _refusal_reasons(_mock_metrics) + # And the countdown was not advanced by the refused poll. + assert src._absent_polls == {} + + def test_type_flip_refused_both_directions(self): + src = _http(bootstrap=("us-east-1", "eu-central-1")) + self._poll(src, [1, 2]) + assert src.current() == ("us-east-1", "eu-central-1") # bootstrap verbatim + + def test_empty_seed_refused_without_bootstrap(self, _mock_metrics): + # REGRESSION (review finding): an empty remote is never an + # acceptable FIRST state — accepting () arms the filter with an + # empty include set and drops every record. + src = _http(bootstrap=None) + self._poll(src, []) + assert src.current() is None + assert "empty" in _refusal_reasons(_mock_metrics) + assert not src._synced.is_set() + + def test_no_bootstrap_first_nonempty_poll_seeds(self): + src = _http(bootstrap=None) + self._poll(src, [7, 8]) + assert src.current() == (7, 8) + assert src._synced.is_set() + + def test_int_homogeneity_survives_updates(self): + src = _http() + self._poll(src, [1, 2, 3, 9]) + assert all(isinstance(v, int) for v in src.current()) + + def test_shadow_reference_diff_gauges(self, _mock_metrics): + src = _http(shadow_reference=(1, 2, 3)) + self._poll(src, [2, 3, 4]) + _mock_metrics.include_values_shadow_only_static.set.assert_called_with(1) # {1} + _mock_metrics.include_values_shadow_only_remote.set.assert_called_with(1) # {4} + + def test_change_metrics_and_size_gauge(self, _mock_metrics): + src = _http(bootstrap=(1, 2), removal_confirm_polls=1) + self._poll(src, [1, 4]) # +4, -2 (1 of 2 allowed) + actions = str(_mock_metrics.include_values_changes_total.labels.call_args_list) + assert "add" in actions and "remove" in actions + _mock_metrics.include_values_size.set.assert_called_with(2) + + def test_pending_removals_gauge_counts_countdown_entries(self, _mock_metrics): + src = _http(removal_confirm_polls=3) + self._poll(src, [1, 2]) + _mock_metrics.include_values_pending_removals.set.assert_called_with(1) + + +class TestStartStop: + def test_start_blocks_and_raises_without_sync_even_with_bootstrap(self): + # REGRESSION (review finding): restart amnesia. Proceeding on the + # static bootstrap after a sync timeout performs de-facto removals + # (values added since the static list was last touched vanish with + # zero polls of evidence). Authoritative mode must halt instead. + src = _http(bootstrap=(1,), poll_interval_s=0.05, startup_timeout_s=0.2) + with patch.object(src, "_fetch", side_effect=OSError("down")): + with pytest.raises(RuntimeError, match="no successful poll"): + src.start() + + def test_start_syncs_from_endpoint(self): + src = _http(bootstrap=(1,), poll_interval_s=0.05, startup_timeout_s=2.0) + with patch.object(src, "_fetch", return_value=[1, 2]): + src.start() + assert src.current() == (1, 2) + src.stop() + + +class TestBuild: + def _cfg(self, **overrides): + cfg = MagicMock() + cfg.filter_values = (1, 2) + cfg.include_values_url = None + cfg.include_values_mode = "shadow" + cfg.include_values_poll_interval_s = 60.0 + cfg.include_values_removal_polls = 5 + cfg.include_values_request_timeout_s = 10.0 + cfg.include_values_startup_timeout_s = 60.0 + cfg.include_values_auth_header_name = None + cfg.include_values_auth_token = None + for k, v in overrides.items(): + setattr(cfg, k, v) + return cfg + + def test_no_url_is_static_only(self): + src = build(self._cfg()) + assert isinstance(src, StaticIncludeValues) + assert src.mode == "static" + assert src.current() == (1, 2) + + def test_shadow_serves_static_and_hides_the_prober(self): + src = build(self._cfg(include_values_url="http://x/v")) + assert isinstance(src, ShadowIncludeValues) + assert src.mode == "shadow" + assert src.current() == (1, 2) + # The prober is private; its polled set must not be part of the + # public surface a future change could mistakenly read. + assert not hasattr(src, "prober") + assert src._prober._shadow_reference == (1, 2) + + def test_authoritative_is_http_with_bootstrap(self): + src = build(self._cfg(include_values_url="http://x/v", include_values_mode="authoritative")) + assert isinstance(src, HttpIncludeValues) + assert src.mode == "authoritative" + assert src.current() == (1, 2) # bootstrap until first sync + assert src._shadow_reference is None + + def test_auth_header_threaded(self): + src = build( + self._cfg( + include_values_url="http://x/v", + include_values_mode="authoritative", + include_values_auth_header_name="X-Internal-Secret", + include_values_auth_token="tok", + ) + ) + assert src._auth_header == ("X-Internal-Secret", "tok") + + def test_shadow_start_does_not_block_on_sync(self): + src = build(self._cfg(include_values_url="http://x/v")) + with patch.object(src._prober, "_fetch", side_effect=OSError("down")): + src.start() # must return immediately despite the dead endpoint + assert src.current() == (1, 2) + src.stop() diff --git a/tests/unit/test_main.py b/tests/unit/test_main.py index 6d51d0d..24843cb 100644 --- a/tests/unit/test_main.py +++ b/tests/unit/test_main.py @@ -314,9 +314,7 @@ def test_no_op_when_unconfigured(self): def test_coerces_when_configured(self): from millpond.main import _coerce_columns - table = pa.table( - {"timestamp": ["2024-01-01 12:00:00.000000"], "project_id": pa.array([None], pa.string())} - ) + table = pa.table({"timestamp": ["2024-01-01 12:00:00.000000"], "project_id": pa.array([None], pa.string())}) pairs = (("timestamp", "timestamptz"), ("project_id", "bigint")) result = _coerce_columns(table, self._cfg(typed_columns=pairs)) assert result.schema.field("timestamp").type == pa.timestamp("us", tz="UTC") @@ -329,17 +327,18 @@ class TestApplyFilter: Prometheus registry. Each test constructs a Config-shaped object only with the fields _apply_filter reads.""" - def _cfg(self, *, keep=None, values=None): + def _cfg(self, *, keep=None): + # values now flow as _apply_filter's third arg (the include-values + # source's current set), not via config. cfg = MagicMock() cfg.filter_keep_field = keep - cfg.filter_values = values return cfg def test_no_op_when_filter_unconfigured(self): from millpond.main import _apply_filter table = pa.table({"team_id": [1, 2, 3]}) - result = _apply_filter(table, self._cfg()) + result = _apply_filter(table, self._cfg(), None) # Same object: no slicing or filtering — short-circuit at the top. assert result is table @@ -349,7 +348,7 @@ def test_int_allowlist_keeps_matching_rows(self, mock_metrics): skip_calls = _capture_skip_calls(mock_metrics) table = pa.table({"team_id": [1, 2, 3, 4, 5], "event": ["a", "b", "c", "d", "e"]}) - result = _apply_filter(table, self._cfg(keep="team_id", values=(2, 4))) + result = _apply_filter(table, self._cfg(keep="team_id"), (2, 4)) assert result.num_rows == 2 assert result.column("team_id").to_pylist() == [2, 4] @@ -364,7 +363,7 @@ def test_string_allowlist_keeps_matching_rows(self, mock_metrics): skip_calls = _capture_skip_calls(mock_metrics) table = pa.table({"region": ["us-east-1", "us-west-2", "eu-central-1"]}) - result = _apply_filter(table, self._cfg(keep="region", values=("us-east-1", "eu-central-1"))) + result = _apply_filter(table, self._cfg(keep="region"), ("us-east-1", "eu-central-1")) assert result.column("region").to_pylist() == ["us-east-1", "eu-central-1"] assert skip_calls == [("filter_excluded", 1)] @@ -383,7 +382,7 @@ def test_int_values_coerce_to_string_column(self, mock_metrics): skip_calls = _capture_skip_calls(mock_metrics) table = pa.table({"team_id": ["1", "2", "3"]}) - result = _apply_filter(table, self._cfg(keep="team_id", values=(2,))) + result = _apply_filter(table, self._cfg(keep="team_id"), (2,)) assert result.num_rows == 1 assert result.column("team_id").to_pylist() == ["2"] @@ -399,7 +398,7 @@ def test_leading_zero_strings_do_not_match_int_values(self, mock_metrics): skip_calls = _capture_skip_calls(mock_metrics) table = pa.table({"team_id": ["02", "2", "003"]}) - result = _apply_filter(table, self._cfg(keep="team_id", values=(2,))) + result = _apply_filter(table, self._cfg(keep="team_id"), (2,)) # Only the unambiguous "2" matches; the leading-zero strings don't. assert result.column("team_id").to_pylist() == ["2"] @@ -411,7 +410,7 @@ def test_missing_field_drops_whole_batch(self, mock_metrics): skip_calls = _capture_skip_calls(mock_metrics) table = pa.table({"event": ["a", "b", "c"]}) - result = _apply_filter(table, self._cfg(keep="team_id", values=(1, 2))) + result = _apply_filter(table, self._cfg(keep="team_id"), (1, 2)) # Column not in schema → whole batch lands in filter_field_missing. assert result.num_rows == 0 @@ -427,7 +426,7 @@ def test_null_values_counted_as_field_missing(self, mock_metrics): skip_calls = _capture_skip_calls(mock_metrics) table = pa.table({"team_id": pa.array([1, None, 2, None, 3], type=pa.int64())}) - result = _apply_filter(table, self._cfg(keep="team_id", values=(1, 2))) + result = _apply_filter(table, self._cfg(keep="team_id"), (1, 2)) assert result.num_rows == 2 assert result.column("team_id").to_pylist() == [1, 2] @@ -440,7 +439,7 @@ def test_all_rows_kept_emits_no_skip_metric(self, mock_metrics): skip_calls = _capture_skip_calls(mock_metrics) table = pa.table({"team_id": [1, 2, 1]}) - result = _apply_filter(table, self._cfg(keep="team_id", values=(1, 2))) + result = _apply_filter(table, self._cfg(keep="team_id"), (1, 2)) assert result.num_rows == 3 assert skip_calls == [] @@ -451,7 +450,7 @@ def test_all_rows_dropped_returns_empty_table_with_same_schema(self, mock_metric _capture_skip_calls(mock_metrics) table = pa.table({"team_id": [9, 10, 11], "event": ["a", "b", "c"]}) - result = _apply_filter(table, self._cfg(keep="team_id", values=(1, 2))) + result = _apply_filter(table, self._cfg(keep="team_id"), (1, 2)) assert result.num_rows == 0 # Schema is preserved — important so a downstream concat doesn't trip @@ -464,7 +463,7 @@ def test_empty_input_batch_is_no_op(self, mock_metrics): skip_calls = _capture_skip_calls(mock_metrics) table = pa.table({"team_id": pa.array([], type=pa.int64())}) - result = _apply_filter(table, self._cfg(keep="team_id", values=(1, 2))) + result = _apply_filter(table, self._cfg(keep="team_id"), (1, 2)) assert result.num_rows == 0 assert skip_calls == [] @@ -486,7 +485,7 @@ def test_multi_chunk_column_is_handled(self, mock_metrics): events = pa.chunked_array([["a", "b"], ["c", "d", "e"], ["f"]]) table = pa.table({"team_id": team_ids, "event": events}) - result = _apply_filter(table, self._cfg(keep="team_id", values=(2,))) + result = _apply_filter(table, self._cfg(keep="team_id"), (2,)) # Two rows match (the two 2s spread across chunks 0 and 2). assert result.column("team_id").to_pylist() == [2, 2] @@ -513,7 +512,7 @@ def test_bool_column_rejected_as_unsupported(self, mock_metrics): skip_calls = _capture_skip_calls(mock_metrics) table = pa.table({"flag": pa.array([True, False, True, False], type=pa.bool_())}) - result = _apply_filter(table, self._cfg(keep="flag", values=(1,))) + result = _apply_filter(table, self._cfg(keep="flag"), (1,)) assert result.num_rows == 0 assert skip_calls == [("filter_field_missing", 4)] @@ -528,7 +527,7 @@ def test_float_column_rejected_as_unsupported(self, mock_metrics): skip_calls = _capture_skip_calls(mock_metrics) table = pa.table({"score": pa.array([1.0, 2.0, 3.0], type=pa.float64())}) - result = _apply_filter(table, self._cfg(keep="score", values=(2,))) + result = _apply_filter(table, self._cfg(keep="score"), (2,)) assert result.num_rows == 0 assert skip_calls == [("filter_field_missing", 3)] @@ -543,7 +542,7 @@ def test_timestamp_column_rejected_as_unsupported(self, mock_metrics): ts_col = pa.array([1700000000_000000, 1700000001_000000], type=pa.timestamp("us")) table = pa.table({"ts": ts_col}) - result = _apply_filter(table, self._cfg(keep="ts", values=(1700000000_000000,))) + result = _apply_filter(table, self._cfg(keep="ts"), (1700000000_000000,)) assert result.num_rows == 0 assert skip_calls == [("filter_field_missing", 2)] @@ -562,7 +561,7 @@ def test_struct_column_rejected_as_unsupported(self, mock_metrics): ) table = pa.table({"team_id": struct_col}) - result = _apply_filter(table, self._cfg(keep="team_id", values=(1, 2))) + result = _apply_filter(table, self._cfg(keep="team_id"), (1, 2)) assert result.num_rows == 0 assert skip_calls == [("filter_field_missing", 3)] @@ -575,7 +574,7 @@ def test_list_column_rejected_as_unsupported(self, mock_metrics): list_col = pa.array([[1, 2], [3], [4, 5, 6]], type=pa.list_(pa.int64())) table = pa.table({"team_id": list_col}) - result = _apply_filter(table, self._cfg(keep="team_id", values=(1, 2))) + result = _apply_filter(table, self._cfg(keep="team_id"), (1, 2)) assert result.num_rows == 0 assert skip_calls == [("filter_field_missing", 3)] @@ -591,7 +590,7 @@ def test_int_values_overflowing_int32_column_skip_batch(self, mock_metrics): skip_calls = _capture_skip_calls(mock_metrics) table = pa.table({"team_id": pa.array([1, 2, 3], type=pa.int32())}) # 2**40 is well outside int32 range. - result = _apply_filter(table, self._cfg(keep="team_id", values=(2**40,))) + result = _apply_filter(table, self._cfg(keep="team_id"), (2**40,)) assert result.num_rows == 0 assert skip_calls == [("filter_field_missing", 3)] diff --git a/tests/unit/test_millpond_logging_config.py b/tests/unit/test_millpond_logging_config.py index ceae46d..86beffb 100644 --- a/tests/unit/test_millpond_logging_config.py +++ b/tests/unit/test_millpond_logging_config.py @@ -63,6 +63,14 @@ def _minimal_config() -> Config: filter_keep_field=None, filter_drop_field=None, filter_values=None, + include_values_url=None, + include_values_mode="shadow", + include_values_poll_interval_s=60.0, + include_values_removal_polls=5, + include_values_request_timeout_s=10.0, + include_values_startup_timeout_s=60.0, + include_values_auth_header_name=None, + include_values_auth_token=None, sort_by=None, typed_columns=None, kafka_config_overrides=(),