Merge lp:~hazmat/pyjuju/lp-616504-provision-this into lp:pyjuju

[Kapil Thangavelu]

Provisioning Agent Implementation!!

[lp:~hazmat/pyjuju/lp-616504-provision-this]

90. By Kapil Thangavelu on 2010-09-16

enable test for removed machine state

[Gustavo Niemeyer (niemeyer)]

That's been reviewed live.

review: Approve

[Gustavo Niemeyer (niemeyer)]

As per the conversation:

[1]

In the concurrent_execution_guard tests, the first block in this test callback should be repeated and ensure that the underlying method was only called twice:

+ def validate_results(results):
+ success, value = results[0]
+ self.assertTrue(success)
+ self.assertEqual(value, 1)
+
+ success, value = results[1]
+ self.assertTrue(success)
+ self.assertEqual(value, False)

[lp:~hazmat/pyjuju/lp-616504-provision-this]

91. By Kapil Thangavelu on 2010-09-17

address review comment, and guard execution value had skipped the concurrent call

[Kapil Thangavelu (hazmat)]

thanks, addressed [1], and merged.

=== modified file 'bin/ensemble-admin' (properties changed: -x to +x)
=== modified file 'ensemble/agents/base.py'
--- ensemble/agents/base.py	2010-09-14 20:03:06 +0000
+++ ensemble/agents/base.py	2010-09-17 14:41:43 +0000
@@ -44,6 +44,9 @@
     name = "ensemble-agent-unknown"
     client = None
 
+    # Flag when enabling persistent topology watches, testing aid.
+    _watch_enabled = True
+
     @classmethod
     def run(cls):
         """Runs the agent as a unix daemon.
@@ -130,8 +133,8 @@
     # conventions.
     @inlineCallbacks
     def startService(self):
-        self.client = yield self.connect()
-        self.start()
+        yield self.connect()
+        yield self.start()
 
     @inlineCallbacks
     def stopService(self):
@@ -141,6 +144,12 @@
             if self.client and self.client.connected:
                 self.client.close()
 
+    def set_watch_enabled(self, flag):
+        self._watch_enabled = bool(flag)
+
+    def get_watch_enabled(self):
+        return self._watch_enabled
+
 
 def setup_default_agent_options(parser, cls):
 

=== modified file 'ensemble/agents/provision.py'
--- ensemble/agents/provision.py	2010-09-14 20:03:06 +0000
+++ ensemble/agents/provision.py	2010-09-17 14:41:43 +0000
@@ -3,8 +3,12 @@
 from twisted.internet.defer import inlineCallbacks, returnValue
 
 from ensemble.environment.config import EnvironmentsConfig
+from ensemble.errors import ProviderInteractionError
+from ensemble.lib.twistutils import concurrent_execution_guard
 from ensemble.state.machine import MachineStateManager
 
+from ensemble.state.errors import MachineStateNotFound, StateChanged
+
 from .base import BaseAgent
 
 
@@ -12,16 +16,25 @@
 
     name = "ensemble-provisoning-agent"
 
-    # @inlineCallbacks
-    # def start(self):
-    #     self.environment = yield self.startup_wait_for_environment()
-    #     self.provider = self.environment.get_machine_provider()
-    #     self.state_manager = MachineStateManager(self.client)
-    #     self.state_manager.watch_machine_changes(self.watch_machine_changes)
-
-    @inlineCallbacks
-    def startup_wait_for_environment(self):
-        """On startup the provisioning agent needs to wait for its environment.
+    _current_machines = ()
+    _machine_watch_enabled = True
+
+    machine_check_period = 60 # time in seconds
+
+    @inlineCallbacks
+    def start(self):
+        self.environment = yield self.configure_environment()
+        self.provider = self.environment.get_machine_provider()
+        self.state_manager = MachineStateManager(self.client)
+        if self.get_watch_enabled():
+            self.state_manager.watch_machine_states(self.watch_machine_changes)
+            from twisted.internet import reactor
+            reactor.callLater(
+                self.machine_check_period, self.periodic_machine_check)
+
+    @inlineCallbacks
+    def configure_environment(self):
+        """The provisioning agent configure its environment on start or change.
 
         The environment contains the configuration th agent needs to interact
         with its machine provider, in order to do its work. This configuration
@@ -30,29 +43,59 @@
         The agent waits for this data to exist before completing its startup.
         """
         try:
-            environment_data, stat = yield self.client.get("/environment")
+            get_d, watch_d = self.client.get_and_watch("/environment")
+            environment_data, stat = yield get_d
+            watch_d.addCallback(self._on_environment_changed)
         except NoNodeException:
             # Wait till the environment node appears. play twisted gymnastics
             exists_d, watch_d = self.client.exists_and_watch("/environment")
             stat = yield exists_d
             if stat:
-                environment = yield self.startup_wait_for_environment()
-                returnValue(environment)
+                environment = yield self.configure_environment()
             else:
                 watch_d.addCallback(
-                    lambda result: self.startup_wait_for_environment())
+                    lambda result: self.configure_environment())
+            if not stat:
                 environment = yield watch_d
