MBSConsoleHandler: show status of ongoing repository downloads

When downloading files from Koji to make a local repository, display
a temporary status of which files are being displayed to the console
appended after any log messages. Updates are done by erasing the status
that was written, adding a log message, then writing the status again.

module_build_service/builder/utils.py

@@ -83,6 +83,8 @@ def create_local_repo_from_koji_tag(config, tag, repo_dir, archs=None):
     # Placed here to avoid py2/py3 conflicts...
     import koji
 
+    log.local_repo_start(tag)
+
     if not archs:
         archs = ["x86_64", "noarch"]
 
@@ -100,6 +102,7 @@ def create_local_repo_from_koji_tag(config, tag, repo_dir, archs=None):
 
     if not builds:
         log.debug("No builds are associated with the tag %r", tag)
+        log.local_repo_done(tag, 'No builds to download')
         return False
 
     # Reformat builds so they are dict with build_id as a key.
@@ -130,9 +133,12 @@ def create_local_repo_from_koji_tag(config, tag, repo_dir, archs=None):
                 os.remove(local_fn)
             repo_changed = True
             url = pathinfo.build(build_info) + "/" + fname
-            download_args.append((url, local_fn))
+            download_args.append((tag, url, local_fn))
 
-    log.info("Downloading %d packages from Koji tag %s to %s" % (len(download_args), tag, repo_dir))
+    if repo_changed:
+        log.local_repo_start_downloads(tag, len(download_args), repo_dir)
+    else:
+        log.local_repo_done(tag, 'All builds already downloaded')
 
     # Create the output directory
     try:
@@ -141,22 +147,26 @@ def create_local_repo_from_koji_tag(config, tag, repo_dir, archs=None):
         if exception.errno != errno.EEXIST:
             raise
 
-    def _download_file(url_and_dest):
+    def _download_file(tag_url_and_dest):
         """
         Download a file in a memory efficient manner
-        :param url_and_dest: a tuple containing the URL and the destination to download to
+        :param url_and_dest: a tuple containing the tag, the URL and the destination to download to
         :return: None
         """
-        log.info("Downloading {0}...".format(url_and_dest[0]))
+        assert len(tag_url_and_dest) == 3, "tag_url_and_dest must have three values"
-        if len(url_and_dest) != 2:
-            raise ValueError("url_and_dest must have two values")
 
-        rv = requests.get(url_and_dest[0], stream=True, timeout=60)
+        tag, url, dest = tag_url_and_dest
-        with open(url_and_dest[1], "wb") as f:
+
+        log.local_repo_start_download(tag, url)
+
+        rv = requests.get(url, stream=True, timeout=60)
+        with open(dest, "wb") as f:
             for chunk in rv.iter_content(chunk_size=1024):
                 if chunk:
                     f.write(chunk)
 
+        log.local_repo_done_download(tag, url)
+
     # Download the RPMs four at a time.
     pool = ThreadPool(4)
     try:
@@ -173,6 +183,8 @@ def create_local_repo_from_koji_tag(config, tag, repo_dir, archs=None):
         log.info("Creating local repository in %s" % repo_dir)
         execute_cmd(["/usr/bin/createrepo_c", repo_dir])
 
+        log.local_repo_done(tag, 'Finished downloading packages')
+
     return True
 
 

module_build_service/common/logger.py

@@ -24,6 +24,8 @@ import os
 import logging
 import logging.handlers
 import inspect
