acts/CI/ccache_snapshot.py

0001 #!/usr/bin/env python3
0002 # /// script
0003 # requires-python = ">=3.12"
0004 # dependencies = ["boto3==1.42.0"]
0005 # ///
0006 """Restore, publish, and prune bounded ccache snapshots; keep compilation lookups local.
0007
0008 Overlapping publishers may leave an older valid snapshot as latest.
0009 Readers are always anonymous. Remote cache failures never fail the build.
0010 """
0011
0012 import argparse
0013 import hashlib
0014 import json
0015 import os
0016 from pathlib import Path, PurePosixPath
0017 import re
0018 import tarfile
0019 import tempfile
0020 import uuid
0021
0022
0023 def report(message):
0024     print(message, flush=True)
0025     if path := os.environ.get("GITHUB_STEP_SUMMARY"):
0026         with open(path, "a") as output:
0027             output.write(f"- {message}\n")
0028
0029
0030 def digest(path):
0031     with path.open("rb") as stream:
0032         return hashlib.file_digest(stream, "sha256").hexdigest()
0033
0034
0035 def read_manifest(client, bucket, prefix):
0036     try:
0037         response = client.get_object(Bucket=bucket, Key=prefix + "latest.json")
0038     except Exception as error:
0039         if getattr(error, "response", {}).get("Error", {}).get("Code") in (
0040             "NoSuchKey",
0041             "404",
0042         ):
0043             return None
0044         raise
0045     with response["Body"] as stream:
0046         manifest = json.loads(stream.read(16385))
0047     if (
0048         manifest["format"] != 1
0049         or not re.fullmatch(
0050             re.escape(prefix) + r"snapshots/[0-9]+-[0-9]+-[0-9a-f]{32}\.tar",
0051             manifest["key"],
0052         )
0053         or not re.fullmatch(r"[0-9a-f]{64}", manifest["sha256"])
0054         or any(
0055             type(manifest[field]) is not int or manifest[field] < 1
0056             for field in ("run_number", "run_attempt")
0057         )
0058     ):
0059         raise ValueError("Invalid ccache snapshot manifest")
0060     return manifest
0061
0062
0063 def extract(archive, destination):
0064     with tarfile.open(archive, "r:") as source:
0065         for member in source:
0066             path = PurePosixPath(member.name)
0067             if (
0068                 path.is_absolute()
0069                 or ".." in path.parts
0070                 or not (member.isfile() or member.isdir())
0071             ):
0072                 raise ValueError("Unsafe ccache archive entry")
0073             source.extract(member, path=destination, filter="data")
0074
0075
0076 def restore(client, bucket, prefix, cache):
0077     # The job uses a fresh, dedicated cache directory. Never combine partial/stale
0078     # restores with a new snapshot, or discard a caller's existing local cache.
0079     cache.mkdir(parents=True, exist_ok=True)
0080     if any(cache.iterdir()):
0081         raise ValueError("Restore requires an empty ccache directory")
0082     manifest = read_manifest(client, bucket, prefix)
0083     if manifest is None:
0084         report("ccache snapshot: no published snapshot; starting empty")
0085         return
0086     with tempfile.TemporaryDirectory(dir=cache.parent) as directory:
0087         staging = Path(directory)
0088         archive = staging / "snapshot.tar"
0089         client.download_file(bucket, manifest["key"], str(archive))
0090         if digest(archive) != manifest["sha256"]:
0091             raise ValueError("ccache snapshot checksum mismatch")
0092         unpacked = staging / "unpacked"
0093         unpacked.mkdir()
0094         extract(archive, unpacked)
0095         cache.rmdir()
0096         unpacked.rename(cache)
0097         report(
0098             f"ccache restored main run {manifest['run_number']}, "
0099             f"attempt {manifest['run_attempt']}"
0100         )
0101
0102
0103 def publish(client, bucket, prefix, cache, run_number, run_id, attempt):
0104     # Best-effort freshness check, not a lock: another publisher may finish
0105     # after this read. An older snapshot is still safe for ccache to use.
0106     order = (run_number, attempt)
0107     current = read_manifest(client, bucket, prefix)
0108     if current and (current["run_number"], current["run_attempt"]) >= order:
0109         report("ccache publication skipped: an equal or newer main run exists")
0110         return
0111     with tempfile.TemporaryDirectory(dir=cache.parent) as directory:
0112         archive = Path(directory) / "snapshot.tar"
0113         # ccache already compresses entries. Avoid a second compression pass.
0114         with tarfile.open(archive, "w:") as output:
0115             for entry in sorted(cache.iterdir()):
0116                 output.add(entry, arcname=entry.name)
0117         manifest = {
0118             "format": 1,
0119             "key": f"{prefix}snapshots/{run_id}-{attempt}-{uuid.uuid4().hex}.tar",
0120             "sha256": digest(archive),
0121             "run_number": run_number,
0122             "run_attempt": attempt,
0123             "commit": os.environ.get("GITHUB_SHA", ""),
0124         }
0125         client.upload_file(str(archive), bucket, manifest["key"])
0126         # A failed/cancelled upload cannot replace the currently usable snapshot.
0127         client.put_object(
0128             Bucket=bucket,
0129             Key=prefix + "latest.json",
0130             Body=json.dumps(manifest).encode(),
0131             ContentType="application/json",
0132             CacheControl="no-cache",
0133         )
0134         report(
0135             f"ccache snapshot published ({archive.stat().st_size / 1024**2:.1f} MiB)"
0136         )
0137
0138
0139 def list_snapshots(client, bucket, prefix):
0140     # Finish listing before reading pointers. Later uploads are not candidates.
0141     variants = {}
0142     pattern = re.compile(
0143         re.escape(prefix) + r".+/snapshots/[0-9]+-[0-9]+-[0-9a-f]{32}\.tar"
0144     )
0145     for page in client.get_paginator("list_objects_v2").paginate(
0146         Bucket=bucket, Prefix=prefix
0147     ):
0148         for item in page.get("Contents", []):
0149             key = item["Key"]
0150             if pattern.fullmatch(key):
0151                 variant = key.rsplit("snapshots/", 1)[0]
0152                 variants.setdefault(variant, {})[key] = item["Size"]
0153
0154     return variants
0155
0156
0157 def prune_variant(client, bucket, variant, objects, dry_run):
0158     manifest = read_manifest(client, bucket, variant)
0159     if manifest is None:
0160         raise ValueError("Missing latest manifest")
0161     protected = {manifest["key"]}
0162     keys = sorted(objects)
0163     for offset in range(0, len(keys), 1000):
0164         # Protect both observed pointers if publication overlaps cleanup.
0165         current = read_manifest(client, bucket, variant)
0166         if current is None:
0167             raise ValueError("Missing latest manifest")
0168         protected.add(current["key"])
0169         candidates = [
0170             key for key in keys[offset : offset + 1000] if key not in protected
0171         ]
0172         if not candidates:
0173             continue
0174         if dry_run:
0175             yield candidates
0176             continue
0177         result = client.delete_objects(
0178             Bucket=bucket,
0179             Delete={"Objects": [{"Key": key} for key in candidates]},
0180         )
0181         deleted = [item["Key"] for item in result.get("Deleted", [])]
0182         yield deleted
0183         if result.get("Errors") or set(deleted) != set(candidates):
0184             raise RuntimeError("Incomplete deletion")
0185
0186
0187 def prune(client, bucket, prefix, dry_run=False):
0188     variants = list_snapshots(client, bucket, prefix)
0189     total = sum(sum(objects.values()) for objects in variants.values())
0190     removed_count = removed_bytes = failures = 0
0191     for variant, objects in sorted(variants.items()):
0192         try:
0193             for deleted in prune_variant(client, bucket, variant, objects, dry_run):
0194                 removed_count += len(deleted)
0195                 removed_bytes += sum(objects[key] for key in deleted)
0196         except Exception as error:
0197             failures += 1
0198             report(f"ccache pruning failed for {variant} ({type(error).__name__})")
0199     verb = "would delete" if dry_run else "deleted"
0200     report(
0201         f"ccache pruning: {verb} {removed_count} archives "
0202         f"({removed_bytes / 1024**2:.1f} MiB); "
0203         f"retained {(total - removed_bytes) / 1024**2:.1f} MiB of listed archives"
0204     )
0205     if failures:
0206         raise RuntimeError(f"ccache pruning encountered {failures} failures")
0207
0208
0209 def make_client(writable):
0210     import boto3
0211     from botocore import UNSIGNED
0212     from botocore.config import Config
0213
0214     options = dict(
0215         connect_timeout=10,
0216         read_timeout=60,
0217         retries={"mode": "standard", "max_attempts": 2},
0218         s3={"addressing_style": "path"},
0219         # CERN's S3 gateway does not need optional streaming checksum trailers.
0220         request_checksum_calculation="when_required",
0221         response_checksum_validation="when_required",
0222     )
0223     if not writable:
0224         options["signature_version"] = UNSIGNED
0225     return boto3.client(
0226         "s3",
0227         endpoint_url=os.environ["CACHE_ENDPOINT"],
0228         region_name=os.environ["CACHE_REGION"],
0229         config=Config(**options),
0230     )
0231
0232
0233 def prune_main(dry_run):
0234     if not (
0235         os.environ.get("GITHUB_REPOSITORY") == "acts-project/acts"
0236         and os.environ.get("GITHUB_REF") == "refs/heads/main"
0237         and os.environ.get("GITHUB_EVENT_NAME") in ("schedule", "workflow_dispatch")
0238         and os.environ.get("CACHE_PREFIX")
0239         == "acts-sccache/ccache-snapshots/acts-project/acts/"
0240     ):
0241         report("ccache pruning refused: requires an authorized main cleanup run")
0242         return 1
0243     try:
0244         prune(
0245             make_client(True),
0246             os.environ["CACHE_BUCKET"],
0247             os.environ["CACHE_PREFIX"],
0248             dry_run,
0249         )
0250     except Exception as error:
0251         report(f"ccache pruning failed ({type(error).__name__})")
0252         return 1
0253     return 0
0254
0255
0256 def main():
0257     parser = argparse.ArgumentParser(description=__doc__)
0258     parser.add_argument("operation", choices=["restore", "publish", "prune"])
0259     parser.add_argument("--dry-run", action="store_true", help="Preview pruning only")
0260     args = parser.parse_args()
0261     if args.dry_run and args.operation != "prune":
0262         parser.error("--dry-run requires prune")
0263     if args.operation == "prune":
0264         return prune_main(args.dry_run)
0265     writable = args.operation == "publish"
0266     if writable and not (
0267         os.environ.get("GITHUB_REF") == "refs/heads/main"
0268         and (
0269             os.environ.get("GITHUB_EVENT_NAME") == "push"
0270             or (
0271                 os.environ.get("CACHE_ALLOW_MAIN_NON_PUSH") == "true"
0272                 and os.environ.get("GITHUB_EVENT_NAME")
0273                 in ("schedule", "workflow_dispatch")
0274             )
0275         )
0276     ):
0277         report("ccache publication skipped: only authorized main runs may publish")
0278         return 0
0279     try:
0280         client = make_client(writable)
0281         bucket = os.environ["CACHE_BUCKET"]
0282         prefix = os.environ["CACHE_PREFIX"].rstrip("/") + "/"
0283         cache = Path(os.environ["CCACHE_DIR"])
0284         if writable:
0285             publish(
0286                 client,
0287                 bucket,
0288                 prefix,
0289                 cache,
0290                 int(os.environ["GITHUB_RUN_NUMBER"]),
0291                 int(os.environ["GITHUB_RUN_ID"]),
0292                 int(os.environ["GITHUB_RUN_ATTEMPT"]),
0293             )
0294         else:
0295             restore(client, bucket, prefix, cache)
0296     except Exception as error:
0297         # Upload credentials are never included in diagnostics.
0298         print(
0299             f"::warning::ccache snapshot {args.operation} failed ({type(error).__name__}); "
0300             "continuing without remote cache updates",
0301             flush=True,
0302         )
0303         report(f"ccache snapshot {args.operation}: failed ({type(error).__name__})")
0304     return 0
0305
0306
0307 if __name__ == "__main__":
0308     raise SystemExit(main())