-                returnValue(environment)
+            returnValue(environment)
 
         config = EnvironmentsConfig()
         config.parse(environment_data)
         returnValue(config.get_default())
 
-    def watch_environment_changes(self, event):
+    @inlineCallbacks
+    def _on_environment_changed(self, event):
         """Reload the environment if its data changes."""
 
+        if event.type_name == "deleted":
+            return
+
+        self.environment = yield self.configure_environment()
+        self.provider = self.environment.get_machine_provider()
+
+    def periodic_machine_check(self):
+        """A periodic checking of machine states and provider machines.
+
+        In addition to the on demand changes to zookeeper states that are
+        monitored by L{watch_machine_changes}, the periodic machine check
+        performs non zookeeper state related verification by periodically
+        checking the last current provider machine states against the
+        last known zookeeper state.
+
+        Primarily this helps in recovering from transient error conditions
+        which may have prevent processing of an individual machine state, as
+        well as verifying the current state of the provider's running machines
+        against the zk state, thus pruning unused resources.
+        """
+        from twisted.internet import reactor
+        d = self.process_machines(self._current_machines)
+        d.addBoth(
+            lambda result: reactor.callLater(
+                self.machine_check_period, self.periodic_machine_check))
+        return d
+
     def watch_machine_changes(self, old_machines, new_machines):
-        """Ensure the currently running machines correspond to state.
+        """Watches and processes machine state changes.
 
         This function is used to subscribe to topology changes, and
         specifically changes to machines within the topology. It performs
@@ -64,21 +107,83 @@
         function will automatically be rescheduled to run whenever a topology
         state change happens that involves machines.
 
+        This functional also caches the current set of machines as an agent
+        instance attribute.
+
+        @param old_machines machine ids as existed in the previous topology.
+        @param new_machines machine ids as exist in the current topology.
+        """
+        self._current_machines = new_machines
+        return self.process_machines(self._current_machines)
+
+    @concurrent_execution_guard("_processing_machines")
+    @inlineCallbacks
+    def process_machines(self, current_machines):
+        """Ensure the currently running machines correspond to state.
+
+        At the end of each process_machines execution, verify that all
+        running machines within the provider correspond to machine_ids within
+        the topology. If they don't then shut them down.
+
+        Utilizes concurrent execution guard, to ensure that this is only being
+        executed at most once per process.
+        """
+        # XXX this is obviously broken, but the margins of 80 columns prevent
+        # me from describing. hint think concurrent agents, and use a lock.
+
+        # map of instance_id -> machine
+        try:
+            provider_machines = yield self.provider.list_machines()
+        except ProviderInteractionError:
+            # XXX log me
+            return
+
+        provider_machines = dict(
+            [(m.instance_id, m) for m in provider_machines])
+
+        instance_ids = []
+        for machine_state_id in current_machines:
+            try:
+                instance_id = yield self.process_machine(
+                    machine_state_id, provider_machines)
+            except (StateChanged,
+                    MachineStateNotFound,
+                    ProviderInteractionError):
+                # XXX log me
+                continue
+            instance_ids.append(instance_id)
+
+        # Terminate all unused ensemble machines running within the cluster.
+        unused = set(provider_machines.keys()) - set(instance_ids)
+        for instance_id in unused:
+            machine = provider_machines[instance_id]
+            try:
+                yield self.provider.shutdown_machine(machine)
+            except ProviderInteractionError:
+                # XXX log me
+                continue
+
+    @inlineCallbacks
+    def process_machine(self, machine_state_id, provider_machine_map):
+        """Ensure a provider machine for a machine state id.
+
         For each machine_id in new machines which represents the current state
         of the topology, check to ensure its state reflects that it has been
         launched. If it hasn't then create the machine and update the state.
-
-        At the end of each watch_machine_change execution, verify that all
-        running machines within the provider correspond to machine_ids within
-        the topology. If they don't then shut them down.
-
-        @param old_machines machine ids as existed in the previous topology.
-        @param new_machines machine ids as exist in the current topology.
-        """
-
-    def process_machine(self, machine_id):
-        """Verify a machine id has state and is running, else launch it."""
-
-    def terminate_unused(self, machine_states):
-        """Terminate all unused ensemble machines running within the cluster.
-        """
+        """
+        # fetch the machine state
+        machine_state = yield self.state_manager.get_machine_state(
+            machine_state_id)
+        instance_id = yield machine_state.get_instance_id()
+
+        # Verify a machine id has state and is running, else launch it.
+        if instance_id is None or not instance_id in provider_machine_map:
+            machines = yield self.provider.start_machine(
+                {"machine_id": machine_state.id})
+            instance_id = machines[0].instance_id
+            yield machine_state.set_instance_id(instance_id)
+
+        returnValue(instance_id)
+
+if __name__ == '__main__':
+    ProvisioningAgent().run()

