sadpandajoe commented on code in PR #43490: URL: https://github.com/apache/superset/pull/43490#discussion_r3856943849
########## superset/commands/deletion_retention/prune_audit.py: ########## @@ -0,0 +1,607 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +"""Deletion-only pruning of the ``purge_audit_log`` table. + +Bounds the audit history's growth without ever weakening its evidentiary +value. Three delete categories, applied in priority order under one shared +per-run batch budget: + +1. **Blocked duplicates** — within an entity's *current* blockage streak + (its ``blocked`` rows newer than the entity's newest streak-breaking + row), only the streak's earliest row survives: later duplicates are + removed regardless of age. The survivor carries the "blocked since" + fact and is never deleted while the streak is current. +2. **Operational expiry** — ``blocked`` rows of *resolved* streaks and + ``failed`` rows older than ``PURGE_AUDIT_OPERATIONAL_RETENTION_DAYS`` + age out. +3. **Evidence expiry** — ``confirmed`` / ``target_absent`` rows are + untouchable unless ``PURGE_AUDIT_EVIDENCE_RETENTION_DAYS`` is + explicitly set (the operator's compliance assertion), and then only + rows older than that window. + +``pending`` rows belong to :func:`audit.reconcile_pending` and rows with +future timestamps (clock skew) are excluded from every category *and* from +streak classification, so a skewed writer cannot reclassify a live streak. + +Three invariants keep the survivor safe without a distributed lock, which +matters because runs can overlap and both ``reconcile_pending`` and the +purge path finalize rows concurrently. A duplicate is only ever deleted +when no concurrent transition could turn it into a survivor first: + +* **A boundary is never removed while it still bounds anything.** Evidence + expiry refuses to delete a row while an older ``blocked`` or ``pending`` + row for the same entity survives. Boundaries therefore never *recede*, + which would otherwise promote resolved rows into a current streak. +* **A blocked row whose classification is unstable is never deleted.** + Finalizing a ``pending`` row resolves it *in place*, keeping its original + timestamp, so an unresolved attempt is a boundary that may appear + mid-history at any moment — and the blocked row after it would become its + streak's survivor. Blocked rows preceded by an unresolved attempt are + therefore skipped by **both** the duplicate and the operational category. + Age does not stabilize the classification: an old blocked row inside a + live streak is exactly the "blocked for years" case FR-009 protects. + Without this, a boundary moving *forward* would demote the current + survivor and promote the next row into its place — possibly a row already + selected for deletion. +* **Deletes are conditional and counted from rowcounts.** A row whose + status changed since selection is not matched, so overlapping runs can + neither double-remove nor double-report. + +Boundaries moving forward past *all* of an entity's blocked rows leave no +survivor to promote, so a selected duplicate may be removed slightly ahead +of its retention window in that case. It was redundant either way and the +streak's earliest row is untouched. + +Candidate selection is embedded in each ``DELETE`` statement. The derived-table +wrapper keeps that shape legal on MySQL while ensuring that a pending or +recovered row committed before the delete is evaluated participates in the +survivor and boundary predicates. No stale list of candidate ids crosses a +transaction boundary. + +Audit creation/recovery and every pruning batch take the same singleton +database write lock before assigning a timestamp or evaluating candidates. +The lock is held through commit, so an audit row cannot become visible in an +already-processed logical past. Automatic pruning still ships disabled by +default so operators explicitly choose their retention policy. +""" + +from __future__ import annotations + +import logging +from collections.abc import Callable +from dataclasses import dataclass, field +from datetime import datetime, timedelta +from functools import partial +from typing import Any, Literal, NamedTuple, TypeAlias + +import sqlalchemy as sa +from flask import current_app + +from superset import db +from superset.commands.deletion_retention.audit import ( + acquire_coordination_lock, + utc_now, +) +from superset.models.purge_audit_log import ( + PurgeAuditLog, + STATUS_BLOCKED, + STATUS_CONFIRMED, + STATUS_FAILED, + STATUS_PENDING, + STATUS_TARGET_ABSENT, +) + +logger: logging.Logger = logging.getLogger(__name__) + +#: Operational records: noise-prone outcomes whose compliance value decays — +#: a blocked or failed purge leaves the object in place (FR-001). +OPERATIONAL_STATUSES: frozenset[str] = frozenset({STATUS_BLOCKED, STATUS_FAILED}) +#: Protected evidence: the only surviving trace of a destroyed object. +PROTECTED_STATUSES: frozenset[str] = frozenset({STATUS_CONFIRMED, STATUS_TARGET_ABSENT}) +#: The outcomes that end a blockage streak — proof the object is gone. +#: +#: ``failed`` is deliberately absent. A failed purge is an infrastructure +#: outcome (the cascade raised), not evidence the blockage cleared: the +#: policy that blocked the entity is untouched, so the blockage continues +#: across it. Treating ``failed`` as a boundary would let one transient +#: error demote the "blocked since" survivor to an ageing duplicate and +#: restate the blockage as beginning after the failure — losing exactly the +#: fact FR-003 exists to preserve. ``pending`` is provisional and likewise +#: neither joins nor breaks streaks. +_STREAK_BREAKING_STATUSES: frozenset[str] = frozenset( + {STATUS_CONFIRMED, STATUS_TARGET_ABSENT} +) + +#: Rows deleted per statement, matching the purge task's batch convention. +BATCH_SIZE: int = 500 +#: One shared budget for the whole run across all three categories; the +#: remaining backlog carries over to the next scheduled run (FR-004/SC-004). +MAX_BATCHES_PER_RUN: int = 10 + +OPERATIONAL_RETENTION_KEY: str = "PURGE_AUDIT_OPERATIONAL_RETENTION_DAYS" +EVIDENCE_RETENTION_KEY: str = "PURGE_AUDIT_EVIDENCE_RETENTION_DAYS" + + +class ResolvedWindow(NamedTuple): + """A retention window resolved from config. + + Distinguishes the three outcomes a caller must tell apart: a usable + window, the deliberate "off" default, and operator error. Collapsing + the last two into a bare ``None`` would force every caller to re-read + config to find out which it got. + """ + + days: int | None + invalid_key: str | None = None + + +def _validated_window(key: str, value: Any) -> ResolvedWindow: + """Validate a configured day count, failing closed on anything odd. + + An invalid value disables its category for the run rather than widening + removal (FR-005/SC-005). + """ + # bool is an int subclass and floats would silently truncate — both are + # config mistakes, not day counts, on a knob that deletes rows. + if not isinstance(value, bool) and not isinstance(value, float): + try: + days: int = int(value) + except (TypeError, ValueError): + days = 0 + if days > 0: + return ResolvedWindow(days) + logger.warning( + "prune_audit: invalid %s=%r; skipping this category for the run " + "(pruning never widens on bad configuration)", + key, + value, + ) + return ResolvedWindow(None, key) + + +def resolve_operational_retention_days() -> ResolvedWindow: + """The operational retention window, or a disabled window when invalid.""" + return _validated_window( + OPERATIONAL_RETENTION_KEY, current_app.config.get(OPERATIONAL_RETENTION_KEY) + ) + + +def resolve_evidence_retention_days() -> ResolvedWindow: + """The evidence expiration window; disabled unless explicitly opted in. + + Unset is the documented "never expire evidence" default (FR-006), not an + error, so it produces a disabled window with no warning. + """ + value: Any = current_app.config.get(EVIDENCE_RETENTION_KEY) + if value is None: + return ResolvedWindow(None) + return _validated_window(EVIDENCE_RETENTION_KEY, value) + + +@dataclass +class PruneRunResult: + """Per-category removal counts and run disposition for one pruning run.""" + + blocked_duplicates: int = 0 + operational_expired: int = 0 + evidence_expired: int = 0 + #: True when the shared batch budget ran out before every category's + #: candidates were drained; the remainder converges on later runs. + carried_over: bool = False + invalid_config_keys: list[str] = field(default_factory=list) + + @property + def total_removed(self) -> int: + """Total rows removed across every category.""" + return ( + self.blocked_duplicates + self.operational_expired + self.evidence_expired + ) + + def as_dict(self) -> dict[str, Any]: + """The task-return / log-line shape of this result.""" + return { + "removed": { + "blocked_duplicates": self.blocked_duplicates, + "operational_expired": self.operational_expired, + "evidence_expired": self.evidence_expired, + }, + "carried_over": self.carried_over, + "invalid_config_keys": list(self.invalid_config_keys), + } + + +def _streak_boundary_subquery(now: datetime) -> sa.Subquery: + """Per entity, the ``created_on`` of its newest streak-breaking row. + + Entities absent from this subquery have never been proven destroyed — + all their blocked rows form one current streak. Future-dated rows are + excluded so a skewed writer clock cannot push the boundary ahead of a + live streak and make its rows look resolved. + """ + table: sa.Table = PurgeAuditLog.__table__ + return ( + sa.select( + table.c.entity_type.label("entity_type"), + table.c.entity_uuid.label("entity_uuid"), + sa.func.max(table.c.created_on).label("boundary"), + ) + .where(table.c.status.in_(_STREAK_BREAKING_STATUSES)) + .where(table.c.entity_uuid.is_not(None)) + .where(table.c.created_on <= now) + .group_by(table.c.entity_type, table.c.entity_uuid) + .subquery("streak_boundary") + ) + + +def _survivor_subquery(now: datetime) -> sa.Subquery: + """Per entity, the ``created_on`` of its current streak's earliest row. + + The survivor is the earliest ``blocked`` row strictly newer than the + entity's boundary (or its earliest blocked row outright when no boundary + exists). Rows sharing that exact timestamp are all treated as survivors — + with microsecond precision ties are pathological, and keeping an extra + row errs on the preserving side. + """ + table: sa.Table = PurgeAuditLog.__table__ + boundary: sa.Subquery = _streak_boundary_subquery(now) + return ( + sa.select( + table.c.entity_type.label("entity_type"), + table.c.entity_uuid.label("entity_uuid"), + sa.func.min(table.c.created_on).label("survivor_created_on"), + ) + .select_from( + table.outerjoin( + boundary, + sa.and_( + table.c.entity_type == boundary.c.entity_type, + table.c.entity_uuid == boundary.c.entity_uuid, + ), + ) + ) + .where(table.c.status == STATUS_BLOCKED) + .where(table.c.entity_uuid.is_not(None)) + .where(table.c.created_on <= now) + .where( + sa.or_( + boundary.c.boundary.is_(None), + table.c.created_on > boundary.c.boundary, + ) + ) + .group_by(table.c.entity_type, table.c.entity_uuid) + .subquery("streak_survivor") + ) + + +def _preceded_by_unresolved_attempt(table: sa.Table) -> sa.ColumnElement[bool]: + """Whether an unresolved (``pending``) attempt precedes this row. + + A ``pending`` row is the only thing that can insert a streak boundary + into *history*: every other write lands at ``now``, newer than every + existing row, whereas reconciliation and the purge path finalize a + pending row **in place**, keeping its original ``created_on``. So a + pending row sitting between two blocked rows is a boundary that may + appear at any moment, and the blocked row after it would become the new + streak's survivor — the very row pruning must never delete. + """ + pending: sa.Table = table.alias("unresolved_attempt") + return sa.exists( + sa.select(sa.literal(1)) + .select_from(pending) + .where( + sa.and_( + pending.c.status == STATUS_PENDING, + pending.c.entity_type == table.c.entity_type, + pending.c.entity_uuid == table.c.entity_uuid, + pending.c.created_on < table.c.created_on, Review Comment: A pending row at the same timestamp is not considered to precede this block. Existing MySQL audit rows can share a timestamp, so `blocked(T0) → pending(T1) → blocked(T1)` lets the later block be pruned; if the pending row later finalizes, that block should become the streak survivor. Could this preserve equal-timestamp successors (or add a deterministic ordering key)? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
