osbuild: add result error reporting for sources

This commit adds error reporting from source download errors
to the monitor. It reuses the `BuildResult` for symmetry but
we probably want to refactor this a bit to make source handling
a bit more similar to stages.

osbuild/monitor.py

@@ -16,9 +16,10 @@ import json
 import os
 import sys
 import time
-from typing import Dict, Optional, Set
+from typing import Dict, Optional, Set, Union
 
 import osbuild
+from osbuild.pipeline import BuildResult, DownloadResult
 from osbuild.util.term import fmt as vt
 
 
@@ -166,7 +167,8 @@ class Progress:
 def log_entry(message: Optional[str] = None,
               context: Optional[Context] = None,
               progress: Optional[Progress] = None,
-              build_result: Optional[osbuild.pipeline.BuildResult] = None) -> dict:
+              result: Union[BuildResult, DownloadResult, None] = None,
+              ) -> dict:
     """
     Create a single log entry dict with a given message, context, and progress objects.
     All arguments are optional. A timestamp is added to the message.
@@ -175,7 +177,7 @@ def log_entry(message: Optional[str] = None,
     # monitors support that
     return omitempty({
         "message": message,
-        "build_result": build_result.as_dict() if build_result else None,
+        "result": result.as_dict() if result else None,
         "context": context.as_dict() if context else None,
         "progress": progress.as_dict() if progress else None,
         "timestamp": time.time(),
@@ -229,7 +231,7 @@ class BaseMonitor(abc.ABC):
     def assembler(self, assembler: osbuild.Stage):
         """Called when an assembler is being built"""
 
-    def result(self, result: osbuild.pipeline.BuildResult):
+    def result(self, result: Union[BuildResult, DownloadResult]) -> None:
         """Called when a module (stage/assembler) is done with its result"""
 
     def log(self, message: str, origin: Optional[str] = None):
@@ -256,7 +258,7 @@ class LogMonitor(BaseMonitor):
         super().__init__(fd, total_steps)
         self.timer_start = 0
 
-    def result(self, result: osbuild.pipeline.BuildResult):
+    def result(self, result: Union[BuildResult, DownloadResult]) -> None:
         duration = int(time.time() - self.timer_start)
         self.out.write(f"\n⏱  Duration: {duration}s\n")
 
@@ -337,7 +339,7 @@ class JSONSeqMonitor(BaseMonitor):
         self.log(f"Starting module {module.name}", origin="osbuild.monitor")
 
     # result is for modules
-    def result(self, result: osbuild.pipeline.BuildResult):
+    def result(self, result: Union[BuildResult, DownloadResult]) -> None:
         # we may need to check pipeline ids here in the future
         if self._progress.sub_progress:
             self._progress.sub_progress.incr()
@@ -349,7 +351,7 @@ class JSONSeqMonitor(BaseMonitor):
             # We should probably remove the "output" key from the result
             # as it is redundant, each output already generates a "log()"
             # message that is streamed to the client.
-            build_result=result,
+            result=result,
         ))
 
     def log(self, message, origin: Optional[str] = None):

osbuild/pipeline.py

@@ -55,6 +55,17 @@ class BuildResult:
         return vars(self)
 
 
+class DownloadResult:
+    def __init__(self, name: str, source_id: str, success: bool) -> None:
+        self.name = name
+        self.id = source_id
+        self.success = success
+        self.output = ""
+
+    def as_dict(self) -> Dict[str, Any]:
+        return vars(self)
+
+
 class Stage:
     def __init__(self, info, source_options, build, base, options, source_epoch):
         self.info = info
@@ -417,8 +428,19 @@ class Manifest:
             for source in self.sources:
                 # Workaround for lack of progress from sources, this
                 # will need to be reworked later.
+                dr = DownloadResult(source.name, source.id, success=True)
                 monitor.begin(source)
-                source.download(mgr, store)
+                try:
+                    source.download(mgr, store)
+                except host.RemoteError as e:
+                    dr.success = False
+                    dr.output = str(e)
+                    monitor.result(dr)
+                    raise e
+                monitor.result(dr)
+                # ideally we would make the whole of download more symmetric
+                # to "build_stages" and return a "results" here in "finish"
+                # as well
                 monitor.finish({"name": source.info.name})
 
     def depsolve(self, store: ObjectStore, targets: Iterable[str]) -> List[str]:

test/mod/test_monitor.py

@@ -10,6 +10,9 @@ import tempfile
 import time
 import unittest
 from collections import defaultdict
+from unittest.mock import Mock, patch
+
+import pytest
 
 import osbuild
 import osbuild.meta
@@ -326,3 +329,90 @@ def test_context_id():
     assert ctx.id == "20bf38c0723b15c2c9a52733c99814c298628526d8b8eabf7c378101cc9a9cf3"
     ctx._origin = "foo"  # pylint: disable=protected-access
     assert ctx.id != "00d202e4fc9d917def414d1c9f284b137287144087ec275f2d146d9d47b3c8bb"
+
+
+def test_monitor_download_happy(tmp_path):
+    store = ObjectStore(tmp_path)
+    tape = TapeMonitor()
+    happy_source = Mock()
+
+    manifest = osbuild.Manifest()
+    manifest.sources = [happy_source]
+    manifest.download(store, tape)
+    assert tape.counter["begin"] == 1
+    assert tape.counter["finish"] == 1
+    assert tape.counter["result"] == 1
+    # no stage was run as part of the download so this is nil
+    assert tape.counter["stages"] == 0
+
+
+def test_monitor_download_error(tmp_path):
+    store = ObjectStore(tmp_path)
+    tape = TapeMonitor()
+    failing_source = Mock()
+    failing_source.download.side_effect = osbuild.host.RemoteError("name", "value", "stack")
+
+    manifest = osbuild.Manifest()
+    manifest.sources = [failing_source]
+    # this is different from stage failures, those do not raise exceptions
+    with pytest.raises(osbuild.host.RemoteError):
+        manifest.download(store, tape)
+    assert tape.counter["begin"] == 1
+    assert tape.counter["result"] == 1
+    # this is different from stage failures that emit a "finish" on failure
+    # here
+    assert tape.counter["finish"] == 0
+
+
+@patch.object(osbuild.sources.Source, "download")
+def test_jsonseq_download_happy(_, tmp_path):
+    store = ObjectStore(tmp_path)
+    index = osbuild.meta.Index(os.curdir)
+    info = index.get_module_info("Source", "org.osbuild.curl")
+    happy_source = osbuild.sources.Source(info, {}, None)
+
+    manifest = osbuild.Manifest()
+    manifest.sources = [happy_source]
+    with tempfile.TemporaryFile() as tf:
+        mon = JSONSeqMonitor(tf.fileno(), 1)
+        manifest.download(store, mon)
+
+        tf.flush()
+        tf.seek(0)
+        log = []
+        for line in tf.read().decode().strip().split("\x1e"):
+            log.append(json.loads(line))
+        assert len(log) == 3
+        assert log[0]["message"] == "Starting pipeline source org.osbuild.curl"
+        assert log[1]["message"] == "Finished module source org.osbuild.curl"
+        assert log[1]["result"]["name"] == "source org.osbuild.curl"
+        assert log[1]["result"]["success"]
+        assert log[2]["message"] == "Finished pipeline org.osbuild.curl"
+
+
+@patch.object(osbuild.sources.Source, "download")
+def test_jsonseq_download_unhappy(mocked_download, tmp_path):
+    store = ObjectStore(tmp_path)
+    index = osbuild.meta.Index(os.curdir)
+    info = index.get_module_info("Source", "org.osbuild.curl")
+    failing_source = osbuild.sources.Source(info, {}, None)
+    mocked_download.side_effect = osbuild.host.RemoteError("RuntimeError", "curl: error download ...", "error stack")
+
+    manifest = osbuild.Manifest()
+    manifest.sources = [failing_source]
+    with tempfile.TemporaryFile() as tf:
+        mon = JSONSeqMonitor(tf.fileno(), 1)
+        with pytest.raises(osbuild.host.RemoteError):
+            manifest.download(store, mon)
+
+        tf.flush()
+        tf.seek(0)
+        log = []
+        for line in tf.read().decode().strip().split("\x1e"):
+            log.append(json.loads(line))
+        assert len(log) == 2
+        assert log[0]["message"] == "Starting pipeline source org.osbuild.curl"
+        assert log[1]["message"] == "Finished module source org.osbuild.curl"
+        assert log[1]["result"]["name"] == "source org.osbuild.curl"
+        assert log[1]["result"]["success"] is False
+        assert log[1]["result"]["output"] == "RuntimeError: curl: error download ...\n error stack"