Merge ~cjwatson/launchpad:buildmaster-fetch-in-process into launchpad:master

[Colin Watson]

Rewrite buildd-manager file fetching using subprocesses

This is similar to commit 8a6418c886433597e8aa113f53bd0b7dad397df2, but using a process pool (via `ampoule`) rather than a thread pool. The threaded approach did a poor job of servicing established connections when we deployed in on production, but it's possible that we were running into trouble with Python's GIL, so let's see if subprocesses will fare any better.

The new process pool is only used if the feature rule "buildmaster.download_in_subprocess" is set, so we can easily switch between the old and the new behaviour until we're confident.

[William Grant (wgrant)]

review: Approve (code)

f9fe0ee... by Colin Watson on 2020-06-10

Add comment about subprocess download strategy

diff --git a/lib/lp/buildmaster/downloader.py b/lib/lp/buildmaster/downloader.py
new file mode 100644
index 0000000..9da2b89
--- /dev/null
+++ b/lib/lp/buildmaster/downloader.py
@@ -0,0 +1,66 @@
+# Copyright 2020 Canonical Ltd.  This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""Download subprocess support for buildd-manager.
+
+To minimise subprocess memory use, this intentionally avoids importing
+anything from the rest of Launchpad.
+"""
+
+from __future__ import absolute_import, print_function, unicode_literals
+
+__metaclass__ = type
+__all__ = [
+    'DownloadCommand',
+    'DownloadProcess',
+    ]
+
+import os.path
+import tempfile
+
+from ampoule.child import AMPChild
+from requests import (
+    RequestException,
+    Session,
+    )
+from requests_toolbelt.downloadutils import stream
+from requests_toolbelt.exceptions import StreamingError
+from twisted.protocols import amp
+
+
+class DownloadCommand(amp.Command):
+
+    arguments = [
+        (b"file_url", amp.Unicode()),
+        (b"path_to_write", amp.Unicode()),
+        (b"timeout", amp.Integer()),
+        ]
+    response = []
+    errors = {
+        RequestException: b"REQUEST_ERROR",
+        StreamingError: b"STREAMING_ERROR",
+        }
+
+
+class DownloadProcess(AMPChild):
+    """A subprocess that downloads a file to disk."""
+
+    @DownloadCommand.responder
+    def downloadCommand(self, file_url, path_to_write, timeout):
+        session = Session()
+        session.trust_env = False
+        response = session.get(file_url, timeout=timeout, stream=True)
+        response.raise_for_status()
+        f = tempfile.NamedTemporaryFile(
+            mode="wb", prefix=os.path.basename(path_to_write) + "_",
+            dir=os.path.dirname(path_to_write), delete=False)
+        try:
+            stream.stream_response_to_file(response, path=f)
+        except Exception:
+            f.close()
+            os.unlink(f.name)
+            raise
+        else:
+            f.close()
+            os.rename(f.name, path_to_write)
+        return {}
diff --git a/lib/lp/buildmaster/interactor.py b/lib/lp/buildmaster/interactor.py
index 7618da5..6d4ee49 100644
--- a/lib/lp/buildmaster/interactor.py
+++ b/lib/lp/buildmaster/interactor.py
@@ -15,6 +15,7 @@ import sys
 import tempfile
 import traceback
 
+from ampoule.pool import ProcessPool
 from six.moves.urllib.parse import urlparse
 import transaction
 from twisted.internet import (
@@ -33,6 +34,10 @@ from zope.security.proxy import (
     removeSecurityProxy,
     )
 
+from lp.buildmaster.downloader import (
+    DownloadCommand,
+    DownloadProcess,
+    )
 from lp.buildmaster.enums import (
     BuilderCleanStatus,
     BuilderResetProtocol,
@@ -47,6 +52,11 @@ from lp.buildmaster.interfaces.buildfarmjobbehaviour import (
     IBuildFarmJobBehaviour,
     )
 from lp.services.config import config
+from lp.services.features import getFeatureFlag
+from lp.services.job.runner import (
+    QuietAMPConnector,
+    VirtualEnvProcessStarter,
+    )
 from lp.services.twistedsupport import cancel_on_timeout
 from lp.services.twistedsupport.processmonitor import ProcessWithTimeout
 from lp.services.webapp import urlappend
@@ -128,6 +138,8 @@ class LimitedHTTPConnectionPool(HTTPConnectionPool):
 
 
 _default_pool = None
+_default_process_pool = None
+_default_process_pool_shutdown = None
 
 
 def default_pool(reactor=None):
@@ -147,6 +159,44 @@ def default_pool(reactor=None):
     return _default_pool
 
 
+def make_download_process_pool(**kwargs):
+    """Make a pool of processes for downloading files."""
+    env = {"PATH": os.environ["PATH"]}
+    if "LPCONFIG" in os.environ:
+        env["LPCONFIG"] = os.environ["LPCONFIG"]
+    starter = VirtualEnvProcessStarter(env=env)
+    starter.connectorFactory = QuietAMPConnector
+    kwargs = dict(kwargs)
+    kwargs.setdefault("max", config.builddmaster.download_connections)
+    return ProcessPool(DownloadProcess, starter=starter, **kwargs)
+
+
+def default_process_pool(reactor=None):
+    global _default_process_pool, _default_process_pool_shutdown
+    if reactor is None:
+        reactor = default_reactor
+    if _default_process_pool is None:
+        _default_process_pool = make_download_process_pool()
+        _default_process_pool.start()
+        shutdown_id = reactor.addSystemEventTrigger(
+            "during", "shutdown", _default_process_pool.stop)
+        _default_process_pool_shutdown = (reactor, shutdown_id)
+    return _default_process_pool
+
+
[e-mail]
+def shut_down_default_process_pool():
+    """Shut down the default process pool.  Used in test cleanup."""
+    global _default_process_pool, _default_process_pool_shutdown
+    if _default_process_pool is not None:
+        yield _default_process_pool.stop()
+        _default_process_pool = None
+    if _default_process_pool_shutdown is not None:
+        reactor, shutdown_id = _default_process_pool_shutdown
+        reactor.removeSystemEventTrigger(shutdown_id)
+        _default_process_pool_shutdown = None
+
+
 class BuilderSlave(object):
     """Add in a few useful methods for the XMLRPC slave.
 
