Lots of changes to distbuild test harness

1 files changed, 234 insertions, 123 deletions

diff --git a/build-log-test.py b/build-log-test.py
index cd92f428..d7f3e9e4 100644
--- a/build-log-test.py
+++ b/build-log-test.py
@@ -14,44 +14,128 @@ MORPH = '/src/morph/morph'
 MORPH_CACHE_SERVER = '/src/morph/morph-cache-server'
 
 
-controller_config = [
-    '--no-default-config',
-    '--controller-initiator-port=7878',
-]
 
-controller_helper_config = [
-    '--no-default-config',
-    # The default behaviour of distbuild-helper is to be a worker helper
-    # process and these connect to port 3434.
-    '--parent-port=5656',
-]
+def subdir(workdir, *args):
+    path = os.path.join(workdir, *args)
+    os.makedirs(path)
+    return path
 
-worker_config = [
-    '--no-default-config',
-]
 
-worker_helper_config = [
-    '--no-default-config',
-]
+class Process(subprocess.Popen):
+    '''Helper for subprocess.Popen with neater commandline argument passing.
 
-initiator_config = [
-    '--no-default-config',
-    '--controller-initiator-address=localhost',
-    '--controller-initiator-port=7878',
-]
+    This allows you to pass a dict of settings, which will be converted to
+    commandline arguments internally.
 
+    '''
 
-build = [
-    'baserock:baserock/definitions',
-    'c7292b7c81cdd7e5b9e85722406371748453c44f',
-    'systems/base-system-x86_64-generic.morph',
-]
+    def __init__(self, name, argv, settings):
+        self.name = name
+        self.argv = argv
+        self.settings = settings
 
+        full_argv = argv + self._format_settings(settings)
+        print('%s commandline: %s' % (name, ' '.join(full_argv)))
+        subprocess.Popen.__init__(self, full_argv)
+        print('%s process ID: %s' % (name, self.pid))
 
-def subdir(workdir, *args):
-    path = os.path.join(workdir, *args)
-    os.makedirs(path)
-    return path
+    def _format_settings(self, arg_dict):
+        def as_string(key, value):
+            if value is True:
+                return '--%s' % key
+            elif value is False:
+                return ''
+            else:
+                return '--%s=%s' % (key, value)
+        return [as_string(k, v) for k, v in arg_dict.iteritems()]
+
+
+class MorphProcess(Process):
+    '''Process helper with special defaults for Morph and related programs.
+
+    Morph and its helper tools all use 'cliapp', which provides a standard
+    interface for configuration and logging.
+
+    '''
+
+    def __init__(self, name, argv, settings, log_path=None):
+        if log_path:
+            settings['log'] = os.path.join(log_path, 'morph-%s.log' % name)
+        settings['no-default-config'] = True
+
+        Process.__init__(self, name, argv, settings)
+
+
+class MorphListenerProcess(MorphProcess):
+    '''Process helper for listening processes.'''
+
+    def __init__(self, name, port_names, argv, settings, log_path=None):
+        for port_name in port_names:
+            self._setup_port_fifo(port_name, settings)
+
+        MorphProcess.__init__(self, name, argv, settings, log_path=log_path)
+
+        for port_name in port_names:
+            port_number = self._read_port_fifo(port_name, settings)
+            print('%s: %s port is %s' % (name, port_name, port_number))
+
+            port_attr = port_name.replace('-', '_')
+            setattr(self, port_attr, port_number)
+
+    def _setup_port_fifo(self, port_name, settings):
+        tempdir = tempfile.mkdtemp()
+        port_file = os.path.join(tempdir, '%s.port' % (port_name))
+        os.mkfifo(port_file)
+
+        # Note that Python passes dicts by reference, so this modifies the
+        # dict that was passed in.
+        settings['%s' % port_name] = 0
+        settings['%s-file' % port_name] = port_file
+
+    def _read_port_fifo(self,  port_name, settings):
+        port_file = settings['%s-file' % port_name]
+
+        print 'Read: %s' % port_file
+        with open(port_file, 'r') as f:
+            # The readline() call will block until a line of data is written.
+            # The process we are starting should only write out the port number
+            # once the process is listening on that port. Thus, we block until
+            # the process is ready, which is very important.
+            port_number = int(f.readline())
+
+        os.unlink(port_file)
+        os.rmdir(os.path.dirname(port_file))
+
+        return port_number
+
+
+class MorphCacheServerProcess(MorphListenerProcess):
+    def __init__(self, name, cache_path, log_path=None, enable_writes=False):
+        ports = ['port']
+        argv = [MORPH_CACHE_SERVER]
+
+        settings = {
+            'artifact-dir': subdir(cache_path, 'artifacts'),
+            'repo-dir': subdir(cache_path, 'gits'),
+            'enable-writes': enable_writes,
+            'no-fcgi': True,
+        }
+
+        MorphListenerProcess.__init__(
+            self, name, ports, argv, settings, log_path=log_path)
+
+
+class MorphWorkerDaemonProcess(MorphListenerProcess):
+    def __init__(self, name, cache_server, log_path=None):
+        ports = ['worker-daemon-port']
+        argv = [MORPH, 'worker-daemon']
+
+        settings = {
+            'artifact-cache-server': 'http://localhost:%s' % cache_server.port,
+        }
+
+        MorphListenerProcess.__init__(
+            self, name, ports, argv, settings, log_path=log_path)
 
 
 class DistbuildTestHarness(object):
