Repository: osbuild/osbuild
Pull request #557 (2021-01-14 18:34:19)

Flatten the pipeline
Instead of having build pipelines nested within the pipeline it is the build pipeline for, the nested structure is transferred into a flat list of pipelines. As a result the recursion is gone and all the pipelines and trees are build one after the other. This is now possible since floating objects are kept alive by the store itself and all trees that are being built are transparently via them. The immediate result dictionary changed accordingly. To keep the JSON output of osbuild the same, the result is now routed through a format specific converter. 

Additionally, the v1 format module gained a function to retrieve the global `tree_id` and `output_id`. With the new models those global ids will go away eventually and thus need to go through the format specific code.


Patch:
diff --git a/osbuild/formats/v1.py b/osbuild/formats/v1.py
index 859438a1..ff35d987 100644
--- a/osbuild/formats/v1.py
+++ b/osbuild/formats/v1.py
@@ -1,6 +1,6 @@
 # Version 1 of the manifest description
 
-from typing import Dict
+from typing import Dict, List, Optional, Tuple
 from osbuild.meta import Index, ValidationResult
 from ..pipeline import Manifest, Pipeline, detect_host_runner
 
@@ -18,7 +18,7 @@ def describe(manifest: Manifest, *, with_id=False) -> Dict:
     def describe_pipeline(pipeline: Pipeline) -> Dict:
         description = {}
         if pipeline.build:
-            build = pipeline.build
+            build = manifest[pipeline.build]
             description["build"] = {
                 "pipeline": describe_pipeline(build),
                 "runner": pipeline.runner
@@ -34,7 +34,7 @@ def describe(manifest: Manifest, *, with_id=False) -> Dict:
         return description
 
     description = {
-        "pipeline": describe_pipeline(manifest.pipeline)
+        "pipeline": describe_pipeline(manifest.pipelines[-1])
     }
 
     if manifest.sources:
@@ -43,24 +43,25 @@ def describe(manifest: Manifest, *, with_id=False) -> Dict:
     return description
 
 
-def load_build(description: Dict, sources_options: Dict):
+def load_build(description: Dict, sources_options: Dict, result: List[Pipeline]):
     pipeline = description.get("pipeline")
     if pipeline:
-        build_pipeline = load_pipeline(pipeline, sources_options)
+        build_pipeline = load_pipeline(pipeline, sources_options, result)
     else:
         build_pipeline = None
 
     return build_pipeline, description["runner"]
 
 
-def load_pipeline(description: Dict, sources_options: Dict) -> Pipeline:
+def load_pipeline(description: Dict, sources_options: Dict, result: List[Pipeline]) -> Pipeline:
     build = description.get("build")
     if build:
-        build_pipeline, runner = load_build(build, sources_options)
+        build_pipeline, runner = load_build(build, sources_options, result)
     else:
         build_pipeline, runner = None, detect_host_runner()
 
-    pipeline = Pipeline(runner, build_pipeline)
+
+    pipeline = Pipeline(runner, build_pipeline and build_pipeline.tree_id)
 
     for s in description.get("stages", []):
         pipeline.add_stage(s["name"], sources_options, s.get("options", {}))
@@ -69,6 +70,8 @@ def load_pipeline(description: Dict, sources_options: Dict) -> Pipeline:
     if a:
         pipeline.set_assembler(a["name"], a.get("options", {}))
 
+    result.append(pipeline)
+
     return pipeline
 
 
@@ -78,12 +81,40 @@ def load(description: Dict) -> Manifest:
     pipeline = description.get("pipeline", {})
     sources = description.get("sources", {})
 
-    pipeline = load_pipeline(pipeline, sources)
-    manifest = Manifest(pipeline, sources)
+    pipelines = []
+
+    load_pipeline(pipeline, sources, pipelines)
+
+    manifest = Manifest(pipelines, sources)
 
     return manifest
 
 
+def get_ids(manifest: Manifest) -> Tuple[Optional[str], Optional[str]]:
+    pipeline = manifest.pipelines[-1]
+    return pipeline.tree_id, pipeline.output_id
+
+
+def output(manifest: Manifest, res: Dict) -> Dict:
+    """Convert a result into the v1 format"""
+
+    def result_for_pipeline(pipeline):
+        current = res[pipeline.id]
+        retval = {"success": current["success"]}
+        if pipeline.build:
+            build = manifest[pipeline.build]
+            retval["build"] = result_for_pipeline(build)
+        stages = current.get("stages")
+        if stages:
+            retval["stages"] = stages
+        assembler = current.get("assembler")
+        if assembler:
+            retval["assembler"] = assembler
+        return retval
+
+    return result_for_pipeline(manifest.pipelines[-1])
+
+
 def validate(manifest: Dict, index: Index) -> ValidationResult:
     """Validate a OSBuild manifest
 
diff --git a/osbuild/main_cli.py b/osbuild/main_cli.py
index 75cd968c..789a8ea6 100644
--- a/osbuild/main_cli.py
+++ b/osbuild/main_cli.py
@@ -14,6 +14,7 @@ import sys
 import osbuild
 import osbuild.meta
 import osbuild.monitor
+from osbuild.objectstore import ObjectStore
 from osbuild.formats import v1 as fmt
 
 
@@ -90,7 +91,6 @@ def osbuild_cli():
         return 2
 
     manifest = fmt.load(desc)
-    pipeline = manifest.pipeline
 
     if args.checkpoint:
         missed = manifest.mark_checkpoints(args.checkpoint)
@@ -114,24 +114,27 @@ def osbuild_cli():
     monitor = osbuild.monitor.make(monitor_name, sys.stdout.fileno())
 
     try:
-        r = manifest.build(
-            args.store,
-            monitor,
-            args.libdir,
-            output_directory=args.output_directory
-        )
+        with ObjectStore(args.store) as object_store:
+            r = manifest.build(
+                object_store,
+                monitor,
+                args.libdir,
+                output_directory=args.output_directory
+            )
     except KeyboardInterrupt:
         print()
         print(f"{RESET}{BOLD}{RED}Aborted{RESET}")
         return 130
 
     if args.json:
+        r = fmt.output(manifest, r)
         json.dump(r, sys.stdout)
         sys.stdout.write("\n")
     else:
         if r["success"]:
-            print("tree id:", pipeline.tree_id)
-            print("output id:", pipeline.output_id)
+            tree_id, output_id = fmt.get_ids(manifest)
+            print("tree id:", tree_id)
+            print("output id:", output_id)
         else:
             print()
             print(f"{RESET}{BOLD}{RED}Failed{RESET}")
diff --git a/osbuild/objectstore.py b/osbuild/objectstore.py
index 3d8e17bb..a24e6fdc 100644
--- a/osbuild/objectstore.py
+++ b/osbuild/objectstore.py
@@ -4,7 +4,6 @@ import hashlib
 import os
 import subprocess
 import tempfile
-import weakref
 from typing import Optional
 
 from osbuild.util.types import PathLike
@@ -49,6 +48,7 @@ class Object:
         self._base = None
         self._workdir = None
         self._tree = None
+        self.id = None
         self.store = store
         self.reset()
 
@@ -72,6 +72,7 @@ class Object:
     def base(self, base_id: Optional[str]):
         self._init = not base_id
         self._base = base_id
+        self.id = base_id
 
     @property
     def treesum(self) -> str:
@@ -97,6 +98,7 @@ class Object:
         self._check_readers()
         self._check_writer()
         self.init()
+        self.id = None
         with self.tempdir("writer") as target:
             mount(self._path, target, ro=False)
             try:
@@ -153,6 +155,7 @@ class Object:
         if self._workdir:
             self._workdir.cleanup()
             self._workdir = None
+        self.id = None
 
     def _check_readers(self):
         """Internal: Raise a ValueError if there are readers"""
@@ -245,11 +248,22 @@ class ObjectStore(contextlib.AbstractContextManager):
         os.makedirs(self.objects, exist_ok=True)
         os.makedirs(self.refs, exist_ok=True)
         os.makedirs(self.tmp, exist_ok=True)
-        self._objs = weakref.WeakSet()
+        self._objs = set()
+
+    def _get_floating(self, object_id: str) -> Optional[Object]:
+        """Internal: get a non-committed object"""
+        for obj in self._objs:
+            if obj.id == object_id:
+                return obj
+        return None
 
     def contains(self, object_id):
         if not object_id:
             return False
+
+        if self._get_floating(object_id):
+            return True
+
         return os.access(self.resolve_ref(object_id), os.F_OK)
 
     def resolve_ref(self, object_id: Optional[str]) -> Optional[str]:
@@ -265,6 +279,10 @@ class ObjectStore(contextlib.AbstractContextManager):
                                            suffix=suffix)
 
     def get(self, object_id):
+        obj = self._get_floating(object_id)
+        if obj:
+            return obj
+
         if not self.contains(object_id):
             return None
 
diff --git a/osbuild/pipeline.py b/osbuild/pipeline.py
index 39914084..0a5e6ee8 100644
--- a/osbuild/pipeline.py
+++ b/osbuild/pipeline.py
@@ -3,7 +3,7 @@ import hashlib
 import json
 import os
 import tempfile
-from typing import Dict
+from typing import Dict, List
 
 from .api import API
 from . import buildroot
@@ -147,6 +147,10 @@ class Pipeline:
         self.stages = []
         self.assembler = None
 
+    @property
+    def id(self):
+        return self.output_id or self.tree_id
+
     @property
     def tree_id(self):
         return self.stages[-1].id if self.stages else None
@@ -156,15 +160,13 @@ class Pipeline:
         return self.assembler.id if self.assembler else None
 
     def add_stage(self, name, sources_options=None, options=None):
-        build = self.build.tree_id if self.build else None
-        stage = Stage(name, sources_options, build, self.tree_id, options or {})
+        stage = Stage(name, sources_options, self.build, self.tree_id, options or {})
         self.stages.append(stage)
         if self.assembler:
             self.assembler.base = stage.id
 
     def set_assembler(self, name, options=None):
-        build = self.build.tree_id if self.build else None
-        self.assembler = Assembler(name, build, self.tree_id, options or {})
+        self.assembler = Assembler(name, self.build, self.tree_id, options or {})
 
     def build_stages(self, object_store, monitor, libdir):
         results = {"success": True}
@@ -178,22 +180,10 @@ class Pipeline:
         if not self.build:
             build_tree = objectstore.HostTree(object_store)
         else:
-            build = self.build
-
-            r, t, tree = build.build_stages(object_store,
-                                            monitor,
-                                            libdir)
-
-            results["build"] = r
-            if not r["success"]:
-                results["success"] = False
-                return results, None, None
-
-            # Cleanup the build tree (`t`) which was used to
-            # build `tree`; it is now not needed anymore
-            t.cleanup()
+            build_tree = object_store.get(self.build)
 
-            build_tree = tree
+        if not build_tree:
+            raise AssertionError(f"build tree {self.build} not found")
 
         # If there are no stages, just return build tree we just
         # obtained and a new, clean `tree`
@@ -201,20 +191,18 @@ class Pipeline:
             tree = object_store.new()
             return results, build_tree, tree
 
-        # Create a new tree. The base is our tree_id because if that
-        # is already in the store, we can short-circuit directly and
-        # exit directly; `tree` is then used to read the tree behind
-        # `self.tree_id`
-        tree = object_store.new(base_id=self.tree_id)
+        # Check if the tree that we are supposed to build does
+        # already exist. If so, short-circuit here
+        tree = object_store.get(self.tree_id)
 
-        if object_store.contains(self.tree_id):
+        if tree:
             return results, build_tree, tree
 
         # Not in the store yet, need to actually build it, but maybe
         # an intermediate checkpoint exists: Find the last stage that
         # already exists in the store and use that as the base.
+        tree = object_store.new()
         base_idx = -1
-        tree.base = None
         for i in reversed(range(len(self.stages))):
             if object_store.contains(self.stages[i].id):
                 tree.base = self.stages[i].id
@@ -249,6 +237,10 @@ class Pipeline:
                 results["success"] = False
                 return results, None, None
 
+            # The content of the tree now corresponds to the stage that
+            # was build and this can can be identified via the id of it
+            tree.id = stage.id
+
             if stage.checkpoint:
                 object_store.commit(tree, stage.id)
 
@@ -294,34 +286,32 @@ class Pipeline:
 
         monitor.begin(self)
 
-        with objectstore.ObjectStore(store) as object_store:
-            # If the final result is already in the store, no need to attempt
-            # building it. Just fetch the cached information. If the associated
-            # tree exists, we return it as well, but we do not care if it is
-            # missing, since it is not a mandatory part of the result and would
-            # usually be needless overhead.
-            obj = object_store.get(self.output_id)
+        # If the final result is already in the store, no need to attempt
+        # building it. Just fetch the cached information. If the associated
+        # tree exists, we return it as well, but we do not care if it is
+        # missing, since it is not a mandatory part of the result and would
+        # usually be needless overhead.
+        obj = store.get(self.output_id)
 
-            if not obj:
-                results, build_tree, tree = self.build_stages(object_store,
-                                                              monitor,
-                                                              libdir)
+        if not obj:
+            results, build_tree, tree = self.build_stages(store,
+                                                          monitor,
+                                                          libdir)
 
-                if not results["success"]:
-                    return results
+            if not results["success"]:
+                return results
 
-                r, obj = self.assemble(object_store,
-                                       build_tree,
-                                       tree,
-                                       monitor,
-                                       libdir)
+            r, obj = self.assemble(store,
+                                   build_tree,
+                                   tree,
+                                   monitor,
+                                   libdir)
 
-                results.update(r)  # This will also update 'success'
+            results.update(r)  # This will also update 'success'
 
-            if obj:
-                if output_directory:
-                    obj.export(output_directory)
-                obj.cleanup()
+        if obj:
+            if output_directory:
+                obj.export(output_directory)
 
         monitor.finish(results)
 
@@ -331,13 +321,21 @@ class Pipeline:
 class Manifest:
     """Representation of a pipeline and its sources"""
 
-    def __init__(self, pipeline: Pipeline, source_options: Dict):
-        self.pipeline = pipeline
+    def __init__(self, pipelines: List[Pipeline], source_options: Dict):
+        self.pipelines = pipelines
         self.sources = source_options
 
     def build(self, store, monitor, libdir, output_directory):
-        return self.pipeline.run(store, monitor, libdir, output_directory)
+        results = {"success": True}
+
+        for pl in self.pipelines:
+            res = pl.run(store, monitor, libdir, output_directory)
+            results[pl.id] = res
+            if not res["success"]:
+                results["success"] = False
+                return results
 
+        return results
 
     def mark_checkpoints(self, checkpoints):
         points = set(checkpoints)
@@ -359,12 +357,18 @@ class Manifest:
                 mark_stage(stage)
             if pl.assembler:
                 mark_assembler(pl.assembler)
-            if pl.build:
-                mark_pipeline(pl.build)
 
-        mark_pipeline(self.pipeline)
+        for pl in self.pipelines:
+            mark_pipeline(pl)
+
         return points
 
+    def __getitem__(self, pipeline_id):
+        for pl in self.pipelines:
+            if pl.id == pipeline_id:
+                return pl
+        raise KeyError("{pipeline_id} not found")
+
 
 def detect_host_runner():
     """Use os-release(5) to detect the runner for the host"""


Test patch:
diff --git a/test/mod/test_monitor.py b/test/mod/test_monitor.py
index d0eac1cb..82593ad1 100644
--- a/test/mod/test_monitor.py
+++ b/test/mod/test_monitor.py
@@ -12,6 +12,7 @@ from collections import defaultdict
 import osbuild
 import osbuild.meta
 from osbuild.monitor import LogMonitor
+from osbuild.objectstore import ObjectStore
 from osbuild.pipeline import detect_host_runner
 from .. import test
 
@@ -67,9 +68,9 @@ class TestMonitor(unittest.TestCase):
 
             logfile = os.path.join(tmpdir, "log.txt")
 
-            with open(logfile, "w") as log:
+            with open(logfile, "w") as log, ObjectStore(storedir) as store:
                 monitor = LogMonitor(log.fileno())
-                res = pipeline.run(storedir,
+                res = pipeline.run(store,
                                    monitor,
                                    libdir=os.path.abspath(os.curdir),
                                    output_directory=outputdir)
@@ -100,10 +101,11 @@ class TestMonitor(unittest.TestCase):
             outputdir = os.path.join(tmpdir, "output")
 
             tape = TapeMonitor()
-            res = pipeline.run(storedir,
-                               tape,
-                               libdir=os.path.abspath(os.curdir),
-                               output_directory=outputdir)
+            with ObjectStore(storedir) as store:
+                res = pipeline.run(store,
+                                   tape,
+                                   libdir=os.path.abspath(os.curdir),
+                                   output_directory=outputdir)
 
         assert res
         self.assertEqual(tape.counter["begin"], 1)
diff --git a/test/mod/test_osbuild.py b/test/mod/test_osbuild.py
index 5c1c96d4..9148ead6 100644
--- a/test/mod/test_osbuild.py
+++ b/test/mod/test_osbuild.py
@@ -82,11 +82,11 @@ class TestDescriptions(unittest.TestCase):
         build = osbuild.Pipeline("org.osbuild.test")
         build.add_stage("org.osbuild.test", {}, {"one": 1})
 
-        pipeline = osbuild.Pipeline("org.osbuild.test", build)
+        pipeline = osbuild.Pipeline("org.osbuild.test", build.tree_id)
         pipeline.add_stage("org.osbuild.test", {}, {"one": 2})
         pipeline.set_assembler("org.osbuild.test")
 
-        manifest = osbuild.Manifest(pipeline, {})
+        manifest = osbuild.Manifest([build, pipeline], {})
 
         self.assertEqual(fmt.describe(manifest), {
             "pipeline": {
diff --git a/test/test.py b/test/test.py
index 7fbe4d85..9a2e1a8b 100644
--- a/test/test.py
+++ b/test/test.py
@@ -355,7 +355,8 @@ class OSBuild(contextlib.AbstractContextManager):
         manifest_json = json.loads(manifest_data)
 
         manifest = fmt.load(manifest_json)
-        return manifest.pipeline.tree_id
+        tree_id, _ = fmt.get_ids(manifest)
+        return tree_id
 
     @contextlib.contextmanager
     def map_object(self, obj):