@@ -159,7 +209,8 @@ class BuilderSlave(object):
     # many false positives in your test run and will most likely break
     # production.
 
-    def __init__(self, proxy, builder_url, vm_host, timeout, reactor, pool):
+    def __init__(self, proxy, builder_url, vm_host, timeout, reactor,
+                 pool=None, process_pool=None):
         """Initialize a BuilderSlave.
 
         :param proxy: An XML-RPC proxy, implementing 'callRemote'. It must
@@ -175,13 +226,21 @@ class BuilderSlave(object):
         if reactor is None:
             reactor = default_reactor
         self.reactor = reactor
+        self._download_in_subprocess = bool(
+            getFeatureFlag('buildmaster.download_in_subprocess'))
         if pool is None:
             pool = default_pool(reactor=reactor)
         self.pool = pool
+        if self._download_in_subprocess:
+            if process_pool is None:
+                process_pool = default_process_pool(reactor=reactor)
+            self.process_pool = process_pool
+        else:
+            self.process_pool = None
 
     @classmethod
     def makeBuilderSlave(cls, builder_url, vm_host, timeout, reactor=None,
-                         proxy=None, pool=None):
+                         proxy=None, pool=None, process_pool=None):
         """Create and return a `BuilderSlave`.
 
         :param builder_url: The URL of the slave buildd machine,
@@ -191,6 +250,7 @@ class BuilderSlave(object):
         :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.
