Function prepare_initrd

utils.py:183–263

Returns a decompressed initial ramdisk. Parameters: architecture (str): Architecture to download image for. rootfs_format (str): Initrd format ('cpio' or 'ext4')

(
    architecture: str,
    rootfs_format: str = 'cpio',
    gh_json_file: Path | None = None,
    modules: Path | None = None,
)

def prepare_initrd(
    architecture: str,
    rootfs_format: str = 'cpio',
    gh_json_file: Path | None = None,
    modules: Path | None = None,
) -> Path:
    """
    Returns a decompressed initial ramdisk.

    Parameters:
        architecture (str): Architecture to download image for.
        rootfs_format (str): Initrd format ('cpio' or 'ext4')
    """
    src = Path(BOOT_UTILS, 'images', architecture, f"rootfs.{rootfs_format}.zst")
    src.parent.mkdir(exist_ok=True, parents=True)

    # If the user supplied a GitHub release JSON file, we do not need to bother
    # querying the GitHub API at all.
    if gh_json_file and gh_json_file != UNINIT_PATH:
        if not gh_json_file.exists():
            msg = f"Provided GitHub JSON file ('{gh_json_file}') does not exist!"
            raise FileNotFoundError(msg)
        gh_json_rel = json.loads(gh_json_file.read_text(encoding='utf-8'))
    else:
        # Make sure that the current user is not rate limited by GitHub,
        # otherwise the next API call will not return valid information.
        gh_json_rl = get_gh_json('https://api.github.com/rate_limit')

        # If we have API calls remaining or have already queried the API previously
        # and cached the result, we can query for the latest release to make sure
        # that we are up to date.
        if (remaining := gh_json_rl['resources']['core']['remaining']) > 0:
            gh_json_rel = get_gh_json(f"https://api.github.com/repos/{REPO}/releases/latest")
        elif not src.exists():
            limit = gh_json_rl['resources']['core']['limit']
            msg = (
                f"Cannot query GitHub API for latest images release due to rate limit (remaining: {remaining}, limit: {limit}) and {src} does not exist already! "
                'Download it manually or supply a GitHub personal access token via the GITHUB_TOKEN environment variable to make an authenticated GitHub API request.'
            )
            raise RuntimeError(msg)

    # Download the ramdisk if it is not already downloaded
    if not src.exists():
        # gh_json_rel cannot be unset when used here because the elif condition
        # above is the same as this one, which causes the script to exit.
        # pylint: disable-next=possibly-used-before-assignment
        download_initrd(gh_json_rel, src)
    # If it is already downloaded, check that it is up to date and download
    # an update only if necessary.
    elif (rel_file := src.with_name('.release')).exists():
        cur_rel = rel_file.read_text(encoding='utf-8')
        supplied_rel = gh_json_rel['tag_name']
        if cur_rel != supplied_rel:
            download_initrd(gh_json_rel, src)

    check_cmd('zstd')
    (dst := src.with_suffix('')).unlink(missing_ok=True)
    subprocess.run(['zstd', '-d', src, '-o', dst, '-q'], check=True)