Merge lp:~mwhudson/launchpad/pull-mirror-branches-separately-bug-520107 into lp:launchpad

[Michael Hudson-Doyle]

Allow an invocation of supermirror-pull.py to only pull certain branch types.

[Michael Hudson-Doyle (mwhudson)]

Mirrored branches are inherently less reliable to access than hosted or imported branches, so it makes sense to be able to handle them in a separate process, and that's what this branch does.

The default behavior is kept the same, we can coordinate with the LOSAs after rollout to set up two different cronjobs.

[Tim Penhey (thumper)]

Good except test_type_filter_hosted_imported_returns_imported which creates a hosted branch instead of an imported one.

review: Approve

=== modified file 'cronscripts/supermirror-pull.py'
--- cronscripts/supermirror-pull.py	2009-10-13 14:38:07 +0000
+++ cronscripts/supermirror-pull.py	2010-02-24 19:04:21 +0000
@@ -38,12 +38,14 @@
 if __name__ == '__main__':
     parser = OptionParser()
     logger_options(parser)
+    parser.add_option('--branch-type', action='append', default=[])
     (options, arguments) = parser.parse_args()
     if arguments:
         parser.error("Unhandled arguments %s" % repr(arguments))
     log = set_up_logging_for_script(options, 'supermirror_puller')
     manager = scheduler.JobScheduler(
-        LoggingProxy(config.codehosting.branch_puller_endpoint, log), log)
+        LoggingProxy(config.codehosting.branch_puller_endpoint, log), log,
+        options.branch_type)
 
     reactor.callWhenRunning(run_mirror, log, manager)
     reactor.run()
 
=== modified file 'lib/lp/code/errors.py'
--- lib/lp/code/errors.py	2009-12-14 06:41:18 +0000
+++ lib/lp/code/errors.py	2010-02-24 19:04:21 +0000
@@ -11,6 +11,7 @@
     'ClaimReviewFailed',
     'InvalidBranchMergeProposal',
     'ReviewNotPending',
+    'UnknownBranchTypeError',
     'UserHasExistingReview',
     'UserNotBranchReviewer',
     'WrongBranchMergeProposal',
@@ -59,3 +60,7 @@
 
 class WrongBranchMergeProposal(Exception):
     """The comment requested is not associated with this merge proposal."""
+
+
+class UnknownBranchTypeError(Exception):
+    """Raised when the user specifies an unrecognized branch type."""
 
=== modified file 'lib/lp/code/interfaces/branch.py'
--- lib/lp/code/interfaces/branch.py	2010-02-18 16:00:24 +0000
+++ lib/lp/code/interfaces/branch.py	2010-02-24 19:04:21 +0000
@@ -119,10 +119,6 @@
     """The branch cannot be deleted at this time."""
 
 
-class UnknownBranchTypeError(Exception):
-    """Raised when the user specifies an unrecognized branch type."""
-
-
 class BranchCreationForbidden(BranchCreationException):
     """A Branch visibility policy forbids branch creation.
 
 
=== modified file 'lib/lp/code/interfaces/branchpuller.py'
--- lib/lp/code/interfaces/branchpuller.py	2010-02-24 01:22:28 +0000
+++ lib/lp/code/interfaces/branchpuller.py	2010-02-24 19:04:21 +0000
@@ -23,9 +23,11 @@
     MIRROR_TIME_INCREMENT = Attribute(
         "How frequently we mirror branches.")
 
-    def acquireBranchToPull():
+    def acquireBranchToPull(*branch_types):
         """Return a Branch to pull and mark it as mirror-started.
 
+        :param branch_types: Only return branches of these types.  Passing no
+            types means consider all types (apart from REMOTE).
         :return: The branch object to pull next, or ``None`` if there is no
             branch to pull.
         """
 
=== modified file 'lib/lp/code/interfaces/codehosting.py'
--- lib/lp/code/interfaces/codehosting.py	2010-02-24 01:57:35 +0000
+++ lib/lp/code/interfaces/codehosting.py	2010-02-24 19:04:21 +0000
@@ -58,9 +58,12 @@
     Published at 'branch_puller' on the private XML-RPC server.
     """
 