=== modified file 'ensemble/agents/tests/test_base.py'
--- ensemble/agents/tests/test_base.py	2010-09-14 19:37:16 +0000
+++ ensemble/agents/tests/test_base.py	2010-09-17 14:41:43 +0000
@@ -272,3 +272,9 @@
         agent = BaseAgent()
         self.assertRaises(
             NoConnection, agent.configure, {"zookeeper_servers": None})
+
+    def test_watch_enabled_accessors(self):
+        agent = BaseAgent()
+        self.assertTrue(agent.get_watch_enabled())
+        agent.set_watch_enabled(False)
+        self.assertFalse(agent.get_watch_enabled())

=== modified file 'ensemble/agents/tests/test_provision.py'
--- ensemble/agents/tests/test_provision.py	2010-09-14 20:03:06 +0000
+++ ensemble/agents/tests/test_provision.py	2010-09-17 14:41:43 +0000
@@ -1,10 +1,10 @@
 import zookeeper
 
-from twisted.internet.defer import inlineCallbacks
+from twisted.internet.defer import inlineCallbacks, fail, succeed
 from twisted.internet import reactor
 
-from txzookeeper import ZookeeperClient
 from txzookeeper.tests.utils import deleteTree
+from txzookeeper.client import ZOO_OPEN_ACL_UNSAFE
 
 from ensemble.agents.provision import ProvisioningAgent
 from ensemble.agents.base import TwistedOptionNamespace
@@ -13,22 +13,24 @@
 from ensemble.environment.errors import EnvironmentsConfigError
 from ensemble.environment.tests.test_config import SAMPLE_ENV
 
-from ensemble.lib.testing import TestCase
-
-
-class ProvisioningAgentStartupTest(TestCase):
+from ensemble.errors import ProviderInteractionError
+from ensemble.lib.mocker import MATCH
+from ensemble.providers.dummy import DummyMachine
+from ensemble.state.machine import MachineStateManager
+from ensemble.state.tests.common import StateTestBase
+
+MATCH_MACHINE = MATCH(lambda x: isinstance(x, DummyMachine))
+
+
+class ProvisioningTestBase(StateTestBase):
 
     @inlineCallbacks
     def setUp(self):
-        zookeeper.set_debug_level(0)
-        self.client = ZookeeperClient()
-        yield self.client.connect("127.0.0.1:2181")
+        yield super(ProvisioningTestBase, self).setUp()
         self.options = TwistedOptionNamespace()
         self.options["zookeeper_servers"] = "127.0.0.1:2181"
-
         self.agent = ProvisioningAgent()
         self.agent.configure(self.options)
-        yield self.agent.connect()
 
     def tearDown(self):
         if self.agent.client and self.agent.client.connected:
@@ -43,13 +45,21 @@
         config.parse(SAMPLE_ENV)
         return config.serialize("myfirstenv")
 
+
+class ProvisioningAgentStartupTest(ProvisioningTestBase):
+
+    @inlineCallbacks
+    def setUp(self):
+        yield super(ProvisioningAgentStartupTest, self).setUp()
+        yield self.agent.connect()
+
     def test_agent_waits_for_environment(self):
         """
         When the agent starts it waits for the /environment node to exist.
         As soon as it does, the agent will fetch the environment, and
         deserialize it into an environment object.
         """
-        env_loaded_deferred = self.agent.startup_wait_for_environment()
+        env_loaded_deferred = self.agent.configure_environment()
 
         def verify_environment(result):
             self.assertTrue(isinstance(result, Environment))
@@ -76,27 +86,291 @@
             self.assertTrue(isinstance(result, Environment))
             self.assertEqual(result.name, "myfirstenv")
 
-        d = self.agent.startup_wait_for_environment()
+        d = self.agent.configure_environment()
         d.addCallback(verify_environment)
         yield d
 
     @inlineCallbacks
     def test_agent_with_invalid_environment(self):
         yield self.client.create("/environment", "WAHOO!")
-        d = self.agent.startup_wait_for_environment()
+        d = self.agent.configure_environment()
         yield self.assertFailure(d, EnvironmentsConfigError)
 
-
-class ProvisoningAgentTest(TestCase):
-
+    def test_agent_with_nonexistent_environment_created_concurrently(self):
+        """
+        If the environment node does not initially exist but it is created
+        while the agent is processing the NoNodeException, it should detect
+        this and configure normally.
+        """
+        data = self.get_serialized_environment()
+        exists_and_watch = self.agent.client.exists_and_watch
+
+        mock_client = self.mocker.patch(self.agent.client)
+        mock_client.exists_and_watch("/environment")
+
+        def inject_creation(path):
+            zookeeper.create(
+                self.agent.client.handle, path, data, [ZOO_OPEN_ACL_UNSAFE])
+            return exists_and_watch(path)
+
+        self.mocker.call(inject_creation)
+        self.mocker.replay()
+
+        def verify_configured(result):
+            self.assertTrue(isinstance(result, Environment))
+            self.assertEqual(result.type, "dummy")
+        # mocker magic test
+        d = self.agent.configure_environment()
+        d.addCallback(verify_configured)
+        return d
+
+
+class ProvisioningAgentTest(ProvisioningTestBase):
+
+    @inlineCallbacks
     def setUp(self):