@@ -64,31 +148,41 @@ class DistbuildTestHarness(object):
         '''Start a distbuild network and run a single build on it.'''
         workdir = tempfile.mkdtemp()
         print('Working directory: %s' % workdir)
+
+        to_build = [
+            'baserock:baserock/definitions',
+            'c7292b7c81cdd7e5b9e85722406371748453c44f',
+            'systems/base-system-x86_64-generic.morph',
+        ]
+
+
         try:
-            self.start_cache_servers(workdir)
-            self.start_distbuild_network(workdir)
-            self.run_build(workdir)
-        finally:
-            self.terminate_processes()
-            print('Test state kept in %s' % workdir)
+            worker_cache, shared_cache = self.start_cache_servers(workdir)
+            controller = self.start_distbuild_network(
+                workdir, worker_cache, shared_cache, n_workers=7)
 
-    def start_process(self, name, argv):
-        '''Start a process that will be cleaned up with the parent process.'''
-        self.process[name] = subprocess.Popen(argv)
-        print('Started %s as pid %i' % (name, self.process[name].pid))
+            initiator1 = self.start_build(controller, to_build,
+                                         log_path=subdir(workdir, '1'))
 
-    def start_process_with_logs(self, workdir, name, argv):
-        '''Start a Morph process that will be cleaned up on exit.
+            initiator2 = self.start_build(controller, to_build,
+                                         log_path=subdir(workdir, '2'))
 
-        Log config will be set so that the process writes to a log file in
-        'workdir'.
+            while initiator1.poll() == None or initiator2.poll() == None:
+                time.sleep(0.1) # FIXME: use select!
+                self.check_processes_are_running()
 
-        '''
-        log_config = [
-            '--log-level=debug',
-            '--log=%s/morph-%s.log' % (workdir, name),
-        ]
-        self.start_process(name, argv + log_config)
+            if initiator1.returncode != 0 or initiator2.returncode != 0:
+                #raise Exception('Initiator fail')
+                print('Initiator fail')
+            else:
+                print('Success!')
+
+        finally:
+            self.terminate_processes()
+            print('Test state kept in %s' % workdir)
+
+    def watch_process(self, process):
+        self.process[process.name] = process
 
     def check_processes_are_running(self):
         '''Check all processes are running.'''
@@ -108,90 +202,107 @@ class DistbuildTestHarness(object):
     def start_cache_servers(self, workdir):
         # We have to make a bit of a kludge here. In a real distbuild setup,
         # each worker machine runs its own instance of morph-cache-server on
-        # port 8080. This port is hardcoded in the controller, so in this
-        # test harness all workers will have to share one cache-server process.
+        # port 8080. The controller uses the same port number for all workers
+        # so in this test harness all workers will have to share one
+        # cache-server process.
         #
         # That's fine, but we still need a separate process for the *shared*
         # artifact cache: artifacts are copied from workers to the shared cache
         # using the /fetch method, which just wouldn't work if it had to fetch
         # from itself.
-        self.worker_git_dir = subdir(workdir, 'worker-cache', 'gits')
-        self.worker_artifacts_dir = subdir(workdir, 'worker-cache',
-                                           'artifacts')
-
-        self.shared_git_dir = subdir(workdir, 'shared-cache', 'gits')
-        self.shared_artifacts_dir = subdir(workdir, 'shared-cache',
-                                           'artifacts')
-
-        self.start_process_with_logs(
-            workdir, 'worker-cache-server',
-            [MORPH_CACHE_SERVER, '--no-default-config',
-             '--repo-dir=%s' % self.worker_git_dir,
-             '--artifact-dir=%s' % self.worker_artifacts_dir,
-             '--no-fcgi'])
-
-        self.start_process_with_logs(
-            workdir, 'shared-cache-server',
-            [MORPH_CACHE_SERVER, '--no-default-config', '--enable-writes',
-             '--repo-dir=%s' % self.shared_git_dir,
-             '--artifact-dir=%s' % self.shared_artifacts_dir,
-             '--no-fcgi', '--port=8081'])
-
-    def start_distbuild_network(self, workdir, n_workers=4):
+        worker_cache = MorphCacheServerProcess(
+            name='worker-cache-server',
+            cache_path=subdir(workdir, 'worker-cache'),
+            log_path=workdir,
+            enable_writes=False)
+        self.watch_process(worker_cache)
+
+        shared_cache = MorphCacheServerProcess(
+            name='shared-cache-server',
+            cache_path=subdir(workdir, 'shared-cache'),
+            log_path=workdir,
+            enable_writes=True)
+        self.watch_process(shared_cache)
+
+        return worker_cache, shared_cache
+
+    def start_distbuild_network(self, workdir, worker_cache, shared_cache,
+                                n_workers=4):
+
+        # This is a hack so we can pass arguments to the Morph used to run
+        # worker-build and serialise-artifact.
+        worker_morph = os.path.join(workdir, 'worker-morph')
+        with open(worker_morph, 'w') as f:
+            cache_dir = os.path.join(workdir, 'worker-cache')
+            log = os.path.join(workdir, 'morph.log')
+            f.write('#!/bin/sh\n')
+            f.write('%s --cachedir=%s --log=%s $@\n' % (MORPH, cache_dir, log))
+        os.chmod(worker_morph, 0755)
+
         workers = []
         for n in range(0, n_workers):
