Merge ~cjwatson/launchpad:codeimport-git-progress into launchpad:master

[Colin Watson]

Enable throttled progress output from git-to-git import workers

This may help with bug 1642699, though I'm not sure yet.

Getting things to work with Python 2's disconnect between file objects and the io module is gratuitously inconvenient.

This is essentially the same as https://code.launchpad.net/~cjwatson/launchpad/codeimport-git-progress/+merge/323902, converted to git and rebased on master.

[Ioana Lasc (ilasc)]

looks good.

review: Approve

[Otto Co-Pilot (otto-copilot)]

Merging failed
https://jenkins.ols.canonical.com/online-services/job/launchpad/649/

[Tom Wardill (twom)]

review: Approve

[Otto Co-Pilot (otto-copilot)]

Merging failed
https://jenkins.ols.canonical.com/online-services/job/launchpad/650/

diff --git a/lib/lp/codehosting/codeimport/tests/test_worker.py b/lib/lp/codehosting/codeimport/tests/test_worker.py
index d0a30cf..9daf357 100644
--- a/lib/lp/codehosting/codeimport/tests/test_worker.py
+++ b/lib/lp/codehosting/codeimport/tests/test_worker.py
@@ -58,6 +58,10 @@ import six
 import subvertpy
 import subvertpy.client
 import subvertpy.ra
