osbuild: add workaround to integrate sources into progress reporting

This commit is somewhat poor, sorry for that. It mostly adds
workaround so that the osbuild sources can emit some progress
reporting as well. Without that the user experience is rather poor
and there is a long delay before any sort of progress can be
reported (even before the normal stages run).

With it the user experience is still not good but slightly better,
i.e. the progress monitor will report that the sources have
started downloading and curl will generated some log output. No
real progress unfortunately (sources subprogress will jump from
zero to 100%).

osbuild/monitor.py

@@ -267,7 +267,8 @@ class LogMonitor(BaseMonitor):
             self.out.write(pipeline.build)
         else:
             self.out.write("<host>")
-        self.out.write(f"\n  runner: {pipeline.runner.name} ({pipeline.runner.exec})")
+        if pipeline.runner:
+            self.out.write(f"\n  runner: {pipeline.runner.name} ({pipeline.runner.exec})")
         source_epoch = pipeline.source_epoch
         if source_epoch is not None:
             timepoint = datetime.datetime.fromtimestamp(source_epoch).strftime('%c')
@@ -308,12 +309,13 @@ class JSONSeqMonitor(BaseMonitor):
     def __init__(self, fd: int, manifest: osbuild.Manifest):
         super().__init__(fd, manifest)
         self._ctx_ids: Set[str] = set()
-        self._progress = Progress("pipelines", len(manifest.pipelines))
+        self._progress = Progress("pipelines/sources", len(manifest.pipelines) + len(manifest.sources))
         self._context = Context(origin="org.osbuild")
 
     def begin(self, pipeline: osbuild.Pipeline):
         self._context.set_pipeline(pipeline)
-        self._progress.sub_progress = Progress("stages", len(pipeline.stages))
+        if pipeline.stages:
+            self._progress.sub_progress = Progress("stages", len(pipeline.stages))
         self.log(f"Starting pipeline {pipeline.name}", origin="org.osbuild.main")
 
     # finish is for pipelines

osbuild/pipeline.py

@@ -412,7 +412,11 @@ class Manifest:
     def download(self, store, monitor, libdir):
         with host.ServiceManager(monitor=monitor) as mgr:
             for source in self.sources:
+                # Workaround for lack of progress from sources, this
+                # will need to be reworked later.
+                monitor.begin(source)
                 source.download(mgr, store, libdir)
+                monitor.finish({"name": source.info.name})
 
     def depsolve(self, store: ObjectStore, targets: Iterable[str]) -> List[str]:
         """Return the list of pipelines that need to be built

osbuild/sources.py

@@ -1,5 +1,6 @@
 import abc
 import contextlib
+import hashlib
 import json
 import os
 import tempfile
@@ -19,6 +20,10 @@ class Source:
         self.info = info
         self.items = items or {}
         self.options = options
+        # compat with pipeline
+        self.build = None
+        self.runner = None
+        self.source_epoch = None
 
     def download(self, mgr: host.ServiceManager, store: ObjectStore, libdir: PathLike):
         source = self.info.name
@@ -47,6 +52,32 @@ class Source:
             f.seek(0)
             yield f.fileno()
 
+    # "name", "id", "stages", "results" is only here to make it looks like a
+    # pipeline for the monitor. This should be revisited at some point
+    # and maybe the monitor should get first-class support for
+    # sources?
+    #
+    # In any case, sources can be represented only poorly right now
+    # by the monitor because the source is called with download()
+    # for all items and there is no way for a stage right now to
+    # report something structured back to the host that runs the
+    # source so it just downloads all sources without any user
+    # visible progress right now
+    @property
+    def name(self):
+        return f"source {self.info.name}"
+
+    @property
+    def id(self):
+        m = hashlib.sha256()
+        m.update(json.dumps(self.info.name, sort_keys=True).encode())
+        m.update(json.dumps(self.items, sort_keys=True).encode())
+        return m.hexdigest()
+
+    @property
+    def stages(self):
+        return []
+
 
 class SourceService(host.Service):
     """Source host service"""

sources/org.osbuild.curl

@@ -192,6 +192,13 @@ class CurlSource(sources.SourceService):
                 os.rename(f"{tmpdir}/{checksum}", f"{self.cache}/{checksum}")
             except FileExistsError:
                 pass
+            # Workaround the lack of structured progress reporting from
+            # stages/sources. It generates messages of the form
+            #   "message": "source/org.osbuild.curl (org.osbuild.curl): Downloaded https://rpmrepo.osbuild.org/v2/mirror/public/f38/f38-x86_64-fedora-20230413/Packages/f/fonts-srpm-macros-2.0.5-11.fc38.noarch.rpm\n
+            #
+            # Without it just a long pause with no progress while curl
+            # downloads.
+            print(f"Downloaded {url}")
 
 
 def main():