Repository: osbuild/osbuild
Pull request #710 (2021-07-08 11:20:56)

Pass arguments to stages as a plain file
Instead of using the JSONCOMM RCP mechanism, write the arguments to a file which is then bind-mounted into the container. Adapt `api.arguments` to just read that file, so nothing changes for the stages.


Patch:
diff --git a/osbuild/api.py b/osbuild/api.py
index d0aedfd7..8ef3521e 100644
--- a/osbuild/api.py
+++ b/osbuild/api.py
@@ -39,6 +39,7 @@ class BaseAPI(abc.ABC):
     called; the default implementation will receive the message
     call `_message.`
     """
+
     def __init__(self, socket_address: Optional[PathLike] = None):
         self.socket_address = socket_address
         self.barrier = threading.Barrier(2)
@@ -138,58 +139,17 @@ class API(BaseAPI):
 
     endpoint = "osbuild"
 
-    def __init__(self, args, monitor, *, socket_address=None):
+    def __init__(self, *, socket_address=None):
         super().__init__(socket_address)
-        self.input = args
-        self._output_data = io.StringIO()
-        self._output_pipe = None
-        self.monitor = monitor
         self.metadata = {}
         self.error = None
 
-    @property
-    def output(self):
-        # Only once the event-loop was stopped, you are guaranteed that the
-        # api-thread scheduled all outstanding events. Therefore, we disallow
-        # asking for the output-data from a running api context. If we happen
-        # to need live streaming access to the output in the future, we need
-        # to redesign the output-handling, anyway.
-        assert not self.running
-
-        return self._output_data.getvalue()
-
-    def _prepare_input(self):
-        with tempfile.TemporaryFile() as fd:
-            fd.write(json.dumps(self.input).encode('utf-8'))
-            # re-open the file to get a read-only file descriptor
-            return open(f"/proc/self/fd/{fd.fileno()}", "r")
-
-    def _prepare_output(self):
-        r, w = os.pipe()
-        self._output_pipe = r
-        self._output_data.truncate(0)
-        self._output_data.seek(0)
-        self.event_loop.add_reader(r, self._output_ready)
-        return os.fdopen(w)
-
-    def _output_ready(self):
-        raw = os.read(self._output_pipe, 4096)
-        data = raw.decode("utf-8")
-        self._output_data.write(data)
-        self.monitor.log(data)
-
     def _set_metadata(self, message, fds):
         fd = message["metadata"]
         with os.fdopen(fds.steal(fd), encoding="utf-8") as f:
             data = json.load(f)
         self.metadata.update(data)
 
-    def _get_arguments(self, sock):
-        with self._prepare_input() as data:
-            fds = []
-            fds.append(data.fileno())
-            sock.send({"type": "fd", "fd": 0}, fds=fds)
-
     def _get_exception(self, message):
         self.error = {
             "type": "exception",
@@ -201,13 +161,6 @@ class API(BaseAPI):
             self._set_metadata(msg, fds)
         elif msg["method"] == 'exception':
             self._get_exception(msg)
-        elif msg["method"] == 'get-arguments':
-            self._get_arguments(sock)
-
-    def _cleanup(self):
-        if self._output_pipe:
-            os.close(self._output_pipe)
-            self._output_pipe = None
 
 
 def exception(e, path="/run/osbuild/api/osbuild"):
@@ -239,32 +192,11 @@ def exception_handler(path="/run/osbuild/api/osbuild"):
         exception(e, path)
 
 
-def arguments(path="/run/osbuild/api/osbuild"):
+def arguments(path="/run/osbuild/api/arguments"):
     """Retrieve the input arguments that were supplied to API"""
-    with jsoncomm.Socket.new_client(path) as client:
-        req = {"method": "get-arguments"}
-        client.send(req)
-        msg, fds, _ = client.recv()
-        assert msg["type"] == "fd"
-        fd = msg["fd"]
-        with os.fdopen(fds.steal(fd), encoding="utf-8") as f:
-            data = json.load(f)
-
-        # Root relative paths: since paths are different on the
-        # host and in the container they are transmitted not as
-        # absolute but relative paths. For all items that have
-        # registered roots, re-root their path entries here
-        for name, root in data.get("paths", {}).items():
-            group = data.get(name)
-            if not group or not isinstance(group, dict):
-                continue
-            for item in group.values():
-                path = item.get("path")
-                if not path:
-                    continue
-                item["path"] = os.path.join(root, path)
-
-        return data
+    with open(path, "r", encoding="utf-8") as fp:
+        data = json.load(fp)
+    return data
 
 
 def metadata(data: Dict, path="/run/osbuild/api/osbuild"):
diff --git a/osbuild/host.py b/osbuild/host.py
index ea1f5ceb..f9182f52 100644
--- a/osbuild/host.py
+++ b/osbuild/host.py
@@ -257,7 +257,10 @@ class Service(abc.ABC):
                 _, val, tb = sys.exc_info()
                 reply = self.protocol.encode_exception(val, tb)
 
-            self.sock.send(reply, fds=fds)
+            try:
+                self.sock.send(reply, fds=fds)
+            except BrokenPipeError:
+                break
 
     def _handle_message(self, msg, fds):
         """
diff --git a/osbuild/pipeline.py b/osbuild/pipeline.py
index 44ec7195..c0ffd5a9 100644
--- a/osbuild/pipeline.py
+++ b/osbuild/pipeline.py
@@ -79,29 +79,56 @@ class Stage:
         self.mounts[name] = mount
         return mount
 
