Repository: osbuild/osbuild
Pull request #470 (2020-07-10 16:39:25)

Introduce pipeline monitoring
Introduce the concept of pipeline monitoring: A new monitor class is passed to the `pipeline.run()` function. The main idea is to separate the monitoring from the code that builds pipeline. Through the build process various methods will be called on that object, representing the different steps and their targets during the build process. This can be used to fully stream the output of the various stages or just indicate the start and finish of the individual stages.

This replaces the 'interactive' argument throughout the pipeline code. The old interactive behavior is replicated via the new `LogMonitor` class that logs the beginning of stages/assembler, but also streams all the output of them to `stdout`. The non-interactive of not reporting anything behavior is done by using the `NullMonitor` class, which in turn does nothing.

A new JSON monitor is added in PR #473 which can be used in clients to follow the progress.

TODO:
 - [x] Tests
 - [x] Integrate `JsonMonitor` via a new command line option

Solves issue #319

teg: I like this a lot! My only questions are about the new JSON stream monitor: what's the usecase we have in mind here? Is the idea that this can contain any (stage-specific) metadata, not just logs (as we have discussed in the past), or would metadata be separate?
gicmo: > I like this a lot! My only questions are about the new JSON stream monitor: what's the usecase we have in mind here? Is the idea that this can contain any (stage-specific) metadata, not just logs (as we have discussed in the past), or would metadata be separate?

The `JsonMonitor` was mostly meant to be a proof-of-concept and a discussion starting point (it is not integrated at all yet). My main idea was to provide feedback to clients (i.e. composer) about the progress of the image building (the info contains the `x of y done` information) maybe with some high level status ("Installing boot loader", see below).

The question is about the separation of monitoring/logging versus result returning & reporting. I feel that monitoring and the actual result reporting should be separated, i.e. you still get the full JSON result dictionary on `stdout` at the end when everything is done (and doing something with the monitor output is optional). And that final result would also contain the metadata. I am very, very tempted to even remove the `output` field from the JSON monitor messages, because I am not sure how useful that is in the non-error case anyway. But I am not 100% sure and the line between monitoring and reporting of results is a bit blurry. 

I also had the idea to add a another level of logging/status reporting to the `API` that can be used by the stages: the stages/assembler could use `api.status("Downloading RPMs")`, `api.status("Installing RPMS")`, `api.status("Installing Bootloader")`, that would then be handed from the stage via the JSON RPC to the monitor via `monitor.status()`, and that be reported back to the client (composer); there it could be used to show the status/progress within a stage in the UI. I feel that is more useful then reporting the output of the tools (which can be a lot of unhelpful gibberish).

Opinions?
teg: > Cool idea indeed. Would those be user-visible (translated) strings?

I'm not sure about best-practices here, but could we somehow make them identifiers, so the strings could live in the UI?
jkozol: Cool stuffs!!

> I also had the idea to add a another level of logging/status reporting to the API that can be used by the stages: the stages/assembler could use api.status("Downloading RPMs"), api.status("Installing RPMS"), api.status("Installing Bootloader"), that would then be handed from the stage via the JSON RPC to the monitor via monitor.status(), and that be reported back to the client (composer); there it could be used to show the status/progress within a stage in the UI. I feel that is more useful then reporting the output of the tools (which can be a lot of unhelpful gibberish).

It would be terrific to show UI users the progress of the different stages.

> I.e., something like api.status("rpm.download"), so the UI sees the identifier rpm.download in the context of the org.osbuild.rpm stage and would map that to the (translatable) string "Downloading RPMs". We could then be very granular in what the stages emit, but the UI would opt in to show those status updates they think are useful.

I think this is a good idea since the UI would definitely map from the identifier to a translated string. I believe that the more granular we are in our status updates, the better a user will understand the progress of the image build.
bcl: > I also had the idea to add a another level of logging/status reporting to the `API` that can be used by the stages: the stages/assembler could use `api.status("Downloading RPMs")`, `api.status("Installing RPMS")`, `api.status("Installing Bootloader")`, that would then be handed from the stage via the JSON RPC to the monitor via `monitor.status()`, and that be reported back to the client (composer); there it could be used to show the status/progress within a stage in the UI. I feel that is more useful then reporting the output of the tools (which can be a lot of unhelpful gibberish).

