77a61da760
The libdir is passed down for sources but it is never used in
any of our sources. As this is confusing and we want to eventually
support multiple libdirs remove this code.
It looks like the libdir for soruces was added a long time ago in 8423da3
but there is no indication if/how it is/was supposed to get used and
AFACT from going over the git history it was very used.
SourceService:dispatch() never sends "libdir" to the actual sources,
so it is not an even technically an API break.
216 lines
8 KiB
Python
216 lines
8 KiB
Python
"""Entrypoints for osbuild
|
|
|
|
This module contains the application and API entrypoints of `osbuild`, the
|
|
command-line-interface to osbuild. The `osbuild_cli()` entrypoint can be safely
|
|
used from tests to run the cli.
|
|
"""
|
|
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
import typing
|
|
from typing import List
|
|
|
|
import osbuild
|
|
import osbuild.meta
|
|
import osbuild.monitor
|
|
from osbuild.meta import ValidationResult
|
|
from osbuild.objectstore import ObjectStore
|
|
from osbuild.pipeline import Manifest
|
|
from osbuild.util.parsing import parse_size
|
|
from osbuild.util.term import fmt as vt
|
|
|
|
|
|
def parse_manifest(path: str) -> dict:
|
|
if path == "-":
|
|
manifest = json.load(sys.stdin)
|
|
else:
|
|
with open(path, encoding="utf8") as f:
|
|
manifest = json.load(f)
|
|
|
|
return manifest
|
|
|
|
|
|
def show_validation(result: ValidationResult, name: str) -> None:
|
|
if name == "-":
|
|
name = "<stdin>"
|
|
|
|
print(f"{vt.bold}{name}{vt.reset} ", end='')
|
|
|
|
if result:
|
|
print(f"is {vt.bold}{vt.green}valid{vt.reset}")
|
|
return
|
|
|
|
print(f"has {vt.bold}{vt.red}errors{vt.reset}:")
|
|
print("")
|
|
|
|
for error in result:
|
|
print(f"{vt.bold}{error.id}{vt.reset}:")
|
|
print(f" {error.message}\n")
|
|
|
|
|
|
def export(name_or_id: str, output_directory: str, store: ObjectStore, manifest: Manifest) -> None:
|
|
pipeline = manifest[name_or_id]
|
|
obj = store.get(pipeline.id)
|
|
dest = os.path.join(output_directory, name_or_id)
|
|
|
|
skip_preserve_owner = \
|
|
os.getenv("OSBUILD_EXPORT_FORCE_NO_PRESERVE_OWNER") == "1"
|
|
os.makedirs(dest, exist_ok=True)
|
|
obj.export(dest, skip_preserve_owner=skip_preserve_owner)
|
|
|
|
|
|
@typing.no_type_check # see https://github.com/python/typeshed/issues/3107
|
|
def parse_arguments(sys_argv: List[str]) -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(prog="osbuild",
|
|
description="Build operating system images")
|
|
|
|
parser.add_argument("manifest_path", metavar="MANIFEST",
|
|
help="json file containing the manifest that should be built, or a '-' to read from stdin")
|
|
parser.add_argument("--store", metavar="DIRECTORY", type=os.path.abspath,
|
|
default=".osbuild",
|
|
help="directory where intermediary os trees are stored")
|
|
parser.add_argument("-l", "--libdir", metavar="DIRECTORY", type=os.path.abspath, default="/usr/lib/osbuild",
|
|
help="directory containing stages, assemblers, and the osbuild library")
|
|
parser.add_argument("--cache-max-size", metavar="SIZE", type=parse_size, default=None,
|
|
help="maximum size of the cache (bytes) or 'unlimited' for no restriction")
|
|
parser.add_argument(
|
|
"--checkpoint",
|
|
metavar="ID",
|
|
action="append",
|
|
type=str,
|
|
default=None,
|
|
help="stage to commit to the object store during build (can be passed multiple times), accepts globs")
|
|
parser.add_argument("--export", metavar="ID", action="append", type=str, default=[],
|
|
help="object to export, can be passed multiple times")
|
|
parser.add_argument("--json", action="store_true",
|
|
help="output results in JSON format")
|
|
parser.add_argument("--output-directory", metavar="DIRECTORY", type=os.path.abspath,
|
|
help="directory where result objects are stored")
|
|
parser.add_argument("--inspect", action="store_true",
|
|
help="return the manifest in JSON format including all the ids")
|
|
parser.add_argument("--monitor", metavar="NAME", default=None,
|
|
help="name of the monitor to be used")
|
|
parser.add_argument("--monitor-fd", metavar="FD", type=int, default=sys.stdout.fileno(),
|
|
help="file descriptor to be used for the monitor")
|
|
parser.add_argument("--stage-timeout", type=int, default=None,
|
|
help="set the maximal time (in seconds) each stage is allowed to run")
|
|
parser.add_argument("--version", action="version",
|
|
help="return the version of osbuild",
|
|
version="%(prog)s " + osbuild.__version__)
|
|
# nargs='?' const='*' means `--break` is equivalent to `--break=*`
|
|
parser.add_argument("--break", dest='debug_break', type=str, nargs='?', const='*',
|
|
help="open debug shell when executing stage. Accepts stage name or id or * (for all)")
|
|
|
|
return parser.parse_args(sys_argv[1:])
|
|
|
|
|
|
# pylint: disable=too-many-branches,too-many-return-statements,too-many-statements
|
|
def osbuild_cli() -> int:
|
|
args = parse_arguments(sys.argv)
|
|
desc = parse_manifest(args.manifest_path)
|
|
|
|
index = osbuild.meta.Index(args.libdir)
|
|
|
|
# detect the format from the manifest description
|
|
info = index.detect_format_info(desc)
|
|
if not info:
|
|
print("Unsupported manifest format")
|
|
return 2
|
|
fmt = info.module
|
|
|
|
# first thing is validation of the manifest
|
|
res = fmt.validate(desc, index)
|
|
if not res:
|
|
if args.json or args.inspect:
|
|
json.dump(res.as_dict(), sys.stdout)
|
|
sys.stdout.write("\n")
|
|
else:
|
|
show_validation(res, args.manifest_path)
|
|
return 2
|
|
|
|
manifest = fmt.load(desc, index)
|
|
|
|
exports = set(args.export)
|
|
unresolved = [e for e in exports if e not in manifest]
|
|
if unresolved:
|
|
for name in unresolved:
|
|
print(f"Export {vt.bold}{name}{vt.reset} not found!")
|
|
print(f"{vt.reset}{vt.bold}{vt.red}Failed{vt.reset}")
|
|
return 1
|
|
|
|
if args.checkpoint:
|
|
marked = manifest.mark_checkpoints(args.checkpoint)
|
|
if not marked:
|
|
print("No checkpoints matched provided patterns!")
|
|
print(f"{vt.reset}{vt.bold}{vt.red}Failed{vt.reset}")
|
|
return 1
|
|
|
|
if args.inspect:
|
|
result = fmt.describe(manifest, with_id=True)
|
|
json.dump(result, sys.stdout)
|
|
sys.stdout.write("\n")
|
|
return 0
|
|
|
|
output_directory = args.output_directory
|
|
|
|
if exports and not output_directory:
|
|
print("Need --output-directory for --export")
|
|
return 1
|
|
|
|
monitor_name = args.monitor
|
|
if not monitor_name:
|
|
monitor_name = "NullMonitor" if args.json else "LogMonitor"
|
|
|
|
try:
|
|
with ObjectStore(args.store) as object_store:
|
|
if args.cache_max_size is not None:
|
|
object_store.maximum_size = args.cache_max_size
|
|
|
|
stage_timeout = args.stage_timeout
|
|
debug_break = args.debug_break
|
|
|
|
pipelines = manifest.depsolve(object_store, exports)
|
|
total_steps = len(manifest.sources) + len(pipelines)
|
|
monitor = osbuild.monitor.make(monitor_name, args.monitor_fd, total_steps)
|
|
monitor.log(f"starting {args.manifest_path}", origin="osbuild.main_cli")
|
|
|
|
manifest.download(object_store, monitor)
|
|
|
|
r = manifest.build(
|
|
object_store,
|
|
pipelines,
|
|
monitor,
|
|
args.libdir,
|
|
debug_break,
|
|
stage_timeout=stage_timeout
|
|
)
|
|
if r["success"]:
|
|
monitor.log(f"manifest {args.manifest_path} finished successfully\n", origin="osbuild.main_cli")
|
|
else:
|
|
# if we had monitor.error() we could use that here
|
|
monitor.log(f"manifest {args.manifest_path} failed\n", origin="osbuild.main_cli")
|
|
|
|
if r["success"] and exports:
|
|
for pid in exports:
|
|
export(pid, output_directory, object_store, manifest)
|
|
|
|
if args.json:
|
|
r = fmt.output(manifest, r, object_store)
|
|
json.dump(r, sys.stdout)
|
|
sys.stdout.write("\n")
|
|
else:
|
|
if r["success"]:
|
|
for name, pl in manifest.pipelines.items():
|
|
print(f"{name + ':': <10}\t{pl.id}")
|
|
else:
|
|
print(f"{vt.reset}{vt.bold}{vt.red}Failed{vt.reset}")
|
|
|
|
return 0 if r["success"] else 1
|
|
|
|
except KeyboardInterrupt:
|
|
print()
|
|
print(f"{vt.reset}{vt.bold}{vt.red}Aborted{vt.reset}")
|
|
return 130
|