Merge ~pappacena/launchpad:mirror-prober-parallelism-control into launchpad:master

[Thiago F. Pappacena]

Adding options to distribution mirror prober to allow us to easily control maximum number of requests in parallel that should be issued.

This change is basically removing the 2 hardcoded constants that control amount of parallel requests (in general and per host) and creating 2 parameters at the script.

The major change was having the RequestManager to create the locks on __init__ instead of a global one, and sharing the same RequestManager on probe_archive_mirror and updateMirrorFreshness.

[Colin Watson (cjwatson)]

review: Approve

[Thiago F. Pappacena (pappacena)]

Pushed the requested change. I'll top-approve the MP now.

diff --git a/cronscripts/distributionmirror-prober.py b/cronscripts/distributionmirror-prober.py
index 6d38880..513ba38 100755
--- a/cronscripts/distributionmirror-prober.py
+++ b/cronscripts/distributionmirror-prober.py
@@ -1,6 +1,6 @@
 #!/usr/bin/python -S
 #
-# Copyright 2009-2018 Canonical Ltd.  This software is licensed under the
+# Copyright 2009-2020 Canonical Ltd.  This software is licensed under the
 # GNU Affero General Public License version 3 (see the file LICENSE).
 
 """Script to probe distribution mirrors and check how up-to-date they are."""
@@ -38,6 +38,23 @@ class DistroMirrorProberScript(LaunchpadCronScript):
             dest='max_mirrors', default=None, action='store', type="int",
             help='Only probe N mirrors.')
 
+        # IMPORTANT: Don't change this unless you really know what you're
+        # doing. Using a too big value can cause spurious failures on lots of
+        # mirrors and a too small one can cause the prober to run for hours.
+        self.parser.add_option('--max-parallel-per-host',
+            dest='max_parallel_per_host', default=2,
+            action='store', type="int",
+            help='Keep maximum N parallel requests per host at a time.'
+                 ' (default=2)')
+
+        # We limit the overall number of simultaneous requests as well to
+        # prevent them from stalling and timing out before they even get a
+        # chance to start connecting.
+        self.parser.add_option('--max-parallel',
+            dest='max_parallel', default=100,
+            action='store', type="int",
+            help='Keep maximum N parallel requests at a time (default=100).')
+
     def main(self):
         if self.options.content_type == 'archive':
             content_type = MirrorContent.ARCHIVE
@@ -52,7 +69,8 @@ class DistroMirrorProberScript(LaunchpadCronScript):
             lambda: config.distributionmirrorprober.timeout)
         DistroMirrorProber(self.txn, self.logger).probe(
             content_type, self.options.no_remote_hosts, self.options.force,
-            self.options.max_mirrors, not self.options.no_owner_notification)
+            self.options.max_mirrors, not self.options.no_owner_notification,
+            self.options.max_parallel, self.options.max_parallel_per_host)
 
 
 if __name__ == '__main__':
diff --git a/lib/lp/registry/scripts/distributionmirror_prober.py b/lib/lp/registry/scripts/distributionmirror_prober.py
index 5b5d424..a6350ca 100644
--- a/lib/lp/registry/scripts/distributionmirror_prober.py
+++ b/lib/lp/registry/scripts/distributionmirror_prober.py
@@ -81,17 +81,6 @@ invalid_certificate_hosts = set()
 
 MAX_REDIRECTS = 3
 
-# Number of simultaneous requests we issue on a given host.
-# IMPORTANT: Don't change this unless you really know what you're doing. Using
-# a too big value can cause spurious failures on lots of mirrors and a too
-# small one can cause the prober to run for hours.
-PER_HOST_REQUESTS = 2
-
-# We limit the overall number of simultaneous requests as well to prevent
-# them from stalling and timing out before they even get a chance to
-# start connecting.
-OVERALL_REQUESTS = 100
-
 
 class LoggingMixin:
     """Common logging class for archive and releases mirror messages."""
@@ -110,12 +99,15 @@ class LoggingMixin:
 
 class RequestManager:
 
-    overall_semaphore = DeferredSemaphore(OVERALL_REQUESTS)
-
     # Yes, I want a mutable class attribute because I want changes done in an
     # instance to be visible in other instances as well.
     host_locks = {}
 
+    def __init__(self, max_parallel, max_parallel_per_host):
+        self.max_parallel = max_parallel
+        self.max_parallel_per_host = max_parallel_per_host
+        self.overall_semaphore = DeferredSemaphore(max_parallel)
+
     def run(self, host, probe_func):
         # Use a MultiLock with one semaphore limiting the overall
         # connections and another limiting the per-host connections.