-    def acquireBranchToPull():
+    def acquireBranchToPull(branch_type_names):
         """Return a Branch to pull and mark it as mirror-started.
 
+        :param branch_type_names: Only consider branches of these type names.
+            An empty list means consider HOSTED, MIRRORED and IMPORTED
+            branches.
         :return: A 5-tuple::
 
               (branch_id, pull_url, unique_name, default_branch, branch_type)
 
=== modified file 'lib/lp/code/model/branchpuller.py'
--- lib/lp/code/model/branchpuller.py	2010-02-24 01:22:28 +0000
+++ lib/lp/code/model/branchpuller.py	2010-02-24 19:04:21 +0000
@@ -28,13 +28,16 @@
     MAXIMUM_MIRROR_FAILURES = 5
     MIRROR_TIME_INCREMENT = timedelta(hours=6)
 
-    def acquireBranchToPull(self):
+    def acquireBranchToPull(self, *branch_types):
         """See `IBranchPuller`."""
+        if not branch_types:
+            branch_types = (
+                BranchType.HOSTED, BranchType.MIRRORED, BranchType.IMPORTED)
         store = getUtility(IStoreSelector).get(MAIN_STORE, DEFAULT_FLAVOR)
         branch = store.find(
             Branch,
             Branch.next_mirror_time <= UTC_NOW,
-            Branch.branch_type != BranchType.REMOTE).order_by(
+            Branch.branch_type.is_in(branch_types)).order_by(
                 Branch.next_mirror_time).first()
         if branch is not None:
             branch.startMirroring()
 
=== modified file 'lib/lp/code/model/tests/test_branchpuller.py'
--- lib/lp/code/model/tests/test_branchpuller.py	2010-02-24 01:22:28 +0000
+++ lib/lp/code/model/tests/test_branchpuller.py	2010-02-24 19:04:21 +0000
@@ -169,12 +169,22 @@
     and `startMirroring` as appropriate.
     """
 
-    def assertNoBranchIsAquired(self):
-        """Assert that there is no branch to pull."""
+    def assertNoBranchIsAquired(self, *branch_types):
+        """Assert that there is no branch to pull.
+
+        :param branch_types: A list of branch types to pass to
+            acquireBranchToPull.  Passing none means consider all types of
+            branch.
+        """
         raise NotImplementedError(self.assertNoBranchIsAquired)
 
-    def assertBranchIsAquired(self, branch):
-        """Assert that ``branch`` is the next branch to be pulled."""
+    def assertBranchIsAquired(self, branch, *branch_types):
+        """Assert that ``branch`` is the next branch to be pulled.
+
+        :param branch_types: A list of branch types to pass to
+            acquireBranchToPull.  Passing none means consider all types of
+            branch.
+        """
         raise NotImplementedError(self.assertBranchIsAquired)
 
     def startMirroring(self, branch):
@@ -202,7 +212,6 @@
         removeSecurityProxy(branch).branch_type = BranchType.REMOTE
         self.assertNoBranchIsAquired()
 
-
     def test_private(self):
         # If there is a private branch that needs mirroring,
         # acquireBranchToPull returns that.
@@ -232,6 +241,33 @@
         naked_second_branch.next_mirror_time -= timedelta(seconds=50)
         self.assertBranchIsAquired(naked_first_branch)
 
