Repository: osbuild/osbuild
Pull request #568 (2021-01-21 19:07:19)

Introduce assembler pipelines and identify pipelines

This converts the assembler phase of the `Pipeline`[1] with a new pipeline that has the previous assembler as the first and only stage. This was possible to do since the `Assembler` previously got converted into a `Stage`, and that stage has the tree as `Input`; so this stage just got moved to a newly created `Pipeline`. 
Additionally, all pipelines in the manifest now have to have a name assigned. This is currently used in the version 1 format code (`formats/v1`) to identify the pipeline that builds the tree and the pipeline that corresponds to the assembler. 
The inputs of a stage are now included in the `id` calculation. NB: This, and the move of the assembler to a new pipeline, changes the `id` of the assembler.

[1] That is the main pipeline, since for build pipelines the assembler is not run, even if defined.

Replacs #554 

gicmo: I will actually re-work the `id` commits.
gicmo: I have updated the PR to completely remove `tree_id` usages. Also instead of having `build-N` for nested build pipelines I am using `build-build-build...` now because I think that makes the nesting clearer (not that anyone would ever actually see that). I added more tests, esp. for the nesting case.

Patch:
diff --git a/osbuild/formats/v1.py b/osbuild/formats/v1.py
index e30e9f5a..2def88cb 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, List, Optional, Tuple
+from typing import Dict, Optional, Tuple
 from osbuild.meta import Index, ValidationResult
 from ..inputs import Input
 from ..pipeline import Manifest, Pipeline, detect_host_runner
@@ -29,14 +29,16 @@ def describe(manifest: Manifest, *, with_id=False) -> Dict:
             stages = [describe_stage(s) for s in pipeline.stages]
             description["stages"] = stages
 
-        if pipeline.assembler:
-            assembler = describe_stage(pipeline.assembler)
-            description["assembler"] = assembler
         return description
 
-    description = {
-        "pipeline": describe_pipeline(manifest.pipelines[-1])
-    }
+    pipeline = describe_pipeline(manifest["tree"])
+
+    assembler = manifest.get("assembler")
+    if assembler:
+        description = describe_stage(assembler.stages[0])
+        pipeline["assembler"] = description
+
+    description = {"pipeline": pipeline}
 
     if manifest.sources:
         description["sources"] = manifest.sources
@@ -44,41 +46,60 @@ def describe(manifest: Manifest, *, with_id=False) -> Dict:
     return description
 
 
-def load_build(description: Dict, index: Index, result: List[Pipeline]):
+def load_assembler(description: Dict, index: Index, manifest: Manifest):
+    pipeline = manifest["tree"]
+
+    build, base, runner = pipeline.build, pipeline.id, pipeline.runner
+    name, options = description["name"], description.get("options", {})
+
+    # Add a pipeline with one stage for our assembler
+    pipeline = manifest.add_pipeline("assembler", runner, build)
+    pipeline.export = True
+
+    info = index.get_module_info("Assembler", name)
+
+    stage = pipeline.add_stage(info, options, {})
+    info = index.get_module_info("Input", "org.osbuild.tree")
+    stage.inputs = {
+        "tree": Input(info, {"pipeline": {"id": base}})
+    }
+
+    return pipeline
+
+
+def load_build(description: Dict, index: Index, manifest: Manifest, n: int):
     pipeline = description.get("pipeline")
     if pipeline:
-        build_pipeline = load_pipeline(pipeline, index, result)
+        build_pipeline = load_pipeline(pipeline, index, manifest, n + 1)
     else:
         build_pipeline = None
 
     return build_pipeline, description["runner"]
 
 
-def load_pipeline(description: Dict, index: Index, result: List[Pipeline]) -> Pipeline:
+def load_pipeline(description: Dict, index: Index, manifest: Manifest, n: int = 0) -> Pipeline:
     build = description.get("build")
     if build:
-        build_pipeline, runner = load_build(build, index, result)
+        build_pipeline, runner = load_build(build, index, manifest, n)
     else:
         build_pipeline, runner = None, detect_host_runner()
 
