#!/usr/local/bin/python # /// script # requires-python = ">=3.9" # dependencies = [ # "click", # "psycopg2-binary", # ] # /// """Reconcile the L7|ESP database against the data volume after a restore. Two reports, both read-only: \b orphans files and pipeline run directories on the data volume with no record in the database dangling database records whose file or pipeline run directory is missing from the data volume Neither command modifies the database or the data volume. Quarantining or recovering files is a separate, deliberate step. Inside an L7|ESP container this runs with no arguments: the data directory comes from LAB7DATA and the database connection from L7ESP_DATABASE_URL or the libpq PG* environment variables. Elsewhere, pass --data-dir and --database-url (or set DATABASE_URL). Exit codes distinguish the three outcomes a runbook has to tell apart: \b 0 the report ran and found nothing 1 the report ran and found something 2 the invocation was wrong, e.g. a --data-dir that is not a data volume 3 the report could not run, e.g. an unreachable database """ from contextlib import contextmanager from datetime import datetime from datetime import timezone from os import environ from os import lstat from os import scandir from os.path import normpath from pathlib import Path from re import IGNORECASE from re import compile as compile_re from sys import exit as sys_exit from typing import Any from typing import Iterable from typing import Iterator from typing import List from typing import NamedTuple from typing import Optional from typing import Set from typing import Tuple from click import Choice from click import ClickException from click import Path as PathType from click import UsageError from click import echo from click import group from click import option from psycopg2 import Error as PsycopgError from psycopg2 import connect from psycopg2.extensions import ISOLATION_LEVEL_REPEATABLE_READ Row = Tuple[str, str, str] UUID_SUFFIX = compile_re(r"\.([0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12})$", IGNORECASE) FILE_RECORDS_SQL = """ SELECT r.uuid::text, r.name, f.file_url, r.archived FROM resource r JOIN lab7_file f ON f.lab7_file_id = r.resource_id """ PIPELINE_RUNS_SQL = """ SELECT r.uuid::text, r.name, r.meta->>'pi_dir', r.archived FROM resource r JOIN task_instance t ON t.instance_id = r.resource_id WHERE r.meta ? 'pi_dir' """ # L7|ESP writes this file into the data directory when it first creates the # database tables: a uuid4 plus a newline, mode 0444. Its presence is what # separates an initialized data volume from an empty directory or a wrong path. SITE_ID_FILE = "site_id.txt" class CheckAborted(ClickException): """The report could not be produced, so its result carries no information. Distinct from finding nothing, which is a real answer. Exits 3 so a runbook never reads a failed connection or an unreadable volume as a clean bill of health, and never reads it as a finding either. """ exit_code = 3 class DanglingReport(NamedTuple): """Missing-path rows, plus counts of the records that could not be checked.""" rows: List[Row] pathless: int # --- Environment defaults --- def default_data_dir() -> Path: """LAB7DATA inside a container, /opt/l7esp/data otherwise.""" return Path(environ.get("LAB7DATA", "/opt/l7esp/data")) def default_database_url() -> str: """DATABASE_URL, then L7ESP_DATABASE_URL, then libpq's PG* variables.""" return environ.get("DATABASE_URL") or environ.get("L7ESP_DATABASE_URL") or "" def database_url_source(database_url: str) -> str: """Name where the connection settings came from, without echoing them. A failed connection has to say which configuration to go and look at, but the URL itself routinely carries a password, so only its origin is named. """ if not database_url: return "the libpq environment variables (PGHOST, PGDATABASE, ...) or their defaults" if database_url == environ.get("DATABASE_URL"): return "$DATABASE_URL" if database_url == environ.get("L7ESP_DATABASE_URL"): return "$L7ESP_DATABASE_URL" return "--database-url" # --- Business logic --- def canonical(path: Path) -> Path: """Collapse redundant separators and up-level references in a path. Applied to both sides of every disk-against-database comparison so the two are normalized the same way. Deliberately does not resolve symlinks: that reads the filesystem, and would answer differently for a path the database recorded than for the same path found on disk. """ return Path(normpath(str(path))) def require_data_volume(data_dir: Path) -> None: """Refuse to run unless data_dir really is an L7|ESP data volume. Catches a wrong or empty --data-dir. """ if (data_dir / SITE_ID_FILE).is_file(): return raise UsageError( "{0} does not look like an L7|ESP data volume: no {1}.\n" "L7|ESP writes that file when it initializes the database, so every\n" "restored volume has one. Inside an L7|ESP container the default comes\n" "from LAB7DATA and needs no flag; elsewhere point --data-dir at the\n" "directory that holds files/ and pipeline/.".format(data_dir, SITE_ID_FILE) ) def walk_files(files_dir: Path) -> Iterator[Path]: """Yield every file under the files/ directory, symlinks included. A missing directory yields nothing rather than raising: an install that has never taken an upload has no files/, and that is not an error. Symlinks are reported by name but never followed, so a link out of the volume or back into it cannot send the walk somewhere else or around a loop. Each directory is closed before descending into its children, so the walk holds one handle at a time rather than one per level. """ if not files_dir.is_dir(): return subdirs: List[Path] = [] try: with scandir(files_dir) as entries: for entry in entries: if entry.is_symlink() or not entry.is_dir(follow_symlinks=False): yield Path(entry.path) else: subdirs.append(Path(entry.path)) except OSError as exc: raise CheckAborted( "could not read {0}: {1}.\n" "A partial walk would under-report orphans while looking complete, " "so nothing is reported.".format(files_dir, exc.strerror or exc) ) for subdir in subdirs: for path in walk_files(subdir): yield path def uuid_of(path: Path) -> Optional[str]: """Return the uuid suffix of a file name, or None if it has none. Lower-cased to match PostgreSQL's rendering of a uuid column, so a name written in upper case still reconciles against its record. """ match = UUID_SUFFIX.search(path.name) return match.group(1).lower() if match else None def mtime_of(path: Path) -> Optional[str]: """The entry's own modification time as an ISO 8601 UTC string. Returns None if the entry is gone, which happens when a file is deleted between the walk and this call; it is no longer an orphan, so its row is dropped rather than aborting the whole report. Does not follow symlinks: the question is what sits on this volume, not what it points at. """ try: return datetime.fromtimestamp(lstat(path).st_mtime, tz=timezone.utc).isoformat() except OSError: return None @contextmanager def read_only_connection(database_url: str) -> Iterator[Any]: """A single repeatable-read, read-only transaction for one command's queries. One connection, so both queries of a command see the same snapshot and the two reports cannot disagree about a row written between them. Read-only and repeatable-read are set on the session, so the guarantee that this tool changes nothing is enforced by the server rather than promised by the code. """ try: conn = connect(database_url) except PsycopgError as exc: raise CheckAborted( "could not connect to the L7|ESP database, so nothing was checked.\n" "{0}\n" "Connection settings came from {1}.".format(str(exc).strip(), database_url_source(database_url)) ) try: conn.set_session(isolation_level=ISOLATION_LEVEL_REPEATABLE_READ, readonly=True) yield conn finally: conn.close() def fetch(conn: Any, sql: str) -> List[tuple]: """Run one query inside the caller's read-only transaction.""" try: with conn.cursor() as cur: cur.execute(sql) return cur.fetchall() except PsycopgError as exc: raise CheckAborted( "could not query the L7|ESP database, so nothing was checked.\n" "{0}\n" "This usually means the schema is not the one this tool expects.".format(str(exc).strip()) ) def local_path(file_url: Optional[str]) -> Optional[Path]: """Strip the file:// scheme from a stored file URL. Returns None when the database recorded no path at all. That happens for a pipeline instance carrying pi_dir as a JSON null, which the `meta ? 'pi_dir'` test in PIPELINE_RUNS_SQL does not exclude, because that operator only asks whether the key is present. """ if not file_url: return None prefix = "file://" stripped = file_url[len(prefix) :] if file_url.startswith(prefix) else file_url return canonical(Path(stripped)) def recorded_paths(records: Iterable[tuple]) -> List[Path]: """Every path a record set actually carries, skipping the pathless ones.""" found = (local_path(record[2]) for record in records) return [path for path in found if path is not None] def orphan_files(files_dir: Path, known_uuids: Set[str]) -> List[Row]: """Files on disk whose uuid is not in the database: (uuid, path, mtime). Matched on the uuid in the file name, not on the path, so this report is unaffected by where the volume is mounted. A file whose name carries no uuid is not something L7|ESP wrote and is left alone. """ rows: List[Row] = [] for path in walk_files(files_dir): uuid = uuid_of(path) if uuid is None or uuid in known_uuids: continue modified = mtime_of(path) if modified is not None: rows.append((uuid, str(path), modified)) return rows def orphan_pipeline_dirs(pipeline_dir: Path, known_dirs: Set[Path]) -> List[Row]: """Run directories on disk not referenced by any pipeline instance. A missing directory yields nothing rather than raising: an install that has never run a pipeline has no pipeline/, and that is not an error. """ rows: List[Row] = [] if not pipeline_dir.is_dir(): return rows try: with scandir(pipeline_dir) as entries: candidates = [Path(entry.path) for entry in entries if entry.is_dir(follow_symlinks=False)] except OSError as exc: raise CheckAborted( "could not read {0}: {1}.\n" "A partial listing would under-report orphans while looking " "complete, so nothing is reported.".format(pipeline_dir, exc.strerror or exc) ) for path in candidates: if canonical(path) in known_dirs: continue modified = mtime_of(path) if modified is not None: rows.append(("", str(path), modified)) return rows def in_scope(path: Path, data_dir: Path, scope: str) -> bool: """Whether a recorded path sits in the subtree --scope asked for. Scope selects a subtree of the data directory, files/ or pipeline/, not a record type: a lab7_file record's path can be under either one. """ if scope == "all": return True return canonical(path).is_relative_to(canonical(data_dir / scope)) def dangling_records( records: Iterable[tuple], include_archived: bool, data_dir: Path, scope: str, ) -> DanglingReport: """Records whose recorded path does not exist on the volume. Out-of-scope records are dropped before the existence test rather than after, so --scope files does not stat every pipeline path only to discard the answer. On a network filesystem each of those stats is a round trip. A record with no path recorded cannot be missing from the volume, so it is counted rather than reported. """ rows: List[Row] = [] pathless = 0 for uuid, name, location, archived in records: if archived and not include_archived: continue path = local_path(location) if path is None: pathless += 1 continue if not in_scope(path, data_dir, scope): continue if not path.exists(): rows.append((uuid, name or "", str(path))) return DanglingReport(rows, pathless) def tsv_field(value: str) -> str: """Escape the characters that would break TSV column or row framing. A file name may legally contain a tab or a newline, which would otherwise silently shift every later column or split one row into two. """ return value.replace("\\", "\\\\").replace("\t", "\\t").replace("\r", "\\r").replace("\n", "\\n") def render(rows: List[Row], headers: Row, output: Optional[Path]) -> None: """Write rows as TSV to output, or to stdout when output is None.""" lines = ["\t".join(headers)] lines += ["\t".join(tsv_field(field) for field in row) for row in rows] if output is None: echo("\n".join(lines)) return if not output.parent.is_dir(): raise UsageError("cannot write {0}: {1} is not a directory.".format(output, output.parent)) output.write_text("\n".join(lines) + "\n") echo("wrote {0} rows to {1}".format(len(rows), output), err=True) def note_unchecked(count: int, reason: str) -> None: """Report records that were skipped, so a clean report is not overread.""" if count: echo("note: {0} records {1} and were not checked".format(count, reason), err=True) def orphans( data_dir: Path, database_url: str, scope: str, output: Optional[Path], ) -> int: """Report files and run directories on disk with no database record.""" require_data_volume(data_dir) rows: List[Row] = [] with read_only_connection(database_url) as conn: if scope != "pipeline": known_uuids = {row[0] for row in fetch(conn, FILE_RECORDS_SQL)} rows += orphan_files(data_dir / "files", known_uuids) if scope != "files": known_dirs = set(recorded_paths(fetch(conn, PIPELINE_RUNS_SQL))) rows += orphan_pipeline_dirs(data_dir / "pipeline", known_dirs) render(rows, ("uuid", "path", "modified"), output) return len(rows) def dangling( data_dir: Path, database_url: str, scope: str, include_archived: bool, output: Optional[Path], ) -> int: """Report database records whose file or run directory is missing. Both stores are queried whichever scope is asked for, because the table a record came from does not say which subtree its path is under: a lab7_file record can point under pipeline/. """ require_data_volume(data_dir) with read_only_connection(database_url) as conn: records = fetch(conn, FILE_RECORDS_SQL) + fetch(conn, PIPELINE_RUNS_SQL) report = dangling_records(records, include_archived, data_dir, scope) note_unchecked(report.pathless, "have no path recorded") render(report.rows, ("uuid", "name", "missing path"), output) return len(report.rows) # --- CLI --- database_url_option = option( "--database-url", default=default_database_url, show_default="$DATABASE_URL, $L7ESP_DATABASE_URL, or libpq PG* variables", help="PostgreSQL connection URL, e.g. postgresql://user:pass@host:5432/l7esp", ) data_dir_option = option( "--data-dir", type=PathType(file_okay=False, exists=True, path_type=Path), default=default_data_dir, show_default="$LAB7DATA or /opt/l7esp/data", help="L7|ESP data directory containing files/ and pipeline/", ) scope_option = option( "--scope", type=Choice(["all", "files", "pipeline"]), default="all", show_default=True, help="Restrict to uploaded files (files/), pipeline runs (pipeline/), or both", ) output_option = option( "--output", type=PathType(dir_okay=False, writable=True, path_type=Path), default=None, help="Write a TSV report to this file instead of stdout", ) @group(help=__doc__) def cli() -> None: pass @cli.command(name="orphans", help="Files and pipeline run directories on disk with no database record.") @database_url_option @data_dir_option @scope_option @output_option def cli_orphans( database_url: str, data_dir: Path, scope: str, output: Optional[Path], ) -> None: sys_exit(1 if orphans(data_dir, database_url, scope, output) else 0) @cli.command(name="dangling", help="Database records whose file or pipeline run directory is missing.") @database_url_option @data_dir_option @scope_option @option("--include-archived", is_flag=True, help="Also check records L7|ESP has soft-deleted") @output_option def cli_dangling( database_url: str, data_dir: Path, scope: str, include_archived: bool, output: Optional[Path], ) -> None: sys_exit(1 if dangling(data_dir, database_url, scope, include_archived, output) else 0) if __name__ == "__main__": cli()