Merge lp:~cjwatson/launchpad/buildmaster-twisted-agent into lp:launchpad

[Colin Watson]

Convert BuilderSlave to twisted.web.client.Agent, causing it to use a connection pool rather than trying to download everything at once.

Convert BuilderSlave to twisted.web.client.Agent, causing it to use a connection pool rather than trying to download everything at once. The old approach could overflow open-file limits.

We'll need to give this a good workout on dogfood.

[Celso Providelo (cprov)]

Thanks Colin, great solution!

From http://twistedmatrix.com/documents/15.2.0/web/howto/client.html#multiple-connections-to-the-same-server it seems that it will sustain at most 2 simultaneous downloads from each slave. Is that reasonable and sufficient ?

review: Approve

[William Grant (wgrant)]

review: Approve (code)

=== modified file 'daemons/buildd-manager.tac'
--- daemons/buildd-manager.tac	2011-12-29 05:29:36 +0000
+++ daemons/buildd-manager.tac	2016-05-25 15:34:01 +0000
@@ -1,14 +1,18 @@
-# Copyright 2009-2011 Canonical Ltd.  This software is licensed under the
+# Copyright 2009-2016 Canonical Ltd.  This software is licensed under the
 # GNU Affero General Public License version 3 (see the file LICENSE).
 
 # Twisted Application Configuration file.
 # Use with "twistd2.4 -y <file.tac>", e.g. "twistd -noy server.tac"
 
+import resource
+
 from twisted.application import service
 from twisted.scripts.twistd import ServerOptions
-from twisted.web import server
 
-from lp.services.config import dbconfig
+from lp.services.config import (
+    config,
+    dbconfig,
+    )
 from lp.services.daemons import readyservice
 from lp.services.scripts import execute_zcml_for_scripts
 from lp.buildmaster.manager import BuilddManager
@@ -21,6 +25,12 @@
 # Should be removed from callsites verified to not need it.
 set_immediate_mail_delivery(True)
 
+# Allow generous slack for database connections, idle download connections,
+# etc.
+soft_nofile = config.builddmaster.download_connections + 1024
+_, hard_nofile = resource.getrlimit(resource.RLIMIT_NOFILE)
+resource.setrlimit(resource.RLIMIT_NOFILE, (soft_nofile, hard_nofile))
+
 options = ServerOptions()
 options.parseOptions()
 
@@ -34,4 +44,3 @@
 # Service for scanning buildd slaves.
 service = BuilddManager()
 service.setServiceParent(application)
-
 
=== modified file 'lib/lp/buildmaster/interactor.py'
--- lib/lp/buildmaster/interactor.py	2015-02-17 05:38:37 +0000
+++ lib/lp/buildmaster/interactor.py	2016-05-25 15:34:01 +0000
@@ -1,4 +1,4 @@
-# Copyright 2009-2014 Canonical Ltd.  This software is licensed under the
+# Copyright 2009-2016 Canonical Ltd.  This software is licensed under the
 # GNU Affero General Public License version 3 (see the file LICENSE).
 
 __metaclass__ = type
@@ -13,9 +13,17 @@
 from urlparse import urlparse
 
 import transaction
-from twisted.internet import defer
+from twisted.internet import (
+    defer,
+    reactor as default_reactor,
+    )
+from twisted.internet.protocol import Protocol
 from twisted.web import xmlrpc