-            worker_port = 3434 + n
-            workers.append('localhost:%i' % worker_port)
-
-            self.start_process_with_logs(
-                workdir, 'worker-%i' % n,
-                [MORPH, 'worker-daemon',
-                 '--worker-daemon-port=%i' % worker_port,
-                 '--artifact-cache-server=http://localhost:8080',
-                 '--morph-instance=%s --cachedir=%s' % (MORPH, self.worker_artifacts_dir)] +
-                worker_config)
-
-            # Again, you'll see 'Connection refused' if you do this too
-            # quickly.
-            time.sleep(0.2)
-
-            self.start_process_with_logs(
-                workdir, 'worker-%i-helper' % n,
-                [DISTBUILD_HELPER, '--parent-port=%i' % worker_port] +
-                 worker_config)
-
-        self.start_process_with_logs(
-            workdir, 'controller',
-            [MORPH, 'controller-daemon', '--worker=%s' % ','.join(workers),
-             '--writeable-cache-server=http://localhost:8081'] +
-             controller_config)
-
-        # distbuild-helper will raise 'Connection refused' if started too soon
-        # after the parent process.
+            worker = MorphWorkerDaemonProcess(
+                name='worker-%i' % n,
+                cache_server=worker_cache,
+                log_path=workdir)
+            self.watch_process(worker)
+
+            workers.append('localhost:%i' % worker.worker_daemon_port)
+
+            worker_helper = MorphProcess(
+                name='worker-%i-helper' % n,
+                argv=[DISTBUILD_HELPER],
+                settings={
+                    'parent-port': worker.worker_daemon_port,
+                },
+                log_path=workdir
+            )
+            self.watch_process(worker_helper)
+
+        controller = MorphListenerProcess(
+            name='controller',
+            # Order is significant -- helper-port must be first!
+            port_names=['controller-helper-port', 'controller-initiator-port'],
+            argv=[MORPH, 'controller-daemon'],
+            settings={
+                'controller-initiator-address': 'localhost',
+                'morph-instance': worker_morph,
+                'worker': ','.join(workers),
+                'worker-cache-server-port': worker_cache.port,
+                'writeable-cache-server': 'http://localhost:%s' % shared_cache.port,
+            },
+            log_path=workdir)
+        self.watch_process(controller)
+
+        controller_helper = MorphProcess(
+            name='controller-helper',
+            argv=[DISTBUILD_HELPER],
+            settings={
+                'parent-port': controller.controller_helper_port,
+            },
+            log_path=workdir
+        )
+
+        self.watch_process(controller_helper)
+
+        # Need to wait for controller-helper to connect to controller.
         time.sleep(0.1)
 
-        self.start_process_with_logs(
-            workdir, 'controller-helper',
-            [DISTBUILD_HELPER] + controller_helper_config)
-
-        time.sleep(1)
         self.check_processes_are_running()
 
-    def run_build(self, workdir):
-        self.start_process_with_logs(
-            workdir, 'initiator',
-            [MORPH, 'distbuild-morphology'] + initiator_config + build)
-
-        while self.process['initiator'].poll() == None:
-            time.sleep(0.1) # FIXME: use select!
-            if self.process['controller'].poll() != None:
-                raise Exception('Controller fail')
-
-        if self.process['initiator'].returncode != 0:
-            #raise Exception('Initiator fail')
-            print('Initiator fail')
-        else:
-            print('Success!')
+        return controller
+
+    def start_build(self, controller, to_build, log_path):
+        port = controller.controller_initiator_port
+        return MorphProcess(
+            name='initiator',
+            argv=[MORPH, 'distbuild-morphology'] + to_build,
+            settings={
+                'controller-initiator-address': 'localhost',
+                'controller-initiator-port': port,
+                'initiator-step-output-dir': log_path,
+            },
+            log_path=log_path)
 
 
 DistbuildTestHarness().run()