I like this idea. Monitoring is useful for telling users 'hey! I'm not dead yet!' but it doesn't need to be super detailed. And if things fail they will have the logs to dig through anyway so I don't see any reason for the monitor to include logs.
gicmo: > > Cool idea indeed. Would those be user-visible (translated) strings?
> I'm not sure about best-practices here, but could we somehow make them identifiers, so the strings could live in the UI?
> 
> I.e., something like `api.status("rpm.download")`, so the UI sees the identifier `rpm.download` in the context of the `org.osbuild.rpm` stage and would map that to the (translatable) string "Downloading RPMs". We could then be very granular in what the stages emit, but the UI would opt in to show those status updates they think are useful.

I love this!
gicmo: 
> I like this idea. Monitoring is useful for telling users 'hey! I'm not dead yet!' but it doesn't need to be super detailed. And if things fail they will have the logs to dig through anyway so I don't see any reason for the monitor to include logs.

Based on all the feedback I have removed it from the `JsonStream` logger. 

I wonder if I should split the inclusion of the `JsonStream` logger out of this PR. I wanted to show them together to give a full picture. But maybe not it is conflating the general monitoring framework idea too much with discussion of that specific monitor. 
teg: > I wonder if I should split the inclusion of the `JsonStream` logger out of this PR.