-        pass
-
-    def test_process_machine_id(self):
-        pass
-
-    def test_terminate_unusued_doesnt_touch_used(self):
-        pass
-
-    def test_terminate_unusued(self):
-        pass
+        yield super(ProvisioningAgentTest, self).setUp()
+        yield self.client.create(
+            "/environment", self.get_serialized_environment())
+        self.agent.set_watch_enabled(False)
+        yield self.agent.startService()
+
+    @inlineCallbacks
+    def test_watch_machine_changes_processes_new_machine_id(self):
+        """The agent should process a new machine id by creating it"""
+        manager = MachineStateManager(self.client)
+        machine_state0 = yield manager.add_machine_state()
+        machine_state1 = yield manager.add_machine_state()
+
+        yield self.agent.watch_machine_changes(
+            None, [machine_state0.id, machine_state1.id])
+
+        machines = yield self.agent.provider.list_machines()
+        self.assertEquals(len(machines), 2)
+
+        instance_id = yield machine_state0.get_instance_id()
+        self.assertEqual(instance_id, 0)
+
+        instance_id = yield machine_state1.get_instance_id()
+        self.assertEqual(instance_id, 1)
+
+    @inlineCallbacks
+    def test_watch_machine_changes_ignores_running_machine(self):
+        manager = MachineStateManager(self.client)
+        machine_state0 = yield manager.add_machine_state()
+        yield self.agent.provider.start_machine()
+        yield machine_state0.set_instance_id(0)
+
+        machine_state1 = yield manager.add_machine_state()
+
+        yield self.agent.watch_machine_changes(
+            None, [machine_state0.id, machine_state1.id])
+
+        machines = yield self.agent.provider.list_machines()
+        self.assertEquals(len(machines), 2)
+
+        instance_id = yield machine_state1.get_instance_id()
+        self.assertEqual(instance_id, 1)
+
+    @inlineCallbacks
+    def test_watch_machine_changes_terminates_unused(self):
+        # start an unused machine within the dummy provider instance
+        yield self.agent.provider.start_machine()
+        yield self.agent.watch_machine_changes(None, [])
+
+        machines = yield self.agent.provider.list_machines()
+        self.assertFalse(machines)
+
+    @inlineCallbacks
+    def test_new_machine_state_removed_while_processing(self):
+        """
+        If the machine state is removed while the event is processing the
+        state, the watch function should process it normally.
+        """
+        yield self.agent.watch_machine_changes(
+            None, [0])
+        machines = yield self.agent.provider.list_machines()
+        self.assertEquals(len(machines), 0)
+
+    @inlineCallbacks
+    def test_process_machines_non_concurrency(self):
+        """
+        Process machines should only be executed serially by an
+        agent.
+        """
+        manager = MachineStateManager(self.client)
+        machine_state0 = yield manager.add_machine_state()
+        machine_state1 = yield manager.add_machine_state()
+
+        call_1 = self.agent.process_machines([machine_state0.id])
+
+        # The second call should return immediately due to the
+        # instance attribute guard.
+        call_2 = self.agent.process_machines([machine_state1.id])
+        self.assertEqual(call_2.called, True)
+        self.assertEqual(call_2.result, False)
+
+        # The first call should have started a provider machine
+        yield call_1
+
+        machines = yield self.agent.provider.list_machines()
+        self.assertEquals(len(machines), 1)
+
+        instance_id_0 = yield machine_state0.get_instance_id()
+        self.assertEqual(instance_id_0, 0)
+
+        instance_id_1 = yield machine_state1.get_instance_id()
+        self.assertEqual(instance_id_1, None)
+
+    def test_new_machine_state_removed_while_processing_get_provider_id(self):
+        """
+        If the machine state is removed while the event is processing the
+        state, the watch function should process it normally.
+        """
+        yield self.agent.watch_machine_changes(
+            None, [0])
+        machines = yield self.agent.provider.list_machines()
+        self.assertEquals(len(machines), 0)
+
+    @inlineCallbacks
+    def test_on_environment_change_agent_reconfigures(self):
+        """
+        If the environment changes the agent reconfigures itself
+        """
+        provider = self.agent.provider
+        data = self.get_serialized_environment()
+        yield self.client.set("/environment", data)
+        yield self.sleep(0.2)
+        self.assertNotIdentical(provider, self.agent.provider)
+
+    @inlineCallbacks
+    def test_machine_state_reflects_invalid_provider_state(self):
+        """
+        If a machine state has an invalid instance_id, it should be detected,
+        and a new machine started and the machine state updated with the
+        new instance_id.
+        """
+        machine_manager = MachineStateManager(self.client)
+        m1 = yield machine_manager.add_machine_state()
+        yield m1.set_instance_id("zebra")
+
+        m2 = yield machine_manager.add_machine_state()
+        yield self.agent.watch_machine_changes(None, [m1.id, m2.id])
+
+        m1_instance_id = yield m1.get_instance_id()
+        self.assertEqual(m1_instance_id, 0)
+
+        m2_instance_id = yield m2.get_instance_id()
+        self.assertEqual(m2_instance_id, 1)
+
+    def test_periodic_task(self):
+        """
+        The agent schedules period checks that execute the process machines
+        call.
+        """
+        mock_reactor = self.mocker.patch(reactor)
+        mock_reactor.callLater(self.agent.machine_check_period,
+                               self.agent.periodic_machine_check)
+        mock_agent = self.mocker.patch(self.agent)
+        mock_agent.process_machines(())
+        self.mocker.result(succeed(None))
+        self.mocker.replay()
+
+        # mocker magic test
+        self.agent.periodic_machine_check()
+
+    @inlineCallbacks
+    def test_transient_provider_error_on_start_machine(self):
+        """
+        If there's an error when processing changes, the agent should log
+        the error and continue.
+        """
+        manager = MachineStateManager(self.client)
+        machine_state0 = yield manager.add_machine_state()
+        machine_state1 = yield manager.add_machine_state()
+
+        mock_provider = self.mocker.patch(self.agent.provider)
+        mock_provider.start_machine({"machine_id": 0})
+        self.mocker.result(fail(ProviderInteractionError(OSError("Bad"))))
+
+        mock_provider.start_machine({"machine_id": 1})
+        self.mocker.passthrough()
+        self.mocker.replay()
+
+        yield self.agent.watch_machine_changes(
+            [], [machine_state0.id, machine_state1.id])
+
+        machine1_instance_id = yield machine_state1.get_instance_id()
+        self.assertEqual(machine1_instance_id, 0)
+
+    @inlineCallbacks
+    def test_transient_provider_error_on_shutdown_machine(self):
+
+        yield self.agent.provider.start_machine()
+        mock_provider = self.mocker.patch(self.agent.provider)
+
+        mock_provider.shutdown_machine(MATCH_MACHINE)
+        self.mocker.result(fail(ProviderInteractionError("Bad")))
+
+        mock_provider.shutdown_machine(MATCH_MACHINE)
+        self.mocker.passthrough()
+
+        self.mocker.replay()
+        try:
+            yield self.agent.process_machines([])
+        except:
+            self.fail("Should not raise")
+
+        machines = yield self.agent.provider.list_machines()
+        self.assertTrue(machines)
+
+        yield self.agent.process_machines([])
+        machines = yield self.agent.provider.list_machines()
+        self.assertFalse(machines)
+
+    @inlineCallbacks
+    def test_transient_provider_error_on_list_machines(self):
+        manager = MachineStateManager(self.client)
+        machine_state0 = yield manager.add_machine_state()
+
+        mock_provider = self.mocker.patch(self.agent.provider)
+        mock_provider.list_machines()
+        self.mocker.result(fail(ProviderInteractionError("Bad")))
+
+        mock_provider.list_machines()
+        self.mocker.passthrough()
+
+        self.mocker.replay()
+        try:
+            yield self.agent.process_machines([machine_state0.id])
+        except:
+            self.fail("Should not raise")
+
+        instance_id = yield machine_state0.get_instance_id()
+        self.assertEqual(instance_id, None)
+
+        yield self.agent.process_machines(
+            [machine_state0.id])
+
+        instance_id = yield machine_state0.get_instance_id()
+        self.assertEqual(instance_id, 0)
+
+    @inlineCallbacks
+    def test_start_agent_with_watch(self):
+        mock_reactor = self.mocker.patch(reactor)
+        mock_reactor.callLater(
+            self.agent.machine_check_period,
+            self.agent.periodic_machine_check)
+        self.mocker.replay()
+
+        self.agent.set_watch_enabled(True)
+        yield self.agent.start()
+
+        manager = MachineStateManager(self.client)
+        machine_state0 = yield manager.add_machine_state()
+
+        # the watch invocation happens out of band, sleep for
+        # a moment so it has a chance to run.
+        yield self.sleep(0.1)
+        instance_id = yield machine_state0.get_instance_id()
+        self.assertEqual(instance_id, 0)

