Mapping YugabyteDB Snapshots Back to Their Databases

				
					#!/usr/bin/env python3
import json
import os
import re
import subprocess
import sys


def yb_admin(args):
    """
    Run yb-admin with the configured master address.

    Environment:
      YB_MASTER = "host:port" or comma-separated list (default 127.0.0.1)
    """
    master_addrs = os.environ.get("YB_MASTER", "127.0.0.1")
    cmd = ["yb-admin", "--init_master_addrs", master_addrs] + args
    return subprocess.check_output(cmd, text=True)


def parse_summary(text):
    """
    Parse `yb-admin list_snapshots` output into a dict keyed by snapshot UUID.

    Expected format (simplified):

      Snapshot UUID                       State      Creation Time                Previous Snapshot Time
      e3cd52fb-...                        COMPLETE   2025-11-23 03:52:56.308819   2112-09-17 23:53:47.370495
    """
    snaps = {}

    # id state snapshot_time [previous_snapshot_time]
    pattern = re.compile(r"^(\S+)\s+(\S+)\s+(\S+\s+\S+)(?:\s+(\S+\s+\S+))?")

    for line in text.splitlines():
        s = line.strip()
        if not s:
            continue

        # skip header/footer lines
        if s.startswith("Snapshot UUID") or s.startswith("No snapshot restorations"):
            continue

        m = pattern.match(s)
        if not m:
            continue

        sid, state, ts, prev_ts = m.groups()
        snaps[sid] = {
            "id": sid,
            "state": state,
            "snapshot_time": ts,
            "previous_snapshot_time": prev_ts or None,
            "details": [],
        }

    return snaps


def parse_details(text):
    """
    Parse `yb-admin list_snapshots show_details` output.

    Example (simplified):

      Snapshot UUID                     State     Creation Time
      e3cd52fb-...                      COMPLETE  2025-11-23 03:52:56.308819
        {"type":"NAMESPACE", ...}
        {"type":"TABLE", ...}
      4444301c-...                      COMPLETE  2025-11-23 03:55:46.552879
        {"type":"NAMESPACE", ...}
        {"type":"TABLE", ...}
      No snapshot restorations
    """
    snap_header = re.compile(r"^([0-9a-f-]+)\s+")
    grouped = {}
    current = None

    for line in text.splitlines():
        s = line.strip()
        if not s:
            continue

        if s.startswith("Snapshot UUID") or s.startswith("No snapshot restorations"):
            continue

        m = snap_header.match(s)
        if m:
            current = m.group(1)
            grouped.setdefault(current, [])
            continue

        if s.startswith("{") and current:
            grouped[current].append(json.loads(s))

    return grouped


def main():
    try:
        summary_txt = yb_admin(["list_snapshots"])
        details_txt = yb_admin(["list_snapshots", "show_details"])
    except subprocess.CalledProcessError as e:
        sys.stderr.write(f"yb-admin failed:\n{e.output}\n")
        sys.exit(1)

    summary = parse_summary(summary_txt)
    details = parse_details(details_txt)

    # stitch details + derive database metadata
    for sid, snap in summary.items():
        dlist = details.get(sid, [])
        snap["details"] = dlist

        # find NAMESPACE entries (YSQL DB or YCQL keyspace)
        namespaces = [d for d in dlist if d.get("type") == "NAMESPACE"]
        if namespaces:
            ns_entry = namespaces[0]
            ns_data = ns_entry.get("data", {})

            snap["database_name"] = ns_data.get("name")
            snap["database_type"] = ns_data.get("database_type")
            snap["database_namespace_id"] = ns_entry.get("id")

    snapshots = list(summary.values())

    print(json.dumps({"snapshots": snapshots}, indent=2))


if __name__ == "__main__":
    main()