+    def test_type_filter_hosted_returns_hosted(self):
+        branch = self.factory.makeAnyBranch(branch_type=BranchType.HOSTED)
+        branch.requestMirror()
+        self.assertBranchIsAquired(branch, BranchType.HOSTED)
+
+    def test_type_filter_hosted_does_not_return_mirrored(self):
+        branch = self.factory.makeAnyBranch(branch_type=BranchType.MIRRORED)
+        branch.requestMirror()
+        self.assertNoBranchIsAquired(BranchType.HOSTED)
+
+    def test_type_filter_mirrored_does_not_return_hosted(self):
+        branch = self.factory.makeAnyBranch(branch_type=BranchType.HOSTED)
+        branch.requestMirror()
+        self.assertNoBranchIsAquired(BranchType.MIRRORED)
+
+    def test_type_filter_hosted_imported_returns_hosted(self):
+        branch = self.factory.makeAnyBranch(branch_type=BranchType.HOSTED)
+        branch.requestMirror()
+        self.assertBranchIsAquired(
+            branch, BranchType.HOSTED, BranchType.IMPORTED)
+
+    def test_type_filter_hosted_imported_returns_imported(self):
+        branch = self.factory.makeAnyBranch(branch_type=BranchType.IMPORTED)
+        branch.requestMirror()
+        self.assertBranchIsAquired(
+            branch, BranchType.HOSTED, BranchType.IMPORTED)
+
 
 class TestAcquireBranchToPullDirectly(TestCaseWithFactory,
                                       AcquireBranchToPullTests):
@@ -239,14 +275,16 @@
 
     layer = DatabaseFunctionalLayer
 
-    def assertNoBranchIsAquired(self):
+    def assertNoBranchIsAquired(self, *branch_types):
         """See `AcquireBranchToPullTests`."""
-        acquired_branch = getUtility(IBranchPuller).acquireBranchToPull()
+        acquired_branch = getUtility(IBranchPuller).acquireBranchToPull(
+            *branch_types)
         self.assertEqual(None, acquired_branch)
 
-    def assertBranchIsAquired(self, branch):
+    def assertBranchIsAquired(self, branch, *branch_types):
         """See `AcquireBranchToPullTests`."""
-        acquired_branch = getUtility(IBranchPuller).acquireBranchToPull()
+        acquired_branch = getUtility(IBranchPuller).acquireBranchToPull(
+            *branch_types)
         login_person(removeSecurityProxy(branch).owner)
         self.assertEqual(branch, acquired_branch)
         self.assertIsNot(None, acquired_branch.last_mirror_attempt)
 
=== modified file 'lib/lp/code/xmlrpc/codehosting.py'
--- lib/lp/code/xmlrpc/codehosting.py	2010-02-24 01:22:28 +0000
+++ lib/lp/code/xmlrpc/codehosting.py	2010-02-24 19:04:21 +0000
@@ -23,6 +23,7 @@
 from zope.security.proxy import removeSecurityProxy
 
 from canonical.launchpad.ftests import login_person, logout
+from lp.code.errors import UnknownBranchTypeError
 from lp.code.enums import BranchType
 from lp.code.interfaces.branch import BranchCreationException
 from lp.code.interfaces.branchlookup import IBranchLookup
@@ -54,9 +55,17 @@
 
     implements(IBranchPuller)
 
-    def acquireBranchToPull(self):
+    def acquireBranchToPull(self, branch_type_names):
         """See `IBranchPuller`."""
-        branch = getUtility(branchpuller.IBranchPuller).acquireBranchToPull()
+        branch_types = []
+        for branch_type_name in branch_type_names:
+            try:
+                branch_types.append(BranchType.items[branch_type_name])
+            except KeyError:
+                raise UnknownBranchTypeError(
+                    'Unknown branch type: %r' % (branch_type_name,))
+        branch = getUtility(branchpuller.IBranchPuller).acquireBranchToPull(
+            *branch_types)
         if branch is not None:
             branch = removeSecurityProxy(branch)
             default_branch = branch.target.default_stacked_on_branch
 
=== modified file 'lib/lp/code/xmlrpc/tests/test_codehosting.py'
--- lib/lp/code/xmlrpc/tests/test_codehosting.py	2010-02-24 01:57:35 +0000
+++ lib/lp/code/xmlrpc/tests/test_codehosting.py	2010-02-24 19:04:21 +0000
@@ -30,6 +30,7 @@
 from canonical.testing import DatabaseFunctionalLayer, FunctionalLayer
 
 from lp.code.enums import BranchType