Makes sense to me. The `JsonStream` looks good to me, but might be good to give that some more time and gather some more input about how it would be used and what it should contain.
gicmo: I changed the `JsonStream` monitor to stream [JSON lines](http://jsonlines.org/) because I could not get `jq` to accept the record separated JSON. Additionally, I added two new command line parameters to `osbuild`:
  - `--monitor`: (`string`) selects which monitor class to be used, e.g. `LogMonitor`, `JsoStream`
  - `--monitor-fd`: (`int`) Use the supplied `fd` to write monitor output to.

The idea would be that clients can open a dedicated file descriptor and pass it to `osbuild` and then this is used for logging.
Not 100% happy with `LogMonitor` yet, maybe `TextLogger`, `TextStream` or something? Happy for input.
gicmo: Removed the command line options, will be done together with the JSON based logger. Added even more tests.
lgtm-com[bot]: This pull request **introduces 1 alert** when merging ec38105e4449bba00d03e36d32dede2c2b8f6a63 into 7e392f819eeac30c7576d56b58b08ee72d7fe713 - [view on LGTM.com](https://lgtm.com/projects/g/osbuild/osbuild/rev/pr-12e3367553769adb4e2340ac37d6e5062730914f)

**new alerts:**

* 1 for Unused local variable

Patch:
diff --git a/.pylintrc b/.pylintrc
index ce564d6d..1a7c8a30 100644
--- a/.pylintrc
+++ b/.pylintrc
@@ -4,3 +4,6 @@ max-line-length=120
 
 [TYPECHECK]
 ignored-classes=osbuild.loop.LoopInfo
+
+[DESIGN]
+max-attributes=10
diff --git a/assemblers/org.osbuild.qemu b/assemblers/org.osbuild.qemu
index 257c43d2..dbd50963 100755
--- a/assemblers/org.osbuild.qemu
+++ b/assemblers/org.osbuild.qemu
@@ -201,7 +201,6 @@ class Filesystem:
         maker(device, self.uuid, self.label)
 
 
-# pylint: disable=too-many-instance-attributes
 class Partition:
     def __init__(self,
                  pttype: str = None,
diff --git a/osbuild/api.py b/osbuild/api.py
index 6fdb11a2..4d77d2b2 100644
--- a/osbuild/api.py
+++ b/osbuild/api.py
@@ -1,25 +1,26 @@
 import asyncio
+import io
 import json
 import os
-import sys
 import tempfile
 import threading
 from .util import jsoncomm
 
 
 class API:
-    def __init__(self, socket_address, args, interactive):
+    def __init__(self, socket_address, args, monitor):
         self.socket_address = socket_address
         self.input = args
-        self.interactive = interactive
-        self._output = None
+        self._output_data = io.StringIO()
+        self._output_pipe = None
+        self.monitor = monitor
         self.event_loop = asyncio.new_event_loop()
         self.thread = threading.Thread(target=self._run_event_loop)
         self.barrier = threading.Barrier(2)
 
     @property
     def output(self):
-        return self._output and self._output.read()
+        return self._output_data.getvalue()
 
     def _prepare_input(self):
         with tempfile.TemporaryFile() as fd:
@@ -28,12 +29,18 @@ class API:
             return open(f"/proc/self/fd/{fd.fileno()}", "r")
 
     def _prepare_output(self):
-        if self.interactive:
-            return os.fdopen(os.dup(sys.stdout.fileno()), 'w')
-        out = tempfile.TemporaryFile(mode="wb")
-        fd = os.open(f"/proc/self/fd/{out.fileno()}", os.O_RDONLY|os.O_CLOEXEC)
-        self._output = os.fdopen(fd)
-        return out
+        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 _setup_stdio(self, server, addr):
         with self._prepare_input() as stdin, \
@@ -70,3 +77,7 @@ class API:
     def __exit__(self, *args):
         self.event_loop.call_soon_threadsafe(self.event_loop.stop)
         self.thread.join()
+        self.event_loop.close()
+        if self._output_pipe:
+            os.close(self._output_pipe)
+            self._output_pipe = None
diff --git a/osbuild/main_cli.py b/osbuild/main_cli.py
index e7104409..5a34ec57 100644
--- a/osbuild/main_cli.py
+++ b/osbuild/main_cli.py
@@ -13,6 +13,7 @@ import sys
 
 import osbuild
 import osbuild.meta
+import osbuild.monitor
 
 
 RESET = "\033[0m"
@@ -145,10 +146,13 @@ def osbuild_cli():
         print("No output directory or checkpoints specified, exited without building.")
         return 0
 
+    monitor_name = "NullMonitor" if args.json else "LogMonitor"
+    monitor = osbuild.monitor.make(monitor_name, sys.stdout.fileno())
+
     try:
         r = pipeline.run(
             args.store,
-            interactive=not args.json,
+            monitor,
             libdir=args.libdir,
             output_directory=args.output_directory
         )
diff --git a/osbuild/monitor.py b/osbuild/monitor.py
new file mode 100644
index 00000000..4d3c9d06
--- /dev/null
+++ b/osbuild/monitor.py
@@ -0,0 +1,126 @@
+"""
+Monitor pipeline activity
+
+The osbuild `Pipeline` class supports monitoring of its activities
+by providing a monitor object that implements the `BaseMonitor`
+interface. During the execution of the pipeline various functions
+are called on the monitor object at certain events. Consult the
+`BaseMonitor` class for the description of all available events.
+"""
+
+import abc
+import json
+import os
+import sys
+
+from typing import Dict
+
+import osbuild
+
+
+RESET = "\033[0m"
+BOLD = "\033[1m"
+
+
+class TextWriter:
+    """Helper class for writing text to file descriptors"""
+    def __init__(self, fd: int):
+        self.fd = fd
+        self.isatty = os.isatty(fd)
+
+    def term(self, text, *, clear=False):
+        """Write text if attached to a terminal."""
+        if not self.isatty:
+            return
+
+        if clear:
+            self.write(RESET)
+
+        self.write(text)
+
+    def write(self, text: str):
+        """Write all of text to the log file descriptor"""
+        data = text.encode("utf-8")
+        n = len(data)
+        while n:
+            k = os.write(self.fd, data)
+            n -= k
+            if n:
+                data = data[n:]
+
+
+class BaseMonitor(abc.ABC):
+    """Base class for all pipeline monitors"""
+
+    def __init__(self, fd: int):
+        """Logging will be done to file descriptor `fd`"""
+        self.out = TextWriter(fd)
+
+    def begin(self, pipeline: osbuild.Pipeline):
+        """Called once at the beginning of a build"""
+
+    def finish(self, result: Dict):
+        """Called at the very end of the build"""
+
+    def stage(self, stage: osbuild.Stage):
+        """Called when a stage is being built"""
+
+    def assembler(self, assembler: osbuild.Assembler):
+        """Called when an assembler is being built"""
+
+    def result(self, result: osbuild.pipeline.BuildResult):
+        """Called when a module is done with its result"""
+
+    def log(self, message: str):
+        """Called for all module log outputs"""
+
+
+class NullMonitor(BaseMonitor):
+    """Monitor class that does not report anything"""
+
+
+class LogMonitor(BaseMonitor):
+    """Monitor that follows show the log output of modules
+
+    This monitor will print a header with `name: id` followed
+    by the options for each module as it is being built. The
+    full log messages of the modules will be print as soon as
+    they become available.
+    The constructor argument `fd` is a file descriptor, where
+    the log will get written to. If `fd`  is a `TTY`, escape
+    sequences will be used to highlight sections of the log.
+    """
+    def stage(self, stage):
+        self.module(stage)
+
+    def assembler(self, assembler):
+        self.out.term(BOLD, clear=True)
+        self.out.write("Assembler ")
+        self.out.term(RESET)
+
+        self.module(assembler)
+
+    def module(self, module):
+        options = module.options or {}
+        title = f"{module.name}: {module.id}"
+
+        self.out.term(BOLD, clear=True)
+        self.out.write(title)
+        self.out.term(RESET)
+        self.out.write(" ")
+
+        json.dump(options, self.out, indent=2)
+        self.out.write("\n")
+
+    def log(self, message):
+        self.out.write(message)
+
+
+def make(name, fd):
+    module = sys.modules[__name__]
+    monitor = getattr(module, name, None)
+    if not monitor:
+        raise ValueError(f"Unknown monitor: {name}")
+    if not issubclass(monitor, BaseMonitor):
+        raise ValueError(f"Invalid monitor: {name}")
+    return monitor(fd)
diff --git a/osbuild/objectstore.py b/osbuild/objectstore.py
index 5a292313..3b6529fd 100644
--- a/osbuild/objectstore.py
+++ b/osbuild/objectstore.py
@@ -40,7 +40,6 @@ def umount(target, lazy=True):
     subprocess.run(["umount"] + args + [target], check=True)
 
 
-# pylint: disable=too-many-instance-attributes
 class Object:
     def __init__(self, store: "ObjectStore"):
         self._init = True
diff --git a/osbuild/pipeline.py b/osbuild/pipeline.py
index 78126786..d80e561c 100644
--- a/osbuild/pipeline.py
+++ b/osbuild/pipeline.py
@@ -13,10 +13,6 @@ from . import sources
 from .util import osrelease
 
 
-RESET = "\033[0m"
-BOLD = "\033[1m"
-
-
 def cleanup(*objs):
     """Call cleanup method for all objects, filters None values out"""
     _ = map(lambda o: o.cleanup(), filter(None, objs))
@@ -34,12 +30,6 @@ class BuildResult:
         return vars(self)
 
 
-def print_header(title, options):
-    print()
-    print(f"{RESET}{BOLD}{title}{RESET} " + json.dumps(options or {}, indent=2))
-    print()
-
-
 class Stage:
     def __init__(self, name, source_options, build, base, options):
         self.name = name
@@ -71,13 +61,11 @@ class Stage:
             runner,
             build_tree,
             cache,
-            interactive=False,
+            monitor,
             libdir=None,
             var="/var/tmp"):
         with buildroot.BuildRoot(build_tree, runner, libdir=libdir, var=var) as build_root, \
             tempfile.TemporaryDirectory(prefix="osbuild-sources-output-", dir=var) as sources_output:
-            if interactive:
-                print_header(f"{self.name}: {self.id}", self.options)
 
             args = {
                 "tree": "/run/osbuild/tree",
@@ -90,7 +78,7 @@ class Stage:
 
             ro_binds = [f"{sources_output}:/run/osbuild/sources"]
 
-            with API(f"{build_root.api}/osbuild", args, interactive) as api, \
+            with API(f"{build_root.api}/osbuild", args, monitor) as api, \
                 sources.SourcesServer(f"{build_root.api}/sources",
                                       libdir or "/usr/lib/osbuild",
                                       self.sources,
@@ -129,10 +117,8 @@ class Assembler:
             description["id"] = self.id
         return description
 
-    def run(self, tree, runner, build_tree, output_dir=None, interactive=False, libdir=None, var="/var/tmp"):
+    def run(self, tree, runner, build_tree, monitor, output_dir=None, libdir=None, var="/var/tmp"):
         with buildroot.BuildRoot(build_tree, runner, libdir=libdir, var=var) as build_root:
-            if interactive:
-                print_header(f"Assembler {self.name}: {self.id}", self.options)
 
             args = {
                 "tree": "/run/osbuild/tree",
@@ -151,7 +137,7 @@ class Assembler:
             ro_binds = [f"{tree}:/run/osbuild/tree"]
 
             with remoteloop.LoopServer(f"{build_root.api}/remoteloop"), \
-                API(f"{build_root.api}/osbuild", args, interactive) as api:
+                API(f"{build_root.api}/osbuild", args, monitor) as api:
                 r = build_root.run(
                     [f"/run/osbuild/lib/assemblers/{self.name}"],
                     binds=binds,
@@ -202,7 +188,7 @@ class Pipeline:
 
         return description
 
-    def build_stages(self, object_store, interactive, libdir):
+    def build_stages(self, object_store, monitor, libdir):
         results = {"success": True}
 
         # We need a build tree for the stages below, which is either
@@ -217,7 +203,7 @@ class Pipeline:
             build = self.build
 
             r, t, tree = build.build_stages(object_store,
-                                            interactive,
+                                            monitor,
                                             libdir)
 
             results["build"] = r
@@ -266,14 +252,19 @@ class Pipeline:
 
         for stage in self.stages[base_idx + 1:]:
             with build_tree.read() as build_path, tree.write() as path:
+
+                monitor.stage(stage)
+
                 r = stage.run(path,
                               self.runner,
                               build_path,
                               object_store.store,
-                              interactive=interactive,
+                              monitor,
                               libdir=libdir,
                               var=object_store.store)
 
+                monitor.result(r)
+
             results["stages"].append(r.as_dict())
             if not r.success:
                 cleanup(build_tree, tree)
@@ -285,7 +276,7 @@ class Pipeline:
 
         return results, build_tree, tree
 
-    def assemble(self, object_store, build_tree, tree, interactive, libdir, output_directory: Optional[str]):
+    def assemble(self, object_store, build_tree, tree, monitor, libdir, output_directory: Optional[str]):
         results = {"success": True}
 
         if not self.assembler:
@@ -297,14 +288,18 @@ class Pipeline:
              tree.read() as input_dir, \
              output.write() as output_dir:
 
+            monitor.assembler(self.assembler)
+
             r = self.assembler.run(input_dir,
                                    self.runner,
                                    build_dir,
+                                   monitor,
                                    output_dir=output_dir,
-                                   interactive=interactive,
                                    libdir=libdir,
                                    var=object_store.store)
 
+            monitor.result(r)
+
         results["assembler"] = r.as_dict()
         if not r.success:
             output.cleanup()
@@ -319,10 +314,12 @@ class Pipeline:
 
         return results
 
-    def run(self, store, interactive=False, libdir=None, output_directory=None):
+    def run(self, store, monitor, libdir=None, output_directory=None):
         os.makedirs("/run/osbuild", exist_ok=True)
         results = {}
 
+        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
@@ -336,7 +333,7 @@ class Pipeline:
                         output.export(output_directory)
             else:
                 results, build_tree, tree = self.build_stages(object_store,
-                                                              interactive,
+                                                              monitor,
                                                               libdir)
 
                 if not results["success"]:
@@ -345,12 +342,14 @@ class Pipeline:
                 r = self.assemble(object_store,
                                   build_tree,
                                   tree,
-                                  interactive,
+                                  monitor,
                                   libdir,
                                   output_directory)
 
                 results.update(r)  # This will also update 'success'
 
+        monitor.finish(results)
+
         return results
 
 
diff --git a/osbuild/remoteloop.py b/osbuild/remoteloop.py
index b014b5f6..75b0eea9 100644
--- a/osbuild/remoteloop.py
+++ b/osbuild/remoteloop.py
@@ -101,6 +101,7 @@ class LoopServer:
     def __exit__(self, *args):
         self.event_loop.call_soon_threadsafe(self.event_loop.stop)
         self.thread.join()
+        self.event_loop.close()
         for lo in self.devs:
             lo.close()
 
diff --git a/osbuild/sources.py b/osbuild/sources.py
index 254aff15..459188fb 100644
--- a/osbuild/sources.py
+++ b/osbuild/sources.py
@@ -6,7 +6,6 @@ from .util import jsoncomm
 
 
 class SourcesServer:
-    # pylint: disable=too-many-instance-attributes
     def __init__(self, socket_address, libdir, options, cache, output):
         self.socket_address = socket_address
         self.libdir = libdir


Test patch:
diff --git a/test/mod/test_monitor.py b/test/mod/test_monitor.py
new file mode 100644
index 00000000..36a2ddd3
--- /dev/null
+++ b/test/mod/test_monitor.py
@@ -0,0 +1,166 @@
+#
+# Test for monitoring classes and integration
+#
+
+import io
+import json
+import os
+import multiprocessing as mp
+import sys
+import tempfile
+import unittest
+from collections import defaultdict
+
+import osbuild
+import osbuild.meta
+from osbuild.api import API
+from osbuild.monitor import LogMonitor
+from osbuild.util import jsoncomm
+from .. import test
+
+
+def setup_stdio(path):
+    """Copy of what the osbuild runners do to setup stdio"""
+    with jsoncomm.Socket.new_client(path) as client:
+        req = {'method': 'setup-stdio'}
+        client.send(req)
+        msg, fds, _ = client.recv()
+        for sio in ['stdin', 'stdout', 'stderr']:
+            target = getattr(sys, sio)
+            source = fds[msg[sio]]
+            os.dup2(source, target.fileno())
+        fds.close()
+
+
+def echo(path):
+    """echo stdin to stdout after setting stdio up via API
+
+    Meant to be called as the main function in a process
+    simulating an osbuild runner and a stage run which does
+    nothing but returns the supplied options to stdout again.
+    """
+    setup_stdio(path)
+    data = json.load(sys.stdin)
+    json.dump(data, sys.stdout)
+    sys.exit(0)
+
+
+class TapeMonitor(osbuild.monitor.BaseMonitor):
+    """Record the usage of all called functions"""
+    def __init__(self):
+        super().__init__(sys.stderr.fileno())
+        self.counter = defaultdict(int)
+        self.stages = set()
+        self.asm = None
+        self.results = set()
+        self.logger = io.StringIO()
+
+    def begin(self, pipeline: osbuild.Pipeline):
+        self.counter["begin"] += 1
+
+    def finish(self, result):
+        self.counter["finish"] += 1
+        self.output = self.logger.getvalue()
+
+    def stage(self, stage: osbuild.Stage):
+        self.counter["stages"] += 1
+        self.stages.add(stage.id)
+
+    def assembler(self, assembler: osbuild.Assembler):
+        self.counter["assembler"] += 1
+        self.asm = assembler.id
+
+    def result(self, result: osbuild.pipeline.BuildResult):
+        self.counter["result"] += 1
+        self.results.add(result.id)
+
+    def log(self, message: str):
+        self.counter["log"] += 1
+        self.logger.write(message)
+
+
+class TestMonitor(unittest.TestCase):
+    def test_log_monitor_api(self):
+        # Basic log and API integration check
+        with tempfile.TemporaryDirectory() as tmpdir:
+            args = {"foo": "bar"}
+            path = os.path.join(tmpdir, "osbuild-api")
+            logfile = os.path.join(tmpdir, "log.txt")
+
+            with open(logfile, "w") as log, \
+                 API(path, args, LogMonitor(log.fileno())) as api:
+                p = mp.Process(target=echo, args=(path, ))
+                p.start()
+                p.join()
+                self.assertEqual(p.exitcode, 0)
+                output = api.output
+                assert output
+
+            self.assertEqual(json.dumps(args), output)
+            with open(logfile) as f:
+                log = f.read()
+            self.assertEqual(log, output)
+
+    @unittest.skipUnless(test.TestBase.can_bind_mount(), "root-only")
+    def test_log_monitor_vfuncs(self):
+        # Checks the basic functioning of the LogMonitor
+        pipeline = osbuild.Pipeline("org.osbuild.linux")
+        pipeline.add_stage("org.osbuild.noop", {}, {
+            "isthisthereallife": False
+        })
+        pipeline.set_assembler("org.osbuild.noop")
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            storedir = os.path.join(tmpdir, "store")
+            outputdir = os.path.join(tmpdir, "output")
+
+            logfile = os.path.join(tmpdir, "log.txt")
+
+            with open(logfile, "w") as log:
+                monitor = LogMonitor(log.fileno())
+                res = pipeline.run(storedir,
+                                   monitor,
+                                   libdir=os.path.abspath(os.curdir),
+                                   output_directory=outputdir)
+
+                with open(logfile) as f:
+                    log = f.read()
+
+            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")
+    def test_monitor_integration(self):
+        # Checks the monitoring API is called properly from the pipeline
+        pipeline = osbuild.Pipeline("org.osbuild.linux")
+        pipeline.add_stage("org.osbuild.noop", {}, {
+            "isthisthereallife": False
+        })
+        pipeline.add_stage("org.osbuild.noop", {}, {
+            "isthisjustfantasy": True
+        })
+        pipeline.set_assembler("org.osbuild.noop")
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            storedir = os.path.join(tmpdir, "store")
+            outputdir = os.path.join(tmpdir, "output")
+
+            tape = TapeMonitor()
+            res = pipeline.run(storedir,
+                               tape,
+                               libdir=os.path.abspath(os.curdir),
+                               output_directory=outputdir)
+
+        assert res
+        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.assertIn(pipeline.stages[0].id, tape.stages)
+        self.assertIn(pipeline.assembler.id, tape.asm)
+        self.assertIn("isthisthereallife", tape.output)
+        self.assertIn("isthisjustfantasy", tape.output)