+from testtools.matchers import (
+    ContainsAll,
+    LessThan,
+    )
 
 import lp.codehosting
 from lp.codehosting.codeimport.tarball import (
@@ -81,6 +85,7 @@ from lp.codehosting.codeimport.worker import (
     ForeignTreeStore,
     get_default_bazaar_branch_store,
     GitImportWorker,
+    GitToGitImportWorker,
     ImportDataStore,
     ToBzrImportWorker,
     )
@@ -1358,6 +1363,53 @@ class TestBzrImport(WorkerTest, TestActualImportMixin,
             branch.repository.get_revision(branch.last_revision()).committer)
 
 
+class TestGitToGitImportWorker(TestCase):
+
+    def test_throttleProgress(self):
+        source_details = self.factory.makeCodeImportSourceDetails(
+            rcstype="git", target_rcstype="git")
+        logger = BufferLogger()
+        worker = GitToGitImportWorker(
+            source_details, logger, AcceptAnythingPolicy())
+        read_fd, write_fd = os.pipe()
+        pid = os.fork()
+        if pid == 0:  # child
+            os.close(read_fd)
+            with os.fdopen(write_fd, "wb") as write:
+                write.write(b"Starting\n")
+                for i in range(50):
+                    time.sleep(0.1)
+                    write.write(("%d ...\r" % i).encode("UTF-8"))
+                    if (i % 10) == 9:
+                        write.write(
+                            ("Interval %d\n" % (i // 10)).encode("UTF-8"))
+                write.write(b"Finishing\n")
+            os._exit(0)
+        else:  # parent
+            os.close(write_fd)
+            with os.fdopen(read_fd, "rb") as read:
+                lines = list(worker._throttleProgress(read, timeout=0.5))
+            os.waitpid(pid, 0)
+            # Matching the exact sequence of lines would be too brittle, but
+            # we require some things to be true:
+            # All the non-progress lines must be present, in the right
+            # order.
+            self.assertEqual(
+                [u"Starting\n", u"Interval 0\n", u"Interval 1\n",
+                 u"Interval 2\n", u"Interval 3\n", u"Interval 4\n",
+                 u"Finishing\n"],
+                [line for line in lines if not line.endswith(u"\r")])
+            # No more than 15 progress lines may be present (allowing some
+            # slack for the child process being slow).
+            progress_lines = [line for line in lines if line.endswith(u"\r")]
+            self.assertThat(len(progress_lines), LessThan(16))
+            # All the progress lines immediately before interval markers
+            # must be present.
+            self.assertThat(
+                progress_lines,
+                ContainsAll([u"%d ...\r" % i for i in (9, 19, 29, 39, 49)]))
+
+
 class CodeImportBranchOpenPolicyTests(TestCase):
 
     def setUp(self):
diff --git a/lib/lp/codehosting/codeimport/worker.py b/lib/lp/codehosting/codeimport/worker.py
index 6259189..5aacf29 100644
--- a/lib/lp/codehosting/codeimport/worker.py
+++ b/lib/lp/codehosting/codeimport/worker.py
@@ -19,10 +19,10 @@ __all__ = [
     'get_default_bazaar_branch_store',
     ]
 
-
 import io
 import os
 import shutil
+import signal
 import subprocess
 
 # FIRST Ensure correct plugins are loaded. Do not delete this comment or the
@@ -962,14 +962,94 @@ class BzrImportWorker(PullingImportWorker):
 class GitToGitImportWorker(ImportWorker):
     """An import worker for imports from Git to Git."""
 
+    def _throttleProgress(self, file_obj, timeout=15.0):
+        """Throttle progress messages from a file object.
+
+        git can produce progress output on stderr, but it produces rather a
+        lot of it and we don't want it all to end up in logs.  Throttle this
+        so that we only produce output every `timeout` seconds, or when we
+        see a line terminated with a newline rather than a carriage return.
+
+        :param file_obj: a file-like object opened in binary mode.
+        :param timeout: emit progress output only after this many seconds
+            have elapsed.
+        :return: an iterator of interesting text lines read from the file.
+        """
+        # newline="" requests universal newlines mode, but without
+        # translation.
+        if six.PY2 and isinstance(file_obj, file):
+            # A Python 2 file object can't be used directly to construct an
+            # io.TextIOWrapper.
+            class _ReadableFileWrapper:
+                def __init__(self, raw):
+                    self._raw = raw
+
+                def __enter__(self):
+                    return self
+
+                def __exit__(self, exc_type, exc_value, exc_tb):
+                    pass
+
+                def readable(self):
+                    return True
+
+                def writable(self):
+                    return False
+
+                def seekable(self):
+                    return True
+
+                def __getattr__(self, name):
+                    return getattr(self._raw, name)
+
+            with _ReadableFileWrapper(file_obj) as readable:
+                with io.BufferedReader(readable) as buffered:
+                    for line in self._throttleProgress(
+                            buffered, timeout=timeout):
+                        yield line
+                    return
+
+        class ReceivedAlarm(Exception):
+            pass
+
+        def alarm_handler(signum, frame):
+            raise ReceivedAlarm()
+
+        old_alarm = signal.signal(signal.SIGALRM, alarm_handler)
+        try:
+            progress_buffer = None
+            with io.TextIOWrapper(
+                    file_obj, encoding="UTF-8", errors="replace",
+                    newline="") as wrapped_file:
+                while True:
+                    try:
+                        signal.setitimer(signal.ITIMER_REAL, timeout)
+                        line = next(wrapped_file)
+                        signal.setitimer(signal.ITIMER_REAL, 0)
+                        if line.endswith(u"\r"):
+                            progress_buffer = line
+                        else:
+                            if progress_buffer is not None:
+                                yield progress_buffer
+                                progress_buffer = None
+                            yield line
+                    except ReceivedAlarm:
+                        if progress_buffer is not None:
+                            yield progress_buffer
+                            progress_buffer = None
+                    except StopIteration:
+                        break
+        finally:
+            signal.setitimer(signal.ITIMER_REAL, 0)
+            signal.signal(signal.SIGALRM, old_alarm)
+
     def _runGit(self, *args, **kwargs):
         """Run git with arguments, sending output to the logger."""
         cmd = ["git"] + list(args)
         git_process = subprocess.Popen(
             cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, **kwargs)
-        for line in git_process.stdout:
-            line = line.decode("UTF-8", "replace").rstrip("\n")
-            self._logger.info(sanitise_urls(line))
+        for line in self._throttleProgress(git_process.stdout):
+            self._logger.info(sanitise_urls(line.rstrip("\r\n")))
         retcode = git_process.wait()
         if retcode:
             raise subprocess.CalledProcessError(retcode, cmd)
@@ -1104,13 +1184,14 @@ class GitToGitImportWorker(ImportWorker):
                 # Push the target of HEAD first to ensure that it is always
                 # available.
                 self._runGit(
-                    "push", target_url, "+%s:%s" % (new_head, new_head),
-                    cwd="repository")
+                    "push", "--progress", target_url,
+                    "+%s:%s" % (new_head, new_head), cwd="repository")
                 try:
                     self._setHead(target_url, new_head)
                 except GitProtocolError as e:
                     self._logger.info("Unable to set default branch: %s" % e)
-            self._runGit("push", "--mirror", target_url, cwd="repository")
+            self._runGit(
+                "push", "--progress", "--mirror", target_url, cwd="repository")
         except subprocess.CalledProcessError as e:
             self._logger.info(
                 "Unable to push to hosting service: git push exited %s" %