@@ -123,7 +115,8 @@ class RequestManager:
             multi_lock = self.host_locks[host]
         else:
             multi_lock = MultiLock(
-                self.overall_semaphore, DeferredSemaphore(PER_HOST_REQUESTS))
+                self.overall_semaphore,
+                DeferredSemaphore(self.max_parallel_per_host))
             self.host_locks[host] = multi_lock
         return multi_lock.run(probe_func)
 
@@ -639,7 +632,7 @@ class ArchiveMirrorProberCallbacks(LoggingMixin):
         self.logMessage(msg)
         return mirror
 
-    def updateMirrorFreshness(self, arch_or_source_mirror):
+    def updateMirrorFreshness(self, arch_or_source_mirror, request_manager):
         """Update the freshness of this MirrorDistro{ArchSeries,SeriesSource}.
 
         This is done by issuing HTTP HEAD requests on that mirror looking for
@@ -664,7 +657,6 @@ class ArchiveMirrorProberCallbacks(LoggingMixin):
             self.deleteMethod(self.series, self.pocket, self.component)
             return
 
-        request_manager = RequestManager()
         deferredList = []
         # We start setting the freshness to unknown, and then we move on
         # trying to find one of the recently published packages mirrored
@@ -836,7 +828,8 @@ def checkComplete(result, key, unchecked_keys):
     return result
 
 
-def probe_archive_mirror(mirror, logfile, unchecked_keys, logger):
+def probe_archive_mirror(mirror, logfile, unchecked_keys, logger,
+                         max_parallel, max_parallel_per_host):
     """Probe an archive mirror for its contents and freshness.
 
     First we issue a set of HTTP HEAD requests on some key files to find out
@@ -850,7 +843,7 @@ def probe_archive_mirror(mirror, logfile, unchecked_keys, logger):
     packages_paths = mirror.getExpectedPackagesPaths()
     sources_paths = mirror.getExpectedSourcesPaths()
     all_paths = itertools.chain(packages_paths, sources_paths)