-from twisted.web.client import downloadPage
+from twisted.web.client import (
+    Agent,
+    HTTPConnectionPool,
+    ResponseDone,
+    )
 from zope.security.proxy import (
     isinstance as zope_isinstance,
     removeSecurityProxy,
@@ -46,6 +54,89 @@
     noisy = False
 
 
+class FileWritingProtocol(Protocol):
+    """A protocol that saves data to a file."""
+
+    def __init__(self, finished, filename):
+        self.finished = finished
+        self.filename = filename
+        self.file = None
+
+    def dataReceived(self, data):
+        if self.file is None:
+            self.file = open(self.filename, "wb")
+        try:
+            self.file.write(data)
+        except IOError:
+            try:
+                self.file.close()
+            except IOError:
+                pass
+            self.file = None
+            self.finished.errback()
+
+    def connectionLost(self, reason):
+        try:
+            self.file.close()
+        except IOError:
+            self.finished.errback()
+        else:
+            if reason.check(ResponseDone):
+                self.finished.callback(None)
+            else:
+                self.finished.errback(reason)
+
+
+class LimitedHTTPConnectionPool(HTTPConnectionPool):
+    """A connection pool with an upper limit on open connections."""
+
+    # XXX cjwatson 2016-05-25: This actually only limits active connections,
+    # and doesn't count idle but open connections towards the limit; this is
+    # because it's very difficult to do the latter with HTTPConnectionPool's
+    # current design.  Users of this pool must therefore expect some
+    # additional file descriptors to be open for idle connections.
+
+    def __init__(self, reactor, limit, persistent=True):
+        super(LimitedHTTPConnectionPool, self).__init__(
+            reactor, persistent=persistent)
+        self._semaphore = defer.DeferredSemaphore(limit)
+
+    def getConnection(self, key, endpoint):
+        d = self._semaphore.acquire()
+        d.addCallback(
+            lambda _: super(LimitedHTTPConnectionPool, self).getConnection(
+                key, endpoint))
+        return d
+
+    def _putConnection(self, key, connection):
+        super(LimitedHTTPConnectionPool, self)._putConnection(key, connection)
+        # Only release the semaphore in the next main loop iteration; if we
+        # release it here then the next request may start using this
+        # connection's parser before this request has quite finished with
+        # it.
+        self._reactor.callLater(0, self._semaphore.release)
+
+
+_default_pool = None
+
+
+def default_pool(reactor=None):
+    global _default_pool
+    if reactor is None:
+        reactor = default_reactor
+    if _default_pool is None:
+        # Circular import.
+        from lp.buildmaster.manager import SlaveScanner
+        # Short cached connection timeout to avoid potential weirdness with
+        # virtual builders that reboot frequently.
+        _default_pool = LimitedHTTPConnectionPool(
+            reactor, config.builddmaster.download_connections)
+        _default_pool.maxPersistentPerHost = (
+            config.builddmaster.idle_download_connections_per_builder)
+        _default_pool.cachedConnectionTimeout = SlaveScanner.SCAN_INTERVAL
+    return _default_pool
+
+
 class BuilderSlave(object):
     """Add in a few useful methods for the XMLRPC slave.
 
@@ -58,7 +149,7 @@
     # many false positives in your test run and will most likely break
     # production.
 
-    def __init__(self, proxy, builder_url, vm_host, timeout, reactor):
+    def __init__(self, proxy, builder_url, vm_host, timeout, reactor, pool):
         """Initialize a BuilderSlave.
 
         :param proxy: An XML-RPC proxy, implementing 'callRemote'. It must
@@ -71,11 +162,16 @@
         self._file_cache_url = urlappend(builder_url, 'filecache')
         self._server = proxy
         self.timeout = timeout
+        if reactor is None:
+            reactor = default_reactor
         self.reactor = reactor
+        if pool is None:
+            pool = default_pool(reactor=reactor)
+        self.pool = pool
 
     @classmethod
     def makeBuilderSlave(cls, builder_url, vm_host, timeout, reactor=None,
-                         proxy=None):
+                         proxy=None, pool=None):
         """Create and return a `BuilderSlave`.
 
         :param builder_url: The URL of the slave buildd machine,
@@ -84,6 +180,7 @@
             here.
         :param reactor: Used by tests to override the Twisted reactor.
         :param proxy: Used By tests to override the xmlrpc.Proxy.
+        :param pool: Used by tests to override the HTTPConnectionPool.
         """
         rpc_url = urlappend(builder_url.encode('utf-8'), 'rpc')
         if proxy is None:
@@ -92,7 +189,7 @@
             server_proxy.queryFactory = QuietQueryFactory
         else:
             server_proxy = proxy
-        return cls(server_proxy, builder_url, vm_host, timeout, reactor)
+        return cls(server_proxy, builder_url, vm_host, timeout, reactor, pool)
 
     def _with_timeout(self, d, timeout=None):
         return cancel_on_timeout(d, timeout or self.timeout, self.reactor)
@@ -138,10 +235,15 @@
             errback with the error string.
         """
         file_url = urlappend(self._file_cache_url, sha_sum).encode('utf8')
-        # If desired we can pass a param "timeout" here but let's leave
-        # it at the default value if it becomes obvious we need to
-        # change it.
-        return downloadPage(file_url, file_to_write, followRedirect=0)
+        d = Agent(self.reactor, pool=self.pool).request("GET", file_url)
+
+        def got_response(response):
+            finished = defer.Deferred()
+            response.deliverBody(FileWritingProtocol(finished, file_to_write))
+            return finished
+
+        d.addCallback(got_response)
+        return d
 
     def getFiles(self, files):
         """Fetch many files from the builder.
 
=== modified file 'lib/lp/buildmaster/tests/mock_slaves.py'
--- lib/lp/buildmaster/tests/mock_slaves.py	2015-09-28 17:38:45 +0000
+++ lib/lp/buildmaster/tests/mock_slaves.py	2016-05-25 15:34:01 +0000
@@ -1,4 +1,4 @@
-# Copyright 2009-2015 Canonical Ltd.  This software is licensed under the
+# Copyright 2009-2016 Canonical Ltd.  This software is licensed under the
 # GNU Affero General Public License version 3 (see the file LICENSE).
 
 """Mock Build objects for tests soyuz buildd-system."""