=== modified file 'ensemble/errors.py'
--- ensemble/errors.py	2010-08-27 20:48:57 +0000
+++ ensemble/errors.py	2010-09-17 14:41:43 +0000
@@ -74,3 +74,13 @@
 
 class ProviderError(EnsembleError):
     """Raised when an exception occurs in a provider."""
+
+
+class ProviderInteractionError(ProviderError):
+
+    def __init__(self, error):
+        self.error = error
+
+    def __str__(self):
+        return "ProviderError: Interaction with machine provider failed: %r" \
+               % self.error

=== modified file 'ensemble/ftests/test_ec2_provider.py'
--- ensemble/ftests/test_ec2_provider.py	2010-09-08 17:03:26 +0000
+++ ensemble/ftests/test_ec2_provider.py	2010-09-17 14:41:43 +0000
@@ -100,7 +100,7 @@
         return provider_instances
 
     @inlineCallbacks
-    def xtest_shutdown(self):
+    def test_shutdown(self):
         """
         Shutting down the provider, terminates all instances associated to
         the provider instance.

=== modified file 'ensemble/lib/testing.py'
--- ensemble/lib/testing.py	2010-08-31 16:53:06 +0000
+++ ensemble/lib/testing.py	2010-09-17 14:41:43 +0000
@@ -64,3 +64,6 @@
             os.environ.update(original_environ)
 
         os.environ.update(kw)
+
+    def assertInstance(self, instance, type):
+        self.assertTrue(isinstance(instance, type))

=== added file 'ensemble/lib/tests/test_twistutils.py'
--- ensemble/lib/tests/test_twistutils.py	1970-01-01 00:00:00 +0000
+++ ensemble/lib/tests/test_twistutils.py	2010-09-17 14:41:43 +0000
@@ -0,0 +1,96 @@
+from twisted.internet.defer import (
+    succeed, fail, Deferred, DeferredList, inlineCallbacks, returnValue)
+from twisted.internet import reactor
+
+from ensemble.lib.testing import TestCase
+from ensemble.lib.twistutils import concurrent_execution_guard
+
+
+class Bar(object):
+
+    def __init__(self):
+        self._count = 0
+
+    @concurrent_execution_guard("guard")
+    def my_function(self, a, b=0):
+        """zebra"""
+        return succeed(a / b)
+
+    @concurrent_execution_guard("other_guard")
+    def other_function(self, a):
+        return fail(OSError("Bad"))
+
+    @concurrent_execution_guard("increment_guard")
+    def slow_increment(self, delay=0.1):
+        deferred = Deferred()
+
+        def _increment():
+            self._count += 1
+            return deferred.callback(self._count)
+
+        reactor.callLater(delay, _increment)
+        return deferred
+
+    @concurrent_execution_guard("inline_guard")
+    @inlineCallbacks
+    def inline_increment(self):
+        result = yield self.slow_increment()
+        returnValue(result * 100)
+
+
+class ExecutionGuardTest(TestCase):
+
+    def test_guarded_function_metadata(self):
+        self.assertEqual(Bar().my_function.__name__, "my_function")
+        self.assertEqual(Bar().my_function.__doc__, "zebra")
+
+    def test_guarded_function_failure(self):
+        foo = Bar()
+        return self.assertFailure(foo.other_function("1"), OSError)
+
+    def test_guarded_function_sync_exception(self):
+        foo = Bar()
+        try:
+            result = foo.my_function(1)
+        except:
+            self.fail("Should not raise exception")
+
+        self.assertFailure(result, ZeroDivisionError)
+        self.assertFailure(foo.my_function(1), ZeroDivisionError)
+        self.assertFalse(foo.guard, False)
+
+    def test_guard_multiple_execution(self):
+        foo = Bar()
+
+        d1 = foo.slow_increment()
+        d2 = foo.slow_increment()
+
+        def validate_results(results):
+            success, value = results[0]
+            self.assertTrue(success)
+            self.assertEqual(value, 1)
+
+            success, value = results[1]
+            self.assertTrue(success)
+            self.assertEqual(value, False)
+            return foo.slow_increment()
+
+        def validate_value(results):
+            # if the guard had not prevent execution the value
+            # would be 3.
+            self.assertEqual(results, 2)
+
+        dlist = DeferredList([d1, d2])
+        dlist.addCallback(validate_results)
+        dlist.addCallback(validate_value)
+        return dlist
+
+    def test_guard_w_inline_callbacks(self):
+        foo = Bar()
+
+        def validate_result(result):
+            self.assertEqual(result, 100)
+
+        d = foo.inline_increment()
+        d.addCallback(validate_result)
+        return d

=== added file 'ensemble/lib/twistutils.py'
--- ensemble/lib/twistutils.py	1970-01-01 00:00:00 +0000
+++ ensemble/lib/twistutils.py	2010-09-17 14:41:43 +0000
@@ -0,0 +1,31 @@
+from twisted.internet.defer import maybeDeferred, succeed
+from twisted.python.util import mergeFunctionMetadata
+
+
+def concurrent_execution_guard(attribute):
+    """Sets attribute to True/False during execution of the decorated method.
+
+    Used to ensure non concurrent execution of the decorated function via
+    an instance attribute. *The underlying function must return a defered*.
+    """
+
+    def guard(f):
+
+        def guard_execute(self, *args, **kw):
+            value = getattr(self, attribute, None)
+            if value:
+                return succeed(False)
+            else:
+                setattr(self, attribute, True)
+
+            d = maybeDeferred(f, self, *args, **kw)
+
+            def post_execute(result):
+                setattr(self, attribute, False)
+                return result
+            d.addBoth(post_execute)
+            return d
+
+        return mergeFunctionMetadata(f, guard_execute)
+
+    return guard

=== modified file 'ensemble/providers/common.py'
--- ensemble/providers/common.py	2010-09-16 17:00:40 +0000
+++ ensemble/providers/common.py	2010-09-17 14:41:43 +0000
@@ -1,4 +1,6 @@
 import os
+from twisted.python.failure import Failure
+from ensemble.errors import EnsembleError, ProviderInteractionError
 
 BOOTSTRAP_PACKAGES = [
     "bzr",
@@ -18,7 +20,40 @@
     "python-zookeeper"]
 
 
+<<<<<<< TREE
 def get_user_authorized_keys(config):
+=======
+def convert_unknown_error(failure):
+    """
+    Convert any non ensemble errors to a provider interaction error.
+
+    Supports both usage from within an except clause, and as an
+    errback handler ie. both the following forms are supported.
+
+    ...
+    try:
+       something()
+    except Exception, e:
+       convert_unknown_errors(e)
+
+    ...
+    d.addErrback(convert_unknown_errors)
+    """
+    if isinstance(failure, Failure):
+        error = failure.value
+    else:
+        error = failure
+
+    if not isinstance(error, EnsembleError):
+        error = ProviderInteractionError(error)
+
+    if isinstance(failure, Failure):
+        return Failure(error)
+    raise error
+
+
+def get_user_public_key(config):
+>>>>>>> MERGE-SOURCE
     """
     Locate a public key for the user. Either one explicitly passed
     in or one in the user's .ssh directory.