+        :param process_pool: Used by tests to override the ProcessPool.
         """
         rpc_url = urlappend(builder_url.encode('utf-8'), 'rpc')
         if proxy is None:
@@ -199,7 +259,9 @@ class BuilderSlave(object):
             server_proxy.queryFactory = QuietQueryFactory
         else:
             server_proxy = proxy
-        return cls(server_proxy, builder_url, vm_host, timeout, reactor, pool)
+        return cls(
+            server_proxy, builder_url, vm_host, timeout, reactor,
+            pool=pool, process_pool=process_pool)
 
     def _with_timeout(self, d, timeout=None):
         return cancel_on_timeout(d, timeout or self.timeout, self.reactor)
@@ -251,11 +313,29 @@ class BuilderSlave(object):
         """
         file_url = self.getURL(sha_sum)
         try:
-            response = yield Agent(self.reactor, pool=self.pool).request(
-                "GET", file_url)
-            finished = defer.Deferred()
-            response.deliverBody(FileWritingProtocol(finished, path_to_write))
-            yield finished
+            # Select download behaviour according to the
+            # buildmaster.download_in_subprocess feature rule: if enabled,
+            # defer the download to a subprocess; if disabled, download the
+            # file asynchronously in Twisted.  We've found that in practice
+            # the asynchronous approach only works well up to a bit over a
+            # hundred builders, and beyond that it struggles to keep up with
+            # incoming packets in time to avoid TCP timeouts (perhaps
+            # because of too much synchronous work being done on the reactor
+            # thread).  The exact reason for this is as yet unproven, so we
+            # use a feature rule to allow us to try out different
+            # approaches.
+            if self._download_in_subprocess:
+                yield self.process_pool.doWork(
+                    DownloadCommand,
+                    file_url=file_url, path_to_write=path_to_write,
+                    timeout=self.timeout)
+            else:
+                response = yield Agent(self.reactor, pool=self.pool).request(
+                    "GET", file_url)
+                finished = defer.Deferred()
+                response.deliverBody(
+                    FileWritingProtocol(finished, path_to_write))
+                yield finished
             if logger is not None:
                 logger.info("Grabbed %s" % file_url)
         except Exception as e:
diff --git a/lib/lp/buildmaster/tests/mock_slaves.py b/lib/lp/buildmaster/tests/mock_slaves.py
index 3a42f8c..9c99137 100644
--- a/lib/lp/buildmaster/tests/mock_slaves.py
+++ b/lib/lp/buildmaster/tests/mock_slaves.py
@@ -315,14 +315,15 @@ class SlaveTestHelpers(fixtures.Fixture):
                 lambda: open(tachandler.logfile, 'r').readlines()))
         return tachandler
 