@@ -290,14 +290,14 @@
                 lambda: open(tachandler.logfile, 'r').readlines()))
         return tachandler
 
-    def getClientSlave(self, reactor=None, proxy=None):
+    def getClientSlave(self, reactor=None, proxy=None, pool=None):
         """Return a `BuilderSlave` for use in testing.
 
         Points to a fixed URL that is also used by `BuilddSlaveTestSetup`.
         """
         return BuilderSlave.makeBuilderSlave(
             self.BASE_URL, 'vmhost', config.builddmaster.socket_timeout,
-            reactor, proxy)
+            reactor=reactor, proxy=proxy, pool=pool)
 
     def makeCacheFile(self, tachandler, filename):
         """Make a cache file available on the remote slave.
 
=== modified file 'lib/lp/buildmaster/tests/test_interactor.py'
--- lib/lp/buildmaster/tests/test_interactor.py	2015-11-04 14:30:09 +0000
+++ lib/lp/buildmaster/tests/test_interactor.py	2016-05-25 15:34:01 +0000
@@ -1,4 +1,4 @@
-# Copyright 2009-2015 Canonical Ltd.  This software is licensed under the
+# Copyright 2009-2016 Canonical Ltd.  This software is licensed under the
 # GNU Affero General Public License version 3 (see the file LICENSE).
 
 """Test BuilderInteractor features."""
@@ -20,9 +20,16 @@
     AsynchronousDeferredRunTestForBrokenTwisted,
     SynchronousDeferredRunTest,
     )
-from testtools.matchers import ContainsAll
+from testtools.matchers import (
+    ContainsAll,
+    HasLength,
+    MatchesDict,
+    )
 from testtools.testcase import ExpectedException
-from twisted.internet import defer
+from twisted.internet import (
+    defer,
+    reactor as default_reactor,
+    )
 from twisted.internet.task import Clock
 from twisted.python.failure import Failure
 from twisted.web.client import getPage
@@ -38,6 +45,7 @@
     BuilderInteractor,
     BuilderSlave,
     extract_vitals_from_db,
+    LimitedHTTPConnectionPool,
     )
 from lp.buildmaster.interfaces.builder import (
     BuildDaemonIsolationError,
@@ -789,6 +797,7 @@
             for sha1, local_file in files:
                 with open(local_file) as f:
                     self.assertEqual(content_map[sha1], f.read())
+            return slave.pool.closeCachedConnections()
 
         def finished_uploading(ignored):
             d = slave.getFiles(files)
@@ -812,3 +821,44 @@
             dl.append(d)
 
         return defer.DeferredList(dl).addCallback(finished_uploading)
+
+    def test_getFiles_open_connections(self):
+        # getFiles honours the configured limit on active download
+        # connections.
+        pool = LimitedHTTPConnectionPool(default_reactor, 2)
+        contents = [self.factory.getUniqueString() for _ in range(10)]
+        self.slave_helper.getServerSlave()
+        slave = self.slave_helper.getClientSlave(pool=pool)
+        files = []
+        content_map = {}
+
+        def got_files(ignored):
+            # Called back when getFiles finishes.  Make sure all the
+            # content is as expected.
+            for sha1, local_file in files:
+                with open(local_file) as f:
+                    self.assertEqual(content_map[sha1], f.read())
+            # Only two connections were used.
+            self.assertThat(
+                slave.pool._connections,
+                MatchesDict({("http", "localhost", 8221): HasLength(2)}))
+            return slave.pool.closeCachedConnections()
+
+        def finished_uploading(ignored):
+            d = slave.getFiles(files)
+            return d.addCallback(got_files)
+
+        # Set up some files on the builder and store details in
+        # content_map so we can compare downloads later.
+        dl = []
+        for content in contents:
+            filename = content + '.txt'
+            lf = self.factory.makeLibraryFileAlias(filename, content=content)
+            content_map[lf.content.sha1] = content
+            files.append((lf.content.sha1, tempfile.mkstemp()[1]))
+            self.addCleanup(os.remove, files[-1][1])
+            self.layer.txn.commit()
+            d = slave.ensurepresent(lf.content.sha1, lf.http_url, "", "")
+            dl.append(d)
+
+        return defer.DeferredList(dl).addCallback(finished_uploading)
 
=== modified file 'lib/lp/services/config/schema-lazr.conf'
--- lib/lp/services/config/schema-lazr.conf	2016-05-18 03:59:44 +0000
+++ lib/lp/services/config/schema-lazr.conf	2016-05-25 15:34:01 +0000
@@ -72,6 +72,14 @@
 # datatype: integer
 virtualized_socket_timeout: 30
 
+# The maximum number of idle file download connections per builder that
+# may be kept open.
+idle_download_connections_per_builder: 10
+
+# The maximum number of file download connections that may be open
+# across all builders.
+download_connections: 2048
+
 # Activate the Build Notification system.
 # datatype: boolean
 send_build_notification: True