+import re
+import signal
 import sys
 
 levels = {
@@ -204,6 +206,95 @@ class ModuleBuildLogs(object):
         del self.handlers[build.id]
 
 
+class LocalRepo(object):
+    def __init__(self, koji_tag):
+        self.koji_tag = koji_tag
+        self.current_downloads = set()
+        self.total_downloads = 0
+        self.completed_downloads = 0
+        self.status = ""
+
+    def start_downloads(self, total):
+        self.status = "Downloading packages"
+        self.total_downloads = total
+
+    def start_download(self, url):
+        self.current_downloads.add(url)
+
+    def done_download(self, url):
+        self.current_downloads.remove(url)
+        self.completed_downloads += 1
+
+    def show_status(self, stream, style):
+        if self.total_downloads > 0:
+            count = " {}/{}".format(self.completed_downloads, self.total_downloads)
+        else:
+            count = ""
+
+        print("{}: {}{}".format(style(self.koji_tag, bold=True), self.status, count),
+              file=stream)
+        for url in self.current_downloads:
+            print("    {}".format(os.path.basename(url)), file=stream)
+
+
+# Used to split aaa\nbbbb\n to ('aaa', '\n', 'bbb', '\n')
+NL_DELIMITER = re.compile('(\n)')
+# Matches *common* ANSI control sequences
+# https://en.wikipedia.org/wiki/ANSI_escape_code#CSI_sequences
+CSI_SEQUENCE = re.compile('\033[0-9;]*[A-Za-z]')
+
+
+class EraseableStream(object):
+    """
+    Wrapper around a terminal stream for writing output that can be
+    erased.
+    """
+
+    def __init__(self, target):
+        self.target = target
+        self.lines_written = 0
+        # We assume that the EraseableStream starts writing at column 0
+        self.column = 0
+        self.resize()
+
+    def resize(self):
+        self.size = os.get_terminal_size(self.target.fileno())
+
+    def write(self, string):
+        # We want to track how many lines we've written so that we can
+        # back up and erase them. Tricky thing is handling wrapping.
+
+        # Strip control sequences
+        plain = CSI_SEQUENCE.sub('', string)
+
+        for piece in NL_DELIMITER.split(plain):
+            if piece == '\n':
+                self.column = 0
+                self.lines_written += 1
+            else:
+                self.column = self.column + len(piece)
+                # self.column == self.size.column doesn't wrap -
+                # normal modern terminals wrap when a character is written
+                # that would be off-screen, not immediately when the
+                # line is full.
+                while self.column > self.size.columns:
+                    self.column -= self.size.columns
+                    self.lines_written += 1
+
+        self.target.write(string)
+
+    def erase(self):
+        if self.column > 0:
+            # move cursor to the beginning of line and delete whole line
+            self.target.write("\033[0G\033[2K")
+        for i in range(0, self.lines_written):
+            # move up cursor and delete whole line
+            self.target.write("\033[1A\033[2K")
+
+        self.lines_written = 0
+        self.column = 0
+
+
 FG_COLORS = {
     'green': '32',
     'red': '91',
@@ -238,8 +329,13 @@ class MBSConsoleHandler(logging.Handler):
         self.stream = stream
 
         self.tty = self.stream.isatty()
+        if self.tty:
+            self.status_stream = EraseableStream(self.stream)
+        else:
+            self.status_stream = None
 
         self.long_running = None
+        self.repos = {}
 
         self.debug_formatter = logging.Formatter(log_format)
         self.info_formatter = logging.Formatter("%(message)s")
@@ -286,6 +382,12 @@ class MBSConsoleHandler(logging.Handler):
                     self.long_running = None
 
                 print(formatted, file=self.stream)
+
+            if self.tty:
+                if self.repos:
+                    print('------------------------------', file=self.status_stream)
+                for repo in self.repos.values():
+                    repo.show_status(self.status_stream, self.style)
         finally:
             self.release()
 
@@ -296,6 +398,10 @@ class MBSConsoleHandler(logging.Handler):
 
         return decorate
 
+    def resize(self):
+        if self.status_stream:
+            self.status_stream.resize()
+
     @console_message("%s ...")
     def long_running_start(self, msg):
         if self.long_running:
@@ -315,6 +421,37 @@ class MBSConsoleHandler(logging.Handler):
             self.long_running = None
             return "{} ... done".format(msg)
 
+    @console_message("%s: Making local repository for Koji tag")
+    def local_repo_start(self, koji_tag):
+        repo = LocalRepo(koji_tag)
+        self.repos[koji_tag] = repo
+
+        return "{}: Making local repository for Koji tag".format(
+            self.style(koji_tag, bold=True))
+
+    @console_message("%s: %s")
+    def local_repo_done(self, koji_tag, message):
+        del self.repos[koji_tag]
+
+        return "{}: {}".format(
+            self.style(koji_tag, bold=True),
+            self.style(message, fg='green', bold=True))
+
+    @console_message("%s: Downloading %d packages from Koji tag to %s")
+    def local_repo_start_downloads(self, koji_tag, num_packages, repo_dir):
+        repo = self.repos[koji_tag]
+        repo.start_downloads(num_packages)
+
+    @console_message("%s: Downloading %s")
+    def local_repo_start_download(self, koji_tag, url):
+        repo = self.repos[koji_tag]
+        repo.start_download(url)
+
+    @console_message("%s: Done downloading %s")
+    def local_repo_done_download(self, koji_tag, url):
+        repo = self.repos[koji_tag]
+        repo.done_download(url)
+
     @classmethod
     def _setup_console_messages(cls):
         for value in cls.__dict__.values():
@@ -434,6 +571,11 @@ def init_logging(conf):
             root_logger.setLevel(conf.log_level)
             handler = MBSConsoleHandler()
             root_logger.addHandler(handler)
+
+            def handle_sigwinch(*args):
+                handler.resize()
+
+            signal.signal(signal.SIGWINCH, handle_sigwinch)
         else:
             logging.basicConfig(level=conf.log_level, format=log_format)
 

module_build_service/common/models.py

@@ -606,6 +606,10 @@ class ModuleBuild(MBSBase):
     def nvr_string(self):
         return kobo.rpmlib.make_nvr(self.nvr)
 
+    @property
+    def nsvc(self):
+        return "{}:{}:{}:{}".format(self.name, self.stream, self.version, self.context)
+
     @classmethod
     def create(
         cls,

module_build_service/scheduler/batches.py

@@ -255,7 +255,9 @@ def start_next_batch_build(config, module, builder, components=None):
         log.info("Skipping build of batch %d, no component to build.", module.batch)
         return start_next_batch_build(config, module, builder)
 
-    log.info("Starting build of next batch %d, %s" % (module.batch, unbuilt_components))
+    log.console("Starting build of next batch %d: %s",
+                module.batch,
+                ', '.join(c.package for c in unbuilt_components))
 
     # Attempt to reuse any components possible in the batch before attempting to build any
     unbuilt_components_after_reuse = []

module_build_service/scheduler/handlers/modules.py

@@ -145,6 +145,8 @@ def done(msg_id, module_build_id, module_build_state):
         build.transition(db_session, conf, state=models.BUILD_STATES["ready"])
         db_session.commit()
 
+    log.console("Finished building %s", build.nsvc)
+
     build_logs.stop(build)
     GenericBuilder.clear_cache(build)
 
@@ -181,7 +183,8 @@ def init(msg_id, module_build_id, module_build_state):
         build_logs.build_logs_dir = mock_resultsdir
 
     build_logs.start(db_session, build)
-    log.info("Start to handle %s which is in init state.", build.mmd().get_nsvc())
+    log.console("Starting to build %s", build.nsvc)
+    log.console("Logging to %s", build_logs.path(db_session, build))
 
     error_msg = ""
     failure_reason = "unspec"
@@ -413,7 +416,8 @@ def wait(msg_id, module_build_id, module_build_state):
         raise
 
     if not build.component_builds:
-        log.info("There are no components in module %r, skipping build" % build)
+        log.console("There are no components in module %s, skipping build",
+                    build.nsvc)
         build.transition(db_session, conf, state=models.BUILD_STATES["build"])
         db_session.add(build)
         db_session.commit()
@@ -426,13 +430,14 @@ def wait(msg_id, module_build_id, module_build_state):
     # If all components in module build will be reused, we don't have to build
     # module-build-macros, because there won't be any build done.
     if attempt_to_reuse_all_components(builder, build):
-        log.info("All components have been reused for module %r, skipping build" % build)
+        log.console("All components have been reused for module %s, skipping build",
+                    build.nsvc)
         build.transition(db_session, conf, state=models.BUILD_STATES["build"])
         db_session.add(build)
         db_session.commit()
         return []
 
-    log.debug("Starting build batch 1")
+    log.console("Starting build of batch 1: module-build-macros")
     build.batch = 1
     db_session.commit()
 

tests/test_common/test_logger.py

@@ -5,6 +5,7 @@ import io
 import logging
 import os
 from os import path
+import re
 import pytest
 import shutil
 import tempfile
@@ -119,6 +120,125 @@ class TestLogger:
         assert log2.path(db_session, build) == "/some/path/build-nginx-1-2.log"
 
 
+class FakeTerminal(object):
+    """
+    Just enough terminal to allow meaningfully testing the terminal fanciness
+    in MBSConsoleHandler
+    """
+
+    def __init__(self):
+        self.serializer = FakeTerminalSerializer(self)
+        self.reset()
+        self.columns = 80
+        self.rows = 24
+
+    def get_size(self):
+        return os.terminal_size((self.columns, self.rows))
+
+    def isatty(self):
+        return True
+
+    def fileno(self):
+        return 42
+
+    def flush(self):
+        pass
+
+    def write(self, raw):
+        for m in re.finditer(r'\n|\033\[[0-9;]*[A-Za-z]|\033|[^\033\n]*', raw):
+            piece = m.group(0)
+            if piece == '\n':
+                self._next_row()
+            elif piece == '\033[0m':
+                self.cur_attr = "x"
+            elif piece == '\033[1m':
+                self.cur_attr = self.cur_attr.upper()
+            elif piece == '\033[32m':
+                self.cur_attr = 'G' if self.cur_attr.isupper() else 'g'
+            elif piece == '\033[91m':
+                self.cur_attr = 'R' if self.cur_attr.isupper() else 'r'
+            elif piece == '\033[0G':
+                self.cursor_column = 0
+            elif piece == '\033[1A':
+                self.cursor_column = 0
+                self.cursor_row = max(self.cursor_row - 1, 0)
+            elif piece == '\033[2K':
+                self.text[self.cursor_row] = ' ' * self.cursor_column
+                self.attrs[self.cursor_row] = 'x' * self.cursor_column
+            elif piece.startswith('\033['):
+                raise RuntimeError("Unhandled CSI sequence: %r", piece)
+            else:
+                pos = 0
+                while len(piece) - pos > self.columns - self.cursor_column:
+                    to_insert = self.columns - self.cursor_column
+                    self._insert(piece[pos:pos + to_insert])
+                    pos += to_insert
+                    self._next_row()
+                self._insert(piece[pos:])
+
+    def _next_row(self):
+        self.cursor_row += 1
+        if self.cursor_row == len(self.text):
+            self.text.append('')
+            self.attrs.append('')
+        self.cursor_column = 0
+
+    def _insert(self, string):
+        text = self.text[self.cursor_row]
+        self.text[self.cursor_row] = (text[0:self.cursor_column]
+                                      + string
+                                      + text[self.cursor_column + len(string):])
+        attrs = self.attrs[self.cursor_row]
+        self.attrs[self.cursor_row] = (attrs[0:self.cursor_column]
+                                       + self.cur_attr * len(string)
+                                       + attrs[self.cursor_column + len(string):])
+        self.cursor_column += len(string)
+
+    def reset(self):
+        self.text = ['']
+        self.attrs = ['']
+        self.cur_attr = 'x'
+        self.cursor_row = 0
+        self.cursor_column = 0
+
+    def serialize(self):
+        return self.serializer.serialize()
+
+
+class FakeTerminalSerializer(object):
+    """Serializes the terminal contents with <R></R> tags to represent attributes"""
+
+    def __init__(self, terminal):
+        self.terminal = terminal
+
+    def serialize(self):
+        self.result = io.StringIO()
+        self.last_attr = 'x'
+
+        for row in range(0, len(self.terminal.text)):
+            text = self.terminal.text[row]
+            attrs = self.terminal.attrs[row]
+
+            for col in range(0, len(text)):
+                self.set_attr(attrs[col])
+                self.result.write(text[col])
+            self.set_attr('x')
+            self.result.write('\n')
+
+        value = self.result.getvalue()
+        self.result = None
+
+        return value
+
+    def set_attr(self, attr):
+        if attr != self.last_attr:
+            if self.last_attr != 'x':
+                self.result.write('</' + self.last_attr + '>')
+            if attr != 'x':
+                self.result.write('<' + attr + '>')
+            self.last_attr = attr
+
+
 class TestConsoleHandler:
     def terminal(test_method):
         test_method.terminal = True
@@ -132,15 +252,18 @@ class TestConsoleHandler:
         return decorate
 
     def setup_method(self, test_method):
-        self.stream = io.StringIO()
-
         if getattr(test_method, 'terminal', False):
+            self.stream = FakeTerminal()
+
             self.get_terminal_size_patcher = patch("os.get_terminal_size")
             mock_get_terminal_size = self.get_terminal_size_patcher.start()
-            mock_get_terminal_size.return_value = os.terminal_size((80, 24))
 
-            self.stream.isatty = lambda: True
+            def get_terminal_size(fileno):
-            self.stream.fileno = lambda: 42
+                return self.stream.get_size()
+
+            mock_get_terminal_size.side_effect = get_terminal_size
+        else:
+            self.stream = io.StringIO()
 
         self.handler = MBSConsoleHandler(stream=self.stream)
         self.handler.level = getattr(test_method, 'level', logging.INFO)
@@ -153,6 +276,14 @@ class TestConsoleHandler:
         if getattr(test_method, 'terminal', False):
             self.get_terminal_size_patcher.stop()
 
+    def current(self):
+        if isinstance(self.stream, FakeTerminal):
+            val = self.stream.serialize()
+        else:
+            val = self.stream.getvalue()
+
+        return val.rstrip() + '\n'
+
     def log_messages(self):
         log.debug("Debug")
         log.info("Info")
@@ -163,35 +294,35 @@ class TestConsoleHandler:
     def test_console_basic(self):
         self.log_messages()
 
-        value = self.stream.getvalue()
+        current = self.current()
-        assert "Debug" not in value
+        assert "Debug" not in current
-        assert "Info" not in value
+        assert "Info" not in current
-        assert "Console" in value
+        assert "Console" in current
-        assert "\nWARNING - Warning" in value
+        assert "\nWARNING - Warning" in current
-        assert "\nERROR - Error" in value
+        assert "\nERROR - Error" in current
 
     @terminal
     def test_console_terminal(self):
         self.log_messages()
 
-        value = self.stream.getvalue()
+        current = self.current()
-        assert "Debug" not in value
+        assert "Debug" not in current
-        assert "Info" not in value
+        assert "Info" not in current
-        assert "Console" in value
+        assert "Console" in current
-        assert "\n\x1b[1m\x1b[91mWARNING\x1b[0m - Warning" in value
+        assert "<R>WARNING</R> - Warning" in current
-        assert "\n\x1b[1m\x1b[91mERROR\x1b[0m - Error" in value
+        assert "<R>ERROR</R> - Error" in current
 
     @level(logging.DEBUG)
     @terminal
     def test_console_debug(self):
         self.log_messages()
 
-        value = self.stream.getvalue()
+        current = self.current()
-        assert "MainThread - MBS - DEBUG - Debug" in value
+        assert "MainThread - MBS - DEBUG - Debug" in current
-        assert "MainThread - MBS - INFO - Info" in value
+        assert "MainThread - MBS - INFO - Info" in current
-        assert "MainThread - MBS - INFO - Console" in value
+        assert "MainThread - MBS - INFO - Console" in current
-        assert "MainThread - MBS - WARNING - Warning" in value
+        assert "MainThread - MBS - WARNING - Warning" in current
-        assert "MainThread - MBS - ERROR - Error" in value
+        assert "MainThread - MBS - ERROR - Error" in current
 
     @terminal
     def test_console_long_running(self):
@@ -209,9 +340,7 @@ class TestConsoleHandler:
         log.long_running_end("Blabbing")
         log.long_running_end("Frobulating")
 
-        print(self.stream.getvalue())
+        assert self.current() == textwrap.dedent("""\
-
-        assert self.stream.getvalue() == textwrap.dedent("""\
             Frobulating ... done
             ---
             Frobulating ...
@@ -222,3 +351,76 @@ class TestConsoleHandler:
             Blabbing ... done
             Frobulating ... done
         """)
+
+    @terminal
+    def test_console_local_repo(self):
+        log.local_repo_start("module-testmodule-build")
+
+        assert self.current() == textwrap.dedent("""\
+            <X>module-testmodule-build</X>: Making local repository for Koji tag
+            ------------------------------
+            <X>module-testmodule-build</X>:
+        """)
+
+        log.local_repo_start_downloads("module-testmodule-build", 42, "/tmp/download-dir")
+        log.local_repo_start_download("module-testmodule-build",
+                                      "https://ftp.example.com/libsomething-1.2.3-1.x86_64.rpm")
+        log.local_repo_start_download("module-testmodule-build",
+                                      "https://ftp.example.com/libother-1.2.3-1.x86_64.rpm")
+        log.local_repo_done_download("module-testmodule-build",
+                                     "https://ftp.example.com/libother-1.2.3-1.x86_64.rpm")
+        assert self.current() == textwrap.dedent("""\
+            <X>module-testmodule-build</X>: Making local repository for Koji tag
+            ------------------------------
+            <X>module-testmodule-build</X>: Downloading packages 1/42
+                libsomething-1.2.3-1.x86_64.rpm
+        """)
+
+        log.local_repo_done_download("module-testmodule-build",
+                                     "https://ftp.example.com/libsomething-1.2.3-1.x86_64.rpm")
+        log.local_repo_done("module-testmodule-build", "downloaded everything")
+        assert self.current() == textwrap.dedent("""\
+            <X>module-testmodule-build</X>: Making local repository for Koji tag
+            <X>module-testmodule-build</X>: <G>downloaded everything</G>
+        """)
+
+    @terminal
+    def test_console_local_repo_wrap(self):
+        log.local_repo_start("module-testmodule-build")
+        long_url = "https://ftp.example.com/lib" + (80 * "z") + "-1.2.3-1.x86_64.rpm"
+        log.local_repo_start_downloads("module-testmodule-build", 42, "/tmp/download-dir")
+        log.local_repo_start_download("module-testmodule-build", long_url)
+        assert self.current() == textwrap.dedent("""\
+            <X>module-testmodule-build</X>: Making local repository for Koji tag
+            ------------------------------
+            <X>module-testmodule-build</X>: Downloading packages 0/42
+                libzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzz
+            zzzzzzz-1.2.3-1.x86_64.rpm
+        """)
+
+        log.local_repo_done_download("module-testmodule-build", long_url)
+        log.local_repo_done("module-testmodule-build", "downloaded everything")
+        assert self.current() == textwrap.dedent("""\
+            <X>module-testmodule-build</X>: Making local repository for Koji tag
+            <X>module-testmodule-build</X>: <G>downloaded everything</G>
+        """)
+
+    @terminal
+    def test_console_partial_line_erase(self):
+        self.handler.status_stream.write("Foo\nBar")
+        self.handler.status_stream.erase()
+        self.handler.status_stream.write("Baz")
+
+        assert self.current() == "Baz\n"
+
+    @terminal
+    def test_console_resize(self):
+        self.stream.columns = 20
+        self.handler.resize()
+
+        self.stream.write("Foo\n")
+        self.handler.status_stream.write(30 * "x")
+        self.handler.status_stream.erase()
+        self.handler.status_stream.write("Baz")
+
+        assert self.current() == "Foo\nBaz\n"