Add ccp.py - compose-critical package list generation script
All checks were successful
/ test (push) Successful in 3s
All checks were successful
/ test (push) Successful in 3s
This adds ccp.py - a script for generating the "compose-critical package list" for a given compose. It's a bit of a WIP but it does already work and give you a fairly useful list (as a flat text file for now). Signed-off-by: Adam Williamson <awilliam@redhat.com>
This commit is contained in:
parent
6ebadaead5
commit
ec9c972fec
1 changed files with 254 additions and 0 deletions
254
quality-assurance/critpath/ccp.py
Executable file
254
quality-assurance/critpath/ccp.py
Executable file
|
|
@ -0,0 +1,254 @@
|
|||
#!/usr/bin/python3
|
||||
#
|
||||
# Copyright (C) Red Hat Inc,
|
||||
# SPDX-License-Identifier: GPL-2.0+
|
||||
#
|
||||
# Authors: Adam Williamson <awilliam@redhat.com>
|
||||
|
||||
# this is a script, not a public module, we don't need docstrings
|
||||
# pylint: disable=missing-module-docstring,missing-class-docstring,missing-function-docstring
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import koji
|
||||
import requests
|
||||
import sys
|
||||
from collections import defaultdict
|
||||
|
||||
KOJI = koji.ClientSession("https://koji.fedoraproject.org/kojihub")
|
||||
RPMD = {}
|
||||
SRCCACHE = {}
|
||||
|
||||
|
||||
def get_source(pkg, arch):
|
||||
"""Get the source package name for a binary package name.
|
||||
"""
|
||||
global SRCCACHE
|
||||
if pkg in SRCCACHE:
|
||||
return SRCCACHE[pkg]
|
||||
for srcnevra in RPMD[arch]:
|
||||
for nevra in RPMD[arch][srcnevra].keys():
|
||||
if pkg == nevra.rsplit("-", 2)[0]:
|
||||
SRCCACHE[pkg] = srcnevra.rsplit("-", 2)[0]
|
||||
return SRCCACHE[pkg]
|
||||
raise ValueError(f"Could not find source package for {pkg}!")
|
||||
|
||||
def get_koji_brnames(task):
|
||||
"""Get the Koji buildroot packages for a task."""
|
||||
brid = KOJI.listBuildroots(taskID=task)[-1]["id"]
|
||||
return {item["name"] for item in KOJI.listRPMs(componentBuildrootID=brid) if item["name"] != "gpg-pubkey"}
|
||||
|
||||
|
||||
def get_ib_pkgs(task, arch, source):
|
||||
"""Get packages from an image-builder task.
|
||||
"""
|
||||
files = KOJI.getTaskResult(task)["files"]
|
||||
# image-builder manages its own buildroot so we have two levels of
|
||||
# buildroot to handle: the koji buildroot (which will have i-b and
|
||||
# its deps) and i-b's own buildroot
|
||||
brmanifestname = [fn for fn in files if fn.endswith("buildroot-build.spdx.json")][0]
|
||||
# NOTE: this is only correct for disk images, if we build any other
|
||||
# critical images with image-builder we will likely have to adjust
|
||||
imanifestname = [fn for fn in files if fn.endswith("image-os.spdx.json")][0]
|
||||
brmanifest = json.loads(KOJI.downloadTaskOutput(task, brmanifestname))
|
||||
imanifest = json.loads(KOJI.downloadTaskOutput(task, imanifestname))
|
||||
# i-b buildroot packages
|
||||
brnames = {pkg["name"] for pkg in brmanifest["packages"]}
|
||||
brsrcs = {pkg["sourceInfo"].replace("Source RPM: ", "").rsplit("-", 2)[0] for pkg in brmanifest["packages"]}
|
||||
# add koji buildroot packages
|
||||
brnames.update(get_koji_brnames(task))
|
||||
inames = {pkg["name"] for pkg in imanifest["packages"]}
|
||||
isrcs = {pkg["sourceInfo"].replace("Source RPM: ", "").rsplit("-", 2)[0] for pkg in imanifest["packages"]}
|
||||
return (brnames, inames, brsrcs, isrcs)
|
||||
|
||||
|
||||
def get_kiwi_pkgs(task, arch, source):
|
||||
"""Get packages from a kiwi task.
|
||||
"""
|
||||
files = KOJI.getTaskResult(task)["files"]
|
||||
imanifestname = [fn for fn in files if fn.endswith(".packages")][0]
|
||||
imanifest = KOJI.downloadTaskOutput(task, imanifestname).decode("utf-8")
|
||||
brsrcs = set()
|
||||
inames = set()
|
||||
isrcs = set()
|
||||
invras = set()
|
||||
for line in imanifest.splitlines():
|
||||
iname = line.split("|")[0]
|
||||
if iname == "gpg-pubkey":
|
||||
continue
|
||||
inames.add(iname)
|
||||
if source:
|
||||
isrcs.add(get_source(iname, arch))
|
||||
brnames = get_koji_brnames(task)
|
||||
if source:
|
||||
brsrcs = {get_source(pkg, arch) for pkg in brnames}
|
||||
return (brnames, inames, brsrcs, isrcs)
|
||||
|
||||
|
||||
def get_lorax_pkgs(task, arch, source):
|
||||
"""Get packages from a lorax task.
|
||||
"""
|
||||
brsrcs = set()
|
||||
inames = set()
|
||||
isrcs = set()
|
||||
runroot = KOJI.downloadTaskOutput(task, "runroot.log").decode("utf-8")
|
||||
# parse runroot.log for image manifest
|
||||
for line in runroot.splitlines():
|
||||
elems = line.split()
|
||||
if elems[-2] == "Install":
|
||||
iname = elems[-1].rsplit("-", 2)[0]
|
||||
if iname == "gpg-pubkey":
|
||||
continue
|
||||
inames.add(iname)
|
||||
if source:
|
||||
isrcs.add(get_source(iname, arch))
|
||||
brnames = get_koji_brnames(task)
|
||||
if source:
|
||||
brsrcs = {get_source(pkg, arch) for pkg in brnames}
|
||||
return (brnames, inames, brsrcs, isrcs)
|
||||
|
||||
|
||||
def get_createiso_pkgs(task, arch, source):
|
||||
"""Get packages from a createiso task."""
|
||||
brsrcs = set()
|
||||
brnames = get_koji_brnames(task)
|
||||
if source:
|
||||
brsrcs = {get_source(pkg, arch) for pkg in brnames}
|
||||
# for fedora, createiso is only used to graft RPMs onto boot.iso
|
||||
# to produce 'DVD' installers; we already know the manifest of
|
||||
# the boot.iso and we don't really need to treat the grafted-on
|
||||
# repo packages as critpath, so we'll only return buildroot here
|
||||
# to capture the tools used for grafting
|
||||
return (brnames, set(), brsrcs, set())
|
||||
|
||||
|
||||
def process_task(task, arch, variant, source, critpkgs, critsrc, getfunc):
|
||||
"""Get critical packages for an image build task."""
|
||||
if variant not in critpkgs:
|
||||
critpkgs[variant] = {arch: {"buildroot": set(), "image": set()}}
|
||||
critsrc[variant] = {arch: {"buildroot": set(), "image": set()}}
|
||||
elif arch not in critpkgs[variant]:
|
||||
critpkgs[variant][arch] = {"buildroot": set(), "image": set()}
|
||||
critsrc[variant][arch] = {"buildroot": set(), "image": set()}
|
||||
brnames, inames, brsrcs, isrcs = getfunc(task, arch, source)
|
||||
critpkgs[variant][arch]["buildroot"].update(brnames)
|
||||
critpkgs[variant][arch]["image"].update(inames)
|
||||
critsrc[variant][arch]["buildroot"].update(brsrcs)
|
||||
critsrc[variant][arch]["image"].update(isrcs)
|
||||
|
||||
|
||||
def source_convert(critpkgs, rpmd):
|
||||
"""Produce a version of the critpkgs dict with source package
|
||||
"""
|
||||
|
||||
|
||||
def parse_args():
|
||||
"""Parse arguments with argparse."""
|
||||
parser = argparse.ArgumentParser(
|
||||
description=(
|
||||
"Produce compose-critical package information for a Fedora compose. "
|
||||
"Operates on latest Rawhide compose if no arguments are passed. "
|
||||
"Outputs to [COMPOSE_ID]-ccp.txt and [COMPOSE_ID]-ccp.json, and "
|
||||
"optionally [COMPOSE_ID]-ccp-source.txt and [COMPOSE_ID]-ccp-source.json."
|
||||
)
|
||||
)
|
||||
parser.add_argument(
|
||||
"-s",
|
||||
"--source",
|
||||
help="Also generate source package lists",
|
||||
required=False,
|
||||
action="store_true",
|
||||
)
|
||||
parser.add_argument(
|
||||
"curl",
|
||||
help="URL to compose top level",
|
||||
nargs="?",
|
||||
default="https://kojipkgs.fedoraproject.org/compose/rawhide/latest-Fedora-Rawhide",
|
||||
)
|
||||
return parser.parse_args()
|
||||
|
||||
def run(args):
|
||||
cidurl = f"{args.curl}/COMPOSE_ID"
|
||||
cid = requests.get(cidurl).text
|
||||
durl = f"{args.curl}/logs/global/deliverables-v2.json"
|
||||
resp = requests.get(durl)
|
||||
if resp.status_code != 200:
|
||||
sys.exit(f"Could not find deliverables at {durl}")
|
||||
deliverables = resp.json()
|
||||
if args.source:
|
||||
# populate global copy of RPM metadata for later use
|
||||
global RPMD
|
||||
RPMD = requests.get(f"{args.curl}/compose/metadata/rpms.json").json()["payload"]["rpms"]["Everything"]
|
||||
|
||||
critpkgs = dict()
|
||||
critsrc = dict()
|
||||
for phase, images in deliverables["required"].items():
|
||||
getfunc, kidmethod = {
|
||||
"buildinstall": (get_lorax_pkgs, ""),
|
||||
"kiwibuild": (get_kiwi_pkgs, "createKiwiImage"),
|
||||
"imageBuilderBuild": (get_ib_pkgs, "imageBuilderBuildArch"),
|
||||
"iso": (get_createiso_pkgs, ""),
|
||||
}[phase]
|
||||
for image in images:
|
||||
variant = image["variant"]
|
||||
if kidmethod:
|
||||
# we need to find the subtasks of the specified method
|
||||
kids = KOJI.getTaskChildren(image["koji_task"])
|
||||
createtasks = [(kid["id"], kid["arch"]) for kid in kids if kid["method"] == kidmethod]
|
||||
for createtask, arch in createtasks:
|
||||
process_task(createtask, arch, variant, args.source, critpkgs, critsrc, getfunc)
|
||||
else:
|
||||
process_task(image["koji_task"], image["arch"], variant, args.source, critpkgs, critsrc, getfunc)
|
||||
|
||||
allcrit = set()
|
||||
if args.source:
|
||||
allscrit = set()
|
||||
for variant in critpkgs:
|
||||
print(f"Variant: {variant}")
|
||||
for arch in critpkgs[variant]:
|
||||
allcrit.update(critpkgs[variant][arch]["buildroot"].union(critpkgs[variant][arch]["image"]))
|
||||
if args.source:
|
||||
allscrit.update(critsrc[variant][arch]["buildroot"].union(critsrc[variant][arch]["image"]))
|
||||
# convert to lists for JSON output
|
||||
critpkgs[variant][arch]["buildroot"] = sorted(critpkgs[variant][arch]["buildroot"])
|
||||
critpkgs[variant][arch]["image"] = sorted(critpkgs[variant][arch]["image"])
|
||||
critsrc[variant][arch]["buildroot"] = sorted(critsrc[variant][arch]["buildroot"])
|
||||
critsrc[variant][arch]["image"] = sorted(critsrc[variant][arch]["image"])
|
||||
brcount = len(critpkgs[variant][arch]["buildroot"])
|
||||
icount = len(critpkgs[variant][arch]["image"])
|
||||
print(f"Arch: {arch} - buildroot {brcount}, image {icount}")
|
||||
if args.source:
|
||||
brscount = len(critsrc[variant][arch]["buildroot"])
|
||||
iscount = len(critsrc[variant][arch]["image"])
|
||||
print(f"Arch: {arch} - buildroot source {brscount}, image source {iscount}")
|
||||
print(f"Total: {len(allcrit)}")
|
||||
if args.source:
|
||||
print(f"Source total: {len(allscrit)}")
|
||||
with open(f"{cid}-ccp.txt", "w", encoding="utf-8") as outfh:
|
||||
outfh.write("\n".join(sorted(allcrit)))
|
||||
print(f"Wrote {cid}-ccp.txt")
|
||||
with open(f"{cid}-ccp.json", "w", encoding="utf-8") as jsonoutfh:
|
||||
json.dump(critpkgs, jsonoutfh)
|
||||
print(f"Wrote {cid}-ccp.json")
|
||||
if args.source:
|
||||
with open(f"{cid}-ccp-source.txt", "w", encoding="utf-8") as soutfh:
|
||||
soutfh.write("\n".join(sorted(allscrit)))
|
||||
print(f"Wrote {cid}-ccp-source.txt")
|
||||
with open(f"{cid}-ccp-source.json", "w", encoding="utf-8") as sjsonoutfh:
|
||||
json.dump(critsrc, sjsonoutfh)
|
||||
print(f"Wrote {cid}-ccp-source.json")
|
||||
|
||||
def main():
|
||||
"""Main loop."""
|
||||
try:
|
||||
args = parse_args()
|
||||
run(args)
|
||||
except KeyboardInterrupt:
|
||||
sys.stderr.write("Interrupted, exiting...\n")
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
||||
# vim: set textwidth=120 ts=8 et sw=4:
|
||||
Loading…
Add table
Add a link
Reference in a new issue