Merge branch 'baserock/richardipsum/merge_distbuild'

Conflicts:
	morphlib/plugins/deploy_plugin.py
	without-test-modules

Reviewed by:
    Richard Maw
    Lars Wirzenius

5 files changed, 368 insertions, 10 deletions

diff --git a/morphlib/buildbranch.py b/morphlib/buildbranch.py
index 546a29d5..d415e7e1 100644
--- a/morphlib/buildbranch.py
+++ b/morphlib/buildbranch.py
@@ -217,7 +217,7 @@ class BuildBranch(object):
         # current HEAD
         if self._push_temporary:
             with morphlib.branchmanager.RemoteRefManager(False) as rrm:
-                for gd, build_ref in self._to_push.iterkeys():
+                for gd, (build_ref, index) in self._to_push.iteritems():
                     remote = gd.get_remote('origin')
                     yield gd, build_ref, remote
                     refspec = morphlib.gitdir.RefSpec(build_ref)
diff --git a/morphlib/buildcommand.py b/morphlib/buildcommand.py
index 862ae715..7ad7909d 100644
--- a/morphlib/buildcommand.py
+++ b/morphlib/buildcommand.py
@@ -20,6 +20,7 @@ import logging
 import tempfile
 
 import morphlib
+import distbuild
 
 
 class MultipleRootArtifactsError(morphlib.Error):
@@ -475,3 +476,32 @@ class BuildCommand(object):
             self.app, staging_area, self.lac, self.rac, self.lrc,
             self.app.settings['max-jobs'], setup_mounts)
         return builder.build_and_cache(artifact)
+
+class InitiatorBuildCommand(BuildCommand):
+
+    def __init__(self, app, addr, port):
+        self.app = app
+        self.addr = addr
+        self.port = port
+        self.app.settings['push-build-branches'] = True
+        super(InitiatorBuildCommand, self).__init__(app)
+
+    def build(self, args):
+        '''Initiate a distributed build on a controller'''
+
+        distbuild.add_crash_conditions(self.app.settings['crash-condition'])
+
+        if len(args) != 3:
+            raise morphlib.Error(
+                'Need repo, ref, morphology triplet to build')
+
+        if self.addr == '':
+            raise morphlib.Error(
+                'Need address of controller to run a distbuild')
+
+        self.app.status(msg='Starting distributed build')
+        loop = distbuild.MainLoop()
+        cm = distbuild.ConnectionMachine(
+            self.addr, self.port, distbuild.Initiator, [self.app] + args)
+        loop.add_state_machine(cm)
+        loop.run()
diff --git a/morphlib/plugins/build_plugin.py b/morphlib/plugins/build_plugin.py
index a58bce20..62009bcd 100644
--- a/morphlib/plugins/build_plugin.py
+++ b/morphlib/plugins/build_plugin.py
@@ -1,4 +1,4 @@
-# Copyright (C) 2012,2013  Codethink Limited
+# Copyright (C) 2012,2013,2014  Codethink Limited
 #
 # This program is free software; you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
@@ -28,9 +28,37 @@ class BuildPlugin(cliapp.Plugin):
                                 arg_synopsis='(REPO REF FILENAME)...')
         self.app.add_subcommand('build', self.build,
                                 arg_synopsis='SYSTEM')
+        self.app.add_subcommand('distbuild', self.distbuild,
+                                arg_synopsis='SYSTEM')
+        self.use_distbuild = False
 
     def disable(self):