+from lp.code.errors import UnknownBranchTypeError
 from lp.code.interfaces.branch import BRANCH_NAME_VALIDATION_ERROR_MESSAGE
 from lp.code.interfaces.branchlookup import IBranchLookup
 from lp.code.interfaces.branchtarget import IBranchTarget
@@ -409,15 +410,17 @@
         self.storage = frontend.getPullerEndpoint()
         self.factory = frontend.getLaunchpadObjectFactory()
 
-    def assertNoBranchIsAquired(self):
+    def assertNoBranchIsAquired(self, *branch_types):
         """See `AcquireBranchToPullTests`."""
-        pull_info = self.storage.acquireBranchToPull()
+        branch_types = tuple(branch_type.name for branch_type in branch_types)
+        pull_info = self.storage.acquireBranchToPull(branch_types)
         self.assertEqual((), pull_info)
 
-    def assertBranchIsAquired(self, branch):
+    def assertBranchIsAquired(self, branch, *branch_types):
         """See `AcquireBranchToPullTests`."""
         branch = removeSecurityProxy(branch)
-        pull_info = self.storage.acquireBranchToPull()
+        branch_types = tuple(branch_type.name for branch_type in branch_types)
+        pull_info = self.storage.acquireBranchToPull(branch_types)
         default_branch = branch.target.default_stacked_on_branch
         if default_branch:
             default_branch_name = default_branch
@@ -437,21 +440,21 @@
     def test_branch_type_returned_hosted(self):
         branch = self.factory.makeAnyBranch(branch_type=BranchType.HOSTED)
         branch.requestMirror()
-        pull_info = self.storage.acquireBranchToPull()
+        pull_info = self.storage.acquireBranchToPull(())
         _, _, _, _, branch_type = pull_info
         self.assertEqual('HOSTED', branch_type)
 
     def test_branch_type_returned_mirrored(self):
         branch = self.factory.makeAnyBranch(branch_type=BranchType.MIRRORED)
         branch.requestMirror()
-        pull_info = self.storage.acquireBranchToPull()
+        pull_info = self.storage.acquireBranchToPull(())
         _, _, _, _, branch_type = pull_info
         self.assertEqual('MIRRORED', branch_type)
 
     def test_branch_type_returned_import(self):
         branch = self.factory.makeAnyBranch(branch_type=BranchType.IMPORTED)
         branch.requestMirror()
-        pull_info = self.storage.acquireBranchToPull()
+        pull_info = self.storage.acquireBranchToPull(())
         _, _, _, _, branch_type = pull_info
         self.assertEqual('IMPORTED', branch_type)
 
@@ -459,7 +462,7 @@
         branch = self.factory.makeProductBranch()
         self.factory.enableDefaultStackingForProduct(branch.product)
         branch.requestMirror()
-        pull_info = self.storage.acquireBranchToPull()
+        pull_info = self.storage.acquireBranchToPull(())
         _, _, _, default_stacked_on_branch, _ = pull_info
         self.assertEqual(
             default_stacked_on_branch,
@@ -475,11 +478,16 @@
         mirrored_branch = self.factory.makeProductBranch(
             branch_type=BranchType.MIRRORED, product=product)
         mirrored_branch.requestMirror()
-        pull_info = self.storage.acquireBranchToPull()
+        pull_info = self.storage.acquireBranchToPull(())
         _, _, _, default_stacked_on_branch, _ = pull_info
         self.assertEqual(
             '', default_stacked_on_branch)
 
+    def test_unknown_branch_type_name_raises(self):
+        self.assertRaises(
+            UnknownBranchTypeError, self.storage.acquireBranchToPull,
+            ('NO_SUCH_TYPE',))
+
 
 class BranchFileSystemTest(TestCaseWithFactory):
     """Tests for the implementation of `IBranchFileSystem`."""
 
=== modified file 'lib/lp/codehosting/inmemory.py'
--- lib/lp/codehosting/inmemory.py	2010-02-24 01:57:35 +0000
+++ lib/lp/codehosting/inmemory.py	2010-02-24 19:04:21 +0000
@@ -18,6 +18,7 @@
 from zope.interface import implementer
 
 from canonical.database.constants import UTC_NOW
+from lp.code.errors import UnknownBranchTypeError
 from lp.code.model.branchnamespace import BranchNamespaceSet
 from lp.code.model.branchtarget import (
     PackageBranchTarget, ProductBranchTarget)
@@ -442,11 +443,20 @@
         self._branch_set = branch_set
         self._script_activity_set = script_activity_set
 
-    def acquireBranchToPull(self):
+    def acquireBranchToPull(self, branch_type_names):
+        if not branch_type_names:
+            branch_type_names = 'HOSTED', 'MIRRORED', 'IMPORTED'
+        branch_types = []
+        for branch_type_name in branch_type_names:
+            try:
+                branch_types.append(BranchType.items[branch_type_name])
+            except KeyError:
+                raise UnknownBranchTypeError(
+                    'Unknown branch type: %r' % (branch_type_name,))
         branches = sorted(
             [branch for branch in self._branch_set
              if branch.next_mirror_time is not None
-             and branch.branch_type != BranchType.REMOTE],
+             and branch.branch_type in branch_types],
             key=operator.attrgetter('next_mirror_time'))
         if branches:
             branch = branches[-1]
 
