Add ccp.py - compose-critical package list generation script

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>

quality-assurance/critpath/ccp.py

@@ -0,0 +1,171 @@
+#!/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 json
+import koji
+import requests
+import sys
+from collections import defaultdict
+
+KOJI = koji.ClientSession("https://koji.fedoraproject.org/kojihub")
+
+
+class CriticalPackages():
+    """Data class for tracking critical package types per arch
+
+    buildroot: packages installed as part of the image creation environment
+    image: packages installed as part of the image itself
+    """
+    def __init__(self):
+        self.buildroot = set()
+        self.image = set()
+
+    def __repr__(self):
+        return f"buildroot: {self.buildroot}\nimage: {self.image}\n"
+
+
+def get_koji_brpkgs(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)}
+
+
+def get_ib_pkgs(task):
+    """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]
+    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
+    brpkgs = {pkg["name"] for pkg in brmanifest["packages"]}
+    # add koji buildroot packages
+    brpkgs.update(get_koji_brpkgs(task))
+    ipkgs = {pkg["name"] for pkg in imanifest["packages"]}
+    return (brpkgs, ipkgs)
+
+
+def get_kiwi_pkgs(task):
+    """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")
+    ipkgs = {line.split("|")[0] for line in imanifest.splitlines()}
+    brpkgs = get_koji_brpkgs(task)
+    return (brpkgs, ipkgs)
+
+
+def get_lorax_pkgs(task):
+    """Get packages from a lorax task.
+    """
+    brpkgs = set()
+    ipkgs = 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":
+            ipkgs.add(elems[-1].rsplit("-", 2)[0])
+    brpkgs = get_koji_brpkgs(task)
+    return (brpkgs, ipkgs)
+
+
+def get_createiso_pkgs(task):
+    """Get packages from a createiso task."""
+    brpkgs = get_koji_brpkgs(task)
+    # 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 (brpkgs, set())
+
+
+def process_task(task, arch, variant, critpkgs, getfunc):
+    """Get critical packages for an image build task."""
+    if variant not in critpkgs:
+        critpkgs[variant] = {arch: CriticalPackages()}
+    elif arch not in critpkgs[variant]:
+        critpkgs[variant][arch] = CriticalPackages()
+    brpkgs, ipkgs = getfunc(task)
+    critpkgs[variant][arch].buildroot.update(brpkgs)
+    critpkgs[variant][arch].image.update(ipkgs)
+
+
+USAGE = f"""Usage: {sys.argv[0]} [COMPOSE_TOP_LEVEL]
+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.
+"""
+
+if len(sys.argv) > 2:
+    print(USAGE)
+    sys.exit("Only one argument can be passed, the compose top level, e.g. https://kojipkgs.fedoraproject.org/compose/rawhide/Fedora-Rawhide-20260706.n.0")
+if len(sys.argv) > 1:
+    curl = sys.argv[1]
+else:
+    curl = "https://kojipkgs.fedoraproject.org/compose/rawhide/latest-Fedora-Rawhide"
+if curl == "--help":
+    print(USAGE)
+    sys.exit()
+
+cidurl = f"{curl}/COMPOSE_ID"
+cid = requests.get(cidurl).text
+durl = f"{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()
+critpkgs = 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, critpkgs, getfunc)
+        else:
+            process_task(image["koji_task"], image["arch"], variant, critpkgs, getfunc)
+
+allcrit = set()
+critbyv = {}
+for variant in critpkgs:
+    print(f"Variant: {variant}")
+    critbyv[variant] = set()
+    for arch in critpkgs[variant]:
+        critbyv[variant].update(critpkgs[variant][arch].buildroot)
+        critbyv[variant].update(critpkgs[variant][arch].image)
+        allcrit.update(critbyv[variant])
+        brcount = len(critpkgs[variant][arch].buildroot)
+        icount = len(critpkgs[variant][arch].image)
+        print(f"Arch: {arch} - buildroot {brcount}, image {icount}")
+print(f"Total: {len(allcrit)}")
+with open(f"{cid}-ccp.txt", "w", encoding="utf-8") as outfh:
+    outfh.write("\n".join(sorted(allcrit)))
+    print(f"Wrote {cid}-ccp.txt")
+for variant, pkgs in critbyv.items():
+    critbyv[variant] = sorted(pkgs)
+with open(f"{cid}-ccp.json", "w", encoding="utf-8") as jsonoutfh:
+    json.dump(critbyv, jsonoutfh)
+    print(f"Wrote {cid}-ccp.json")