-    def getClientSlave(self, reactor=None, proxy=None, pool=None):
+    def getClientSlave(self, reactor=None, proxy=None,
+                       pool=None, process_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=reactor, proxy=proxy, pool=pool)
+            reactor=reactor, proxy=proxy, pool=pool, process_pool=process_pool)
 
     def makeCacheFile(self, tachandler, filename, contents=b'something'):
         """Make a cache file available on the remote slave.
diff --git a/lib/lp/buildmaster/tests/test_buildfarmjobbehaviour.py b/lib/lp/buildmaster/tests/test_buildfarmjobbehaviour.py
index cc77f56..f837bb0 100644
--- a/lib/lp/buildmaster/tests/test_buildfarmjobbehaviour.py
+++ b/lib/lp/buildmaster/tests/test_buildfarmjobbehaviour.py
@@ -13,6 +13,7 @@ import os
 import shutil
 import tempfile
 
+from testscenarios import WithScenarios
 from testtools import ExpectedException
 from testtools.twistedsupport import AsynchronousDeferredRunTest
 from twisted.internet import defer
@@ -24,7 +25,10 @@ from lp.buildmaster.enums import (
     BuildBaseImageType,
     BuildStatus,
     )
-from lp.buildmaster.interactor import BuilderInteractor
+from lp.buildmaster.interactor import (
+    BuilderInteractor,
+    shut_down_default_process_pool,
+    )
 from lp.buildmaster.interfaces.builder import BuildDaemonError
 from lp.buildmaster.interfaces.buildfarmjobbehaviour import (
     IBuildFarmJobBehaviour,
@@ -40,6 +44,7 @@ from lp.buildmaster.tests.mock_slaves import (
     )
 from lp.registry.interfaces.pocket import PackagePublishingPocket
 from lp.services.config import config
+from lp.services.features.testing import FeatureFixture
 from lp.services.log.logger import BufferLogger
 from lp.soyuz.interfaces.binarypackagebuild import IBinaryPackageBuildSet
 from lp.testing import (
@@ -316,8 +321,17 @@ class TestVerifySuccessfulBuildMixin:
         self.assertRaises(AssertionError, behaviour.verifySuccessfulBuild)
 
 
-class TestHandleStatusMixin:
-    """Tests for `IPackageBuild`s handleStatus method."""
+class TestHandleStatusMixin(WithScenarios):
+    """Tests for `IPackageBuild`s handleStatus method.
+
+    This should be run in a test file with
+    `load_tests = load_tests_apply_scenarios`.
+    """
+
+    scenarios = [
+        ('download_in_twisted', {'download_in_subprocess': False}),
+        ('download_in_subprocess', {'download_in_subprocess': True}),
+        ]
 
     layer = LaunchpadZopelessLayer
     run_tests_with = AsynchronousDeferredRunTest.make_factory(timeout=30)
@@ -328,6 +342,9 @@ class TestHandleStatusMixin:
 
     def setUp(self):
         super(TestHandleStatusMixin, self).setUp()
+        if self.download_in_subprocess:
+            self.useFixture(FeatureFixture(
+                {'buildmaster.download_in_subprocess': 'on'}))
         self.factory = LaunchpadObjectFactory()
         self.build = self.makeBuild()
         # For the moment, we require a builder for the build so that
@@ -339,6 +356,7 @@ class TestHandleStatusMixin:
         self.interactor = BuilderInteractor()
         self.behaviour = self.interactor.getBuildBehaviour(
             self.build.buildqueue_record, self.builder, self.slave)
+        self.addCleanup(shut_down_default_process_pool)
 
         # We overwrite the buildmaster root to use a temp directory.
         tempdir = tempfile.mkdtemp()
diff --git a/lib/lp/buildmaster/tests/test_interactor.py b/lib/lp/buildmaster/tests/test_interactor.py
index 76b958c..0c4cf14 100644
--- a/lib/lp/buildmaster/tests/test_interactor.py
+++ b/lib/lp/buildmaster/tests/test_interactor.py
@@ -18,6 +18,10 @@ import tempfile
 from lpbuildd.slave import BuilderStatus
 from lpbuildd.tests.harness import BuilddSlaveTestSetup
 from six.moves import xmlrpc_client
+from testscenarios import (
+    load_tests_apply_scenarios,
+    WithScenarios,
+    )
 from testtools.matchers import (
     ContainsAll,
     HasLength,
@@ -49,6 +53,8 @@ from lp.buildmaster.interactor import (
     BuilderSlave,
     extract_vitals_from_db,
     LimitedHTTPConnectionPool,
+    make_download_process_pool,
+    shut_down_default_process_pool,
     )
 from lp.buildmaster.interfaces.builder import (
     BuildDaemonIsolationError,
@@ -67,6 +73,7 @@ from lp.buildmaster.tests.mock_slaves import (
     WaitingSlave,
     )
 from lp.services.config import config
+from lp.services.features.testing import FeatureFixture
 from lp.services.twistedsupport.testing import TReqFixture
 from lp.services.twistedsupport.treq import check_status
 from lp.soyuz.model.binarypackagebuildbehaviour import (
@@ -123,6 +130,10 @@ class TestBuilderInteractor(TestCase):
 
     run_tests_with = AsynchronousDeferredRunTest.make_factory(timeout=10)
 
+    def setUp(self):
+        super(TestBuilderInteractor, self).setUp()
+        self.addCleanup(shut_down_default_process_pool)
+
     def test_extractBuildStatus_baseline(self):
         # extractBuildStatus picks the name of the build status out of a
         # dict describing the slave's status.
@@ -478,6 +489,7 @@ class TestSlave(TestCase):
     def setUp(self):
         super(TestSlave, self).setUp()
         self.slave_helper = self.useFixture(SlaveTestHelpers())
+        self.addCleanup(shut_down_default_process_pool)
 
     def test_abort(self):
         slave = self.slave_helper.getClientSlave()
@@ -679,6 +691,7 @@ class TestSlaveTimeouts(TestCase):
         self.proxy = DeadProxy("url")
         self.slave = self.slave_helper.getClientSlave(
             reactor=self.clock, proxy=self.proxy)
+        self.addCleanup(shut_down_default_process_pool)
 
     def assertCancelled(self, d, timeout=None):
         self.clock.advance((timeout or config.builddmaster.socket_timeout) + 1)
@@ -719,6 +732,7 @@ class TestSlaveConnectionTimeouts(TestCase):
         super(TestSlaveConnectionTimeouts, self).setUp()
         self.slave_helper = self.useFixture(SlaveTestHelpers())
         self.clock = Clock()
+        self.addCleanup(shut_down_default_process_pool)
 
     def tearDown(self):
         # We need to remove any DelayedCalls that didn't actually get called.
@@ -744,16 +758,26 @@ class TestSlaveConnectionTimeouts(TestCase):
         return assert_fails_with(d, defer.CancelledError)
 
 
-class TestSlaveWithLibrarian(TestCaseWithFactory):
+class TestSlaveWithLibrarian(WithScenarios, TestCaseWithFactory):
     """Tests that need more of Launchpad to run."""
 
+    scenarios = [
+        ('download_in_twisted', {'download_in_subprocess': False}),
+        ('download_in_subprocess', {'download_in_subprocess': True}),
+        ]
+
     layer = LaunchpadZopelessLayer
     run_tests_with = AsynchronousDeferredRunTestForBrokenTwisted.make_factory(
         timeout=20)
 
     def setUp(self):
         super(TestSlaveWithLibrarian, self).setUp()
+        if self.download_in_subprocess:
+            self.useFixture(FeatureFixture(
+                {'buildmaster.download_in_subprocess': 'on'}))
         self.slave_helper = self.useFixture(SlaveTestHelpers())
+        if self.download_in_subprocess:
+            self.addCleanup(shut_down_default_process_pool)
 
     def test_ensurepresent_librarian(self):
         # ensurepresent, when given an http URL for a file will download the
@@ -835,10 +859,17 @@ class TestSlaveWithLibrarian(TestCaseWithFactory):
     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)
+        pool = LimitedHTTPConnectionPool(default_reactor, 2)
+        if self.download_in_subprocess:
+            process_pool = make_download_process_pool(min=1, max=2)
+            process_pool.start()
+            self.addCleanup(process_pool.stop)
+        else:
+            process_pool = None
+        slave = self.slave_helper.getClientSlave(
+            pool=pool, process_pool=process_pool)
         files = []
         content_map = {}
 
@@ -848,11 +879,15 @@ class TestSlaveWithLibrarian(TestCaseWithFactory):
             for sha1, local_file in files:
                 with open(local_file) as f:
                     self.assertEqual(content_map[sha1], f.read())
-            # Only two connections were used.
             port = BuilddSlaveTestSetup().daemon_port
-            self.assertThat(
-                slave.pool._connections,
-                MatchesDict({("http", "localhost", port): HasLength(2)}))
+            if self.download_in_subprocess:
+                # Only two workers were used.
+                self.assertEqual(2, len(process_pool.processes))
+            else:
+                # Only two connections were used.
+                self.assertThat(
+                    slave.pool._connections,
+                    MatchesDict({("http", "localhost", port): HasLength(2)}))
             return slave.pool.closeCachedConnections()
 
         def finished_uploading(ignored):
@@ -887,3 +922,6 @@ class TestSlaveWithLibrarian(TestCaseWithFactory):
         with open(temp_name) as f:
             self.assertEqual(b'', f.read())
         yield slave.pool.closeCachedConnections()
+
+
+load_tests = load_tests_apply_scenarios
diff --git a/lib/lp/buildmaster/tests/test_manager.py b/lib/lp/buildmaster/tests/test_manager.py
index 97de637..b0d56ed 100644
--- a/lib/lp/buildmaster/tests/test_manager.py
+++ b/lib/lp/buildmaster/tests/test_manager.py
@@ -37,6 +37,7 @@ from lp.buildmaster.interactor import (
     BuilderInteractor,
     BuilderSlave,
     extract_vitals_from_db,
+    shut_down_default_process_pool,
     )
 from lp.buildmaster.interfaces.builder import (
     BuildDaemonIsolationError,
@@ -116,6 +117,7 @@ class TestSlaveScannerScan(TestCaseWithFactory):
         hoary = ubuntu.getSeries('hoary')
         self.test_publisher.setUpDefaultDistroSeries(hoary)
         self.test_publisher.addFakeChroots(db_only=True)
+        self.addCleanup(shut_down_default_process_pool)
 
     def _resetBuilder(self, builder):
         """Reset the given builder and its job."""
@@ -653,6 +655,10 @@ class TestSlaveScannerWithLibrarian(TestCaseWithFactory):
     layer = LaunchpadZopelessLayer
     run_tests_with = AsynchronousDeferredRunTest.make_factory(timeout=20)
 
+    def setUp(self):
+        super(TestSlaveScannerWithLibrarian, self).setUp()
+        self.addCleanup(shut_down_default_process_pool)
+
     @defer.inlineCallbacks
     def test_end_to_end(self):
         # Test that SlaveScanner.scan() successfully finds, dispatches,
@@ -884,6 +890,10 @@ class TestSlaveScannerWithoutDB(TestCase):
 
     run_tests_with = AsynchronousDeferredRunTest
 
+    def setUp(self):
+        super(TestSlaveScannerWithoutDB, self).setUp()
+        self.addCleanup(shut_down_default_process_pool)
+
     def getScanner(self, builder_factory=None, interactor=None, slave=None,
                    behaviour=None):
         if builder_factory is None:
@@ -1079,6 +1089,7 @@ class TestCancellationChecking(TestCaseWithFactory):
         builder_name = BOB_THE_BUILDER_NAME
         self.builder = getUtility(IBuilderSet)[builder_name]
         self.builder.virtualized = True
+        self.addCleanup(shut_down_default_process_pool)
 
     @property
     def vitals(self):
diff --git a/lib/lp/code/model/tests/test_recipebuilder.py b/lib/lp/code/model/tests/test_recipebuilder.py
index 8acc5b8..1ef74fe 100644
--- a/lib/lp/code/model/tests/test_recipebuilder.py
+++ b/lib/lp/code/model/tests/test_recipebuilder.py
@@ -11,6 +11,7 @@ import os.path
 import shutil
 import tempfile
 
+from testscenarios import load_tests_apply_scenarios
 from testtools.matchers import MatchesListwise
 from testtools.twistedsupport import AsynchronousDeferredRunTest
 import transaction
@@ -457,3 +458,6 @@ class TestVerifySuccessfulBuildForSPRBuild(
 class TestHandleStatusForSPRBuild(
     MakeSPRecipeBuildMixin, TestHandleStatusMixin, TestCaseWithFactory):
     """IPackageBuild.handleStatus works with SPRecipe builds."""
+
+
+load_tests = load_tests_apply_scenarios
diff --git a/lib/lp/oci/tests/test_ocirecipebuildbehaviour.py b/lib/lp/oci/tests/test_ocirecipebuildbehaviour.py
index cf0056b..bdfbf89 100644
--- a/lib/lp/oci/tests/test_ocirecipebuildbehaviour.py
+++ b/lib/lp/oci/tests/test_ocirecipebuildbehaviour.py
@@ -19,6 +19,10 @@ import uuid
 import fixtures
 from fixtures import MockPatch
 from six.moves.urllib_parse import urlsplit
+from testscenarios import (
+    load_tests_apply_scenarios,
+    WithScenarios,
+    )
 from testtools import ExpectedException
 from testtools.matchers import (
     AfterPreprocessing,
@@ -41,7 +45,10 @@ from lp.buildmaster.enums import (
     BuildBaseImageType,
     BuildStatus,
     )
-from lp.buildmaster.interactor import BuilderInteractor
+from lp.buildmaster.interactor import (
+    BuilderInteractor,
+    shut_down_default_process_pool,
+    )
 from lp.buildmaster.interfaces.builder import (
     BuildDaemonError,
     CannotBuild,
@@ -393,7 +400,8 @@ class TestAsyncOCIRecipeBuildBehaviour(MakeOCIBuildMixin, TestCaseWithFactory):
         self.assertEqual(distroseries.name, slave.call_log[1][5]['series'])
 
 
-class TestHandleStatusForOCIRecipeBuild(MakeOCIBuildMixin,
+class TestHandleStatusForOCIRecipeBuild(WithScenarios,
+                                        MakeOCIBuildMixin,
                                         TestCaseWithFactory):
     # This is mostly copied from TestHandleStatusMixin, however
     # we can't use all of those tests, due to the way OCIRecipeBuildBehaviour
@@ -401,6 +409,11 @@ class TestHandleStatusForOCIRecipeBuild(MakeOCIBuildMixin,
     # available. There's also some differences in the filemap handling, as
     # we need a much more complex filemap here.
 
+    scenarios = [
+        ('download_in_twisted', {'download_in_subprocess': False}),
+        ('download_in_subprocess', {'download_in_subprocess': True}),
+        ]
+
     layer = LaunchpadZopelessLayer
     run_tests_with = AsynchronousDeferredRunTestForBrokenTwisted.make_factory(
         timeout=20)
@@ -414,7 +427,10 @@ class TestHandleStatusForOCIRecipeBuild(MakeOCIBuildMixin,
     def setUp(self):
         super(TestHandleStatusForOCIRecipeBuild, self).setUp()
         self.useFixture(fixtures.FakeLogger())
-        self.useFixture(FeatureFixture({OCI_RECIPE_ALLOW_CREATE: 'on'}))
+        features = {OCI_RECIPE_ALLOW_CREATE: 'on'}
+        if self.download_in_subprocess:
+            features['buildmaster.download_in_subprocess'] = 'on'
+        self.useFixture(FeatureFixture(features))
         self.build = self.makeBuild()
         # For the moment, we require a builder for the build so that
         # handleStatus_OK can get a reference to the slave.
@@ -425,6 +441,7 @@ class TestHandleStatusForOCIRecipeBuild(MakeOCIBuildMixin,
         self.interactor = BuilderInteractor()
         self.behaviour = self.interactor.getBuildBehaviour(
             self.build.buildqueue_record, self.builder, self.slave)
+        self.addCleanup(shut_down_default_process_pool)
 
         # We overwrite the buildmaster root to use a temp directory.
         tempdir = tempfile.mkdtemp()
@@ -683,3 +700,6 @@ class TestGetUploadMethodsForOCIRecipeBuild(
     def setUp(self):
         self.useFixture(FeatureFixture({OCI_RECIPE_ALLOW_CREATE: 'on'}))
         super(TestGetUploadMethodsForOCIRecipeBuild, self).setUp()
+
+
+load_tests = load_tests_apply_scenarios
diff --git a/lib/lp/services/job/runner.py b/lib/lp/services/job/runner.py
index c7b6284..8f80294 100644
--- a/lib/lp/services/job/runner.py
+++ b/lib/lp/services/job/runner.py
@@ -12,7 +12,9 @@ __all__ = [
     'celery_enabled',
     'JobRunner',
     'JobRunnerProcess',
+    'QuietAMPConnector',
     'TwistedJobRunner',
+    'VirtualEnvProcessStarter',
     ]
 
 
diff --git a/lib/lp/snappy/tests/test_snapbuildbehaviour.py b/lib/lp/snappy/tests/test_snapbuildbehaviour.py
index 81344b1..883ad47 100644
--- a/lib/lp/snappy/tests/test_snapbuildbehaviour.py
+++ b/lib/lp/snappy/tests/test_snapbuildbehaviour.py
@@ -19,6 +19,7 @@ import fixtures
 from pymacaroons import Macaroon
 import pytz
 from six.moves.urllib_parse import urlsplit
+from testscenarios import load_tests_apply_scenarios
 from testtools import ExpectedException
 from testtools.matchers import (
     AfterPreprocessing,
@@ -844,3 +845,6 @@ class TestVerifySuccessfulBuildForSnapBuild(
 class TestHandleStatusForSnapBuild(
     MakeSnapBuildMixin, TestHandleStatusMixin, TestCaseWithFactory):
     """IPackageBuild.handleStatus works with Snap builds."""
+
+
+load_tests = load_tests_apply_scenarios
diff --git a/lib/lp/soyuz/tests/test_binarypackagebuildbehaviour.py b/lib/lp/soyuz/tests/test_binarypackagebuildbehaviour.py
index aa79d81..d445091 100644
--- a/lib/lp/soyuz/tests/test_binarypackagebuildbehaviour.py
+++ b/lib/lp/soyuz/tests/test_binarypackagebuildbehaviour.py
@@ -13,6 +13,7 @@ import shutil
 import tempfile
 
 from storm.store import Store
+from testscenarios import load_tests_apply_scenarios
 from testtools.matchers import MatchesListwise
 from testtools.twistedsupport import AsynchronousDeferredRunTest
 import transaction
@@ -635,3 +636,6 @@ class TestVerifySuccessfulBuildForBinaryPackageBuild(
 class TestHandleStatusForBinaryPackageBuild(
     MakeBinaryPackageBuildMixin, TestHandleStatusMixin, TestCaseWithFactory):
     """IPackageBuild.handleStatus works with binary builds."""
+
+
+load_tests = load_tests_apply_scenarios
diff --git a/lib/lp/soyuz/tests/test_livefsbuildbehaviour.py b/lib/lp/soyuz/tests/test_livefsbuildbehaviour.py
index 4820132..3c3ee29 100644
--- a/lib/lp/soyuz/tests/test_livefsbuildbehaviour.py
+++ b/lib/lp/soyuz/tests/test_livefsbuildbehaviour.py
@@ -11,6 +11,7 @@ from datetime import datetime
 import os.path
 
 import pytz
+from testscenarios import load_tests_apply_scenarios
 from testtools.matchers import MatchesListwise
 from testtools.twistedsupport import AsynchronousDeferredRunTest
 import transaction
@@ -342,3 +343,6 @@ class TestVerifySuccessfulBuildForLiveFSBuild(
 class TestHandleStatusForLiveFSBuild(
     MakeLiveFSBuildMixin, TestHandleStatusMixin, TestCaseWithFactory):
     """IPackageBuild.handleStatus works with LiveFS builds."""
+
+
+load_tests = load_tests_apply_scenarios