duckdb/scripts/github_cache_usage.py

#!/usr/bin/env python3
"""
Convert `gh cache list` output into a detailed CSV with bucket mappings.

Example: generate the CSV directly from GitHub cache output

    gh cache list --repo duckdb/duckdb --limit 1000 | \
      python3 scripts/github_cache_usage.py

Example: summarize entry count, total GiB, and percentage by bucket

    duckdb -c "
      SELECT
        bucket,
        COUNT(*) AS entry_count,
        ROUND(SUM(size_gib), 3) AS total_gib,
        ROUND(100.0 * SUM(size_gib) / SUM(SUM(size_gib)) OVER ()) AS pct_of_total,
        MAX(last_accessed_at) AS last_usage,
      FROM read_csv_auto('github_cache_usage.csv')
      GROUP BY bucket
      ORDER BY total_gib DESC;
    "
"""

import argparse
import sys
import csv
import re
from pathlib import Path
from typing import Dict, Iterable, List, Optional, Tuple

MAIN = {
    "linux-relassert",
    "linux-configs",
    "vector-sizes",
    "threadsan",
    "no-string-inline",
}

NIGHTLYTESTS = {
    "valgrind",
    "release-assert",
    "release-assert-clang",
    "release-assert-osx-storage",
    "smaller-binary",
    "hash-zero",
    "extension-updating",
    "regression-test-memory-safety",
    "vector-and-block-sizes",
    "linux-relassert-configs",
    "force-blocking-sink-source",
    "storage-initialization",
}

EXTENSIONS_EXPLICIT = {
    "autoload-tests",
    "check-load-install-extensions",
}


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Parse `gh cache list` from stdin and classify github cache items to CI workflows."
    )
    parser.add_argument(
        "--output",
        default="github_cache_usage.csv",
        help="Output CSV path (default: github_cache_usage.csv).",
    )
    parser.add_argument(
        "--workflows-dir",
        default=".github/workflows",
        help="Workflow directory used to map ccache keys to jobs/workflows.",
    )
    return parser.parse_args()


def parse_size_to_gib(size_raw: str) -> float:
    parts = size_raw.strip().split()
    if len(parts) != 2:
        return 0.0
    value_str, unit = parts
    try:
        value = float(value_str)
    except ValueError:
        return 0.0
    if unit == "GiB":
        return value
    if unit == "MiB":
        return value / 1024.0
    if unit == "KiB":
        return value / (1024.0 * 1024.0)
    return 0.0


def normalize_name(cache_key: str) -> str:
    name = re.sub(r"-202\d-\d{2}-\d{2}.*$", "", cache_key)
    while name.startswith("ccache-"):
        name = name[len("ccache-") :]
    return name


def classify_bucket(normalized: str) -> Tuple[str, str]:
    if normalized.startswith("extension-distribution-"):
        return "extensions", "prefix rule: extension-distribution-*"
    if normalized in EXTENSIONS_EXPLICIT:
        return "extensions", "explicit extensions allowlist"
    if normalized in MAIN:
        return "main", "explicit main job list"
    if normalized in NIGHTLYTESTS:
        return "nightlytests", "explicit nightlytests job list"
    if normalized in {"xcode-debug", "xcode-release"}:
        return "osx", "explicit osx job list"
    if normalized == "bundle-osx-static-libs":
        return "bundlestaticlibs", "explicit bundle static libs job"
    if normalized == "linux-release":
        return "linux-release", "explicit linux-release job list"
    return "unmapped/other", "no mapping rule matched"


def extract_ccache_job_mapping(workflows_dir: Path) -> Dict[str, str]:
    mapping: Dict[str, str] = {}
    for workflow in sorted(workflows_dir.glob("*.yml")):
        current_job: Optional[str] = None
        in_jobs = False
        for raw_line in workflow.read_text(encoding="utf-8").splitlines():
            line = raw_line.rstrip("\n")
            if line.startswith("jobs:"):
                in_jobs = True
                current_job = None
                continue
            if in_jobs and line and not line.startswith(" "):
                in_jobs = False
                current_job = None
                continue
            if not in_jobs:
                continue
            match = re.match(r"^(\s{1,2})([A-Za-z0-9_-]+):\s*$", line)
            if match:
                current_job = match.group(2)
                continue
            if "uses:" in line and "ccache-action" in line and current_job:
                mapping.setdefault(current_job, str(workflow))
    return mapping


def parse_gh_cache_lines(lines: Iterable[str]) -> List[Dict[str, str]]:
    rows: List[Dict[str, str]] = []
    for line in lines:
        text = line.strip()
        if not text:
            continue
        parts = text.split("\t")
        if len(parts) < 5:
            continue
        cache_id, cache_key, size_raw, created_at, last_accessed_at = parts[:5]
        rows.append(
            {
                "cache_id": cache_id,
                "cache_key": cache_key,
                "size_raw": size_raw,
                "created_at": created_at,
                "last_accessed_at": last_accessed_at,
            }
        )
    return rows


def main() -> None:
    args = parse_args()
    output_path = Path(args.output)
    workflows_dir = Path(args.workflows_dir)

    cache_lines = sys.stdin.read().splitlines()
    parsed_rows = parse_gh_cache_lines(cache_lines)
    job_to_workflow = extract_ccache_job_mapping(workflows_dir)

    fieldnames = [
        "cache_id",
        "cache_key",
        "size_raw",
        "size_gib",
        "created_at",
        "last_accessed_at",
        "normalized_name",
        "bucket",
        "workflow_file",
        "mapped_job",
        "mapping_reason",
    ]

    with output_path.open("w", encoding="utf-8", newline="") as out:
        writer = csv.DictWriter(out, fieldnames=fieldnames)
        writer.writeheader()
        for row in parsed_rows:
            normalized = normalize_name(row["cache_key"])
            bucket, rule_reason = classify_bucket(normalized)
            mapped_job = normalized if normalized in job_to_workflow else ""
            workflow_file = job_to_workflow.get(normalized, "")
            reasons = [rule_reason]
            if mapped_job:
                reasons.append("job-name match")
            if row["cache_key"].startswith("ccache-ccache-"):
                reasons.append("double-ccache normalized")

            writer.writerow(
                {
                    "cache_id": row["cache_id"],
                    "cache_key": row["cache_key"],
                    "size_raw": row["size_raw"],
                    "size_gib": f"{parse_size_to_gib(row['size_raw']):.6f}",
                    "created_at": row["created_at"],
                    "last_accessed_at": row["last_accessed_at"],
                    "normalized_name": normalized,
                    "bucket": bucket,
                    "workflow_file": workflow_file,
                    "mapped_job": mapped_job,
                    "mapping_reason": "; ".join(reasons),
                }
            )
    print("wrote CSV to {}".format(str(output_path)))


if __name__ == "__main__":
    main()