pungi/pungi/metadata.py
Haibo Lin 4c297beb65 Include images info in composeinfo.json
This change requires https://github.com/release-engineering/productmd/pull/150

JIRA: RHELCMP-2296
Signed-off-by: Haibo Lin <hlin@redhat.com>
2020-10-14 17:57:30 +08:00

543 lines
20 KiB
Python

# -*- coding: utf-8 -*-
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; version 2 of the License.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Library General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, see <https://gnu.org/licenses/>.
import copy
import os
import time
import productmd.composeinfo
import productmd.treeinfo
from productmd.common import get_major_version
from kobo.shortcuts import relative_path, compute_file_checksums
from pungi.compose_metadata.discinfo import write_discinfo as create_discinfo
from pungi.compose_metadata.discinfo import write_media_repo as create_media_repo
def get_description(compose, variant, arch):
if "release_discinfo_description" in compose.conf:
result = compose.conf["release_discinfo_description"]
elif variant.type == "layered-product":
# we need to make sure the layered product behaves as it was composed separately
result = "%s %s for %s %s" % (
variant.release_name,
variant.release_version,
compose.conf["release_name"],
get_major_version(compose.conf["release_version"]),
)
else:
result = "%s %s" % (
compose.conf["release_name"],
compose.conf["release_version"],
)
if compose.conf.get("base_product_name", ""):
result += " for %s %s" % (
compose.conf["base_product_name"],
compose.conf["base_product_version"],
)
result = result % {"variant_name": variant.name, "arch": arch}
return result
def write_discinfo(compose, arch, variant):
if variant.type == "addon":
return
os_tree = compose.paths.compose.os_tree(arch, variant)
path = os.path.join(os_tree, ".discinfo")
# description = get_volid(compose, arch, variant)
description = get_description(compose, variant, arch)
return create_discinfo(path, description, arch)
def write_media_repo(compose, arch, variant, timestamp=None):
if variant.type == "addon":
return
os_tree = compose.paths.compose.os_tree(arch, variant)
path = os.path.join(os_tree, "media.repo")
# description = get_volid(compose, arch, variant)
description = get_description(compose, variant, arch)
return create_media_repo(path, description, timestamp)
def compose_to_composeinfo(compose):
ci = productmd.composeinfo.ComposeInfo()
# compose
ci.compose.id = compose.compose_id
ci.compose.type = compose.compose_type
ci.compose.date = compose.compose_date
ci.compose.respin = compose.compose_respin
ci.compose.label = compose.compose_label
ci.compose.final = compose.supported
# product
ci.release.name = compose.conf["release_name"]
ci.release.version = compose.conf["release_version"]
ci.release.short = compose.conf["release_short"]
ci.release.is_layered = True if compose.conf.get("base_product_name", "") else False
ci.release.type = compose.conf["release_type"].lower()
ci.release.internal = bool(compose.conf["release_internal"])
# base product
if ci.release.is_layered:
ci.base_product.name = compose.conf["base_product_name"]
ci.base_product.version = compose.conf["base_product_version"]
ci.base_product.short = compose.conf["base_product_short"]
ci.base_product.type = compose.conf["base_product_type"].lower()
def dump_variant(variant, parent=None):
var = productmd.composeinfo.Variant(ci)
tree_arches = compose.conf.get("tree_arches")
if tree_arches and not (set(variant.arches) & set(tree_arches)):
return None
# variant details
# remove dashes from variant ID, rely on productmd verification
var.id = variant.id.replace("-", "")
var.uid = variant.uid
var.name = variant.name
var.type = variant.type
var.arches = set(variant.arches)
if var.type == "layered-product":
var.release.name = variant.release_name
var.release.short = variant.release_short
var.release.version = variant.release_version
var.release.is_layered = True
var.release.type = ci.release.type
for arch in variant.arches:
# paths: binaries
var.paths.os_tree[arch] = relative_path(
compose.paths.compose.os_tree(
arch=arch, variant=variant, create_dir=False
).rstrip("/")
+ "/",
compose.paths.compose.topdir().rstrip("/") + "/",
).rstrip("/")
var.paths.repository[arch] = relative_path(
compose.paths.compose.repository(
arch=arch, variant=variant, create_dir=False
).rstrip("/")
+ "/",
compose.paths.compose.topdir().rstrip("/") + "/",
).rstrip("/")
var.paths.packages[arch] = relative_path(
compose.paths.compose.packages(
arch=arch, variant=variant, create_dir=False
).rstrip("/")
+ "/",
compose.paths.compose.topdir().rstrip("/") + "/",
).rstrip("/")
iso_dir = (
compose.paths.compose.iso_dir(
arch=arch, variant=variant, create_dir=False
)
or ""
)
if iso_dir and os.path.isdir(
os.path.join(compose.paths.compose.topdir(), iso_dir)
):
var.paths.isos[arch] = relative_path(
iso_dir, compose.paths.compose.topdir().rstrip("/") + "/"
).rstrip("/")
image_dir = compose.paths.compose.image_dir(variant=variant) or ""
if image_dir:
image_dir = image_dir % {"arch": arch}
if os.path.isdir(image_dir):
var.paths.images[arch] = relative_path(
image_dir, compose.paths.compose.topdir().rstrip("/") + "/"
).rstrip("/")
jigdo_dir = (
compose.paths.compose.jigdo_dir(
arch=arch, variant=variant, create_dir=False
)
or ""
)
if jigdo_dir and os.path.isdir(
os.path.join(compose.paths.compose.topdir(), jigdo_dir)
):
var.paths.jigdos[arch] = relative_path(
jigdo_dir, compose.paths.compose.topdir().rstrip("/") + "/"
).rstrip("/")
# paths: sources
var.paths.source_tree[arch] = relative_path(
compose.paths.compose.os_tree(
arch="source", variant=variant, create_dir=False
).rstrip("/")
+ "/",
compose.paths.compose.topdir().rstrip("/") + "/",
).rstrip("/")
var.paths.source_repository[arch] = relative_path(
compose.paths.compose.repository(
arch="source", variant=variant, create_dir=False
).rstrip("/")
+ "/",
compose.paths.compose.topdir().rstrip("/") + "/",
).rstrip("/")
var.paths.source_packages[arch] = relative_path(
compose.paths.compose.packages(
arch="source", variant=variant, create_dir=False
).rstrip("/")
+ "/",
compose.paths.compose.topdir().rstrip("/") + "/",
).rstrip("/")
source_iso_dir = (
compose.paths.compose.iso_dir(
arch="source", variant=variant, create_dir=False
)
or ""
)
if source_iso_dir and os.path.isdir(
os.path.join(compose.paths.compose.topdir(), source_iso_dir)
):
var.paths.source_isos[arch] = relative_path(
source_iso_dir, compose.paths.compose.topdir().rstrip("/") + "/"
).rstrip("/")
source_jigdo_dir = (
compose.paths.compose.jigdo_dir(
arch="source", variant=variant, create_dir=False
)
or ""
)
if source_jigdo_dir and os.path.isdir(
os.path.join(compose.paths.compose.topdir(), source_jigdo_dir)
):
var.paths.source_jigdos[arch] = relative_path(
source_jigdo_dir, compose.paths.compose.topdir().rstrip("/") + "/"
).rstrip("/")
# paths: debug
var.paths.debug_tree[arch] = relative_path(
compose.paths.compose.debug_tree(
arch=arch, variant=variant, create_dir=False
).rstrip("/")
+ "/",
compose.paths.compose.topdir().rstrip("/") + "/",
).rstrip("/")
var.paths.debug_repository[arch] = relative_path(
compose.paths.compose.debug_repository(
arch=arch, variant=variant, create_dir=False
).rstrip("/")
+ "/",
compose.paths.compose.topdir().rstrip("/") + "/",
).rstrip("/")
var.paths.debug_packages[arch] = relative_path(
compose.paths.compose.debug_packages(
arch=arch, variant=variant, create_dir=False
).rstrip("/")
+ "/",
compose.paths.compose.topdir().rstrip("/") + "/",
).rstrip("/")
"""
# XXX: not supported (yet?)
debug_iso_dir = (
compose.paths.compose.debug_iso_dir(arch=arch, variant=variant) or ""
)
if debug_iso_dir:
var.debug_iso_dir[arch] = relative_path(
debug_iso_dir, compose.paths.compose.topdir().rstrip("/") + "/"
).rstrip("/")
debug_jigdo_dir = (
compose.paths.compose.debug_jigdo_dir(arch=arch, variant=variant) or ""
)
if debug_jigdo_dir:
var.debug_jigdo_dir[arch] = relative_path(
debug_jigdo_dir, compose.paths.compose.topdir().rstrip("/") + "/"
).rstrip("/")
"""
for v in variant.get_variants(recursive=False):
x = dump_variant(v, parent=variant)
if x is not None:
var.add(x)
return var
for variant_id in sorted(compose.variants):
variant = compose.variants[variant_id]
v = dump_variant(variant)
if v is not None:
ci.variants.add(v)
return ci
def write_compose_info(compose):
ci = compose_to_composeinfo(compose)
msg = "Writing composeinfo"
compose.log_info("[BEGIN] %s" % msg)
path = compose.paths.compose.metadata("composeinfo.json")
# make a copy of composeinfo and modify the copy
# if any path in variant paths doesn't exist or just an empty
# dir, set it to None, then it won't be dumped.
ci_copy = copy.deepcopy(ci)
for variant in ci_copy.variants.variants.values():
for field in variant.paths._fields:
field_paths = getattr(variant.paths, field)
for arch, dirpath in field_paths.items():
dirpath = os.path.join(compose.paths.compose.topdir(), dirpath)
if not os.path.isdir(dirpath):
# If the directory does not exist, do not include the path
# in metadata.
field_paths[arch] = None
ci_copy.dump(path)
compose.log_info("[DONE ] %s" % msg)
def write_tree_info(compose, arch, variant, timestamp=None, bi=None):
if variant.type in ("addon",) or variant.is_empty:
return
compose.log_debug(
"on arch '%s' looking at variant '%s' of type '%s'"
% (arch, variant, variant.type)
)
if not timestamp:
timestamp = int(time.time())
else:
timestamp = int(timestamp)
os_tree = (
compose.paths.compose.os_tree(arch=arch, variant=variant).rstrip("/") + "/"
)
ti = productmd.treeinfo.TreeInfo()
# load from buildinstall .treeinfo
if variant.type == "layered-product":
# we need to make sure the layered product behaves as it was composed separately
# release
# TODO: read from variants.xml
ti.release.name = variant.release_name
ti.release.version = variant.release_version
ti.release.short = variant.release_short
ti.release.is_layered = True
ti.release.type = compose.conf["release_type"].lower()
# base product
ti.base_product.name = compose.conf["release_name"]
if "." in compose.conf["release_version"]:
# remove minor version if present
ti.base_product.version = get_major_version(compose.conf["release_version"])
else:
ti.base_product.version = compose.conf["release_version"]
ti.base_product.short = compose.conf["release_short"]
else:
# release
ti.release.name = compose.conf["release_name"]
ti.release.version = compose.conf.get(
"treeinfo_version", compose.conf["release_version"]
)
ti.release.short = compose.conf["release_short"]
ti.release.is_layered = (
True if compose.conf.get("base_product_name", "") else False
)
ti.release.type = compose.conf["release_type"].lower()
# base product
if ti.release.is_layered:
ti.base_product.name = compose.conf["base_product_name"]
ti.base_product.version = compose.conf["base_product_version"]
ti.base_product.short = compose.conf["base_product_short"]
# tree
ti.tree.arch = arch
ti.tree.build_timestamp = timestamp
# ti.platforms
# main variant
var = productmd.treeinfo.Variant(ti)
if variant.type == "layered-product":
var.id = variant.parent.id
var.uid = variant.parent.uid
var.name = variant.parent.name
var.type = "variant"
else:
# remove dashes from variant ID, rely on productmd verification
var.id = variant.id.replace("-", "")
var.uid = variant.uid
var.name = variant.name
var.type = variant.type
var.paths.packages = (
relative_path(
compose.paths.compose.packages(
arch=arch, variant=variant, create_dir=False
).rstrip("/")
+ "/",
os_tree,
).rstrip("/")
or "."
)
var.paths.repository = (
relative_path(
compose.paths.compose.repository(
arch=arch, variant=variant, create_dir=False
).rstrip("/")
+ "/",
os_tree,
).rstrip("/")
or "."
)
ti.variants.add(var)
repomd_path = os.path.join(var.paths.repository, "repodata", "repomd.xml")
createrepo_checksum = compose.conf["createrepo_checksum"]
if os.path.isfile(repomd_path):
ti.checksums.add(repomd_path, createrepo_checksum, root_dir=os_tree)
for i in variant.get_variants(types=["addon"], arch=arch):
addon = productmd.treeinfo.Variant(ti)
addon.id = i.id
addon.uid = i.uid
addon.name = i.name
addon.type = i.type
compose.log_debug(
"variant '%s' inserting addon uid '%s' type '%s'"
% (variant, addon.uid, addon.type)
)
os_tree = compose.paths.compose.os_tree(arch=arch, variant=i).rstrip("/") + "/"
addon.paths.packages = (
relative_path(
compose.paths.compose.packages(
arch=arch, variant=i, create_dir=False
).rstrip("/")
+ "/",
os_tree,
).rstrip("/")
or "."
)
addon.paths.repository = (
relative_path(
compose.paths.compose.repository(
arch=arch, variant=i, create_dir=False
).rstrip("/")
+ "/",
os_tree,
).rstrip("/")
or "."
)
var.add(addon)
repomd_path = os.path.join(addon.paths.repository, "repodata", "repomd.xml")
if os.path.isfile(repomd_path):
ti.checksums.add(repomd_path, createrepo_checksum, root_dir=os_tree)
class LoraxProduct(productmd.treeinfo.Release):
def _validate_short(self):
# HACK: set self.short so .treeinfo produced by lorax can be read
if not self.short:
self.short = compose.conf["release_short"]
class LoraxTreeInfo(productmd.treeinfo.TreeInfo):
def __init__(self, *args, **kwargs):
super(LoraxTreeInfo, self).__init__(*args, **kwargs)
self.release = LoraxProduct(self)
# images
if variant.type == "variant" and bi.succeeded(variant, arch):
os_tree = compose.paths.compose.os_tree(arch, variant)
# clone all but 'general' sections from buildinstall .treeinfo
bi_dir = compose.paths.work.buildinstall_dir(arch)
if compose.conf.get("buildinstall_method") == "lorax":
# The .treeinfo file produced by lorax is nested in variant
# subdirectory. Legacy buildinstall runs once per arch, so there is
# only one file.
bi_dir = os.path.join(bi_dir, variant.uid)
bi_treeinfo = os.path.join(bi_dir, ".treeinfo")
if os.path.exists(bi_treeinfo):
bi_ti = LoraxTreeInfo()
bi_ti.load(bi_treeinfo)
# stage2 - mainimage
if bi_ti.stage2.mainimage:
ti.stage2.mainimage = bi_ti.stage2.mainimage
ti.checksums.add(
ti.stage2.mainimage, createrepo_checksum, root_dir=os_tree
)
# stage2 - instimage
if bi_ti.stage2.instimage:
ti.stage2.instimage = bi_ti.stage2.instimage
ti.checksums.add(
ti.stage2.instimage, createrepo_checksum, root_dir=os_tree
)
# images
for platform in bi_ti.images.images:
ti.images.images[platform] = {}
ti.tree.platforms.add(platform)
for image, path in bi_ti.images.images[platform].items():
if not path:
# The .treeinfo file contains an image without a path.
# We can't add that.
continue
ti.images.images[platform][image] = path
ti.checksums.add(path, createrepo_checksum, root_dir=os_tree)
path = os.path.join(
compose.paths.compose.os_tree(arch=arch, variant=variant), ".treeinfo"
)
compose.log_info("Writing treeinfo: %s" % path)
ti.dump(path)
def populate_extra_files_metadata(
metadata, variant, arch, topdir, files, checksum_types, relative_root=None
):
"""
:param metadata: an instance of productmd.extra_files.ExtraFiles to
populate with the current files
:param Variant variant: under which variant should the files be listed
:param str arch: under which arch should the files be listed
:param topdir: directory where files are located
:param files: list of file paths relative to topdir
:param checksum_types: list of checksums to compute
:param relative_root: ancestor directory of topdir, this will be removed
from paths written to local metadata file
"""
for copied_file in files:
full_path = os.path.join(topdir, copied_file)
size = os.path.getsize(full_path)
try:
checksums = compute_file_checksums(full_path, checksum_types)
except IOError as exc:
raise RuntimeError(
"Failed to calculate checksum for %s: %s" % (full_path, exc)
)
if relative_root:
copied_file = os.path.relpath(full_path, relative_root)
metadata.add(variant.uid, arch, copied_file, size, checksums)
strip_prefix = (
(os.path.relpath(topdir, relative_root) + "/") if relative_root else ""
)
with open(os.path.join(topdir, "extra_files.json"), "w") as f:
metadata.dump_for_tree(f, variant.uid, arch, strip_prefix)