+    def prepare_arguments(self, args, location):
+        args["options"] = self.options
+        args["meta"] = {
+            "id": self.id
+        }
+
+        # Root relative paths: since paths are different on the
+        # host and in the container they need to be mapped to
+        # their path within the container. For all items that
+        # have registered roots, re-root their path entries here
+        for name, root in args.get("paths", {}).items():
+            group = args.get(name)
+            if not group or not isinstance(group, dict):
+                continue
+            for item in group.values():
+                path = item.get("path")
+                if not path:
+                    continue
+                item["path"] = os.path.join(root, path)
+
+        with open(location, "w", encoding="utf-8") as fp:
+            json.dump(args, fp)
+
     def run(self, tree, runner, build_tree, store, monitor, libdir):
         with contextlib.ExitStack() as cm:
 
-            build_root = buildroot.BuildRoot(
-                build_tree, runner, libdir, store.tmp)
+            build_root = buildroot.BuildRoot(build_tree, runner, libdir, store.tmp)
             cm.enter_context(build_root)
 
-            inputs_tmpdir = store.tempdir(prefix="inputs-")
-            inputs_tmpdir = cm.enter_context(inputs_tmpdir)
+            tmpdir = store.tempdir(prefix="buildroot-tmp-")
+            tmpdir = cm.enter_context(tmpdir)
+
+            inputs_tmpdir = os.path.join(tmpdir, "inputs")
+            os.makedirs(inputs_tmpdir)
             inputs_mapped = "/run/osbuild/inputs"
             inputs = {}
 
             devices_mapped = "/dev"
             devices = {}
 
-            mounts_tmpdir = store.tempdir(prefix="mounts-")
-            mounts_tmpdir = cm.enter_context(mounts_tmpdir)
+            mounts_tmpdir = os.path.join(tmpdir, "mounts")
+            os.makedirs(mounts_tmpdir)
             mounts_mapped = "/run/osbuild/mounts"
             mounts = {}
 
+            os.makedirs(os.path.join(tmpdir, "api"))
+            args_path = os.path.join(tmpdir, "api", "arguments")
+
             args = {
                 "tree": "/run/osbuild/tree",
-                "options": self.options,
                 "paths": {
                     "devices": devices_mapped,
                     "inputs": inputs_mapped,
@@ -110,14 +137,12 @@ class Stage:
                 "devices": devices,
                 "inputs": inputs,
                 "mounts": mounts,
-                "meta": {
-                    "id": self.id
-                }
             }
 
             ro_binds = [
                 f"{self.info.path}:/run/osbuild/bin/{self.name}",
-                f"{inputs_tmpdir}:{inputs_mapped}"
+                f"{inputs_tmpdir}:{inputs_mapped}",
+                f"{args_path}:/run/osbuild/api/arguments"
             ]
 
             binds = [
@@ -145,7 +170,9 @@ class Stage:
                 data = mount.mount(mgr, abspath, mounts_tmpdir)
                 mounts[key] = data
 
-            api = API(args, monitor)
+            self.prepare_arguments(args, args_path)
+
+            api = API()
             build_root.register_api(api)
 
             rls = remoteloop.LoopServer()


Test patch:
diff --git a/test/mod/test_api.py b/test/mod/test_api.py
index 1576d774..7ce85536 100644
--- a/test/mod/test_api.py
+++ b/test/mod/test_api.py
@@ -4,7 +4,6 @@
 
 import os
 import multiprocessing as mp
-import sys
 import tempfile
 import unittest
 
@@ -68,29 +67,17 @@ class TestAPI(unittest.TestCase):
                 with api:
                     pass
 
-    def test_get_arguments(self):
-        tmpdir = self.tmp.name
-        path = os.path.join(tmpdir, "osbuild-api")
-        args = {"options": {"answer": 42}}
-        monitor = osbuild.monitor.BaseMonitor(sys.stderr.fileno())
-
-        with osbuild.api.API(args, monitor, socket_address=path) as _:
-            data = osbuild.api.arguments(path=path)
-            self.assertEqual(data, args)
-
     def test_exception(self):
         # Check that 'api.exception' correctly sets 'API.exception'
         tmpdir = self.tmp.name
         path = os.path.join(tmpdir, "osbuild-api")
-        args = {}
-        monitor = osbuild.monitor.BaseMonitor(sys.stderr.fileno())
 
         def exception(path):
             with osbuild.api.exception_handler(path):
                 raise ValueError("osbuild test exception")
             assert False, "api.exception should exit process"
 
-        api = osbuild.api.API(args, monitor, socket_address=path)
+        api = osbuild.api.API(socket_address=path)
         with api:
             p = mp.Process(target=exception, args=(path, ))
             p.start()
@@ -111,15 +98,13 @@ class TestAPI(unittest.TestCase):
         # set correctly
         tmpdir = self.tmp.name
         path = os.path.join(tmpdir, "osbuild-api")
-        args = {}
-        monitor = osbuild.monitor.BaseMonitor(sys.stderr.fileno())
 
         def metadata(path):
             data = {"meta": "42"}
             osbuild.api.metadata(data, path=path)
             return 0
 
-        api = osbuild.api.API(args, monitor, socket_address=path)
+        api = osbuild.api.API(socket_address=path)
         with api:
             p = mp.Process(target=metadata, args=(path, ))
             p.start()