-    build_id = build_pipeline and build_pipeline.tree_id
-    pipeline = Pipeline(runner, build_id)
+    # the "main" pipeline is called `tree`, since it is building the
+    # tree that will later be used by the `assembler`. Nested build
+    # pipelines will get call "build", and "build-build-...", where
+    # the number of repetitions is equal their level of nesting
+    if not n:
+        name = "tree"
+    else:
+        name = "-".join(["build"] * n)
+
+    build_id = build_pipeline and build_pipeline.id
+    pipeline = manifest.add_pipeline(name, runner, build_id)
 
     for s in description.get("stages", []):
         info = index.get_module_info("Stage", s["name"])
         pipeline.add_stage(info, s.get("options", {}))
 
-    a = description.get("assembler")
-    if a:
-        info = index.get_module_info("Assembler", a["name"])
-        asm = pipeline.set_assembler(info, a.get("options", {}))
-        info = index.get_module_info("Input", "org.osbuild.tree")
-        asm.inputs = {
-            "tree": Input(info, {"pipeline": {"id": pipeline.tree_id}})
-        }
-
-    result.append(pipeline)
-
     return pipeline
 
 
@@ -88,22 +109,26 @@ def load(description: Dict, index: Index) -> Manifest:
     pipeline = description.get("pipeline", {})
     sources = description.get("sources", {})
 
-    pipelines = []
+    manifest = Manifest(sources)
 
-    load_pipeline(pipeline, index, pipelines)
+    load_pipeline(pipeline, index, manifest)
 
-    for pipeline in pipelines:
+    # load the assembler, if any
+    assembler = pipeline.get("assembler")
+    if assembler:
+        load_assembler(assembler, index, manifest)
+
+    for pipeline in manifest.pipelines.values():
         for stage in pipeline.stages:
             stage.sources = sources
 
-    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
+    pipeline = manifest["tree"]
+    assembler = manifest.get("assembler")
+    return pipeline.id, assembler and assembler.id
 
 
 def output(manifest: Manifest, res: Dict) -> Dict:
@@ -129,7 +154,15 @@ def output(manifest: Manifest, res: Dict) -> Dict:
             retval["assembler"] = assembler
         return retval
 
-    return result_for_pipeline(manifest.pipelines[-1])
+    result = result_for_pipeline(manifest["tree"])
+
+    assembler = manifest.get("assembler")
+    if assembler:
+        current = res.get(assembler.id)
+        if current:
+            result["assembler"] = current["stages"][0]
+
+    return result
 
 
 def validate(manifest: Dict, index: Index) -> ValidationResult:
diff --git a/osbuild/inputs.py b/osbuild/inputs.py
index f0e250f1..7503de34 100644
--- a/osbuild/inputs.py
+++ b/osbuild/inputs.py
@@ -17,6 +17,7 @@ osbuild is the path. The input options are just passed to the
 """
 
 
+import hashlib
 import importlib
 import json
 import os
@@ -36,7 +37,15 @@ class Input:
     def __init__(self, info: ModuleInfo, options: Dict):
         self.info = info
         self.options = options or {}
+        self.id = self.calc_id()
 
+    def calc_id(self):
+        m = hashlib.sha256()
+        m.update(json.dumps(self.name, sort_keys=True).encode())
+        m.update(json.dumps(self.options, sort_keys=True).encode())
+        return m.hexdigest()
+
+    @property
     def name(self) -> str:
         return self.info.name
 
diff --git a/osbuild/pipeline.py b/osbuild/pipeline.py
index 4973d17d..33dd831d 100644
--- a/osbuild/pipeline.py
+++ b/osbuild/pipeline.py
@@ -1,8 +1,9 @@
+import collections
 import contextlib
 import hashlib
 import json
 import os
-from typing import Dict, List
+from typing import Dict, Iterator, Optional
 
 from .api import API
 from . import buildroot
@@ -52,6 +53,9 @@ class Stage:
         m.update(json.dumps(self.build, sort_keys=True).encode())
         m.update(json.dumps(self.base, sort_keys=True).encode())
         m.update(json.dumps(self.options, sort_keys=True).encode())
+        if self.inputs:
+            data = {n: i.id for n, i in self.inputs.items()}
+            m.update(json.dumps(data, sort_keys=True).encode())
         return m.hexdigest()
 
     def run(self, tree, runner, build_tree, store, monitor, libdir):
@@ -111,35 +115,36 @@ class Stage:
 
 
 class Pipeline:
-    def __init__(self, runner=None, build=None):
+    def __init__(self, name: str, runner=None, build=None):
+        self.name = name
         self.build = build
         self.runner = runner
         self.stages = []
         self.assembler = None
+        self.export = False
 
     @property
     def id(self):
-        return self.output_id or self.tree_id
-
-    @property
-    def tree_id(self):
+        """
+        Pipeline id: corresponds to the `id` of the last stage
+
+        In contrast to `name` this identifies the pipeline via
+        the tree, i.e. the content, it produces. Therefore two
+        pipelines that produce the same `tree`, i.e. have the
+        same exact stages and build pipeline, will have the
+        same `id`; thus the `id`, in contrast to `name` does
+        not uniquely identify a pipeline.
+        In case a Pipeline has no stages, its `id` is `None`.
+        """
         return self.stages[-1].id if self.stages else None
 
-    @property
-    def output_id(self):
-        return self.assembler.id if self.assembler else None
-
     def add_stage(self, info, options, sources_options=None):
-        stage = Stage(info, sources_options, self.build, self.tree_id, options or {})
+        stage = Stage(info, sources_options, self.build, self.id, options or {})
         self.stages.append(stage)
         if self.assembler:
             self.assembler.base = stage.id
         return stage
 
-    def set_assembler(self, info, options=None):
-        self.assembler = Stage(info, {}, self.build, self.tree_id, options or {})
-        return self.assembler
-
     def build_stages(self, object_store, monitor, libdir):
         results = {"success": True}
 
@@ -165,7 +170,7 @@ class Pipeline:
 
         # 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)
+        tree = object_store.get(self.id)
 
         if tree:
             return results, build_tree, tree
@@ -217,39 +222,6 @@ class Pipeline:
 
         return results, build_tree, tree
 
-    def assemble(self, object_store, build_tree, monitor, libdir):
-        results = {"success": True}
-
-        if not self.assembler:
-            return results, None
-
-        output = object_store.new()
-
-        with build_tree.read() as build_dir, \
-             output.write() as output_dir:
-
-            monitor.assembler(self.assembler)
-
-            r = self.assembler.run(output_dir,
-                                   self.runner,
-                                   build_dir,
-                                   object_store,
-                                   monitor,
-                                   libdir)
-
-            monitor.result(r)
-
-        results["assembler"] = r.as_dict()
-        if not r.success:
-            output.cleanup()
-            results["success"] = False
-            return results, None
-
-        if self.assembler.checkpoint:
-            object_store.commit(output, self.assembler.id)
-
-        return results, output
-
     def run(self, store, monitor, libdir, output_directory):
         results = {"success": True}
 
@@ -260,22 +232,15 @@ class Pipeline:
         # 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)
+        obj = store.get(self.id)
 
         if not obj:
-            results, build_tree, _ = self.build_stages(store, monitor, libdir)
+            results, _, obj = self.build_stages(store, monitor, libdir)
 
             if not results["success"]:
                 return results
 
-            r, obj = self.assemble(store,
-                                   build_tree,
-                                   monitor,
-                                   libdir)
-
-            results.update(r)  # This will also update 'success'
-
-        if obj:
+        if self.export and obj:
             if output_directory:
                 obj.export(output_directory)
 
@@ -287,14 +252,21 @@ class Pipeline:
 class Manifest:
     """Representation of a pipeline and its sources"""
 
-    def __init__(self, pipelines: List[Pipeline], source_options: Dict):
-        self.pipelines = pipelines
+    def __init__(self, source_options: Dict):
+        self.pipelines = collections.OrderedDict()
         self.sources = source_options
 
+    def add_pipeline(self, name: str, runner: str, build: str) -> Pipeline:
+        pipeline = Pipeline(name, runner, build)
+        if name in self.pipelines:
+            raise ValueError(f"Name {name} already exists")
+        self.pipelines[name] = pipeline
+        return pipeline
+
     def build(self, store, monitor, libdir, output_directory):
         results = {"success": True}
 
-        for pl in self.pipelines:
+        for pl in self.pipelines.values():
             res = pl.run(store, monitor, libdir, output_directory)
             results[pl.id] = res
             if not res["success"]:
@@ -312,28 +284,35 @@ class Manifest:
                 stage.checkpoint = True
                 points.remove(c)
 
-        def mark_assembler(assembler):
-            c = assembler.id
-            if c in points:
-                assembler.checkpoint = True
-                points.remove(c)
-
         def mark_pipeline(pl):
             for stage in pl.stages:
                 mark_stage(stage)
-            if pl.assembler:
-                mark_assembler(pl.assembler)
 
-        for pl in self.pipelines:
+        for pl in self.pipelines.values():
             mark_pipeline(pl)
 
         return points
 
-    def __getitem__(self, pipeline_id):
-        for pl in self.pipelines:
-            if pl.id == pipeline_id:
+    def get(self, name_or_id: str) -> Optional[Pipeline]:
+        pl = self.pipelines.get(name_or_id)
+        if pl:
+            return pl
+        for pl in self.pipelines.values():
+            if pl.id == name_or_id:
                 return pl
-        raise KeyError("{pipeline_id} not found")
+        return None
+
+    def __contains__(self, name_or_id: str) -> bool:
+        return self.get(name_or_id) is not None
+
+    def __getitem__(self, name_or_id: str) -> Pipeline:
+        pl = self.get(name_or_id)
+        if pl:
+            return pl
+        raise KeyError(f"'{name_or_id}' not found")
+
+    def __iter__(self) -> Iterator[Pipeline]:
+        return iter(self.pipelines.values())
 
 
 def detect_host_runner():


Test patch:
diff --git a/test/mod/test_fmt_v1.py b/test/mod/test_fmt_v1.py
index 1791780d..62a1c6bf 100644
--- a/test/mod/test_fmt_v1.py
+++ b/test/mod/test_fmt_v1.py
@@ -21,6 +21,17 @@ BASIC_PIPELINE = {
     "pipeline": {
         "build": {
             "pipeline": {
+                "build": {
+                    "pipeline": {
+                        "stages": [
+                            {
+                                "name": "org.osbuild.noop",
+                                "options": {"zero": 0}
+                            }
+                        ]
+                    },
+                    "runner": "org.osbuild.linux"
+                },
                 "stages": [
                     {
                         "name": "org.osbuild.test",
@@ -51,7 +62,6 @@ class TestFormatV1(unittest.TestCase):
         storedir = pathlib.Path(tmpdir, "store")
         monitor = NullMonitor(sys.stderr.fileno())
         libdir = os.path.abspath(os.curdir)
-        print(libdir)
         store = ObjectStore(storedir)
         outdir = pathlib.Path(tmpdir, "out")
         outdir.mkdir()
@@ -83,7 +93,7 @@ class TestFormatV1(unittest.TestCase):
         # Load a pipeline and check the resulting manifest
         def check_stage(have: osbuild.Stage, want: Dict):
             self.assertEqual(have.name, want["name"])
-            self.assertEqual(have.options, want["options"])
+            self.assertEqual(have.options, want.get("options", {}))
 
         index = osbuild.meta.Index(os.curdir)
 
@@ -95,29 +105,44 @@ class TestFormatV1(unittest.TestCase):
         manifest = fmt.load(description, index)
         self.assertIsNotNone(manifest)
 
-        # We have to have a build pipeline and a main pipeline
+        # We have to have two build pipelines and a main pipeline
         self.assertTrue(manifest.pipelines)
-        self.assertTrue(len(manifest.pipelines) == 2)
+        self.assertTrue(len(manifest.pipelines) == 4)
+
+        # access the individual pipelines via their names
+
+        # the inner most build pipeline
+        build = description["pipeline"]["build"]["pipeline"]["build"]
+        pl = manifest["build-build"]
+        have = pl.stages[0]
+        want = build["pipeline"]["stages"][0]
+        check_stage(have, want)
+
+        runner = build["runner"]
 
+        # the build pipeline for the 'tree' pipeline
         build = description["pipeline"]["build"]
-        pl = manifest.pipelines[0]
+        pl = manifest["build"]
         have = pl.stages[0]
         want = build["pipeline"]["stages"][0]
+        self.assertEqual(pl.runner, runner)
         check_stage(have, want)
 
         runner = build["runner"]
 
-        # main pipeline is the next one
-        pl = manifest.pipelines[1]
+        # the main, aka 'tree', pipeline
+        pl = manifest["tree"]
         have = pl.stages[0]
         want = description["pipeline"]["stages"][0]
         self.assertEqual(pl.runner, runner)
         check_stage(have, want)
 
-        # the assembler
-        have = pl.assembler
+        # the assembler pipeline
+        pl = manifest["assembler"]
+        have = pl.stages[0]
         want = description["pipeline"]["assembler"]
-        self.assertEqual(have.name, want["name"])
+        self.assertEqual(pl.runner, runner)
+        check_stage(have, want)
 
 
     def test_describe(self):
diff --git a/test/mod/test_monitor.py b/test/mod/test_monitor.py
index cb549769..3816e074 100644
--- a/test/mod/test_monitor.py
+++ b/test/mod/test_monitor.py
@@ -38,10 +38,6 @@ class TapeMonitor(osbuild.monitor.BaseMonitor):
         self.counter["stages"] += 1
         self.stages.add(stage.id)
 
-    def assembler(self, assembler: osbuild.Stage):
-        self.counter["assembler"] += 1
-        self.asm = assembler.id
-
     def result(self, result: osbuild.pipeline.BuildResult):
         self.counter["result"] += 1
         self.results.add(result.id)
@@ -58,13 +54,11 @@ class TestMonitor(unittest.TestCase):
         index = osbuild.meta.Index(os.curdir)
 
         runner = detect_host_runner()
-        pipeline = osbuild.Pipeline(runner=runner)
+        pipeline = osbuild.Pipeline("pipeline", runner=runner)
         info = index.get_module_info("Stage", "org.osbuild.noop")
         pipeline.add_stage(info, {
             "isthisthereallife": False
         })
-        info = index.get_module_info("Assembler", "org.osbuild.noop")
-        pipeline.set_assembler(info)
 
         with tempfile.TemporaryDirectory() as tmpdir:
             storedir = os.path.join(tmpdir, "store")
@@ -84,7 +78,6 @@ class TestMonitor(unittest.TestCase):
 
             assert res
             self.assertIn(pipeline.stages[0].id, log)
-            self.assertIn(pipeline.assembler.id, log)
             self.assertIn("isthisthereallife", log)
 
     @unittest.skipUnless(test.TestBase.can_bind_mount(), "root-only")
@@ -93,7 +86,7 @@ class TestMonitor(unittest.TestCase):
         runner = detect_host_runner()
         index = osbuild.meta.Index(os.curdir)
 
-        pipeline = osbuild.Pipeline(runner=runner)
+        pipeline = osbuild.Pipeline("pipeline", runner=runner)
         noop_info = index.get_module_info("Stage", "org.osbuild.noop")
         pipeline.add_stage(noop_info, {
             "isthisthereallife": False
@@ -101,8 +94,6 @@ class TestMonitor(unittest.TestCase):
         pipeline.add_stage(noop_info, {
             "isthisjustfantasy": True
         })
-        info = index.get_module_info("Assembler", "org.osbuild.noop")
-        pipeline.set_assembler(info)
 
         with tempfile.TemporaryDirectory() as tmpdir:
             storedir = os.path.join(tmpdir, "store")
@@ -119,10 +110,8 @@ class TestMonitor(unittest.TestCase):
         self.assertEqual(tape.counter["begin"], 1)
         self.assertEqual(tape.counter["finish"], 1)
         self.assertEqual(tape.counter["stages"], 2)
-        self.assertEqual(tape.counter["assembler"], 1)
         self.assertEqual(tape.counter["stages"], 2)
-        self.assertEqual(tape.counter["result"], 3)
+        self.assertEqual(tape.counter["result"], 2)
         self.assertIn(pipeline.stages[0].id, tape.stages)
-        self.assertIn(pipeline.assembler.id, tape.asm)
         self.assertIn("isthisthereallife", tape.output)
         self.assertIn("isthisjustfantasy", tape.output)
diff --git a/test/mod/test_osbuild.py b/test/mod/test_osbuild.py
index 6dc64427..b51f1e98 100644
--- a/test/mod/test_osbuild.py
+++ b/test/mod/test_osbuild.py
@@ -13,7 +13,7 @@ import osbuild
 import osbuild.meta
 from osbuild.monitor import NullMonitor
 from osbuild.objectstore import ObjectStore
-from osbuild.pipeline import detect_host_runner
+from osbuild.pipeline import Manifest, detect_host_runner
 from .. import test
 
 
@@ -41,6 +41,73 @@ class TestDescriptions(unittest.TestCase):
         self.assertEqual(res.success, True)
         self.assertEqual(res.id, stage.id)
 
+    def test_manifest(self):
+        index = osbuild.meta.Index(os.curdir)
+
+        info = index.get_module_info("Stage", "org.osbuild.noop")
+
+        manifest = Manifest({})
+
+        # each pipeline gets a noop stage with different
+        # options so that their ids are different
+        build = manifest.add_pipeline("build", None, None)
+        build.add_stage(info, {"option": 1})
+
+        tree = manifest.add_pipeline("tree",
+                                     "org.osbuild.linux",
+                                     build.id)
+        tree.add_stage(info, {"option": 2})
+
+        assmelber = manifest.add_pipeline("assembler",
+                                          "org.osbuild.inux",
+                                          build.id)
+        assmelber.add_stage(info, {"option": 3})
+
+        self.assertEqual(len(manifest.pipelines), 3)
+
+        self.assertIn("build", manifest.pipelines)
+        self.assertIn("tree", manifest.pipelines)
+        self.assertIn("assembler", manifest.pipelines)
+
+        self.assertIn("build", manifest)
+        self.assertIn("tree", manifest)
+        self.assertIn("assembler", manifest)
+
+        # make sure the order is correct
+        lst = ["build", "tree", "assembler"]
+        for a, b in zip(manifest.pipelines, lst):
+            self.assertEqual(a, b)
+
+        for a, b in zip(manifest, [build, tree, assmelber]):
+            self.assertEqual(a.name, b.name)
+
+        # check we get exceptions on unknown names
+        with self.assertRaises(KeyError):
+            _ = manifest.pipelines["foo"]
+
+        with self.assertRaises(KeyError):
+            _ = manifest["foo"]
+
+        # check helper functions
+        #  access by name
+        check = manifest["build"]
+        self.assertEqual(build.name, check.name)
+
+        check = manifest["tree"]
+        self.assertEqual(tree.name, check.name)
+
+        check = manifest["assembler"]
+        self.assertEqual(assmelber.name, check.name)
+
+        #  `None` return for unknown items
+        check = manifest.get("foo")
+        self.assertIsNone(check)
+
+        #  id based access
+        for i in [build, tree, assmelber]:
+            check = manifest[i.id]
+            self.assertEqual(i.name, check.name)
+
     def test_moduleinfo(self):
         index = osbuild.meta.Index(os.curdir)