=== modified file 'ensemble/providers/ec2/__init__.py'
--- ensemble/providers/ec2/__init__.py	2010-09-17 14:09:20 +0000
+++ ensemble/providers/ec2/__init__.py	2010-09-17 14:41:43 +0000
@@ -3,6 +3,7 @@
 
 from txaws.service import AWSServiceRegion, REGION_US
 
+<<<<<<< TREE
 from ensemble.environment.errors import EnvironmentsConfigError
 from ensemble.providers.common import get_user_authorized_keys
 
@@ -12,6 +13,16 @@
 from launch import EC2LaunchMachine, EC2Bootstrap
 from shutdown import EC2Shutdown, EC2ShutdownMachine
 from state import EC2SaveState, EC2LoadState
+=======
+from ensemble.providers.common import convert_unknown_error
+
+from .connect import EC2Connect
+from .files import FileStorage
+from .iterate import EC2MachineIteration
+from .launch import EC2LaunchMachine, EC2Bootstrap
+from .shutdown import EC2Shutdown, EC2ShutdownMachine
+from .state import EC2SaveState, EC2LoadState
+>>>>>>> MERGE-SOURCE
 
 
 class MachineProvider(object):
@@ -35,9 +46,9 @@
                 "and authorized-keys-path. Pick one!")
 
     def get_serialization_data(self):
