From aed7cfec05773dbe8ecc3a55b4164f048806d24f Mon Sep 17 00:00:00 2001 From: fuziontech Date: Mon, 3 Aug 2026 22:46:53 +0000 Subject: [PATCH] feat(discovery): configurable table_field for persons pipelines MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Discovery previously hardcoded the team payload's events_table as the destination table. Streamed persons replication runs two more pipelines (persons + persons_distinct_ids), each needing its own resolved table from the same payload. discovery.table_field selects the field (events_table default; persons_table / persons_distinct_ids_table for persons pipelines), validated against the known payload fields at startup — a typo failing open to events_table would route person rows into events tables. The value is threaded through map_payload/classify_payload and the DriftWatcher so startup and drift polls classify against the same field. Missing-field rows keep the events_table contract: mentioned, not poisoned. Pairs with the duckgres discovery change serving persons_distinct_ids_table. Co-authored-by: Shelley --- README.md | 3 +- tests/unit/test_config.py | 62 ++++++++++++++++++++++++++++++++++++ tests/unit/test_discovery.py | 41 ++++++++++++++++++++++++ viaduck/config.py | 13 ++++++++ viaduck/discovery.py | 39 ++++++++++++++--------- viaduck/main.py | 3 +- 6 files changed, 144 insertions(+), 17 deletions(-) diff --git a/README.md b/README.md index 28912d5..ac00def 100644 --- a/README.md +++ b/README.md @@ -361,7 +361,7 @@ A destination paused while it has never been seeded (cursor 0) skips seeding at ## CP-Driven Destination Discovery (additive) -With `discovery.enabled`, viaduck polls the duckgres control plane's read-only endpoint (`GET /api/v1/warehouses`, authenticated with the scoped read-only secret via `discovery.auth_header_name` + `discovery.auth_token_env`) at startup and extends the static destination set: one destination per (warehouse, team), id `org--team-`, routing value = team id, **table = the payload's `events_table` verbatim** (the CP owns naming; renames are not allowed upstream). Metadata-store passwords resolve via a direct Kubernetes Secret read with viaduck's ServiceAccount (`password_secret_ref` — RBAC into the tenant namespace, no secret copies, no plaintext payloads). +With `discovery.enabled`, viaduck polls the duckgres control plane's read-only endpoint (`GET /api/v1/warehouses`, authenticated with the scoped read-only secret via `discovery.auth_header_name` + `discovery.auth_token_env`) at startup and extends the static destination set: one destination per (warehouse, team), id `org--team-`, routing value = team id, **table = a payload table field verbatim** — `events_table` by default; a persons pipeline sets `discovery.table_field` to `persons_table` or `persons_distinct_ids_table` (the CP owns naming; renames are not allowed upstream). Metadata-store passwords resolve via a direct Kubernetes Secret read with viaduck's ServiceAccount (`password_secret_ref` — RBAC into the tenant namespace, no secret copies, no plaintext payloads). Semantics (M4): **additive, static wins, fixed set.** A static destination beats a discovered one on routing-value collision (cutover = delete the static entry). Discovered destinations initialize at the source head (`seed_mode`-independent: discovery starts the stream, never backfills) with convention defaults — schema projection on, `captured_at` dropped, `memory_limit` from `discovery.defaults`. The destination set is fixed at startup; a background poller detects drift (`viaduck_discovery_drift_destinations{kind}`) and a restart applies it. Fail-open at startup (CP unreachable → static-only + `viaduck_discovery_synced` 0); fail-safe per entry (`viaduck_discovery_broken_entries_total{reason}` — a broken tenant never takes down the rest; non-writable/resharding warehouses are skipped). @@ -373,6 +373,7 @@ discovery: url: "http://duckgres-admin.duckgres.svc.cluster.local:8080/api/v1/warehouses" auth_header_name: "X-Duckgres-Internal-Secret" auth_token_env: "VIADUCK_DISCOVERY_TOKEN" # the read-only secret, never the internal secret + table_field: "events_table" # persons pipeline: persons_table / persons_distinct_ids_table poll_interval_s: 60 defaults: memory_limit: "8GB" diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py index 59e97f0..2639bbd 100644 --- a/tests/unit/test_config.py +++ b/tests/unit/test_config.py @@ -1043,3 +1043,65 @@ def test_destination_buffer_max_bytes_loader(tmp_path: Path): def test_destination_buffer_max_bytes_default_zero(config_file: Path): cfg = load(config_file) assert cfg.destinations[0].buffer_max_bytes == 0 + + +def test_discovery_table_field_loads_from_yaml(tmp_path): + import yaml + + from viaduck.config import load + + cfg_yaml = { + "source": {"name": "s", "postgres_uri_env": "SRC", "data_path": "/d", "table": "t"}, + "routing": {"field": "company", "mode": "append_only"}, + "destinations": [ + { + "id": "d1", + "routing_value": "a", + "name": "n", + "postgres_uri_env": "DST", + "data_path": "/d", + "table": "t", + } + ], + "discovery": {"enabled": True, "url": "http://cp/api/v1/warehouses", "table_field": "persons_table"}, + } + p = tmp_path / "viaduck.yaml" + p.write_text(yaml.safe_dump(cfg_yaml)) + cfg = load(str(p)) + assert cfg.discovery.table_field == "persons_table" + + +def test_discovery_table_field_defaults_to_events_table(config_file): + from viaduck.config import load + + cfg = load(config_file) + assert cfg.discovery.table_field == "events_table" + + +def test_discovery_table_field_rejects_unknown_field(tmp_path): + import pytest + import yaml + + from viaduck.config import ConfigError, load + + cfg_yaml = { + "source": {"name": "s", "postgres_uri_env": "SRC", "data_path": "/d", "table": "t"}, + "routing": {"field": "company", "mode": "append_only"}, + "destinations": [ + { + "id": "d1", + "routing_value": "a", + "name": "n", + "postgres_uri_env": "DST", + "data_path": "/d", + "table": "t", + } + ], + # A typo here must fail fast: silently falling back to + # events_table would route person rows into the events tables. + "discovery": {"table_field": "person_table"}, + } + p = tmp_path / "viaduck.yaml" + p.write_text(yaml.safe_dump(cfg_yaml)) + with pytest.raises(ConfigError, match="table_field"): + load(str(p)) diff --git a/tests/unit/test_discovery.py b/tests/unit/test_discovery.py index efd697a..b432d5a 100644 --- a/tests/unit/test_discovery.py +++ b/tests/unit/test_discovery.py @@ -69,6 +69,47 @@ def test_happy_path_events_table_verbatim(self): assert m.pg_endpoint == "cnpg-shard-1-rw.ducklings.svc" assert m.secret_namespace == "ducklings" + def test_table_field_selects_persons_table(self): + # A persons pipeline routes into the payload's persons_table, + # still verbatim — the CP owns naming for every streamed table. + wh = _warehouse( + teams=[ + { + "team_id": 7, + "schema_name": "t7", + "enabled": True, + "events_table": "t7.events", + "persons_table": "t7.persons_ab12", + } + ] + ) + mapped = discovery.map_payload(_payload([wh]), table_field="persons_table") + assert [m.table for m in mapped] == ["t7.persons_ab12"] + + def test_table_field_selects_persons_distinct_ids_table(self): + wh = _warehouse( + teams=[ + { + "team_id": 7, + "schema_name": "t7", + "enabled": True, + "events_table": "t7.events", + "persons_distinct_ids_table": "t7.persons_distinct_ids_ab12", + } + ] + ) + mapped = discovery.map_payload(_payload([wh]), table_field="persons_distinct_ids_table") + assert [m.table for m in mapped] == ["t7.persons_distinct_ids_ab12"] + + def test_table_field_missing_is_mentioned_not_poison(self): + # Same contract as a missing events_table: the id is nameable, so + # the row is degraded, never absent and never view-poisoning. + wh = _warehouse() + view = discovery.classify_payload(_payload([wh]), table_field="persons_table") + e = view.entries["org-acme-team-666"] + assert not e.startable + assert not view.parse_poisoned + def test_disabled_team_still_included(self): # `enabled` is the QUERY-SERVING switch (duckgres migration # 000024): deriving ingestion-stop from it turns a serving hold diff --git a/viaduck/config.py b/viaduck/config.py index 20d7613..b72e811 100644 --- a/viaduck/config.py +++ b/viaduck/config.py @@ -586,6 +586,12 @@ class DiscoveryConfig: url: str | None = None auth_header_name: str | None = None auth_token_env: str | None = None + # Which team-payload field discovery reads as the destination table. + # events_table for the events pipeline; persons_table / + # persons_distinct_ids_table for the two streamed-persons pipelines. + # Validated against the known payload fields: silently defaulting a + # typo to events_table would route person rows into events tables. + table_field: str = "events_table" poll_interval_s: float = 60.0 request_timeout_s: float = 10.0 # C3 reconciler (viaduck/reconciler.py). Default OFF: the classified @@ -636,9 +642,15 @@ class DiscoveryConfig: secret_cache_ttl_s: float = 300.0 defaults: dict = field(default_factory=dict) + # Team-payload fields that name a destination table. Kept in sync + # with the duckgres discovery payload (resolveTeamTables). + TABLE_FIELDS = ("events_table", "persons_table", "persons_distinct_ids_table") + def __post_init__(self): if self.enabled and not self.url: raise ConfigError("discovery.enabled requires discovery.url") + if self.table_field not in self.TABLE_FIELDS: + raise ConfigError(f"discovery.table_field must be one of {self.TABLE_FIELDS}, got {self.table_field!r}") if (self.auth_header_name is None) != (self.auth_token_env is None): raise ConfigError("discovery.auth_header_name and discovery.auth_token_env must be set together") if self.poll_interval_s <= 0 or self.request_timeout_s <= 0 or self.materialize_deadline_s <= 0: @@ -1006,6 +1018,7 @@ def load(path: str | Path) -> ViaduckConfig: url=disc_raw.get("url"), auth_header_name=disc_raw.get("auth_header_name"), auth_token_env=disc_raw.get("auth_token_env"), + table_field=str(disc_raw.get("table_field", "events_table")), poll_interval_s=float(disc_raw.get("poll_interval_s", 60.0)), request_timeout_s=float(disc_raw.get("request_timeout_s", 10.0)), min_destinations=int(disc_raw.get("min_destinations", 1)), diff --git a/viaduck/discovery.py b/viaduck/discovery.py index 9a8008b..ec44df0 100644 --- a/viaduck/discovery.py +++ b/viaduck/discovery.py @@ -5,10 +5,12 @@ secret) and maps (warehouse, team) pairs onto destination configs: - id ``org--team-``; routing value = the team id. -- **Table = the payload's ``events_table`` VERBATIM** — the CP owns - naming (schema-per-team + legacy bare-name overrides resolve there); - viaduck never derives a table name. Renames are not allowed upstream, - so the table is immutable for a destination's lifetime. +- **Table = a payload table field VERBATIM** — ``events_table`` by + default; a persons pipeline sets ``discovery.table_field`` to + ``persons_table``/``persons_distinct_ids_table``. The CP owns naming + (schema-per-team + legacy bare-name overrides resolve there); viaduck + never derives a table name. Renames are not allowed upstream, so the + table is immutable for a destination's lifetime. - Metadata-store credentials come from the payload's connection fields plus a Kubernetes Secret reference (``password_secret_ref``) resolved by reading the Secret directly with viaduck's ServiceAccount (RBAC @@ -244,18 +246,19 @@ def __post_init__(self) -> None: object.__setattr__(self, "entries", MappingProxyType(first)) -def classify_payload(payload: dict, *, count_broken: bool = True) -> ClassifiedView: +def classify_payload(payload: dict, *, count_broken: bool = True, table_field: str = "events_table") -> ClassifiedView: """Classify every enumerable destination id in the raw payload as startable (mapped config) or merely mentioned (fenced, degraded — the reason is counted, not carried). Never raises on data problems: un-enumerable content poisons the view (see ClassifiedView). `count_broken` keeps the startup/loud vs drift-poll/quiet split of - map_payload.""" + map_payload. `table_field` names the team-payload field read as the + destination table (see DiscoveryConfig.table_field).""" entries: list[ClassifiedEntry] = [] poisoned = False for wh in payload.get("warehouses", []): try: - if not _classify_warehouse(wh, entries, count_broken): + if not _classify_warehouse(wh, entries, count_broken, table_field): poisoned = True except Exception as e: _broken("malformed", f"warehouse entry unparseable: {e!r}", count=count_broken) @@ -290,16 +293,18 @@ def derive_absent(view: ClassifiedView | None, registry_snapshot) -> frozenset[s return frozenset(d for d in routable_discovered if d not in view.entries) -def map_payload(payload: dict, *, count_broken: bool = True) -> list[MappedDestination]: +def map_payload( + payload: dict, *, count_broken: bool = True, table_field: str = "events_table" +) -> list[MappedDestination]: """STARTABLE entries of the classified view, in payload order with duplicates preserved (materialize() owns dedupe + its counter). Startup-compatible shape; the classification is the single parsing path so the two can never drift.""" - view = classify_payload(payload, count_broken=count_broken) + view = classify_payload(payload, count_broken=count_broken, table_field=table_field) return [e.mapped for e in view.entry_list if e.mapped is not None] -def _classify_warehouse(wh: dict, entries: list[ClassifiedEntry], count_broken: bool) -> bool: +def _classify_warehouse(wh: dict, entries: list[ClassifiedEntry], count_broken: bool, table_field: str) -> bool: """Classify one warehouse's teams into `entries`. Returns False when any content was UN-ENUMERABLE (a team id we cannot even name — that id would falsely read as ABSENT, so the caller poisons the view). @@ -352,7 +357,7 @@ def _classify_warehouse(wh: dict, entries: list[ClassifiedEntry], count_broken: for team in teams: team_id = team.get("team_id") - events_table = team.get("events_table") + table = team.get(table_field) if team_id is None: # An id we cannot name — the poison case. Deliberate cadence # change vs the pre-v6 parser: this fires for unstartable @@ -363,8 +368,8 @@ def _classify_warehouse(wh: dict, entries: list[ClassifiedEntry], count_broken: enumerable = False continue dest_id = f"org-{org}-team-{team_id}" - if startable and not events_table: - _broken("bad_team_row", f"org {org} team row missing events_table", count=count_broken) + if startable and not table: + _broken("bad_team_row", f"org {org} team row missing {table_field}", count=count_broken) entries.append(ClassifiedEntry(dest_id=dest_id, mapped=None)) continue if not startable: @@ -381,7 +386,7 @@ def _classify_warehouse(wh: dict, entries: list[ClassifiedEntry], count_broken: dest_id=dest_id, org_id=org, team_id=team_id, - table=events_table, + table=table, data_path=f"s3://{bucket}/", pg_endpoint=ms["endpoint"], pg_port=ms.get("port") or 5432, @@ -574,6 +579,10 @@ class DriftWatcher: poll_interval_s: float baseline: dict[str, MappedDestination] startup_generation: int + # Team-payload field read as the destination table — must match the + # field the startup map_payload used, or every discovered entry reads + # as changed drift. + table_field: str = "events_table" # True when the C3 reconciler is applying views (discovery. # apply_enabled): the vs-STARTUP drift comparison below is then # permanently wrong after the first applied change (an applied add @@ -636,7 +645,7 @@ def _poll_once(self) -> None: # static-only alert (round-2 review). metrics.discovery_config_generation.set(payload["config_generation"]) metrics.discovery_last_success_timestamp_seconds.set_to_current_time() - view = classify_payload(payload, count_broken=False) + view = classify_payload(payload, count_broken=False, table_field=self.table_field) with self._view_lock: self._view = view startable = sum(1 for e in view.entries.values() if e.startable) diff --git a/viaduck/main.py b/viaduck/main.py index f827929..daabd18 100644 --- a/viaduck/main.py +++ b/viaduck/main.py @@ -748,7 +748,7 @@ def run(cfg: config.ViaduckConfig) -> None: time.sleep(min(2**attempt, 5)) if payload is None: raise last_err # type: ignore[misc] - mapped = disco.map_payload(payload) + mapped = disco.map_payload(payload, table_field=cfg.discovery.table_field) if len(mapped) < cfg.discovery.min_destinations: raise disco.DiscoveryError( f"payload mapped {len(mapped)} destination(s) < discovery.min_destinations " @@ -821,6 +821,7 @@ def run(cfg: config.ViaduckConfig) -> None: baseline=baseline, startup_generation=generation, apply_mode=cfg.discovery.apply_enabled, + table_field=cfg.discovery.table_field, ) # The registry is built from the post-merge cfg and is THE runtime # resolution path for destination configs — the pool, the poll cycle,