-        pass
+        self.use_distbuild = False
+
+    def distbuild(self, args):
+        '''Distbuild a system image in the current system branch
+
+        Command line arguments:
+
+        * `SYSTEM` is the name of the system to build.
+
+        This command launches a distributed build, to use this command
+        you must first set up a distbuild cluster.
+
+        Artifacts produced during the build will be stored on your trove.
+
+        Once the build completes you can use morph deploy to the deploy
+        your system, the system artifact will be copied from your trove
+        and cached locally.
+
+        Example:
+
+            morph distbuild devel-system-x86_64-generic
+
+        '''
+
+        self.use_distbuild = True
+        self.build(args)
 
     def build_morphology(self, args):
         '''Build a system, outside of a system branch.
@@ -65,8 +93,6 @@ class BuildPlugin(cliapp.Plugin):
             self.app.settings['cachedir-min-space'])
 
         build_command = morphlib.buildcommand.BuildCommand(self.app)
-        build_command = self.app.hookmgr.call('new-build-command',
-                                              build_command)
         build_command.build(args)
 
     def build(self, args):
@@ -114,9 +140,15 @@ class BuildPlugin(cliapp.Plugin):
 
         build_uuid = uuid.uuid4().hex
 
-        build_command = morphlib.buildcommand.BuildCommand(self.app)
-        build_command = self.app.hookmgr.call('new-build-command',
-                                              build_command)
+        if self.use_distbuild:
+            addr = self.app.settings['controller-initiator-address']
+            port = self.app.settings['controller-initiator-port']
+
+            build_command = morphlib.buildcommand.InitiatorBuildCommand(
+                self.app, addr, port)
+        else:
+            build_command = morphlib.buildcommand.BuildCommand(self.app)
+
         loader = morphlib.morphloader.MorphologyLoader()
         push = self.app.settings['push-build-branches']
         name = morphlib.git.get_user_name(self.app.runcmd)
diff --git a/morphlib/plugins/deploy_plugin.py b/morphlib/plugins/deploy_plugin.py
index cc2076cd..ae62b75d 100644
--- a/morphlib/plugins/deploy_plugin.py
+++ b/morphlib/plugins/deploy_plugin.py
@@ -1,4 +1,4 @@
-# Copyright (C) 2013-2014  Codethink Limited
+# Copyright (C) 2013, 2014  Codethink Limited
 #
 # This program is free software; you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
@@ -444,7 +444,8 @@ class DeployPlugin(cliapp.Plugin):
             if build_command.lac.has(artifact):
                 f = build_command.lac.get(artifact)
             elif build_command.rac.has(artifact):
-                f = build_command.rac.get(artifact)
+                build_command.cache_artifacts_locally([artifact])
+                f = build_command.lac.get(artifact)
             else:
                 raise cliapp.AppException('Deployment failed as system is'
                                           ' not yet built.\nPlease ensure'
diff --git a/morphlib/plugins/distbuild_plugin.py b/morphlib/plugins/distbuild_plugin.py
new file mode 100644
index 00000000..27d87e35
--- /dev/null
+++ b/morphlib/plugins/distbuild_plugin.py
@@ -0,0 +1,295 @@
+# distbuild_plugin.py -- Morph distributed build plugin
+#
+# Copyright (C) 2014  Codethink Limited
+# 
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; version 2 of the License.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License along
+# with this program; if not, write to the Free Software Foundation, Inc.,
+# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA..
+
+
+import cliapp
+import logging
+import sys
+
+import morphlib
+import distbuild
+
+
+group_distbuild = 'Distributed Build Options'
+
+class DistbuildOptionsPlugin(cliapp.Plugin):
+
+    def enable(self):
+        self.app.settings.string_list(
+            ['crash-condition'],
+            'add FILENAME:FUNCNAME:MAXCALLS to list of crash conditions '
+                '(this is for testing only)',
+            metavar='FILENAME:FUNCNAME:MAXCALLS',
+            group=group_distbuild)
+
+    def disable(self):
+        pass
+
+
+class SerialiseArtifactPlugin(cliapp.Plugin):
+
+    def enable(self):
+        self.app.add_subcommand('serialise-artifact', self.serialise_artifact, 
+                                arg_synopsis='REPO REF MORPHOLOGY')
+
+    def disable(self):
+        pass
+
+    def serialise_artifact(self, args):
+        '''Internal use only: Serialise Artifact build graph as JSON.'''
+        
+        distbuild.add_crash_conditions(self.app.settings['crash-condition'])
+        
+        if len(args) != 3:
+            raise cliapp.AppException('Must get triplet')
+        
+        repo_name, ref, morph_name = args
+        filename = '%s.morph' % morph_name
+        build_command = morphlib.buildcommand.BuildCommand(self.app)
+        srcpool = build_command.create_source_pool(repo_name, ref, filename)
+        artifact = build_command.resolve_artifacts(srcpool)
+        self.app.output.write(distbuild.serialise_artifact(artifact))
+        self.app.output.write('\n')
+
+
+class WorkerBuild(cliapp.Plugin):
+
+    def enable(self):
+        self.app.add_subcommand(
+            'worker-build', self.worker_build, arg_synopsis='')
+
+    def disable(self):
+        pass
+    
+    def worker_build(self, args):
+        '''Internal use only: Build an artifact in a worker.
+        
+        All build dependencies are assumed to have been built already
+        and available in the local or remote artifact cache.
+        
+        '''
+        
+        distbuild.add_crash_conditions(self.app.settings['crash-condition'])
+
+        serialized = sys.stdin.readline()
+        artifact = distbuild.deserialise_artifact(serialized)
+        
+        bc = morphlib.buildcommand.BuildCommand(self.app)
+        
+        # We always, unconditionally clear the local artifact cache
+        # to avoid it growing boundlessly on a worker. Especially system
+        # artifacts are big (up to gigabytes), and having a new one for
+        # every build eats up a lot of disk space.
+        bc.lac.clear()
+
+        arch = artifact.arch
+        bc.build_artifact(artifact, bc.new_build_env(arch))
+
+
+class WorkerDaemon(cliapp.Plugin):
+
+    def enable(self):
+        self.app.settings.string(
+            ['worker-daemon-address'],
+            'listen for connections on ADDRESS (domain / IP address)',
+            default='',
+            group=group_distbuild)
+        self.app.settings.integer(
+            ['worker-daemon-port'],
+            'listen for connections on PORT',
+            default=3434,
+            group=group_distbuild)
+        self.app.add_subcommand(
+            'worker-daemon',
+            self.worker_daemon,
+            arg_synopsis='')
+    
+    def disable(self):
+        pass
+        
+    def worker_daemon(self, args):
+        '''Daemon that controls builds on a single worker node.'''
+
+        distbuild.add_crash_conditions(self.app.settings['crash-condition'])
+
+        address = self.app.settings['worker-daemon-address']
+        port = self.app.settings['worker-daemon-port']
+        router = distbuild.ListenServer(address, port, distbuild.JsonRouter)
+        loop = distbuild.MainLoop()
+        loop.add_state_machine(router)
+        loop.run()
+
+
+class ControllerDaemon(cliapp.Plugin):
+
+    def enable(self):
+        self.app.settings.string(
+            ['controller-initiator-address'],
+            'listen for initiator connections on ADDRESS '
+                '(domain / IP address)',
+            default='',
+            group=group_distbuild)
+        self.app.settings.integer(
+            ['controller-initiator-port'],
+            'listen for initiator connections on PORT',
+            default=7878,
+            group=group_distbuild)
+
+        self.app.settings.string(
+            ['controller-helper-address'],
+            'listen for helper connections on ADDRESS (domain / IP address)',
+            default='localhost',
+            group=group_distbuild)
+        self.app.settings.integer(
+            ['controller-helper-port'],
+            'listen for helper connections on PORT',
+            default=5656,
+            group=group_distbuild)
+
+        self.app.settings.string_list(
+            ['worker'],
+            'specify a build worker (WORKER is ADDRESS or ADDRESS:PORT, '
+                'with PORT defaulting to 3434)',
+            metavar='WORKER',
+            default=[],
+            group=group_distbuild)
+        self.app.settings.integer(
+            ['worker-cache-server-port'],
+            'port number for the artifact cache server on each worker',
+            metavar='PORT',
+            default=8080,
+            group=group_distbuild)
+        self.app.settings.string(
+            ['writeable-cache-server'],
+            'specify the shared cache server writeable instance '
+                '(SERVER is ADDRESS or ADDRESS:PORT, with PORT defaulting '
+                'to 80',
+            metavar='SERVER',
+            group=group_distbuild)
+
+        self.app.settings.string(
+            ['morph-instance'],
+            'use FILENAME to invoke morph (default: %default)',
+            metavar='FILENAME',
+            default='morph',
+            group=group_distbuild)
+
+        self.app.add_subcommand(
+            'controller-daemon', self.controller_daemon, arg_synopsis='')
+
+    def disable(self):
+        pass
+        
+    def controller_daemon(self, args):
+        '''Daemon that gives jobs to worker daemons.'''
+        
+        distbuild.add_crash_conditions(self.app.settings['crash-condition'])
+
+        artifact_cache_server = (
+            self.app.settings['artifact-cache-server'] or
+            self.app.settings['cache-server'])
+        writeable_cache_server = self.app.settings['writeable-cache-server']
+        worker_cache_server_port = \
+            self.app.settings['worker-cache-server-port']
+        morph_instance = self.app.settings['morph-instance']
+
+        listener_specs = [
+            ('controller-helper-address', 'controller-helper-port', 
+             distbuild.HelperRouter, []),
+            ('controller-initiator-address', 'controller-initiator-port',
+             distbuild.InitiatorConnection, 
+             [artifact_cache_server, morph_instance]),
+        ]
+
+        loop = distbuild.MainLoop()
+        
+        queuer = distbuild.WorkerBuildQueuer()
+        loop.add_state_machine(queuer)
+
+        for addr, port, sm, extra_args in listener_specs:
+            addr = self.app.settings[addr]
+            port = self.app.settings[port]
+            listener = distbuild.ListenServer(
+                addr, port, sm, extra_args=extra_args)
+            loop.add_state_machine(listener)
+
+        for worker in self.app.settings['worker']:
+            if ':' in worker:
+                addr, port = worker.split(':', 1)
+                port = int(port)
+            else:
+                addr = worker
+                port = 3434
+            cm = distbuild.ConnectionMachine(
+                addr, port, distbuild.WorkerConnection, 
+                [writeable_cache_server, worker_cache_server_port,
+                 morph_instance])
+            loop.add_state_machine(cm)
+
+        loop.run()
+
+class GraphStateMachines(cliapp.Plugin):
+
+    def enable(self):
+        self.app.add_subcommand(
+            'graph-state-machines',
+            self.graph_state_machines,
+            arg_synopsis='')
+
+    def disable(self):
+        pass
+
+    def graph_state_machines(self, args):
+        cm = distbuild.ConnectionMachine(None, None, None, None)
+        cm._start_connect = lambda *args: None
+        self.graph_one(cm)
+
+        self.graph_one(distbuild.BuildController(None, None, None))
+        self.graph_one(distbuild.HelperRouter(None))
+        self.graph_one(distbuild.InitiatorConnection(None, None, None))
+        self.graph_one(distbuild.JsonMachine(None))
+        self.graph_one(distbuild.WorkerBuildQueuer())
+
+        # FIXME: These need more mocking to work.
+        # self.graph_one(distbuild.Initiator(None, None,
+        #    self, None, None, None))
+        # self.graph_one(distbuild.JsonRouter(None))
+        # self.graph_one(distbuild.SocketBuffer(None, None))
+        # self.graph_one(distbuild.ListenServer(None, None, None))
+
+    def graph_one(self, sm):
+        class_name = sm.__class__.__name__.split('.')[-1]
+        filename = '%s.gv' % class_name
+        sm.mainloop = self
+        sm.setup()
+        sm.dump_dot(filename)
+
+    # Some methods to mock this class as other classes, which the
+    # state machine class need to access, just enough to allow the
+    # transitions to be set up for graphing.
+
+    def queue_event(self, *args, **kwargs):
+        pass
+
+    def add_event_source(self, *args, **kwargs):
+        pass
+
+    def add_state_machine(self, sm):
+        pass
+
+    def status(self, *args, **kwargs):
+        pass