-        """
-        Return a yaml serialization of the provider configuration within
-        the environment.
+        """Return a dictionary serialization of the provider configuration.
+
+        Additionally this extracts crednetial information from the environment.
         """
         data = copy.deepcopy(self.config)
         data["secret-key"] = self.config.get(
@@ -48,6 +59,11 @@
         data.pop("authorized-keys-path", None)
         return data
 
+    def _run_operation(self, operation, *args, **kw):
+        d = operation.run(*args, **kw)
+        d.addErrback(convert_unknown_error)
+        return d
+
     def connect(self):
         """
         Connect to the zookeeper ensemble running in the machine provider.
@@ -56,7 +72,7 @@
         C{ensemble.storage.connection.TunnelProtocol}
         """
         connect = EC2Connect(self)
-        return connect.run()
+        return self._run_operation(connect)
 
     def get_file_storage(self):
         """Retrieve the provider C{FileStorage} abstraction."""

=== modified file 'ensemble/providers/ec2/launch.py'
--- ensemble/providers/ec2/launch.py	2010-09-16 17:00:40 +0000
+++ ensemble/providers/ec2/launch.py	2010-09-17 14:41:43 +0000
@@ -231,4 +231,9 @@
             admin_identity)
 
         variables["scripts"].append(initialize_script)
+
+        provision_agent_start = "python -m %s -n --zookeeper-servers %s" %(
+            "ensemble.agents.provision", "127.0.0.1:2181")
+        variables["scripts"].append(provision_agent_start)
+
         return variables

=== modified file 'ensemble/providers/ec2/tests/test_launch.py'
--- ensemble/providers/ec2/tests/test_launch.py	2010-09-16 17:00:40 +0000
+++ ensemble/providers/ec2/tests/test_launch.py	2010-09-17 14:41:43 +0000
@@ -287,12 +287,16 @@
             self.assertEqual(
                 config["packages"],
                 list(DEFAULT_PACKAGES) + BOOTSTRAP_PACKAGES)
-            self.failUnlessIn("admin-identity", config["runcmd"][-1])
+            self.failUnlessIn("admin-identity", config["runcmd"][-2])
 
             script = '%s initialize --admin-identity="%s"' % (
                 "sudo /usr/local/bin/ensemble-admin",
                 admin_identity)
 
+            self.assertEqual(config["runcmd"][-2], script)
+
+            script = "python -m %s -n --zookeeper-servers %s" % (
+                "ensemble.agents.provision", "127.0.0.1:2181")
             self.assertEqual(config["runcmd"][-1], script)
             return True
 
@@ -443,4 +447,4 @@
         bootstrap = EC2Bootstrap(provider)
         variables = bootstrap.get_machine_variables()
         identity = make_identity("admin:%s" % config["admin-secret"])
-        self.failUnlessIn(identity, variables["scripts"][-1])
+        self.failUnlessIn(identity, variables["scripts"][-2])

=== modified file 'ensemble/providers/tests/test_common.py'
--- ensemble/providers/tests/test_common.py	2010-09-17 14:09:20 +0000
+++ ensemble/providers/tests/test_common.py	2010-09-17 14:41:43 +0000
@@ -1,7 +1,15 @@
 import os
 
+from twisted.python.failure import Failure
+
 from ensemble.environment.tests.test_config import EnvironmentsConfigTestBase
+<<<<<<< TREE
 from ensemble.providers.common import get_user_authorized_keys
+=======
+from ensemble.errors import ProviderInteractionError, EnsembleError
+from ensemble.providers.common import (
+    get_user_public_key, convert_unknown_error)
+>>>>>>> MERGE-SOURCE
 
 
 class CommonProviderTests(EnvironmentsConfigTestBase):
@@ -46,5 +54,44 @@
 
     def test_invalid_key_specified(self):
         """If an invalid key is specified, a LookupError is raised."""
+<<<<<<< TREE
         config = {"authorized-keys-path": "zebra_moon.pub"}
         self.assertRaises(LookupError, get_user_authorized_keys, config)
+=======
+        config = {"ssh_public_key": "zebra_moon.pub"}
+        self.assertRaises(LookupError, get_user_public_key, config)
+
+    def test_convert_unknown_error(self):
+        error = self.assertRaises(
+            ProviderInteractionError,
+            convert_unknown_error,
+            OSError("Bad"))
+        self.assertInstance(error, EnsembleError)
+        self.assertEqual(
+            str(error),
+            "ProviderError: Interaction with machine provider failed: "
+            "OSError('Bad',)")
+
+    def test_convert_unknown_error_ignores_ensemble_error(self):
+        error = self.assertRaises(
+            EnsembleError,
+            convert_unknown_error,
+            EnsembleError("Magic"))
+        self.assertEqual(
+            str(error),
+            "Magic")
+
+    def test_convert_unknown_error_ignores_ensemble_failure(self):
+        failure = convert_unknown_error(Failure(EnsembleError("Magic")))
+        self.assertTrue(isinstance(failure, Failure))
+        self.assertEqual(failure.value.__class__, EnsembleError)
+
+    def test_convert_unknown_error_with_failure(self):
+        failure = convert_unknown_error(Failure(OSError("Bad")))
+        self.assertTrue(isinstance(failure, Failure))
+        self.assertInstance(failure.value, ProviderInteractionError)
+        self.assertEqual(
+            str(failure.value),
+            "ProviderError: Interaction with machine provider failed: "
+            "OSError('Bad',)")
+>>>>>>> MERGE-SOURCE

=== modified file 'ensemble/state/tests/common.py'
--- ensemble/state/tests/common.py	2010-09-15 13:21:49 +0000
+++ ensemble/state/tests/common.py	2010-09-17 14:41:43 +0000
@@ -57,4 +57,4 @@
         """Non-blocking sleep."""
         deferred = Deferred()
         reactor.callLater(delay, deferred.callback, None)
-        return deferred 
+        return deferred

=== modified file 'ensemble/tests/test_errors.py'
--- ensemble/tests/test_errors.py	2010-08-27 20:48:57 +0000
+++ ensemble/tests/test_errors.py	2010-09-17 14:41:43 +0000
@@ -1,6 +1,6 @@
 from ensemble.errors import (
     EnsembleError, FileNotFound, FileAlreadyExists, InvalidEnsembleHeaderValue,
-    NoConnection, ProviderError)
+    NoConnection, ProviderError, ProviderInteractionError)
 
 from ensemble.lib.testing import TestCase
 
@@ -37,3 +37,11 @@
     def test_ProviderError(self):
         error = ProviderError("Invalid credentials")
         self.assertIsEnsembleError(error)
+
+    def test_ProviderInteractionError(self):
+        error = ProviderInteractionError(OSError("Bad Stuff"))
+        self.assertIsEnsembleError(error)
+        self.assertEquals(
+            str(error),
+            "ProviderError: Interaction with machine provider failed: "
+            "OSError('Bad Stuff',)")