-    request_manager = RequestManager()
+    request_manager = RequestManager(max_parallel, max_parallel_per_host)
     for series, pocket, component, path in all_paths:
         url = urljoin(base_url, path)
         callbacks = ArchiveMirrorProberCallbacks(
@@ -864,13 +857,14 @@ def probe_archive_mirror(mirror, logfile, unchecked_keys, logger):
         deferred.addCallbacks(
             callbacks.ensureMirrorSeries, callbacks.deleteMirrorSeries)
 
-        deferred.addCallback(callbacks.updateMirrorFreshness)
+        deferred.addCallback(callbacks.updateMirrorFreshness, request_manager)
         deferred.addErrback(logger.error)
 
         deferred.addBoth(checkComplete, url, unchecked_keys)
 
 
-def probe_cdimage_mirror(mirror, logfile, unchecked_keys, logger):
+def probe_cdimage_mirror(mirror, logfile, unchecked_keys, logger,
+                         max_parallel, max_parallel_per_host):
     """Probe a cdimage mirror for its contents.
 
     This is done by checking the list of files for each flavour and series
@@ -900,7 +894,7 @@ def probe_cdimage_mirror(mirror, logfile, unchecked_keys, logger):
         mirror_key = (series, flavour)
         unchecked_keys.append(mirror_key)
         deferredList = []
-        request_manager = RequestManager()
+        request_manager = RequestManager(max_parallel, max_parallel_per_host)
         for path in paths:
             url = urljoin(base_url, path)
             # Use a RedirectAwareProberFactory because CD mirrors are allowed
@@ -971,9 +965,17 @@ class DistroMirrorProber:
         mirror.newProbeRecord(log_file)
 
     def probe(self, content_type, no_remote_hosts, ignore_last_probe,
-              max_mirrors, notify_owner):
+              max_mirrors, notify_owner, max_parallel=100,
+              max_parallel_per_host=2):
         """Probe distribution mirrors.
 
+        You should control carefully the parallelism here. Increasing too
+        much the number of max_parallel_per_host could make the mirrors take
+        too much to answer or deny our requests.
+
+        If we increase too much the max_parallel, we might experience timeouts
+        because of our production proxy or internet bandwidth.
+
         :param content_type: The type of mirrored content, as a
             `MirrorContent`.
         :param no_remote_hosts: If True, restrict access to localhost.
@@ -983,6 +985,10 @@ class DistroMirrorProber:
             no maximum.
         :param notify_owner: Send failure notification to the owners of the
             mirrors.
+        :param max_parallel: Maximum number of requests happening
+            simultaneously.
+        :param max_parallel_per_host: Maximum number of requests to the same
+            host happening simultaneously.
         """
         if content_type == MirrorContent.ARCHIVE:
             probe_function = probe_archive_mirror
@@ -1033,7 +1039,8 @@ class DistroMirrorProber:
             probed_mirrors.append(mirror)
             logfile = StringIO()
             logfiles[mirror_id] = logfile
-            probe_function(mirror, logfile, unchecked_keys, self.logger)
+            probe_function(mirror, logfile, unchecked_keys, self.logger,
+                           max_parallel, max_parallel_per_host)
 
         if probed_mirrors:
             reactor.run()
diff --git a/lib/lp/registry/tests/test_distributionmirror_prober.py b/lib/lp/registry/tests/test_distributionmirror_prober.py
index 82db551..c674ac0 100644
--- a/lib/lp/registry/tests/test_distributionmirror_prober.py
+++ b/lib/lp/registry/tests/test_distributionmirror_prober.py
@@ -56,8 +56,6 @@ from lp.registry.scripts.distributionmirror_prober import (
     MIN_REQUESTS_TO_CONSIDER_RATIO,
     MirrorCDImageProberCallbacks,
     MultiLock,
-    OVERALL_REQUESTS,
-    PER_HOST_REQUESTS,
     probe_archive_mirror,
     probe_cdimage_mirror,
     ProberFactory,
@@ -1052,7 +1050,7 @@ class TestProbeFunctionSemaphores(TestCase):
         # Note that calling this function won't actually probe any mirrors; we
         # need to call reactor.run() to actually start the probing.
         with default_timeout(15.0):
-            probe_cdimage_mirror(mirror, StringIO(), [], logging)
+            probe_cdimage_mirror(mirror, StringIO(), [], logging, 100, 2)
         self.assertEqual(0, len(mirror.cdimage_series))
 
     def test_archive_mirror_probe_function(self):
@@ -1081,43 +1079,45 @@ class TestProbeFunctionSemaphores(TestCase):
         The given probe_function must be either probe_cdimage_mirror or
         probe_archive_mirror.
         """
-        request_manager = RequestManager()
+        max_per_host_requests = 2
+        max_requests = 100
+        request_manager = RequestManager(max_requests, max_per_host_requests)
         mirror1_host = URI(mirror1.base_url).host
         mirror2_host = URI(mirror2.base_url).host
         mirror3_host = URI(mirror3.base_url).host
 
-        probe_function(mirror1, StringIO(), [], logging)
+        probe_function(mirror1, StringIO(), [], logging, 100, 2)
         # Since we have a single mirror to probe we need to have a single
-        # DeferredSemaphore with a limit of PER_HOST_REQUESTS, to ensure we
+        # DeferredSemaphore with a limit of max_per_host_requests, to ensure we
         # don't issue too many simultaneous connections on that host.
         self.assertEqual(len(request_manager.host_locks), 1)
         multi_lock = request_manager.host_locks[mirror1_host]
-        self.assertEqual(multi_lock.host_lock.limit, PER_HOST_REQUESTS)
+        self.assertEqual(multi_lock.host_lock.limit, max_per_host_requests)
         # Note that our multi_lock contains another semaphore to control the
         # overall number of requests.
-        self.assertEqual(multi_lock.overall_lock.limit, OVERALL_REQUESTS)
+        self.assertEqual(multi_lock.overall_lock.limit, max_requests)
 
-        probe_function(mirror2, StringIO(), [], logging)
+        probe_function(mirror2, StringIO(), [], logging, 100, 2)
         # Now we have two mirrors to probe, but they have the same hostname,
         # so we'll still have a single semaphore in host_semaphores.
         self.assertEqual(mirror2_host, mirror1_host)
         self.assertEqual(len(request_manager.host_locks), 1)
         multi_lock = request_manager.host_locks[mirror2_host]
-        self.assertEqual(multi_lock.host_lock.limit, PER_HOST_REQUESTS)
+        self.assertEqual(multi_lock.host_lock.limit, max_per_host_requests)
 
-        probe_function(mirror3, StringIO(), [], logging)
+        probe_function(mirror3, StringIO(), [], logging, 100, 2)
         # This third mirror is on a separate host, so we'll have a second
         # semaphore added to host_semaphores.
         self.assertTrue(mirror3_host != mirror1_host)
         self.assertEqual(len(request_manager.host_locks), 2)
         multi_lock = request_manager.host_locks[mirror3_host]
-        self.assertEqual(multi_lock.host_lock.limit, PER_HOST_REQUESTS)
+        self.assertEqual(multi_lock.host_lock.limit, max_per_host_requests)
 
         # When using an http_proxy, even though we'll actually connect to the
         # proxy, we'll use the mirror's host as the key to find the semaphore
         # that should be used
         self.pushConfig('launchpad', http_proxy='http://squid.internal:3128/')
-        probe_function(mirror3, StringIO(), [], logging)
+        probe_function(mirror3, StringIO(), [], logging, 100, 2)
         self.assertEqual(len(request_manager.host_locks), 2)