=== modified file 'lib/lp/codehosting/puller/scheduler.py'
--- lib/lp/codehosting/puller/scheduler.py	2009-12-22 00:26:49 +0000
+++ lib/lp/codehosting/puller/scheduler.py	2010-02-24 19:04:21 +0000
@@ -417,9 +417,10 @@
     branches.
     """
 
-    def __init__(self, branch_puller_endpoint, logger):
+    def __init__(self, branch_puller_endpoint, logger, branch_type_names):
         self.branch_puller_endpoint = branch_puller_endpoint
         self.logger = logger
+        self.branch_type_names = branch_type_names
         self.actualLock = None
         self.name = 'branch-puller'
         self.lockfilename = '/var/lock/launchpad-%s.lock' % self.name
@@ -455,7 +456,7 @@
 
     def _poll(self):
         deferred = self.branch_puller_endpoint.callRemote(
-            'acquireBranchToPull')
+            'acquireBranchToPull', self.branch_type_names)
         deferred.addCallback(self._turnJobTupleIntoTask)
         return deferred
 
 
=== modified file 'lib/lp/codehosting/puller/tests/test_acceptance.py'
--- lib/lp/codehosting/puller/tests/test_acceptance.py	2010-01-20 20:56:29 +0000
+++ lib/lp/codehosting/puller/tests/test_acceptance.py	2010-02-24 19:04:21 +0000
@@ -122,7 +122,7 @@
         output, error = process.communicate()
         return process.returncode, output, error
 
-    def runPuller(self):
+    def runPuller(self, *args):
         """Run the puller script for the given branch type.
 
         :param branch_type: One of 'upload', 'mirror' or 'import'
