diff --git a/.github/images.yml b/.github/images.yml index c04fb7055..fafcecff5 100644 --- a/.github/images.yml +++ b/.github/images.yml @@ -23,9 +23,6 @@ images: - path: ./images/python-aws-bash image: ghcr.io/spack/python-aws-bash:0.0.2 - - path: ./images/snapshot-release-tags - image: ghcr.io/spack/snapshot-release-tags:0.0.4 - - path: ./images/cache-indexer image: ghcr.io/spack/cache-indexer:0.0.6 @@ -35,8 +32,8 @@ images: - path: ./images/ci-prune-buildcache image: ghcr.io/spack/ci-prune-buildcache:0.0.5 - - path: ./images/protected-publish - image: ghcr.io/spack/protected-publish:0.0.9 + - path: ./images/buildcache-tools + image: ghcr.io/spack/buildcache-tools:0.0.1 - path: ./images/retry-trigger-jobs image: ghcr.io/spack/retry-trigger-jobs:0.0.2 diff --git a/images/protected-publish/Dockerfile b/images/buildcache-tools/Dockerfile similarity index 84% rename from images/protected-publish/Dockerfile rename to images/buildcache-tools/Dockerfile index 73503dd07..3836e2130 100644 --- a/images/protected-publish/Dockerfile +++ b/images/buildcache-tools/Dockerfile @@ -13,7 +13,8 @@ RUN uv pip install --system --no-cache-dir -r /srcs/requirements.txt COPY pkg /srcs/pkg COPY --chmod=755 migrate.sh /srcs/migrate.sh +COPY --chmod=755 run.sh /srcs/run.sh ENV PYTHONUNBUFFERED=1 WORKDIR /srcs -ENTRYPOINT ["python", "-m", "pkg.publish"] +ENTRYPOINT ["bash", "/srcs/run.sh"] diff --git a/images/protected-publish/README.md b/images/buildcache-tools/README.md similarity index 100% rename from images/protected-publish/README.md rename to images/buildcache-tools/README.md diff --git a/images/protected-publish/TESTING.txt b/images/buildcache-tools/TESTING.txt similarity index 100% rename from images/protected-publish/TESTING.txt rename to images/buildcache-tools/TESTING.txt diff --git a/images/protected-publish/migrate.sh b/images/buildcache-tools/migrate.sh similarity index 100% rename from images/protected-publish/migrate.sh rename to images/buildcache-tools/migrate.sh diff --git a/images/protected-publish/migrate_job.yaml b/images/buildcache-tools/migrate_job.yaml similarity index 100% rename from images/protected-publish/migrate_job.yaml rename to images/buildcache-tools/migrate_job.yaml diff --git a/images/protected-publish/migrate_service_account.yaml b/images/buildcache-tools/migrate_service_account.yaml similarity index 100% rename from images/protected-publish/migrate_service_account.yaml rename to images/buildcache-tools/migrate_service_account.yaml diff --git a/images/protected-publish/pkg/__init__.py b/images/buildcache-tools/pkg/__init__.py similarity index 100% rename from images/protected-publish/pkg/__init__.py rename to images/buildcache-tools/pkg/__init__.py diff --git a/images/buildcache-tools/pkg/common.py b/images/buildcache-tools/pkg/common.py new file mode 100644 index 000000000..157e09ff0 --- /dev/null +++ b/images/buildcache-tools/pkg/common.py @@ -0,0 +1,712 @@ +import contextlib +import hashlib +import json +import logging +import os +import re +import requests +import shutil +import stat +import subprocess +import tempfile +from collections import defaultdict +from concurrent.futures import as_completed, ThreadPoolExecutor +from datetime import datetime, timezone, timedelta +from typing import Dict, List, Optional + +import boto3 +import boto3.session +from boto3.s3.transfer import TransferConfig + +try: + import sentry_sdk + sentry_sdk.init( + # This cron job only runs once weekly, + # so just record all transactions. + traces_sample_rate=1.0, + ) +except ImportError: + print("Sentry Disabled") + + +logging.basicConfig(level=logging.INFO) +logging.getLogger("boto3").setLevel(logging.ERROR) +logging.getLogger("botocore").setLevel(logging.ERROR) +logging.getLogger("urllib3").setLevel(logging.ERROR) + + +SPACK_REPO = "https://github.com/spack/spack" +PACKAGES_REPO = "https://github.com/spack/spack-packages" + +TIMESTAMP_AND_SIZE = r"^[\d]{4}-[\d]{2}-[\d]{2}\s[\d]{2}:[\d]{2}:[\d]{2}\s+\d+\s+" +TIMESTAMP_PATTERN = "%Y-%m-%d %H:%M:%S" + +SPACK_PUBLIC_KEY_LOCATION = "https://spack.github.io/keys" +SPACK_PUBLIC_KEY_NAME = "spack-public-binary-key.pub" +TARBALL_MEDIA_TYPE = "application/vnd.spack.install.v2.tar+gzip" +SPEC_METADATA_MEDIA_TYPE = "application/vnd.spack.spec.v5+json" + +REGEX_LISTING_DATA = re.compile(r"^([\d]{4}-[\d]{2}-[\d]{2}\s[\d]{2}:[\d]{2}:[\d]{2})\s+(\d+)\s+(.+)") + +#: regular expressions designed to match "aws s3 ls" output +REGEX_V2_SIGNED_SPECFILE_RELATIVE = re.compile( + rf"(.+)(/build_cache/.+-)([^\.]+)(\.spec\.json\.sig)$" +) +REGEX_V2_ARCHIVE_RELATIVE = re.compile( + rf"(.+)(/build_cache/.+-)([^\.]+)(\.spack)$" +) +REGEX_V3_SIGNED_SPECFILE_RELATIVE = re.compile( + rf"(.+)(/v3/manifests/spec/.+-)([^-\.]+)(\.spec\.manifest\.json)$" +) + +#: Regular expression to pull spec contents out of clearsigned signature +#: file. +CLEARSIGN_FILE_REGEX = re.compile( + ( + r"^-----BEGIN PGP SIGNED MESSAGE-----" + r"\s+Hash:\s+[^\s]+\s+(.+)-----BEGIN PGP SIGNATURE-----" + ), + re.MULTILINE | re.DOTALL, +) + +#: regex to capture bucket name from an s3 url +REGEX_S3_BUCKET = re.compile(r"s3://([^/]+)/") + +#: Values used to config multi-part s3 copies +MB = 1024**2 +MULTIPART_THRESHOLD = 100 * MB +MULTIPART_CHUNKSIZE = 20 * MB +MAX_CONCURRENCY = 10 +USE_THREADS = True + +SNAPSHOT_TAG_REGEXES = [ + re.compile(r"^develop-[\d]{4}-[\d]{2}-[\d]{2}$"), + re.compile(r"^v([\d]+)\.([\d]+)\.[\d]+$"), +] + +PROTECTED_BRANCH_REGEXES = [ + re.compile(r"^develop$"), + re.compile(r"^releases/v[\d]+\.[\d]+$"), +] + + +LOGGER = logging.getLogger(__name__) + + +def download_and_import_key(gpg_home: str, tmpdir: str, force: bool) -> str | None: + """Download spack public signing key and import it""" + if os.path.isdir(gpg_home): + if force is True: + shutil.rmtree(gpg_home) + else: + return None + + mode_owner_rwe = stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR + os.makedirs(gpg_home, mode=mode_owner_rwe) + + public_key_url = f"{SPACK_PUBLIC_KEY_LOCATION}/{SPACK_PUBLIC_KEY_NAME}" + public_key_id = "2C8DD3224EF3573A42BD221FA8E0CA3C1C2ADA2F" + + # Fetch the public key and write it to a file to be imported + tmp_key_path = os.path.join(tmpdir, SPACK_PUBLIC_KEY_NAME) + response = requests.get(public_key_url) + with open(tmp_key_path, "w") as f: + f.write(response.text) + + # Also write an ownertrust file to be imported + ownertrust_path = os.path.join(tmpdir, "trustfile") + with open(ownertrust_path, "w") as f: + f.write(f"{public_key_id}:6:\n") + + env = {"GNUPGHOME": gpg_home} + + # Import the key + subprocess.run(["gpg", "--no-tty", "--import", tmp_key_path], env=env, check=True) + + # Trust it ultimately + subprocess.run( + ["gpg", "--no-tty", "--import-ownertrust", ownertrust_path], + env=env, + check=True, + ) + + return tmp_key_path + + +def tag_source_branch(tag): + """Parse a tag and return the source branch + """ + m = SNAPSHOT_TAG_REGEXES[0].match(tag) + if m: + return "develop" + + m = SNAPSHOT_TAG_REGEXES[1].match(tag) + if m: + major, minor = m.groups() + return f"releases/v{major}.{minor}" + + return None + + +################################################################################ +# Encapsulate information about a built spec in a mirror +class BuiltSpec: + def __init__( + self, + hash: Optional[str] = None, + stack: Optional[str] = None, + prefix: Optional[str] = None, + meta: Optional[str] = None, + archive: Optional[str] = None, + manifest_prefix: Optional[str] = None, + manifest_path: Optional[str] = None, + ): + self.hash = hash + self.stack = stack + self.prefix = prefix + self.meta = meta + self.archive = archive + self.manifest_prefix = manifest_prefix + self.manifest_path = manifest_path + + +################################################################################ +# +def bucket_name_from_s3_url(url): + m = REGEX_S3_BUCKET.search(url) + if m: + return m.group(1) + return "" + + +################################################################################ +# +def spec_catalogs_from_listing_v2(bucket: str, ref: str) -> Dict[str, Dict[str, BuiltSpec]]: + """Return a complete catalog of all the built specs in the listing + + Return a complete catalog of all the built specs for every prefix in the + listing. The returned dictionary of catalogs is keyed by unique prefix. + """ + list_url = f"s3://{bucket}/{ref}/" + all_catalogs: Dict[str, Dict[str, BuiltSpec]] = defaultdict( + lambda: defaultdict(BuiltSpec) + ) + + for date, size, key in list_prefix_contents(list_url): + m = REGEX_V2_SIGNED_SPECFILE_RELATIVE.search(key) + if m: + prefix = m.group(1) + middle_bit = m.group(2) + hash = m.group(3) + end_bit = m.group(4) + spec = all_catalogs[prefix][hash] + spec.hash = hash + spec.meta = f"{prefix}{middle_bit}{hash}{end_bit}" + continue + + m = REGEX_V2_ARCHIVE_RELATIVE.search(key) + if m: + prefix = m.group(1) + middle_bit = m.group(2) + hash = m.group(3) + end_bit = m.group(4) + spec = all_catalogs[prefix][hash] + spec.hash = hash + spec.archive = f"{prefix}{middle_bit}{hash}{end_bit}" + continue + + # else it must be a public key, an index, or a hash of an index + + return all_catalogs + + +################################################################################ +# +def spec_catalogs_from_listing_v3(bucket: str, ref: str) -> Dict[str, Dict[str, BuiltSpec]]: + list_url = f"s3://{bucket}/{ref}/" + all_catalogs: Dict[str, Dict[str, BuiltSpec]] = defaultdict( + lambda: defaultdict(BuiltSpec) + ) + for date, size, key in list_prefix_contents(list_url): + m = REGEX_V3_SIGNED_SPECFILE_RELATIVE.search(key) + if m: + prefix = m.group(1) + middle_bit = m.group(2) + hash = m.group(3) + end_bit = m.group(4) + spec = all_catalogs[prefix][hash] + spec.hash = hash + spec.manifest_prefix = f"{prefix}{middle_bit}{hash}{end_bit}" + continue + + return all_catalogs + + +################################################################################ +# +def generate_spec_catalogs_v2( + bucket: str, ref: str, exclude: List[str] = [] +) -> tuple[Dict[str, Dict[str, BuiltSpec]], Dict[str, BuiltSpec]]: + """Return information about specs in stacks and at the root + + Read the listing file, populate and return a tuple of dicts indicating which + specs exist in stacks, and which exist in the top-level buildcache. Stacks + appearing in the ``exclude`` list are ignoreed. + + Returns a tuple like the following: + + ( + # First element of tuple is the stack specs + { + : { + : , + ... + }, + ... + }, + # Followed by specs at the top level + { + : , + ... + } + ) + """ + stack_prefix_regex = re.compile(rf"{ref}/(.+)") + stack_specs: Dict[str, Dict[str, BuiltSpec]] = defaultdict( + lambda: defaultdict(BuiltSpec) + ) + all_catalogs = spec_catalogs_from_listing_v2(bucket, ref) + top_level_specs = all_catalogs[ref] + + for prefix in all_catalogs: + m = stack_prefix_regex.search(prefix) + if not m: + continue + + stack = m.group(1) + if stack in exclude: + continue + + for spec_hash, built_spec in all_catalogs[prefix].items(): + stack_specs[stack][spec_hash] = built_spec + + return stack_specs, top_level_specs + + +def format_blob_url(prefix: str, blob_record: Dict[str, str]) -> str: + """Use prefix and algorithm/checksum from record to build full prefix""" + hash_algo = blob_record.get("checksumAlgorithm", None) + checksum = blob_record.get("checksum", None) + + if not hash_algo: + raise MalformedManifestError("Missing 'checksumAlgorithm'") + + if not checksum: + raise MalformedManifestError("Missing 'checksum'") + + return f"{prefix}/blobs/{hash_algo}/{checksum[:2]}/{checksum}" + + +def find_data_with_media_type( + data: List[Dict[str, str]], mediaType: str +) -> Dict[str, str]: + """Return data element with matching mediaType, or else raise""" + for elt in data: + if elt["mediaType"] == mediaType: + return elt + raise NoSuchMediaTypeError(mediaType) + + +################################################################################ +# +def generate_spec_catalogs_v3( + bucket: str, + ref: str, + exclude: List[str] = [], + include: List[str] = [], + parallel: int = 8, + workdir: Optional[str] = None, +) -> tuple[Dict[str, Dict[str, BuiltSpec]], Dict[str, BuiltSpec]]: + """Return information about specs in stacks and at the root""" + stack_prefix_regex = re.compile(rf"{ref}/(.+)") + stack_specs: Dict[str, Dict[str, BuiltSpec]] = defaultdict( + lambda: defaultdict(BuiltSpec) + ) + all_catalogs = spec_catalogs_from_listing_v3(bucket, ref) + top_level_specs = all_catalogs[ref] + + task_list = [] + delete_on_exit = False + if not workdir: + delete_on_exit = True + workdir = tempfile.mkdtemp() + + tmpdir = workdir + + for prefix in all_catalogs: + m = stack_prefix_regex.search(prefix) + if not m: + continue + + stack = m.group(1) + if stack in exclude: + continue + + if include and stack not in include: + continue + + stack_manifests_dir = os.path.join(tmpdir, stack) + os.makedirs(stack_manifests_dir, exist_ok=True) + stack_manifest_sync_cmd = [ + "aws", + "s3", + "sync", + "--exclude", + "*", + "--include", + "*.spec.manifest.json", + f"s3://{bucket}/{prefix}/v3/manifests/spec", + stack_manifests_dir, + ] + + start_time = datetime.now() + + try: + LOGGER.debug(f"Downloading manifests for stack {stack}") + # subprocess.run( + # stack_manifest_sync_cmd, + # check=True, + # stdout=subprocess.DEVNULL, + # stderr=subprocess.DEVNULL, + # ) + except subprocess.CalledProcessError as cpe: + error_msg = getattr(cpe, "message", cpe) + LOGGER.error(f"Failed to download manifests for {stack} due to: {error_msg}") + continue + + end_time = datetime.now() + elapsed = end_time - start_time + LOGGER.info(f"Downloaded manifests for stack {stack}, elapsed time: {elapsed}") + + for spec_hash, built_spec in all_catalogs[prefix].items(): + stack_specs[stack][spec_hash] = built_spec + task_list.append((built_spec.hash, stack)) + + def _process_manifest_fn(spec_hash, stack): + download_dir = os.path.join(tmpdir, stack) + LOGGER.debug(f"searching {download_dir} for spec /{spec_hash}") + find_cmd = ["find", download_dir, "-type", "f", "-name", f"*{spec_hash}*"] + find_result = subprocess.run( + find_cmd, + capture_output=True, + ) + + # Check for an error searching for the spec + manifest_path = find_result.stdout.decode("utf-8").strip() + if not manifest_path or find_result.returncode != 0: + LOGGER.error(f"[{find_cmd}] failed to find manifest for /{spec_hash} in {stack}") + return (None, None, None, None) + + manifest_dict = extract_json_from_clearsig(manifest_path) + return (spec_hash, stack, manifest_dict, manifest_path) + + with ThreadPoolExecutor(max_workers=parallel) as executor: + futures = [executor.submit(_process_manifest_fn, *task) for task in task_list] + for future in as_completed(futures): + try: + spec_hash, stack, manifest_dict, manifest_path = future.result() + if not spec_hash or not stack or not manifest_dict or not manifest_path: + continue + + stack_specs[stack][spec_hash].stack = stack + stack_specs[stack][spec_hash].manifest_path = manifest_path + stack_specs[stack][spec_hash].meta = format_blob_url( + f"{ref}/{stack}", + find_data_with_media_type( + manifest_dict["data"], SPEC_METADATA_MEDIA_TYPE + ), + ) + stack_specs[stack][spec_hash].archive = format_blob_url( + f"{ref}/{stack}", + find_data_with_media_type( + manifest_dict["data"], TARBALL_MEDIA_TYPE + ), + ) + except Exception as exc: + LOGGER.error(f"Exception processing manifests: {exc}") + + # Cleanup the tmpdir + if delete_on_exit: + shutil.rmtree(tmpdir) + return stack_specs, top_level_specs + + +################################################################################ +# If the cli didn't provide a working directory, we will create (and clean up) +# a temporary directory. +def get_workdir_context(workdir: Optional[str] = None): + if not workdir: + return tempfile.TemporaryDirectory() + + return contextlib.nullcontext(workdir) + + +listing_prefix = os.environ.get("LISTING_CACHE_PREFIX", ".") + +def listing_file(url: str) -> str: + if not listing_prefix: + listing_prefix = tempfile.mkdtemp() + # Store the listing has the checksum of the url + h = hashlib.sha256() + h.update(url.encode()) + return os.path.join(listing_prefix, h.hexdigest()) + +################################################################################ +# Given a url and a file path to use for writing, get a recursive listing of +# everything under the prefix defined by the url, and write it to disk using the +# supplied path. +def list_prefix_contents(url: str, output_prefix: Optional[str] = None, force: bool = False, iterator: bool = True): + + # Auto caching of listing file + global listing_prefix + if not output_prefix: + if not listing_prefix: + listing_prefix = tempfile.mkdtemp() + output_prefix = listing_prefix + + # Store the listing has the checksum of the url + h = hashlib.sha256() + h.update(url.encode()) + output_file = os.path.join(output_prefix, h.hexdigest()) + LOGGER.debug(f"caching listing: {url} {output_file}") + + dt_format = "%Y-%m-%d %H:%M:%S" + if not os.path.isfile(output_file) or force: + if iterator: + client = s3_create_client() + purl = urllib.parse.urlparse(url) + prefix = re.sub("^/*", "/") + list_args = dict(Bucket=url.netloc, Prefix=prefix) + # Local buffer of objects to cache to a file + all_objects = [] + while True: + resp = client.list_objects_v2(**list_args) + + all_objects.extend(resp.get("Contents", [])) + obj = None + for obj in resp.get("Contents", []): + yield obj["LastModified"], obj["Size"], obj["Key"] + + if resp.get("IsTruncated", False) and obj: + list_args.update({ + "StartAfter": obj + }) + else: + break + + # Write the listing in the same format used by "aws s3 ls" + msize = max([obj["Size"] for obj in all_objects]) + msize = math.ceil(math.log(msize) / math.log(10)) + 1 + with open(output_file, "w", encoding="utf=8") as fd: + for obj in all_objects: + date_time = obj["LastModified"].strftime(dt_format) + size = int(obj["Size"]) + key = obj["Key"] + fd.write(f"{date_time} {size:msize} {key}\n") + + else: + LOGGER.info(f"Writing cached listfile for {url} to {output_file}") + list_cmd = ["aws", "s3", "ls", "--recursive", url] + with open(output_file, "w") as f: + subprocess.run(list_cmd, stdout=f, stderr=subprocess.DEVNULL, check=True) + elif iterator: + with open(output_file, "r", encoding="utf-8") as fd: + for line in fd: + m = REGEX_LISTING_DATA.search(line) + if m: + datestr = m.group(1).strip() + size = int(m.group(2).strip()) + key = m.group(3).strip() + yield datetime.strptime(datestr, dt_format), size, key + + +################################################################################ +# +def extract_json_from_clearsig(file_path): + with open(file_path) as fd: + data = fd.read() + + m = CLEARSIGN_FILE_REGEX.search(data) + if not m: + return {} + + return json.loads(m.group(1)) + + +################################################################################ +# Each mirror we might publish was built with a particular version of spack, and +# in order to be able update the index for one of those mirrors, we need to +# clone the matching version of spack. +# +# Clones the version of spack specified by ref to the root of the file system +def clone_spack(spack_ref: str = "develop", packages_ref: str = "develop", spack_repo: str = SPACK_REPO, packages_repo: str = PACKAGES_REPO, clone_dir: str = "/"): + spack_path = f"{clone_dir}/spack" + packages_path = f"{clone_dir}/spack-packages" + + if os.path.isdir(spack_path): + shutil.rmtree(spack_path) + + if os.path.isdir(packages_path): + shutil.rmtree(packages_path) + + owd = os.getcwd() + + try: + os.chdir(clone_dir) + subprocess.run( + [ + "git", + "clone", + "--depth", + "1", + "--single-branch", + "--branch", + f"{spack_ref}", + f"{spack_repo}", + spack_path, + ], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + check=True, + ) + subprocess.run( + [ + "git", + "clone", + "--depth", + "1", + "--single-branch", + "--branch", + f"{packages_ref}", + f"{packages_repo}", + packages_path, + ], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + check=True, + ) + # Configure the repo destination + subprocess.run( + [ + "spack/bin/spack", + "repo", + "set", + "builtin", + "--destination", + packages_path, + ], + check=True, + stdout=subprocess.DEVNULL, + ) + finally: + os.chdir(owd) + + +################################################################################ +# Download a file from s3 +def s3_download_file(bucket: str, prefix: str, save_path: str, force: bool = False): + if not os.path.isfile(save_path) or force is True: + session = boto3.session.Session() + s3_resource = session.resource("s3") + s3_client = s3_resource.meta.client + + with open(save_path, "wb") as f: + s3_client.download_fileobj(bucket, prefix, f) + + return save_path + +################################################################################ +# Create and return a new s3 client by first creating a Session, using that to +# create a new "s3" resource, and return the client stored within the resources +# metadata. +def s3_create_client(): + session = boto3.session.Session() + s3_resource = session.resource("s3") + return s3_resource.meta.client + +################################################################################ +# Copy objects between s3 buckets/prefixes +def s3_copy_file(copy_source: Dict[str, str], bucket: str, dest_prefix: str, client=None): + if client: + s3_client = client + else: + session = boto3.session.Session() + s3_resource = session.resource("s3") + s3_client = s3_resource.meta.client + + config = TransferConfig( + multipart_threshold=MULTIPART_THRESHOLD, + multipart_chunksize=MULTIPART_CHUNKSIZE, + max_concurrency=MAX_CONCURRENCY, + use_threads=USE_THREADS, + ) + + s3_client.copy(copy_source, bucket, dest_prefix, Config=config) + + +################################################################################ +# +def s3_upload_file(file_path: str, bucket: str, prefix: str, client=None): + if client: + s3_client = client + else: + session = boto3.session.Session() + s3_resource = session.resource("s3") + s3_client = s3_resource.meta.client + + with open(file_path, "rb") as fd: + s3_client.upload_fileobj(fd, bucket, prefix) + + +def s3_object_exists(bucket: str, key: str, client=None): + """Check if an s3 object exists""" + + if client: + s3_client = client + else: + s3_client = s3_create_client() + + try: + _ = s3_client.head_object(Bucket=bucket, Key=key) + return True + except Exception: + return False + + +################################################################################ +# +def compute_checksum(input_file: str, buf_size: int = 65536) -> str: + sha256 = hashlib.sha256() + + with open(input_file, 'rb') as f: + while True: + data = f.read(buf_size) + if not data: + break + sha256.update(data) + + return sha256.hexdigest() + + +################################################################################ +# +class NoSuchMediaTypeError(Exception): + pass + + +class MalformedManifestError(Exception): + pass + + +class UnexpectedURLFormatError(Exception): + pass diff --git a/images/protected-publish/pkg/migrate.py b/images/buildcache-tools/pkg/migrate.py similarity index 98% rename from images/protected-publish/pkg/migrate.py rename to images/buildcache-tools/pkg/migrate.py index 8dead3874..bd401a125 100644 --- a/images/protected-publish/pkg/migrate.py +++ b/images/buildcache-tools/pkg/migrate.py @@ -11,9 +11,7 @@ from datetime import datetime from typing import NamedTuple -import sentry_sdk - -from .common import ( +from pkg.common import ( BuiltSpec, TIMESTAMP_AND_SIZE, TIMESTAMP_PATTERN, @@ -28,8 +26,6 @@ spec_catalogs_from_listing_v2, ) -sentry_sdk.init(traces_sample_rate=1.0) - class MigrationResult(NamedTuple): #: False unless a spec was actually migrated @@ -411,16 +407,15 @@ def migrate(mirror_url: str, workdir: str, force: bool = False, parallel: int = force: Determines whether to migrate already-migrate specs parallel: The number of concurrent threads to use in processing """ - listing_file = os.path.join(workdir, "full_listing.txt") + listing_file = listing_file(mirror_url) tmp_storage_dir = os.path.join(workdir, "specfiles") if not os.path.isdir(tmp_storage_dir): os.makedirs(tmp_storage_dir) - if not os.path.isfile(listing_file) or force: - list_prefix_contents(f"{mirror_url}/", listing_file) + url = urllib.parse.urlparse(mirror_url) - all_catalogs = spec_catalogs_from_listing_v2(listing_file) + all_catalogs = spec_catalogs_from_listing_v2(url.netloc, url.path) target_prefix = None print(f"Looking for {mirror_url} in the catalogs...") diff --git a/images/buildcache-tools/pkg/publish.py b/images/buildcache-tools/pkg/publish.py new file mode 100644 index 000000000..57cf626e0 --- /dev/null +++ b/images/buildcache-tools/pkg/publish.py @@ -0,0 +1,535 @@ +import argparse +import logging +import os +import re +import shutil +import stat +import subprocess +import tempfile + +from collections import defaultdict +from concurrent.futures import as_completed, ThreadPoolExecutor +from datetime import datetime, timezone, timedelta +from typing import Callable, Dict, List, Optional + +import botocore.exceptions + +import github + +from boto3.s3.transfer import TransferConfig + +from pkg.common import ( + clone_spack, + download_and_import_key, + extract_json_from_clearsig, + get_workdir_context, + s3_copy_file, + s3_create_client, + s3_download_file, + generate_spec_catalogs_v2, + generate_spec_catalogs_v3, + BuiltSpec, + MalformedManifestError, + NoSuchMediaTypeError, + UnexpectedURLFormatError, + SNAPSHOT_TAG_REGEXES, + PROTECTED_BRANCH_REGEXES, +) + +GITHUB_PROJECT = "spack/spack-packages" +PREFIX_REGEX_V2 = re.compile(r"/(build_cache/.+)$") +PROTECTED_REF_REGEXES = SNAPSHOT_TAG_REGEXES + PROTECTED_BRANCH_REGEXES + +LOGGER = logging.getLogger(__name__) + +################################################################################ +# +def is_ref_protected(ref): + """Check if given ref matches expected protected ref pattern + + Returns: True if ref matches a protected ref pattern, False otherwise + """ + for regex in PROTECTED_REF_REGEXES: + m = regex.match(ref) + if m: + return True + return False + + +################################################################################ +# +def publish_spec_v2(built_spec, bucket, prefix_from, prefix_to, force, gpg_home, tmpdir): + """Publish a single spec from a stack to the root""" + hash = built_spec.hash + meta_suffix = built_spec.meta + archive_suffix = built_spec.archive + + specfile_path = os.path.join(tmpdir, f"{hash}.spec.json.sig") + + try: + s3_download_file(bucket, meta_suffix, specfile_path, force=force) + except Exception as error: + error_msg = getattr(error, "message", error) + error_msg = f"Failed to download {meta_suffix} due to {error_msg}" + return False, error_msg + + # Verify the signature of the locally downloaded metadata file + if gpg_home: + try: + env = {"GNUPGHOME": gpg_home} + subprocess.run(["gpg", "--verify", specfile_path], env=env, check=True) + except subprocess.CalledProcessError as cpe: + error_msg = getattr(cpe, "message", cpe) + LOGGER.error(f"Failed to verify signature of {meta_suffix} due to {error_msg}") + return False, error_msg + + # Finally, copy the files directly from source to dest, starting with the tarball + for suffix in [archive_suffix, meta_suffix]: + m = PREFIX_REGEX_V2.search(suffix) + if m: + dest_prefix = f"{prefix_to}/{m.group(1)}" + try: + copy_source = { + "Bucket": bucket, + "Key": suffix, + } + s3_copy_file(copy_source, bucket, dest_prefix) + except Exception as error: + error_msg = getattr(error, "message", error) + error_msg = f"Failed to copy_object({suffix}) due to {error_msg}" + return False, error_msg + + return True, f"Published {meta_suffix} and {archive_suffix} to s3://{bucket}/{ref}/" + + +################################################################################ +# +def publish_spec_v3(built_spec, bucket, prefix_from, prefix_to, force, gpg_home, tmpdir): + """Publish a single spec from a stack to the root""" + spec_hash = built_spec.hash + stack_manifest_prefix = built_spec.manifest_prefix + stack_meta_prefix = built_spec.meta + stack_archive_prefix = built_spec.archive + + # In v3 land, we already had to download this file in order to access + # the content-address of the tarball and metadata. + manifest_path = built_spec.manifest_path + + # Verify the signature of the previously downloaded manifest file + if gpg_home: + try: + env = {"GNUPGHOME": gpg_home} + subprocess.run( + ["gpg", "--verify", manifest_path], + env=env, + check=True, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + except subprocess.CalledProcessError as cpe: + error_msg = getattr(cpe, "message", cpe) + LOGGER.error(f"Failed to verify signature of {stack_meta_prefix} due to {error_msg}") + return False, error_msg + + from_regex = re.compile(rf"^{prefix_from}/(.+)$") + + m = from_regex.match(stack_manifest_prefix) + if not m: + raise UnexpectedURLFormatError(stack_manifest_prefix) + top_level_manifest_prefix = f"{prefix_to}/{m.group(1)}" + + m = from_regex.match(stack_meta_prefix) + if not m: + raise UnexpectedURLFormatError(stack_meta_prefix) + top_level_meta_prefix = f"{prefix_to}/{m.group(1)}" + + m = from_regex.match(stack_archive_prefix) + if not m: + raise UnexpectedURLFormatError(stack_archive_prefix) + top_level_archive_prefix = f"{prefix_to}/{m.group(1)}" + + things_to_copy = [ + (stack_archive_prefix, top_level_archive_prefix), + (stack_meta_prefix, top_level_meta_prefix), + (stack_manifest_prefix, top_level_manifest_prefix), + ] + + s3_client = s3_create_client() + + # Finally, copy the files directly from source to dest, starting with the tarball + errs = [] + for src_prefix, dest_prefix in things_to_copy: + try: + copy_source = {"Bucket": bucket, "Key": src_prefix} + s3_copy_file(copy_source, bucket, dest_prefix, client=s3_client) + except Exception as error: + error_msg = getattr(error, "message", error) + error_msg = f"Failed to copy_object({src_prefix}) due to {error_msg}" + errs.append(error_msg) + + if errs: + msg = f"Failed to publish /{built_spec.hash}" + for err in errs: + msg = msg + f"\n\t{err}" + return False, msg + + return ( + True, + f"Published {stack_manifest_prefix}, {stack_meta_prefix}, and {stack_archive_prefix} to s3://{bucket}/{prefix_to}/", + ) + + +################################################################################ +# +def publish( + bucket: str, + ref: str, + exclude: List[str] = [], + verify: bool = True, + force: bool = False, + parallel: int = 8, + workdir: str = "/work", + layout_version: int = 3, +): + """Publish all specs present in stacks but missing at the root + + Main steps of the publish algorithm: + 1) Get a listing of the bucket contents. This will include entries for + metadata and archive files for all specs at the root as well as in all + stacks + 2) Use regular expressions to build dictionaries of all hashes in the + stack mirrors, as well as all hashes at the root. Stored information + for each includes url (path) to metadata and archive file. + 3) Determine which specs are missing from the root (should contain union + of all specs in stacks) + 4) If no specs are missing from the top level, quit + 5) Download and trust the public part of the reputational signing key + 6) In parallel, publish any missing specs: + 6a) Download meta file from stack mirror + 6b) Verify signature of metadata file + 6c) If not valid signature, QUIT + 6d) Try to copy archive file from src to dst, and quit if you can't + 6e) Try to copy metadata file from src to dst + 7) Once all threads complete, rebuild the remote mirror index + """ + print(workdir) + tmp_storage_dir = os.path.join(workdir, "specfiles") + + if not os.path.isdir(tmp_storage_dir): + os.makedirs(tmp_storage_dir) + + # Build dictionaries of specs existing at the root and within stacks + + if layout_version == 2: + all_stack_specs, top_level_specs = generate_spec_catalogs_v2( + bucket, ref, exclude=exclude + ) + publish_fn = publish_spec_v2 + elif layout_version == 3: + all_stack_specs, top_level_specs = generate_spec_catalogs_v3( + bucket, ref, exclude=exclude, parallel=parallel, workdir=workdir + ) + publish_fn = publish_spec_v3 + else: + LOGGER.error(f"Unrecognized layout version: {layout_version}") + return + + # Build dictionary of specs in stacks but missing from the root + missing_at_top = find_top_level_missing(all_stack_specs, top_level_specs) + + print_summary(missing_at_top) + + if not missing_at_top: + LOGGER.info(f"No specs missing from s3://{bucket}/{ref}, nothing to do.") + return + + gnu_pg_home = os.path.join(workdir, ".gnupg") + download_and_import_key(gnu_pg_home, workdir, force) + + # Build a list of tasks for threads + task_list = [ + ( + # Duplicates are effectively identical, just take the "first" one + next(iter(stacks_dict.values())), + bucket, + f"{ref}/{next(iter(stacks_dict.values())).stack}", + ref, + force, + gnu_pg_home if verify else "", + tmp_storage_dir, + ) + for (_, stacks_dict) in missing_at_top.items() + ] + + # Dispatch work tasks + with ThreadPoolExecutor(max_workers=parallel) as executor: + futures = [executor.submit(publish_fn, *task) for task in task_list] + for future in as_completed(futures): + try: + result = future.result() + except Exception as exc: + LOGGER.error(f"Exception: {exc}") + else: + if not result[0]: + LOGGER.error(f"Publishing failed: {result[1]}") + else: + LOGGER.info(result[1]) + + publish_keys(f"s3://{bucket}/{ref}", gnu_pg_home) + + # When all the tasks are finished, rebuild the top-level index + LOGGER.info("Publishing complete") + + +def publish_keys(mirror_url, gnu_pg_home, ref: str = "develop"): + # Clone spack version appropriate to what we're publishing + with tempfile.TemporaryDirectory() as workdir: + spack_root = os.environ.get("SPACK_ROOT") + if not spack_root: + clone_spack(packages_ref="develop", clone_dir=workdir) + spack_root = f"{workdir}/spack" + + spack_exe = f"{spack_root}/bin/spack" + + gnu_pg_home = os.path.abspath(gnu_pg_home) + # Can be useful for testing to clone a custom spack to somewhere other than "/" + # clone_spack( + # packages_ref=ref, + # spack_ref="content-addressable-tarballs-2", + # spack_repo="https://github.com/scottwittenburg/spack.git", + # clone_dir=workdir, + # ) + # spack_exe = f"{workdir}/spack/bin/spack" + + # Publish the key used for verification + LOGGER.info(f"Publishing trusted keys to {mirror_url} ({gnu_pg_home})") + my_env = os.environ.copy() + my_env["SPACK_GNUPGHOME"] = gnu_pg_home + subprocess.run( + [spack_exe, "gpg", "publish", "--mirror-url", mirror_url], + env=my_env, + check=True, + ) + + # Rebuild the package and key index + LOGGER.info(f"Rebuilding index at {mirror_url}") + subprocess.run( + [spack_exe, "buildcache", "update-index", "--keys", mirror_url], + stdout=subprocess.DEVNULL, + check=True, + ) + + +################################################################################ +# +################################################################################ +# +def find_top_level_missing( + all_stack_specs: Dict[str, Dict[str, BuiltSpec]], + top_level_specs: Dict[str, BuiltSpec], +) -> Dict[str, Dict[str, BuiltSpec]]: + """Return a dictionary of all specs missing at the top level + + Return a dictionary keyed by hashes missing from the top-level mirror, along + with all the stacks that contain each missing hash. Only complete entries + (i.e. those with both metadata and compressed archive) within the stacks + are considered missing at the top level. + + missing_at_top = { + : { + : , + ... + }, + ... + } + """ + missing_at_top: Dict[str, Dict[str, BuiltSpec]] = defaultdict( + lambda: defaultdict(BuiltSpec) + ) + + for stack, stack_specs in all_stack_specs.items(): + for hash, built_spec in stack_specs.items(): + if hash not in top_level_specs: + # Only if at least one stack has a "complete" (both + # meta and archive are present) version of the spec + # do we really consider it missing from the root. + if built_spec.meta and built_spec.archive: + missing_at_top[hash][stack] = built_spec + + return missing_at_top + + +################################################################################ +# +def print_summary(missing_at_top: Dict[str, Dict[str, BuiltSpec]]): + total_missing = len(missing_at_top) + incomplete_pairs = {} + + LOGGER.info(f"There are {total_missing} specs missing from the top-level:") + for hash, stacks_dict in missing_at_top.items(): + viable_stacks = [] + nonviable_stacks = [] + for stack, built_spec in stacks_dict.items(): + if built_spec.meta and built_spec.archive: + viable_stacks.append(stack) + else: + nonviable_stacks.append(stack) + + if viable_stacks: + viables = ",".join(viable_stacks) + LOGGER.info(f" {hash} is available from {viables}") + + if nonviable_stacks: + incomplete_pairs[hash] = nonviable_stacks + + if incomplete_pairs: + LOGGER.info(f"Stacks with incomplete pairs, by hash:") + for hash, stacks in incomplete_pairs.items(): + borked_stacks = ",".join(stacks) + LOGGER.info(f" {hash}: {borked_stacks}") + + +################################################################################ +# +def get_recently_run_protected_refs(last_n_days): + """Query Github for recently updated refs + + Filter through pipelines updated over the last_n_days to find all protected + branches that had a pipeline run. + """ + gh = github.Github() + repo = gh.get_repo(GITHUB_PROJECT) + + recent_protected_refs = set() + now = datetime.now(timezone.utc) + previous = now - timedelta(days=last_n_days) + for branch in repo.get_branches(): + if not branch.protected: + continue + if branch.commit.commit.author.date < previous: + continue + recent_protected_refs.add(branch.name) + + for tag in repo.get_tags(): + if tag.last_modified_datetime < previous: + continue + recent_protected_refs.add(tag.name) + + return list(recent_protected_refs) + + +################################################################################ +# +def main(): + start_time = datetime.now() + LOGGER.info(f"Publish script started at {start_time}") + + parser = argparse.ArgumentParser( + prog="publish.py", + description="Publish specs from stack-specific mirrors to the root", + ) + + parser.add_argument( + "-b", "--bucket", default="spack-binaries", help="Bucket to operate on" + ) + parser.add_argument( + "-r", + "--ref", + action="append", + help=( + "A single protected ref to publish, or else 'recent', to " + "publish any protected refs that had a pipeline recently" + ), + ) + parser.add_argument( + "-d", + "--days", + type=int, + default=1, + help=( + "Number of days to look backward for recent protected " + "pipelines (only used if `--ref recent` is provided)" + ), + ) + parser.add_argument( + "-f", + "--force", + default=False, + action="store_true", + help="Refetch files if they already exist", + ) + parser.add_argument( + "-p", "--parallel", default=8, type=int, help="Thread parallelism level" + ) + parser.add_argument( + "-w", + "--workdir", + default=None, + help="A scratch directory, defaults to a tmp dir", + ) + parser.add_argument( + "-v", + "--version", + type=int, + default=3, + help=("Target layout version to publish (either 2 or 3, defaults to 2)"), + ) + parser.add_argument( + "-x", + "--exclude", + nargs="+", + default=[], + help="Optional list of stacks to exclude", + ) + + args = parser.parse_args() + + refs = [] + if not args.ref: + refs = ["develop"] + + if "recent" in args.ref: + refs = get_recently_run_protected_refs(args.days) + args.ref.remove("recent") + + if args.ref: + refs.extend(list(args.ref)) + + exceptions = [] + + for ref in refs: + # If the cli didn't provide a working directory, we will create (and clean up) + # a temporary directory using this workdir context + with get_workdir_context(args.workdir) as workdir: + print(workdir) + LOGGER.info(f"Publishing missing specs for {args.bucket} / {ref}") + try: + publish( + args.bucket, + ref, + exclude=args.exclude, + force=args.force, + parallel=args.parallel, + workdir=workdir, + layout_version=args.version, + ) + except Exception as e: + # Swallow exceptions here so we can proceed with remaining refs, + # but save the exceptions to raise at the end. + LOGGER.error(f"Error publishing specs for {args.bucket} / {ref} due to {e}") + raise RuntimeError('') from e + exceptions.append(e) + + end_time = datetime.now() + elapsed = end_time - start_time + LOGGER.info(f"Publish script finished at {end_time}, elapsed time: {elapsed}") + + if exceptions: + # Re-raise the first exception encountered, so we can see it in Sentry. + raise exceptions[0] + + +################################################################################ +# +if __name__ == "__main__": + main() diff --git a/images/buildcache-tools/pkg/snapshot.py b/images/buildcache-tools/pkg/snapshot.py new file mode 100644 index 000000000..4d73edff7 --- /dev/null +++ b/images/buildcache-tools/pkg/snapshot.py @@ -0,0 +1,280 @@ +#!/usr/bin/env python3 + +import argparse +import github +import json +import os +import re +import subprocess +import tempfile +import urllib.request +import logging + +from datetime import datetime, timezone +from concurrent.futures import as_completed, ThreadPoolExecutor +from github import Github, InputGitAuthor +from gitlab import Gitlab + +from pkg.common import ( + download_and_import_key, + generate_spec_catalogs_v3, + s3_create_client, + s3_object_exists, + tag_source_branch, + SNAPSHOT_TAG_REGEXES, + PROTECTED_BRANCH_REGEXES, +) + +from .publish import ( + publish_spec_v3, + publish, + publish_keys, +) + + +DRYRUN = False +WORKDIR = os.environ.get("SNAPSHOT_WORKDIR") + +GL = Gitlab("https://gitlab.spack.io") +DEFAULT_GITLAB_PROJECT = os.environ.get("SNAPSHOT_GITLAB_REPO", "spack/spack-packages") + +GITHUB_TOKEN = os.environ.get('GITHUB_TOKEN') +GH = Github(auth=github.Auth.Token(GITHUB_TOKEN)) +DEFAULT_GITHUB_PROJECT = os.environ.get("SNAPSHOT_GITHUB_REPO", "spack/spack-packages") + +LOGGER = logging.getLogger("snapshot.__main__" if __name__ == "__main__" else "snapshot") + +def gl_last_successful_pipeline(project, branch): + """Return the commit sha associated with the last successful pipeline for + a given branch in a project. + + project: project slug (ie. spack/spack) + branch: name of the branch (ie. develop) + """ + if isinstance(project, str): + project = GL.projects.get(project, lazy=True) + + pipeline = project.pipelines.list(get_all=False, per_page=1, ref=branch, status="success") + + if pipeline: + return pipeline[0].sha + + return None + + +def create_develop_snapshot_tag(project): + global DRYRUN + gl_project = GL.projects.get(DEFAULT_GITLAB_PROJECT, lazy=True) + + # Get the sha to snapshot + sha = gl_last_successful_pipeline(gl_project, "develop") + if not sha: + LOGGER.warning("No successful develop pipelines found!") + + # Check to see if this ref has already been used as a snapshot + commit = gl_project.commits.get(sha) + tags = commit.refs("tag") + snapshot_tag = None + for t in tags: + if re.match(t.get("name", ""), "develop-.*"): + snapshot_tag = t + break + + if snapshot_tag: + LOGGER.warning(f"Skipping SHA ({sha}) already associated with snapshot tag {snapshot_tag}") + return + + # Now that we found a new commit, tag it for snapshot + date_str = datetime.now(timezone.utc).strftime("%Y-%m-%d") + tag_name = f"develop-{date_str}" + tag_msg = f"Snapshot release {date_str}" + + # Use the GitHub API to create a tag for this commit of develop. + py_gh_repo = GH.get_repo(project, lazy=True) + spackbot_author = InputGitAuthor("spackbot", "noreply@spack.io") + LOGGER.info(f"Pushing tag {tag_name} for commit {sha} ({project})") + + if not DRYRUN: + try: + tag = py_gh_repo.create_git_tag( + tag=tag_name, + message=tag_msg, + object=sha, + type="commit", + tagger=spackbot_author) + + py_gh_repo.create_git_ref( + ref=f"refs/tags/{tag_name}", + sha=tag.sha) + + LOGGER.info("Push done!") + except github.GithubException as e: + LOGGER.info(str(e)) + else: + LOGGER.info("DRYRUN: No tags pushed!") + + +def create_snapshot(bucket: str, tag: github.GitTag, workdir: str, parallel: int = 8): + """Create a snapshot mirror associated with a tag and branch""" + global DRYRUN + + branch = tag_source_branch(t.name) + if not branch: + LOGGER.warning(f"Skipping snapshot for {tag.name}, cannot determine base branch") + return False + + if s3_object_exists(bucket, "{tag.name}/v3/layout.json"): + LOGGER.info(f"Skipping snapshot for {tag.name} as it already exists") + return True + + gl_project = DEFAULT_GITLAB_PROJECT + if isinstance(gl_project, str): + gl_project = GL.projects.get(gl_project, lazy=True) + + pipeline = gl_project.pipelines.list(get_all=False, per_page=1, sha=tag.commit.sha, ref=branch, status="success") + if not pipeline: + LOGGER.warning(f"Skipping {tag.name}: Could not find corresponding successful pipeline for {branch}") + return False + + LOGGER.info(f"Creating snapshot for: {t.name} from {branch} using pipeline {pipeline[0].id}") + + # Assuming all snapshots are v3 only now + include_stacks = [] + all_specs_catalog, _ = generate_spec_catalogs_v3(bucket, branch, workdir=workdir, include=include_stacks) + + gnu_pg_home = os.path.join(workdir, ".gnupg") + if not DRYRUN: + download_and_import_key(gnu_pg_home, workdir, False) + else: + LOGGER.info("DRYRUN: download_and_import_key...") + + # Get the lockfile artifacts for the generate jobs + for j in pipeline[0].jobs.list(iterator=True, scope="success"): + if not j.stage == 'generate': + continue + + stack = j.name.replace("-generate", "") + + if include_stacks and stack not in include_stacks: + continue + + if s3_object_exists(bucket, "{tag.name}/{stack}/v3/layout.json"): + LOGGER.info(f"Skipping snapshot for {tag.name}/{stack} as it already exists") + continue + + # Get the lockfile/concrete hashes to sync to snapshot mirror + job = gl_project.jobs.get(j.id, lazy=True) + artifact_path = f"jobs_scratch_dir/{stack}/concrete_environment/spack.lock" + LOGGER.info(f"Fetching artifacts for job {j.id}: {artifact_path}") + artifact = job.artifact(artifact_path) + lockfile = json.loads(artifact) + + snapshot_hashes = list(iter(lockfile["concrete_specs"].keys())) + + task_list = [ + ( + built_spec, + bucket, + f"{branch}/{stack}", + f"{tag.name}/{stack}", + False, + None, #gnu_pg_home, + workdir, + ) + for hash, built_spec in all_specs_catalog[stack].items() + if hash in snapshot_hashes + ] + + publish_fn = publish_spec_v3 + if DRYRUN: + def dryrun_publish(spec, bucket, source, dest, force, gpg_home, workdir): + LOGGER.debug(f""" +DRYRUN: publish + prefix: {spec.manifest_prefix} + bucket: {bucket} + source: {source} + dest: {dest} +""") + return True, None + publish_fn = dryrun_publish + + with ThreadPoolExecutor(max_workers=parallel) as executor: + futures = [executor.submit(publish_fn, *task) for task in task_list] + for future in as_completed(futures): + try: + result = future.result() + except Exception as exc: + LOGGER.error(f"Exception: {exc}") + else: + if not result[0]: + LOGGER.error(f"Publishing failed: {result[1]}") + else: + if result[1]: + LOGGER.debug(result[1]) + + mirror_url = f"s3://{bucket}/{tag.name}/{stack}" + if DRYRUN: + LOGGER.info("DRYRUN: Skipping key publish") + else: + publish_keys(mirror_url, gnu_pg_home) + + return True + + +if __name__ == "__main__": + + parser = argparse.ArgumentParser() + + parser.add_argument("-n", "--dryrun", + action="store_true", + help="Bucket to create snapshot mirrors in", + ) + parser.add_argument("-b", "--bucket", + default="spack-binaries", + help="Bucket to create snapshot mirrors in", + ) + parser.add_argument("-t" ,"--tag", + action="append", + help="Tags to snapshot", + ) + parser.add_argument("-p", "--project", + default=DEFAULT_GITHUB_PROJECT, + help="Github project to get/push snapshot tags", + ) + parser.add_argument("--workdir", action="store") + + logging.basicConfig(level=logging.DEBUG) + logging.getLogger("boto3").setLevel(logging.ERROR) + logging.getLogger("botocore").setLevel(logging.ERROR) + logging.getLogger("urllib3").setLevel(logging.ERROR) + + args = parser.parse_args() + + if args.dryrun: + DRYRUN=True + + # Create a new develop snapshot if one is created + if not args.tag: + create_develop_snapshot_tag(args.project) + + # Iterate all of the project tags and attempt to create a + # snaptshot if it is needed + py_gh_repo = GH.get_repo(args.project) + tempdir = args.workdir or WORKDIR or tempfile.mkdtemp("snapshot") + if not os.path.exists(tempdir): + os.makedirs(tempdir) + + LOGGER.info(f"workdir: {tempdir}") + for t in py_gh_repo.get_tags(): + if args.tag and t.name not in args.tag: + LOGGER.debug(f"Skipping tag {t.name}") + continue + + try: + if create_snapshot(args.bucket, t, tempdir): + # Now use publish to create the top level mirror if the snapshot exists + # Don't re-verify everything, it was already done by create_snapshot + publish(args.bucket, t.name, verify=False, workdir=tempdir) + except Exception as e: + raise Exception from e + LOGGER.error(f"Failed to create snapshot for {t.name}: {e}") diff --git a/images/protected-publish/pkg/validate_index.py b/images/buildcache-tools/pkg/validate.py similarity index 97% rename from images/protected-publish/pkg/validate_index.py rename to images/buildcache-tools/pkg/validate.py index 8bb5c181b..dcde30f1e 100644 --- a/images/protected-publish/pkg/validate_index.py +++ b/images/buildcache-tools/pkg/validate.py @@ -7,9 +7,9 @@ import botocore.exceptions import boto3.session -import sentry_sdk -sentry_sdk.init(traces_sample_rate=1.0) +# Import to init sentry if available +import pkg.common ################################################################################ @@ -57,7 +57,7 @@ def validate_s3_index(url, layout_version=2): if layout_version == 2: prefix = f"{m.group(2)}/build_cache/index.json" elif layout_version == 3: - prefix = f"{m.group(2)}/v3/specs/index.json" + prefix = f"{m.group(2)}/v3/manifests/index/index.json" else: print(f"Unrecognized layout_version given ({layout_version}): must be 2 or 3") sys.exit(1) diff --git a/images/protected-publish/requirements.txt b/images/buildcache-tools/requirements.txt similarity index 100% rename from images/protected-publish/requirements.txt rename to images/buildcache-tools/requirements.txt diff --git a/images/buildcache-tools/run.sh b/images/buildcache-tools/run.sh new file mode 100644 index 000000000..d5cbf7dee --- /dev/null +++ b/images/buildcache-tools/run.sh @@ -0,0 +1,23 @@ +#/bin/bash + +# TODO: Enable running chain of commands +# buildcache-tools publish ... -- snapshot ... -- validate-index ... + +cmd=$1 +shift + +case "$cmd" in + *pub*) + py_cmd="pkg.publish";; + *mig*) + exec /srcs/migrate.sh;; + *snap*) + py_cmd="pkg.snapshot";; + *val*) + py_cmd="pkg.validate_index";; + *) + echo "Unknown command: $cmd" + exit 1;; +esac + +python -m $py_cmd $@ diff --git a/images/protected-publish/test_stacks.txt b/images/buildcache-tools/test_stacks.txt similarity index 100% rename from images/protected-publish/test_stacks.txt rename to images/buildcache-tools/test_stacks.txt diff --git a/images/protected-publish/pkg/common.py b/images/protected-publish/pkg/common.py deleted file mode 100644 index 996c8b800..000000000 --- a/images/protected-publish/pkg/common.py +++ /dev/null @@ -1,300 +0,0 @@ -import contextlib -import hashlib -import json -import os -import re -import shutil -import subprocess -import tempfile -from collections import defaultdict -from typing import Dict, Optional - -import boto3 -import boto3.session -from boto3.s3.transfer import TransferConfig - - -SPACK_REPO = "https://github.com/spack/spack" - -TIMESTAMP_AND_SIZE = r"^[\d]{4}-[\d]{2}-[\d]{2}\s[\d]{2}:[\d]{2}:[\d]{2}\s+\d+\s+" -TIMESTAMP_PATTERN = "%Y-%m-%d %H:%M:%S" - -#: regular expressions designed to match "aws s3 ls" output -REGEX_V2_SIGNED_SPECFILE_RELATIVE = re.compile( - rf"{TIMESTAMP_AND_SIZE}(.+)(/build_cache/.+-)([^\.]+)(\.spec\.json\.sig)$" -) -REGEX_V2_ARCHIVE_RELATIVE = re.compile( - rf"{TIMESTAMP_AND_SIZE}(.+)(/build_cache/.+-)([^\.]+)(\.spack)$" -) -REGEX_V3_SIGNED_SPECFILE_RELATIVE = re.compile( - rf"{TIMESTAMP_AND_SIZE}(.+)(/v3/manifests/spec/.+-)([^-\.]+)(\.spec\.manifest\.json)$" -) - -#: Regular expression to pull spec contents out of clearsigned signature -#: file. -CLEARSIGN_FILE_REGEX = re.compile( - ( - r"^-----BEGIN PGP SIGNED MESSAGE-----" - r"\s+Hash:\s+[^\s]+\s+(.+)-----BEGIN PGP SIGNATURE-----" - ), - re.MULTILINE | re.DOTALL, -) - -#: regex to capture bucket name from an s3 url -REGEX_S3_BUCKET = re.compile(r"s3://([^/]+)/") - -#: Values used to config multi-part s3 copies -MB = 1024**2 -MULTIPART_THRESHOLD = 100 * MB -MULTIPART_CHUNKSIZE = 20 * MB -MAX_CONCURRENCY = 10 -USE_THREADS = True - - -################################################################################ -# Encapsulate information about a built spec in a mirror -class BuiltSpec: - def __init__( - self, - hash: Optional[str] = None, - stack: Optional[str] = None, - prefix: Optional[str] = None, - meta: Optional[str] = None, - archive: Optional[str] = None, - manifest_prefix: Optional[str] = None, - manifest_path: Optional[str] = None, - ): - self.hash = hash - self.stack = stack - self.prefix = prefix - self.meta = meta - self.archive = archive - self.manifest_prefix = manifest_prefix - self.manifest_path = manifest_path - - -################################################################################ -# -def bucket_name_from_s3_url(url): - m = REGEX_S3_BUCKET.search(url) - if m: - return m.group(1) - return "" - - -################################################################################ -# -def spec_catalogs_from_listing_v2(listing_path: str) -> Dict[str, Dict[str, BuiltSpec]]: - """Return a complete catalog of all the built specs in the listing - - Return a complete catalog of all the built specs for every prefix in the - listing. The returned dictionary of catalogs is keyed by unique prefix. - """ - all_catalogs: Dict[str, Dict[str, BuiltSpec]] = defaultdict( - lambda: defaultdict(BuiltSpec) - ) - - with open(listing_path) as f: - for line in f: - m = REGEX_V2_SIGNED_SPECFILE_RELATIVE.search(line) - if m: - # print("matched a specfile") - prefix = m.group(1) - middle_bit = m.group(2) - hash = m.group(3) - end_bit = m.group(4) - spec = all_catalogs[prefix][hash] - spec.hash = hash - spec.meta = f"{prefix}{middle_bit}{hash}{end_bit}" - continue - - m = REGEX_V2_ARCHIVE_RELATIVE.search(line) - if m: - # print("matched an archive file") - prefix = m.group(1) - middle_bit = m.group(2) - hash = m.group(3) - end_bit = m.group(4) - spec = all_catalogs[prefix][hash] - spec.hash = hash - spec.archive = f"{prefix}{middle_bit}{hash}{end_bit}" - continue - - # else it must be a public key, an index, or a hash of an index - - return all_catalogs - - -################################################################################ -# -def spec_catalogs_from_listing_v3(listing_path: str) -> Dict[str, Dict[str, BuiltSpec]]: - all_catalogs: Dict[str, Dict[str, BuiltSpec]] = defaultdict( - lambda: defaultdict(BuiltSpec) - ) - - with open(listing_path) as f: - for line in f: - m = REGEX_V3_SIGNED_SPECFILE_RELATIVE.search(line) - if m: - prefix = m.group(1) - middle_bit = m.group(2) - hash = m.group(3) - end_bit = m.group(4) - spec = all_catalogs[prefix][hash] - spec.hash = hash - spec.manifest_prefix = f"{prefix}{middle_bit}{hash}{end_bit}" - continue - - return all_catalogs - - -################################################################################ -# If the cli didn't provide a working directory, we will create (and clean up) -# a temporary directory. -def get_workdir_context(workdir: Optional[str] = None): - if not workdir: - return tempfile.TemporaryDirectory() - - return contextlib.nullcontext(workdir) - - -################################################################################ -# Given a url and a file path to use for writing, get a recursive listing of -# everything under the prefix defined by the url, and write it to disk using the -# supplied path. -def list_prefix_contents(url: str, output_file: str): - list_cmd = ["aws", "s3", "ls", "--recursive", url] - - with open(output_file, "w") as f: - subprocess.run(list_cmd, stdout=f, check=True) - - -################################################################################ -# -def extract_json_from_clearsig(file_path): - with open(file_path) as fd: - data = fd.read() - - m = CLEARSIGN_FILE_REGEX.search(data) - if not m: - return {} - - return json.loads(m.group(1)) - - -################################################################################ -# Each mirror we might publish was built with a particular version of spack, and -# in order to be able update the index for one of those mirrors, we need to -# clone the matching version of spack. -# -# Clones the version of spack specified by ref to the root of the file system -def clone_spack(ref: str = "develop", repo: str = SPACK_REPO, clone_dir: str = "/"): - spack_path = f"{clone_dir}/spack" - - if os.path.isdir(spack_path): - shutil.rmtree(spack_path) - - owd = os.getcwd() - - try: - os.chdir(clone_dir) - subprocess.run( - [ - "git", - "clone", - "--depth", - "1", - "--single-branch", - "--branch", - f"{ref}", - f"{repo}", - ], - check=True, - ) - finally: - os.chdir(owd) - - -################################################################################ -# Download a file from s3 -def s3_download_file(bucket: str, prefix: str, save_path: str, force: bool = False): - if not os.path.isfile(save_path) or force is True: - session = boto3.session.Session() - s3_resource = session.resource("s3") - s3_client = s3_resource.meta.client - - with open(save_path, "wb") as f: - s3_client.download_fileobj(bucket, prefix, f) - - return save_path - -################################################################################ -# Create and return a new s3 client by first creating a Session, using that to -# create a new "s3" resource, and return the client stored within the resources -# metadata. -def s3_create_client(): - session = boto3.session.Session() - s3_resource = session.resource("s3") - return s3_resource.meta.client - -################################################################################ -# Copy objects between s3 buckets/prefixes -def s3_copy_file(copy_source: Dict[str, str], bucket: str, dest_prefix: str, client=None): - if client: - s3_client = client - else: - session = boto3.session.Session() - s3_resource = session.resource("s3") - s3_client = s3_resource.meta.client - - config = TransferConfig( - multipart_threshold=MULTIPART_THRESHOLD, - multipart_chunksize=MULTIPART_CHUNKSIZE, - max_concurrency=MAX_CONCURRENCY, - use_threads=USE_THREADS, - ) - - s3_client.copy(copy_source, bucket, dest_prefix, Config=config) - - -################################################################################ -# -def s3_upload_file(file_path: str, bucket: str, prefix: str, client=None): - if client: - s3_client = client - else: - session = boto3.session.Session() - s3_resource = session.resource("s3") - s3_client = s3_resource.meta.client - - with open(file_path, "rb") as fd: - s3_client.upload_fileobj(fd, bucket, prefix) - - -################################################################################ -# -def compute_checksum(input_file: str, buf_size: int = 65536) -> str: - sha256 = hashlib.sha256() - - with open(input_file, 'rb') as f: - while True: - data = f.read(buf_size) - if not data: - break - sha256.update(data) - - return sha256.hexdigest() - - -################################################################################ -# -class NoSuchMediaTypeError(Exception): - pass - - -class MalformedManifestError(Exception): - pass - - -class UnexpectedURLFormatError(Exception): - pass diff --git a/images/protected-publish/pkg/publish.py b/images/protected-publish/pkg/publish.py deleted file mode 100644 index dd55d19ad..000000000 --- a/images/protected-publish/pkg/publish.py +++ /dev/null @@ -1,725 +0,0 @@ -import argparse -import os -import re -import shutil -import stat -import subprocess - -from collections import defaultdict -from concurrent.futures import as_completed, ThreadPoolExecutor -from datetime import datetime, timedelta -from typing import Callable, Dict, List, Optional - -import botocore.exceptions - -import gitlab -import requests -import sentry_sdk -from boto3.s3.transfer import TransferConfig - -from .common import ( - clone_spack, - extract_json_from_clearsig, - get_workdir_context, - list_prefix_contents, - s3_copy_file, - s3_create_client, - s3_download_file, - spec_catalogs_from_listing_v2, - spec_catalogs_from_listing_v3, - BuiltSpec, - MalformedManifestError, - NoSuchMediaTypeError, - UnexpectedURLFormatError, -) - -sentry_sdk.init(traces_sample_rate=1.0) - -GITLAB_URL = "https://gitlab.spack.io" -GITLAB_PROJECT = "spack/spack" -PREFIX_REGEX_V2 = re.compile(r"/build_cache/(.+)$") -PROTECTED_REF_REGEXES = [ - re.compile(r"^develop$"), - re.compile(r"^v[\d]+\.[\d]+\.[\d]+$"), - re.compile(r"^releases/v[\d]+\.[\d]+$"), - re.compile(r"^develop-[\d]{4}-[\d]{2}-[\d]{2}$"), -] - -SPACK_PUBLIC_KEY_LOCATION = "https://spack.github.io/keys" -SPACK_PUBLIC_KEY_NAME = "spack-public-binary-key.pub" -TARBALL_MEDIA_TYPE = "application/vnd.spack.install.v2.tar+gzip" -SPEC_METADATA_MEDIA_TYPE = "application/vnd.spack.spec.v5+json" - - -################################################################################ -# -def is_ref_protected(ref): - """Check if given ref matches expected protected ref pattern - - Returns: True if ref matches a protected ref pattern, False otherwise - """ - for regex in PROTECTED_REF_REGEXES: - m = regex.match(ref) - if m: - return True - return False - - -################################################################################ -# -def publish_missing_spec_v2(built_spec, bucket, ref, force, gpg_home, tmpdir): - """Publish a single spec from a stack to the root""" - hash = built_spec.hash - meta_suffix = built_spec.meta - archive_suffix = built_spec.archive - - specfile_path = os.path.join(tmpdir, f"{hash}.spec.json.sig") - - try: - s3_download_file(bucket, meta_suffix, specfile_path, force=force) - except Exception as error: - error_msg = getattr(error, "message", error) - error_msg = f"Failed to download {meta_suffix} due to {error_msg}" - return False, error_msg - - # Verify the signature of the locally downloaded metadata file - try: - env = {"GNUPGHOME": gpg_home} - subprocess.run(["gpg", "--verify", specfile_path], env=env, check=True) - except subprocess.CalledProcessError as cpe: - error_msg = getattr(cpe, "message", cpe) - print(f"Failed to verify signature of {meta_suffix} due to {error_msg}") - return False, error_msg - - # Finally, copy the files directly from source to dest, starting with the tarball - for suffix in [archive_suffix, meta_suffix]: - m = PREFIX_REGEX_V2.search(suffix) - if m: - dest_prefix = f"{ref}/build_cache/{m.group(1)}" - try: - copy_source = { - "Bucket": bucket, - "Key": suffix, - } - s3_copy_file(copy_source, bucket, dest_prefix) - except Exception as error: - error_msg = getattr(error, "message", error) - error_msg = f"Failed to copy_object({suffix}) due to {error_msg}" - return False, error_msg - - return True, f"Published {meta_suffix} and {archive_suffix} to s3://{bucket}/{ref}/" - - -################################################################################ -# -def publish_missing_spec_v3(built_spec, bucket, ref, force, gpg_home, tmpdir): - """Publish a single spec from a stack to the root""" - spec_hash = built_spec.hash - stack = built_spec.stack - stack_manifest_prefix = built_spec.manifest_prefix - stack_meta_prefix = built_spec.meta - stack_archive_prefix = built_spec.archive - - # In v3 land, we already had to download this file in order to access - # the content-address of the tarball and metadata. - manifest_path = built_spec.manifest_path - - # Verify the signature of the previously downloaded manifest file - try: - env = {"GNUPGHOME": gpg_home} - subprocess.run(["gpg", "--verify", manifest_path], env=env, check=True) - except subprocess.CalledProcessError as cpe: - error_msg = getattr(cpe, "message", cpe) - print(f"Failed to verify signature of {stack_meta_prefix} due to {error_msg}") - return False, error_msg - - stack_regex = re.compile(rf"^{ref}/{stack}/(.+)$") - - m = stack_regex.match(stack_manifest_prefix) - if not m: - raise UnexpectedURLFormatError(stack_manifest_prefix) - top_level_manifest_prefix = f"{ref}/{m.group(1)}" - - m = stack_regex.match(stack_meta_prefix) - if not m: - raise UnexpectedURLFormatError(stack_meta_prefix) - top_level_meta_prefix = f"{ref}/{m.group(1)}" - - m = stack_regex.match(stack_archive_prefix) - if not m: - raise UnexpectedURLFormatError(stack_archive_prefix) - top_level_archive_prefix = f"{ref}/{m.group(1)}" - - things_to_copy = [ - (stack_archive_prefix, top_level_archive_prefix), - (stack_meta_prefix, top_level_meta_prefix), - (stack_manifest_prefix, top_level_manifest_prefix), - ] - - s3_client = s3_create_client() - - # Finally, copy the files directly from source to dest, starting with the tarball - for src_prefix, dest_prefix in things_to_copy: - try: - copy_source = {"Bucket": bucket, "Key": src_prefix} - s3_copy_file(copy_source, bucket, dest_prefix, client=s3_client) - except Exception as error: - error_msg = getattr(error, "message", error) - error_msg = f"Failed to copy_object({src_prefix}) due to {error_msg}" - return False, error_msg - - return ( - True, - f"Published {stack_manifest_prefix}, {stack_meta_prefix}, and {stack_archive_prefix} to s3://{bucket}/{ref}/", - ) - - -################################################################################ -# -def publish( - bucket: str, - ref: str, - exclude: List[str], - force: bool = False, - parallel: int = 8, - workdir: str = "/work", - layout_version: int = 2, -): - """Publish all specs present in stacks but missing at the root - - Main steps of the publish algorithm: - 1) Get a listing of the bucket contents. This will include entries for - metadata and archive files for all specs at the root as well as in all - stacks - 2) Use regular expressions to build dictionaries of all hashes in the - stack mirrors, as well as all hashes at the root. Stored information - for each includes url (path) to metadata and archive file. - 3) Determine which specs are missing from the root (should contain union - of all specs in stacks) - 4) If no specs are missing from the top level, quit - 5) Download and trust the public part of the reputational signing key - 6) In parallel, publish any missing specs: - 6a) Download meta file from stack mirror - 6b) Verify signature of metadata file - 6c) If not valid signature, QUIT - 6d) Try to copy archive file from src to dst, and quit if you can't - 6e) Try to copy metadata file from src to dst - 7) Once all threads complete, rebuild the remote mirror index - """ - list_url = f"s3://{bucket}/{ref}/" - listing_file = os.path.join(workdir, "full_listing.txt") - tmp_storage_dir = os.path.join(workdir, "specfiles") - - if not os.path.isdir(tmp_storage_dir): - os.makedirs(tmp_storage_dir) - - if not os.path.isfile(listing_file) or force: - list_prefix_contents(list_url, listing_file) - - # Build dictionaries of specs existing at the root and within stacks - - if layout_version == 2: - all_stack_specs, top_level_specs = generate_spec_catalogs_v2( - ref, listing_file, exclude - ) - publish_fn = publish_missing_spec_v2 - elif layout_version == 3: - all_stack_specs, top_level_specs = generate_spec_catalogs_v3( - bucket, ref, listing_file, exclude, tmp_storage_dir, parallel - ) - publish_fn = publish_missing_spec_v3 - else: - print(f"Unrecognized layout version: {layout_version}") - return - - # Build dictionary of specs in stacks but missing from the root - missing_at_top = find_top_level_missing(all_stack_specs, top_level_specs) - - print_summary(missing_at_top) - - if not missing_at_top: - print(f"No specs missing from s3://{bucket}/{ref}, nothing to do.") - return - - gnu_pg_home = os.path.join(workdir, ".gnupg") - download_and_import_key(gnu_pg_home, workdir, force) - - # Build a list of tasks for threads - task_list = [ - ( - # Duplicates are effectively identical, just take the "first" one - next(iter(stacks_dict.values())), - bucket, - ref, - force, - gnu_pg_home, - tmp_storage_dir, - ) - for (_, stacks_dict) in missing_at_top.items() - ] - - # Dispatch work tasks - with ThreadPoolExecutor(max_workers=parallel) as executor: - futures = [executor.submit(publish_fn, *task) for task in task_list] - for future in as_completed(futures): - try: - result = future.result() - except Exception as exc: - print(f"Exception: {exc}") - else: - if not result[0]: - print(f"Publishing failed: {result[1]}") - else: - print(result[1]) - - mirror_url = f"s3://{bucket}/{ref}" - - # When all the tasks are finished, rebuild the top-level index - print("Publishing complete") - - # Clone spack version appropriate to what we're publishing - clone_spack(ref, clone_dir=workdir) - spack_exe = f"{workdir}/spack/bin/spack" - - # Can be useful for testing to clone a custom spack to somewhere other than "/" - # clone_spack( - # ref="content-addressable-tarballs-2", - # repo="https://github.com/scottwittenburg/spack.git", - # clone_dir=workdir, - # ) - # spack_exe = f"{workdir}/spack/bin/spack" - - # Publish the key used for verification - print(f"Publishing trusted keys to {mirror_url}") - my_env = os.environ.copy() - my_env["SPACK_GNUPGHOME"] = gnu_pg_home - subprocess.run( - [spack_exe, "gpg", "publish", "--mirror-url", mirror_url], - env=my_env, - check=True, - ) - - # Rebuild the package and key index - print(f"Rebuilding index at {mirror_url}") - subprocess.run( - [spack_exe, "buildcache", "update-index", "--keys", mirror_url], - check=True, - ) - - -################################################################################ -# -def download_and_import_key(gpg_home: str, tmpdir: str, force: bool) -> str | None: - """Download spack public signing key and import it""" - if os.path.isdir(gpg_home): - if force is True: - shutil.rmtree(gpg_home) - else: - return None - - mode_owner_rwe = stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR - os.makedirs(gpg_home, mode=mode_owner_rwe) - - public_key_url = f"{SPACK_PUBLIC_KEY_LOCATION}/{SPACK_PUBLIC_KEY_NAME}" - public_key_id = "2C8DD3224EF3573A42BD221FA8E0CA3C1C2ADA2F" - - # Fetch the public key and write it to a file to be imported - tmp_key_path = os.path.join(tmpdir, SPACK_PUBLIC_KEY_NAME) - response = requests.get(public_key_url) - with open(tmp_key_path, "w") as f: - f.write(response.text) - - # Also write an ownertrust file to be imported - ownertrust_path = os.path.join(tmpdir, "trustfile") - with open(ownertrust_path, "w") as f: - f.write(f"{public_key_id}:6:\n") - - env = {"GNUPGHOME": gpg_home} - - # Import the key - subprocess.run(["gpg", "--no-tty", "--import", tmp_key_path], env=env, check=True) - - # Trust it ultimately - subprocess.run( - ["gpg", "--no-tty", "--import-ownertrust", ownertrust_path], - env=env, - check=True, - ) - - return tmp_key_path - - -################################################################################ -# -def find_top_level_missing( - all_stack_specs: Dict[str, Dict[str, BuiltSpec]], - top_level_specs: Dict[str, BuiltSpec], -) -> Dict[str, Dict[str, BuiltSpec]]: - """Return a dictionary of all specs missing at the top level - - Return a dictionary keyed by hashes missing from the top-level mirror, along - with all the stacks that contain each missing hash. Only complete entries - (i.e. those with both metadata and compressed archive) within the stacks - are considered missing at the top level. - - missing_at_top = { - : { - : , - ... - }, - ... - } - """ - missing_at_top: Dict[str, Dict[str, BuiltSpec]] = defaultdict( - lambda: defaultdict(BuiltSpec) - ) - - for hash, stack_specs in all_stack_specs.items(): - if hash not in top_level_specs: - for stack, built_spec in stack_specs.items(): - # Only if at least one stack has a "complete" (both - # meta and archive are present) version of the spec - # do we really consider it missing from the root. - if built_spec.meta and built_spec.archive: - missing_at_top[hash][stack] = built_spec - - return missing_at_top - - -################################################################################ -# -def print_summary(missing_at_top: Dict[str, Dict[str, BuiltSpec]]): - total_missing = len(missing_at_top) - incomplete_pairs = {} - - print(f"There are {total_missing} specs missing from the top-level:") - for hash, stacks_dict in missing_at_top.items(): - viable_stacks = [] - nonviable_stacks = [] - for stack, built_spec in stacks_dict.items(): - if built_spec.meta and built_spec.archive: - viable_stacks.append(stack) - else: - nonviable_stacks.append(stack) - - if viable_stacks: - viables = ",".join(viable_stacks) - print(f" {hash} is available from {viables}") - - if nonviable_stacks: - incomplete_pairs[hash] = nonviable_stacks - - if incomplete_pairs: - print(f"Stacks with incomplete pairs, by hash:") - for hash, stacks in incomplete_pairs.items(): - borked_stacks = ",".join(stacks) - print(f" {hash}: {borked_stacks}") - - -################################################################################ -# -def generate_spec_catalogs_v2( - ref: str, listing_path: str, exclude: List[str] -) -> tuple[Dict[str, Dict[str, BuiltSpec]], Dict[str, BuiltSpec]]: - """Return information about specs in stacks and at the root - - Read the listing file, populate and return a tuple of dicts indicating which - specs exist in stacks, and which exist in the top-level buildcache. Stacks - appearing in the ``exclude`` list are ignoreed. - - Returns a tuple like the following: - - ( - # First element of tuple is the stack specs - { - : { - : , - ... - }, - ... - }, - # Followed by specs at the top level - { - : , - ... - } - ) - """ - stack_prefix_regex = re.compile(rf"{ref}/(.+)") - stack_specs: Dict[str, Dict[str, BuiltSpec]] = defaultdict( - lambda: defaultdict(BuiltSpec) - ) - all_catalogs = spec_catalogs_from_listing_v2(listing_path) - top_level_specs = all_catalogs[ref] - - for prefix in all_catalogs: - m = stack_prefix_regex.search(prefix) - if not m: - continue - - stack = m.group(1) - if stack in exclude: - continue - - for spec_hash, built_spec in all_catalogs[prefix].items(): - stack_specs[spec_hash][stack] = built_spec - - return stack_specs, top_level_specs - - -def find_data_with_media_type( - data: List[Dict[str, str]], mediaType: str -) -> Dict[str, str]: - """Return data element with matching mediaType, or else raise""" - for elt in data: - if elt["mediaType"] == mediaType: - return elt - raise NoSuchMediaTypeError(mediaType) - - -def format_blob_url(prefix: str, blob_record: Dict[str, str]) -> str: - """Use prefix and algorithm/checksum from record to build full prefix""" - hash_algo = blob_record.get("checksumAlgorithm", None) - checksum = blob_record.get("checksum", None) - - if not hash_algo: - raise MalformedManifestError("Missing 'checksumAlgorithm'") - - if not checksum: - raise MalformedManifestError("Missing 'checksum'") - - return f"{prefix}/blobs/{hash_algo}/{checksum[:2]}/{checksum}" - - -################################################################################ -# -def generate_spec_catalogs_v3( - bucket: str, - ref: str, - listing_path: str, - exclude: List[str], - specfiles_dir: str, - parallel: int = 8, -) -> tuple[Dict[str, Dict[str, BuiltSpec]], Dict[str, BuiltSpec]]: - """Return information about specs in stacks and at the root""" - stack_prefix_regex = re.compile(rf"{ref}/(.+)") - stack_specs: Dict[str, Dict[str, BuiltSpec]] = defaultdict( - lambda: defaultdict(BuiltSpec) - ) - all_catalogs = spec_catalogs_from_listing_v3(listing_path) - top_level_specs = all_catalogs[ref] - - task_list = [] - - for prefix in all_catalogs: - m = stack_prefix_regex.search(prefix) - if not m: - continue - - stack = m.group(1) - if stack in exclude: - continue - - stack_manifests_dir = os.path.join(specfiles_dir, stack) - os.makedirs(stack_manifests_dir) - stack_manifest_sync_cmd = [ - "aws", - "s3", - "sync", - "--exclude", - "*", - "--include", - "*.spec.manifest.json", - f"s3://{bucket}/{prefix}/v3/manifests/spec", - stack_manifests_dir, - ] - - start_time = datetime.now() - - try: - print(f"Downloading manifests for stack {stack}") - subprocess.run(stack_manifest_sync_cmd, check=True) - except subprocess.CalledProcessError as cpe: - error_msg = getattr(cpe, "message", cpe) - print(f"Failed to download manifests for {stack} due to: {error_msg}") - continue - - end_time = datetime.now() - elapsed = end_time - start_time - print(f"Downloaded manifests for stack {stack}, elapsed time: {elapsed}") - - for spec_hash, built_spec in all_catalogs[prefix].items(): - stack_specs[spec_hash][stack] = built_spec - task_list.append((built_spec.hash, stack)) - - def _process_manifest_fn(spec_hash, stack): - download_dir = os.path.join(specfiles_dir, stack) - find_cmd = ["find", download_dir, "-type", "f", "-name", f"*{spec_hash}*"] - find_result = subprocess.run(find_cmd, capture_output=True) - - if find_result.returncode != 0: - print(f"[{find_cmd}] failed to find manifest for {spec_hash} in {stack}") - return (None, None, None, None) - - manifest_path = find_result.stdout.decode("utf-8").strip() - manifest_dict = extract_json_from_clearsig(manifest_path) - return (spec_hash, stack, manifest_dict, manifest_path) - - with ThreadPoolExecutor(max_workers=parallel) as executor: - futures = [executor.submit(_process_manifest_fn, *task) for task in task_list] - for future in as_completed(futures): - try: - spec_hash, stack, manifest_dict, manifest_path = future.result() - if not spec_hash or not stack or not manifest_dict or not manifest_path: - continue - - stack_specs[spec_hash][stack].stack = stack - stack_specs[spec_hash][stack].manifest_path = manifest_path - stack_specs[spec_hash][stack].meta = format_blob_url( - f"{ref}/{stack}", - find_data_with_media_type( - manifest_dict["data"], SPEC_METADATA_MEDIA_TYPE - ), - ) - stack_specs[spec_hash][stack].archive = format_blob_url( - f"{ref}/{stack}", - find_data_with_media_type( - manifest_dict["data"], TARBALL_MEDIA_TYPE - ), - ) - except Exception as exc: - print(f"Exception processing manifests: {exc}") - - return stack_specs, top_level_specs - - -################################################################################ -# -def get_recently_run_protected_refs(last_n_days): - """Query gitlab pipelines to get recently run refs - - Filter through pipelines updated over the last_n_days to find all protected - branches that had a pipeline run. - """ - gl = gitlab.Gitlab(GITLAB_URL) - project = gl.projects.get(GITLAB_PROJECT) - now = datetime.now() - previous = now - timedelta(days=last_n_days) - recent_protected_refs = set() - print(f"Piplines in the last {last_n_days} day(s):") - for pipeline in project.pipelines.list( - iterator=True, updated_before=now, updated_after=previous - ): - print(f" {pipeline.id}: {pipeline.ref}") - if is_ref_protected(pipeline.ref): - recent_protected_refs.add(pipeline.ref) - return list(recent_protected_refs) - - -################################################################################ -# -def main(): - start_time = datetime.now() - print(f"Publish script started at {start_time}") - - parser = argparse.ArgumentParser( - prog="publish.py", - description="Publish specs from stack-specific mirrors to the root", - ) - - parser.add_argument( - "-b", "--bucket", default="spack-binaries", help="Bucket to operate on" - ) - parser.add_argument( - "-r", - "--ref", - default="develop", - help=( - "A single protected ref to publish, or else 'recent', to " - "publish any protected refs that had a pipeline recently" - ), - ) - parser.add_argument( - "-d", - "--days", - type=int, - default=1, - help=( - "Number of days to look backward for recent protected " - "pipelines (only used if `--ref recent` is provided)" - ), - ) - parser.add_argument( - "-f", - "--force", - default=False, - action="store_true", - help="Refetch files if they already exist", - ) - parser.add_argument( - "-p", "--parallel", default=8, type=int, help="Thread parallelism level" - ) - parser.add_argument( - "-w", - "--workdir", - default=None, - help="A scratch directory, defaults to a tmp dir", - ) - parser.add_argument( - "-v", - "--version", - type=int, - default=2, - help=("Target layout version to publish (either 2 or 3, defaults to 2)"), - ) - parser.add_argument( - "-x", - "--exclude", - nargs="+", - default=[], - help="Optional list of stacks to exclude", - ) - - args = parser.parse_args() - - if args.ref == "recent": - refs = get_recently_run_protected_refs(args.days) - else: - refs = [args.ref] - - exceptions = [] - - for ref in refs: - # If the cli didn't provide a working directory, we will create (and clean up) - # a temporary directory using this workdir context - with get_workdir_context(args.workdir) as workdir: - print(f"Publishing missing specs for {args.bucket} / {ref}") - try: - publish( - args.bucket, - ref, - args.exclude, - args.force, - args.parallel, - workdir, - args.version, - ) - except Exception as e: - # Swallow exceptions here so we can proceed with remaining refs, - # but save the exceptions to raise at the end. - print(f"Error publishing specs for {args.bucket} / {ref} due to {e}") - exceptions.append(e) - - end_time = datetime.now() - elapsed = end_time - start_time - print(f"Publish script finished at {end_time}, elapsed time: {elapsed}") - - if exceptions: - # Re-raise the first exception encountered, so we can see it in Sentry. - raise exceptions[0] - - -################################################################################ -# -if __name__ == "__main__": - main() diff --git a/images/snapshot-release-tags/Dockerfile b/images/snapshot-release-tags/Dockerfile deleted file mode 100644 index e8ad126cd..000000000 --- a/images/snapshot-release-tags/Dockerfile +++ /dev/null @@ -1,12 +0,0 @@ -FROM python:3.11-slim - -WORKDIR /app - -COPY requirements.txt /app/requirements.txt - -RUN pip install --upgrade uv -RUN uv pip install --system -r requirements.txt - -COPY . . - -CMD [ "python", "./snapshot_release_tags.py" ] diff --git a/images/snapshot-release-tags/requirements.txt b/images/snapshot-release-tags/requirements.txt deleted file mode 100644 index f16b4a991..000000000 --- a/images/snapshot-release-tags/requirements.txt +++ /dev/null @@ -1,14 +0,0 @@ -certifi==2023.5.7 -cffi==1.15.1 -charset-normalizer==3.1.0 -cryptography==40.0.2 -Deprecated==1.2.13 -idna==3.4 -pycparser==2.21 -PyGithub==1.58.2 -PyJWT==2.7.0 -PyNaCl==1.5.0 -requests==2.30.0 -sentry-sdk -urllib3==2.0.2 -wrapt==1.15.0 diff --git a/images/snapshot-release-tags/snapshot_release_tags.py b/images/snapshot-release-tags/snapshot_release_tags.py deleted file mode 100644 index 5ed695482..000000000 --- a/images/snapshot-release-tags/snapshot_release_tags.py +++ /dev/null @@ -1,62 +0,0 @@ -#!/usr/bin/env python3 - -from datetime import datetime, timezone -from github import Github, InputGitAuthor -import json -import os -import re -import sentry_sdk -import subprocess -import tempfile -import urllib.request - -sentry_sdk.init( - # This cron job only runs once weekly, - # so just record all transactions. - traces_sample_rate=1.0, -) - - -if __name__ == "__main__": - if "GITHUB_TOKEN" not in os.environ: - raise Exception("GITHUB_TOKEN environment is not set") - - # Use the GitLab API to get the most recent successful develop pipeline. - gitlab_api_url = "https://gitlab.spack.io/api/v4/projects/2" - pipeline_api_url = f"{gitlab_api_url}/pipelines?ref=develop&status=success" - request = urllib.request.Request(pipeline_api_url) - response = urllib.request.urlopen(request) - response_data = response.read() - try: - pipelines = json.loads(response_data) - except json.decoder.JSONDecodeError: - raise Exception("Failed to parse response as json ({0})".format(response_data)) - - if len(pipelines) == 0: - raise Exception("No successful develop pipelines found!") - - sha = pipelines[0]["sha"] - - date_str = datetime.now(timezone.utc).strftime("%Y-%m-%d") - tag_name = f"develop-{date_str}" - tag_msg = f"Snapshot release {date_str}" - - # Use the GitHub API to create a tag for this commit of develop. - github_token = os.environ.get('GITHUB_TOKEN') - py_github = Github(github_token) - py_gh_repo = py_github.get_repo("spack/spack", lazy=True) - spackbot_author = InputGitAuthor("spackbot", "noreply@spack.io") - print(f"Pushing tag {tag_name} for commit {sha}") - - tag = py_gh_repo.create_git_tag( - tag=tag_name, - message=tag_msg, - object=sha, - type="commit", - tagger=spackbot_author) - - py_gh_repo.create_git_ref( - ref=f"refs/tags/{tag_name}", - sha=tag.sha) - - print("Push done!") diff --git a/k8s/production/custom/protected-publish/cron-jobs.yaml b/k8s/production/custom/protected-publish/cron-jobs.yaml index 5a873f571..7df2ebb16 100644 --- a/k8s/production/custom/protected-publish/cron-jobs.yaml +++ b/k8s/production/custom/protected-publish/cron-jobs.yaml @@ -17,7 +17,7 @@ spec: restartPolicy: Never containers: - name: protected-publish - image: ghcr.io/spack/protected-publish:0.0.9 + image: ghcr.io/spack/buildcache-tools:0.0.1 imagePullPolicy: IfNotPresent resources: requests: @@ -28,6 +28,7 @@ spec: - configMapRef: name: python-scripts-sentry-config args: + - "publish" - "--bucket" - "spack-binaries" - "--ref" diff --git a/k8s/production/custom/snapshot-release-tags/cron-jobs.yaml b/k8s/production/custom/snapshot-release-tags/cron-jobs.yaml index deded8bab..205e4165b 100644 --- a/k8s/production/custom/snapshot-release-tags/cron-jobs.yaml +++ b/k8s/production/custom/snapshot-release-tags/cron-jobs.yaml @@ -16,7 +16,7 @@ spec: restartPolicy: Never containers: - name: snapshot-release-tags - image: ghcr.io/spack/snapshot-release-tags:0.0.4 + image: ghcr.io/spack/buildcache-tools:0.0.1 imagePullPolicy: IfNotPresent resources: requests: @@ -31,5 +31,12 @@ spec: envFrom: - configMapRef: name: python-scripts-sentry-config + args: + - "snapshot" + - "--bucket" + - "spack-binaries" + - "--project" + - "spack/spack-packages" + nodeSelector: spack.io/node-pool: base diff --git a/terraform/modules/spack_aws_k8s/iam_service_accounts.tf b/terraform/modules/spack_aws_k8s/iam_service_accounts.tf index 358f61d16..514ae5d8f 100644 --- a/terraform/modules/spack_aws_k8s/iam_service_accounts.tf +++ b/terraform/modules/spack_aws_k8s/iam_service_accounts.tf @@ -1,3 +1,40 @@ +module "buildcache_snapshoter" { + source = "../iam_service_account" + + deployment_name = var.deployment_name + deployment_stage = var.deployment_stage + + service_account_iam_policies = [ + jsonencode({ + "Version" : "2012-10-17", + "Statement" : [ + { + "Effect" : "Allow", + "Action" : "s3:PutObject", + "Resource" : "${module.protected_binary_mirror.bucket_arn}/v*" + }, + { + "Effect" : "Allow", + "Action" : "s3:PutObject", + "Resource" : "${module.protected_binary_mirror.bucket_arn}/develop-*" + }, + { + "Effect" : "Allow", + "Action" : "s3:GetObject", + "Resource" : "${module.protected_binary_mirror.bucket_arn}/releases/*" + }, + { + "Effect" : "Allow", + "Action" : "s3:GetObject", + "Resource" : "${module.protected_binary_mirror.bucket_arn}/develop/*" + } + ] + }), + ] + service_account_name = "buildcache-snapshot" + service_account_namespace = "custom" +} + module "build_cache_pruner" { source = "../iam_service_account"