@@ -132,7 +132,7 @@
             stdout and stderr respectively.
         """
         command = [
-            '%s/bin/py' % config.root, self._puller_script, '-q']
+            '%s/bin/py' % config.root, self._puller_script, '-q'] + list(args)
         retcode, output, error = self.runSubprocess(command)
         return command, retcode, output, error
 
@@ -430,6 +430,23 @@
         command, retcode, output, error = self.runPuller()
         self.assertRanSuccessfully(command, retcode, output, error)
 
+    def test_type_filtering(self):
+        # When run with --branch-type arguments, the puller only mirrors those
+        # branches of the specified types.
+        hosted_branch = self.factory.makeAnyBranch(
+            branch_type=BranchType.HOSTED)
+        mirrored_branch = self.factory.makeAnyBranch(
+            branch_type=BranchType.MIRRORED)
+        mirrored_branch.requestMirror()
+        transaction.commit()
+        self.pushBranch(hosted_branch)
+        command, retcode, output, error = self.runPuller(
+            '--branch-type', 'HOSTED')
+        self.assertRanSuccessfully(command, retcode, output, error)
+        self.assertMirrored(hosted_branch)
+        self.assertIsNot(
+            None, mirrored_branch.next_mirror_time)
+
     def test_records_script_activity(self):
         # A record gets created in the ScriptActivity table.
         script_activity_set = getUtility(IScriptActivitySet)
 
=== modified file 'lib/lp/codehosting/puller/tests/test_scheduler.py'
--- lib/lp/codehosting/puller/tests/test_scheduler.py	2010-01-08 03:07:25 +0000
+++ lib/lp/codehosting/puller/tests/test_scheduler.py	2010-02-24 19:04:21 +0000
@@ -8,7 +8,6 @@
 from datetime import datetime
 import logging
 import os
-import sys
 import textwrap
 import unittest
 
@@ -42,8 +41,7 @@
 
 class FakePullerEndpointProxy:
 
-    def __init__(self, branch_queues=None):
-        self.branch_queues = branch_queues
+    def __init__(self):
         self.calls = []
 
     def callRemote(self, method_name, *args):
@@ -58,8 +56,8 @@
     def _default(self, *args):
         return defer.succeed(None)
 
-    def _remote_getBranchPullQueue(self, branch_type):
-        return defer.succeed(self.branch_queues[branch_type])
+    def _remote_acquireBranchToPull(self, *args):
+        return defer.succeed(0)
 
     def _remote_setStackedOn(self, branch_id, stacked_on_location):
         if stacked_on_location == 'raise-branch-not-found':
@@ -70,7 +68,7 @@
         return defer.succeed(None)
 
 
-class TestJobScheduler(unittest.TestCase):
+class TestJobScheduler(TrialTestCase):
 
     def setUp(self):
         self.masterlock = 'master.lock'
@@ -80,8 +78,9 @@
         if os.path.exists(self.masterlock):
             os.unlink(self.masterlock)
 
-    def makeJobScheduler(self):
-        return scheduler.JobScheduler(None, logging.getLogger())
+    def makeJobScheduler(self, branch_type_names=()):
+        return scheduler.JobScheduler(
+            FakePullerEndpointProxy(), logging.getLogger(), branch_type_names)
 
     def testManagerCreatesLocks(self):
         manager = self.makeJobScheduler()
@@ -100,6 +99,13 @@
         self.failUnless(os.path.exists(self.masterlock))
         manager.unlock()
 
+    def test_run_calls_acquireBranchToPull(self):
+        manager = self.makeJobScheduler(('HOSTED',))
+        manager.run()
+        self.assertEqual(
+            [('acquireBranchToPull', ('HOSTED',))],
+            manager.branch_puller_endpoint.calls)
+
 
 class TestPullerWireProtocol(TrialTestCase):
     """Tests for the `PullerWireProtocol`.
@@ -516,7 +522,6 @@
     layer = TwistedLayer
 
     def setUp(self):
-        from twisted.internet import reactor
         self.factory = ObjectFactory()
         self.available_oops_prefixes = set(['foo'])
         self.eventHandler = self.makePullerMaster(
@@ -526,7 +531,6 @@
         self.commands_spawned = []
 
     def tearDown(self):
-        from twisted.internet import reactor
         reactor.spawnProcess = self._realSpawnProcess
 
     def makePullerMaster(self, branch_type_name, default_stacked_on_url=None,
 
=== modified file 'lib/lp/testing/factory.py'
--- lib/lp/testing/factory.py	2010-02-23 20:02:07 +0000
+++ lib/lp/testing/factory.py	2010-02-24 19:04:21 +0000
@@ -93,7 +93,7 @@
     BranchType, CodeImportMachineState, CodeImportReviewStatus,
     CodeImportResultStatus, CodeReviewNotificationLevel,
     RevisionControlSystems)
-from lp.code.interfaces.branch import UnknownBranchTypeError
+from lp.code.errors import UnknownBranchTypeError
 from lp.code.interfaces.branchmergequeue import IBranchMergeQueueSet
 from lp.code.interfaces.branchnamespace import get_branch_namespace
 from lp.code.interfaces.codeimport import ICodeImportSet