Merge lp:~hazmat/pyjuju/unit-agent-resolved into lp:pyjuju

[Kapil Thangavelu]

This implements the unit lifecycle resolving unit relations, and additional changes to enable resolution (transition actions receiving hook execution flags).

The branch is large enough that i'm going to split the remaining unit agent resolving unit relations into another branch.

[lp:~hazmat/pyjuju/unit-agent-resolved]

269. By Kapil Thangavelu on 2011-05-03

update unit tests to use relation workflow accessor on lifecycle.

270. By Kapil Thangavelu on 2011-05-04

remove passing action transition/state variables

271. By Kapil Thangavelu on 2011-05-04

separate transitions for retry with hook

272. By Kapil Thangavelu on 2011-05-04

merge trunk resolve conflict.

273. By Kapil Thangavelu on 2011-05-04

resolve merge conflict

274. By Kapil Thangavelu on 2011-05-04

expand out additional recovery transition actions

275. By Kapil Thangavelu on 2011-05-04

complete the change over to more transitions, fill out coverage

276. By Kapil Thangavelu on 2011-05-05

Merged ensemble-resolved into unit-agent-resolved.

277. By Kapil Thangavelu on 2011-05-05

Merged ensemble-resolved into unit-agent-resolved.

278. By Kapil Thangavelu on 2011-05-05

Merged ensemble-resolved into unit-agent-resolved.

279. By Kapil Thangavelu on 2011-05-05

address review comments re formatting

Unmerged revisions

=== modified file 'Makefile'
--- Makefile	2011-02-11 14:17:18 +0000
+++ Makefile	2011-05-02 21:57:29 +0000
@@ -31,7 +31,7 @@
 	@test -n "$(modified)" && echo $(modified) | xargs pyflakes
 
 
-modified=$(shell bzr status -S -r ancestor:../trunk |grep -P '^\s*M' | awk '{print $$2;}'| grep -P ".py$$")
+modified=$(shell bzr status -S -r ancestor:../../trunk |grep -P '^\s*M' | awk '{print $$2;}'| grep -P ".py$$")
 review:
 	@test -n "$(modified)" && echo $(modified) | xargs $(PEP8) --repeat
 	@test -n "$(modified)" && echo $(modified) | xargs pyflakes
 
=== modified file 'ensemble/control/__init__.py'
--- ensemble/control/__init__.py	2011-04-25 17:53:19 +0000
+++ ensemble/control/__init__.py	2011-05-02 21:57:29 +0000
@@ -17,6 +17,7 @@
 import open_tunnel
 import remove_relation
 import remove_unit
+import resolved
 import status
 import shutdown
 import ssh
@@ -39,6 +40,7 @@
     open_tunnel,
     remove_relation,
     remove_unit,
+    resolved,
     status,
     ssh,
     shutdown,
 
=== added file 'ensemble/control/resolved.py'
--- ensemble/control/resolved.py	1970-01-01 00:00:00 +0000
+++ ensemble/control/resolved.py	2011-05-02 21:57:29 +0000
@@ -0,0 +1,110 @@
+"""Implementation of resolved subcommand"""
+
+
+from twisted.internet.defer import inlineCallbacks, returnValue
+
+from ensemble.control.utils import get_environment
+
+from ensemble.state.service import ServiceStateManager, RETRY_HOOKS, NO_HOOKS
+from ensemble.state.relation import RelationStateManager
+from ensemble.state.errors import RelationStateNotFound
+from ensemble.unit.workflow import is_unit_running, is_relation_running
+
+
+def configure_subparser(subparsers):
+    """Configure resolved subcommand"""
+    sub_parser = subparsers.add_parser(
+        "resolved", help=command.__doc__, description=resolved.__doc__)
+    sub_parser.add_argument(
+        "--retry", "-r", action="store_true",
+        help="Retry failed hook."),
+    sub_parser.add_argument(
+        "--environment", "-e",
+        help="Ensemble environment to operate in.")
+    sub_parser.add_argument(
+        "service_unit_name",
+        help="Name of the service unit that should be resolved")
+    sub_parser.add_argument(
+       "relation_name", nargs="?", default=None,
+        help="Name of the unit relation that should be resolved")
+    return sub_parser
+
+
+def command(options):
+    """Mark an error as resolved in a unit or unit relation."""
+    environment = get_environment(options)
+    return resolved(
+        options.environments,
+        environment,
+        options.verbose,
+        options.log,
+        options.service_unit_name,
+        options.relation_name,
+        options.retry)
+
+
+@inlineCallbacks
+def resolved(
+    config, environment, verbose, log, unit_name, relation_name, retry):
+    """Mark an error as resolved in a unit or unit relation.
+
+    If one of a unit's formula non-relation hooks returns a non-zero exit
+    status, the entire unit can be considered to be in a non-running state.
+
+    As a resolution, the the unit can be manually returned a running state
+    via the ensemble resolved command. Optionally this command can also
+    rerun the failed hook.
+
+    This resolution also applies separately to each of the unit's relations.
+    If one of the relation-hooks failed. In that case there is no
+    notion of retrying (the change is gone), but resolving will allow
+    additional relation hooks for that relation to proceed.
+    """
+    provider = environment.get_machine_provider()
+    client = yield provider.connect()
+    service_manager = ServiceStateManager(client)
+    relation_manager = RelationStateManager(client)
+
+    unit_state = yield service_manager.get_unit_state(unit_name)
+    service_state = yield service_manager.get_service_state(
+        unit_name.split("/")[0])
+
+    retry = retry and RETRY_HOOKS or NO_HOOKS
+
+    if not relation_name:
+        running, workflow_state = yield is_unit_running(client, unit_state)
+        if running:
+            log.info("Unit %r already running: %s", unit_name, workflow_state)
+            client.close()
+            returnValue(False)
+
+        yield unit_state.set_resolved(retry)
+        log.info("Marked unit %r as resolved", unit_name)
+        returnValue(True)
+
+    # Check for the matching relations
+    service_relations = yield relation_manager.get_relations_for_service(
+        service_state)
+    service_relations = [
+        sr for sr in service_relations if sr.relation_name == relation_name]
+    if not service_relations:
+        raise RelationStateNotFound()
+
+    # Verify the relations are in need of resolution.
+    resolve_relations = {}
+    for service_relation in service_relations:
+        unit_relation = yield service_relation.get_unit_state(unit_state)
+        running, state = yield is_relation_running(client, unit_relation)
+        if not running:
+            resolve_relations[unit_relation.internal_relation_id] = retry
+
+    if not resolve_relations:
+        log.warning("Matched relations are all running")
+        client.close()
+        returnValue(False)
+
+    # Mark the relations for resolution.
+    yield unit_state.set_relation_resolved(resolve_relations)
+    log.info(
+        "Marked unit %r relation %r as resolved", unit_name, relation_name)
+    client.close()
 
=== added file 'ensemble/control/tests/test_resolved.py'
--- ensemble/control/tests/test_resolved.py	1970-01-01 00:00:00 +0000
+++ ensemble/control/tests/test_resolved.py	2011-05-02 21:57:29 +0000
@@ -0,0 +1,390 @@
+from twisted.internet.defer import inlineCallbacks, returnValue
+from yaml import dump
+
+from ensemble.control import main
+from ensemble.control.resolved import resolved
+from ensemble.control.tests.common import ControlToolTest
+from ensemble.formula.tests.test_repository import RepositoryTestBase
+
+from ensemble.state.service import RETRY_HOOKS, NO_HOOKS
+from ensemble.state.tests.test_service import ServiceStateManagerTestBase
+from ensemble.state.errors import ServiceStateNotFound
+
+from ensemble.unit.workflow import UnitWorkflowState, RelationWorkflowState
+from ensemble.unit.lifecycle import UnitRelationLifecycle
+from ensemble.hooks.executor import HookExecutor
+
+
+class ControlResolvedTest(
+    ServiceStateManagerTestBase, ControlToolTest, RepositoryTestBase):
+
+    @inlineCallbacks
+    def setUp(self):
+        yield super(ControlResolvedTest, self).setUp()
+        config = {
+            "ensemble": "environments",
+            "environments": {"firstenv": {"type": "dummy"}}}
+
+        self.write_config(dump(config))
+        self.config.load()
+
+        yield self.add_relation_state("wordpress", "mysql")
+        yield self.add_relation_state("wordpress", "varnish")
+
+        self.service1 = yield self.service_state_manager.get_service_state(
+            "mysql")
+        self.service_unit1 = yield self.service1.add_unit_state()
+        self.service_unit2 = yield self.service1.add_unit_state()
+
+        self.unit1_workflow = UnitWorkflowState(
+            self.client, self.service_unit1, None, self.makeDir())
+        yield self.unit1_workflow.set_state("started")
+
+        self.environment = self.config.get_default()
+        self.provider = self.environment.get_machine_provider()
+
+        self.output = self.capture_logging()
+        self.stderr = self.capture_stream("stderr")
+        self.executor = HookExecutor()
+
+    @inlineCallbacks
+    def add_relation_state(self, *service_names):
+        for service_name in service_names:
+            try:
+                yield self.service_state_manager.get_service_state(
+                    service_name)
+            except ServiceStateNotFound:
+                yield self.add_service_from_formula(service_name)
+
+        endpoint_pairs = yield self.service_state_manager.join_descriptors(
+            *service_names)
+        endpoints = endpoint_pairs[0]
+        endpoints = endpoint_pairs[0]
+        if endpoints[0] == endpoints[1]:
+            endpoints = endpoints[0:1]
+        relation_state = (yield self.relation_state_manager.add_relation_state(
+            *endpoints))[0]
+        returnValue(relation_state)
+
+    @inlineCallbacks
+    def get_named_service_relation(self, service_state, relation_name):
+        if isinstance(service_state, str):
+            service_state = yield self.service_state_manager.get_service_state(
+                service_state)
+
+        rels = yield self.relation_state_manager.get_relations_for_service(
+            service_state)
+
+        rels = [sr for sr in rels if sr.relation_name == relation_name]
+        if len(rels) == 1:
+            returnValue(rels[0])
+        returnValue(rels)
+
+    @inlineCallbacks
+    def setup_unit_relations(self, service_relation, *units):
+        """
+        Given a service relation and set of unit tuples in the form
+        unit_state, unit_relation_workflow_state, will add unit relations
+        for these units and update their workflow state to the desired/given
+        state.
+        """
+        for unit, state in units:
+            unit_relation = yield service_relation.add_unit_state(unit)
+            lifecycle = UnitRelationLifecycle(
+                self.client, unit_relation, service_relation.relation_name,
+                self.makeDir(), self.executor)
+            workflow_state = RelationWorkflowState(
+                self.client, unit_relation, lifecycle, self.makeDir())
+            yield workflow_state.set_state(state)
+
+    @inlineCallbacks
+    def test_resolved(self):
+        """
+        'ensemble resolved <unit_name>' will schedule a unit for
+        retrying from an error state.
+        """
+        # Push the unit into an error state
+        yield self.unit1_workflow.set_state("start_error")
+        self.setup_exit(0)
+        finished = self.setup_cli_reactor()
+        self.mocker.replay()
+
+        self.assertEqual(
+            (yield self.service_unit1.get_resolved()), None)
+
+        main(["resolved", "mysql/0"])
+        yield finished
+
+        self.assertEqual(
+            (yield self.service_unit1.get_resolved()), {"retry": NO_HOOKS})
+        self.assertIn(
+            "Marked unit 'mysql/0' as resolved",
+            self.output.getvalue())
+
+    @inlineCallbacks
+    def test_resolved_retry(self):
+        """
+        'ensemble resolved --retry <unit_name>' will schedule a unit
+        for retrying from an error state with a retry of hooks
+        executions.
+        """
+        yield self.unit1_workflow.set_state("start_error")
+        self.setup_exit(0)
+        finished = self.setup_cli_reactor()
+        self.mocker.replay()
+
+        self.assertEqual(
+            (yield self.service_unit1.get_resolved()), None)
+
+        main(["resolved", "--retry", "mysql/0"])
+        yield finished
+
+        self.assertEqual(
+            (yield self.service_unit1.get_resolved()), {"retry": RETRY_HOOKS})
+        self.assertIn(
+            "Marked unit 'mysql/0' as resolved",
+            self.output.getvalue())
+
+    @inlineCallbacks
+    def test_relation_resolved(self):
+        """
+        'ensemble relation <unit_name> <rel_name>' will schedule
+        the broken unit relations for being resolved.
+        """
+        service_relation = yield self.get_named_service_relation(
+            self.service1, "server")
+
+        self.setup_unit_relations(
+            service_relation,
+            (self.service_unit1, "down"),
+            (self.service_unit2, "up"))
+
+        yield self.unit1_workflow.set_state("start_error")
+        self.setup_exit(0)
+        finished = self.setup_cli_reactor()
+        self.mocker.replay()
+
+        self.assertEqual(
+            (yield self.service_unit1.get_relation_resolved()), None)
+
+        main(["resolved", "--retry", "mysql/0",
+              service_relation.relation_name])
+        yield finished
+
+        self.assertEqual(
+            (yield self.service_unit1.get_relation_resolved()),
+            {service_relation.internal_relation_id: RETRY_HOOKS})
+        self.assertEqual(
+            (yield self.service_unit2.get_relation_resolved()),
+            None)
+        self.assertIn(
+            "Marked unit 'mysql/0' relation 'server' as resolved",
+            self.output.getvalue())
+
+    @inlineCallbacks
+    def test_resolved_relation_some_already_resolved(self):
+        """
+        'ensemble resolved <service_name> <rel_name>' will mark
+        resolved all down units that are not already marked resolved.
+        """
+
+        service2 = yield self.service_state_manager.get_service_state(
+            "wordpress")
+        service_unit1 = yield service2.add_unit_state()
+
+        service_relation = yield self.get_named_service_relation(
+            service2, "db")
+        yield self.setup_unit_relations(
+            service_relation, (service_unit1, "down"))
+
+        service_relation2 = yield self.get_named_service_relation(
+            service2, "cache")
+        yield self.setup_unit_relations(
+            service_relation2, (service_unit1, "down"))
+
+        yield service_unit1.set_relation_resolved(
+            {service_relation.internal_relation_id: NO_HOOKS})
+
+        self.setup_exit(0)
+        finished = self.setup_cli_reactor()
+        self.mocker.replay()
+
+        main(["resolved", "--retry", "wordpress/0", "cache"])
+        yield finished
+
+        self.assertEqual(
+            (yield service_unit1.get_relation_resolved()),
+            {service_relation.internal_relation_id: NO_HOOKS,
+             service_relation2.internal_relation_id: RETRY_HOOKS})
+
+        self.assertIn(
+            "Marked unit 'wordpress/0' relation 'cache' as resolved",
+            self.output.getvalue())
+
+    @inlineCallbacks
+    def test_resolved_relation_some_already_resolved_conflict(self):
+        """
+        'ensemble resolved <service_name> <rel_name>' will mark
+        resolved all down units that are not already marked resolved.
+        """
+
+        service2 = yield self.service_state_manager.get_service_state(
+            "wordpress")
+        service_unit1 = yield service2.add_unit_state()
+
+        service_relation = yield self.get_named_service_relation(
+            service2, "db")
+        yield self.setup_unit_relations(
+            service_relation, (service_unit1, "down"))
+
+        yield service_unit1.set_relation_resolved(
+            {service_relation.internal_relation_id: NO_HOOKS})
+
+        self.setup_exit(0)
+        finished = self.setup_cli_reactor()
+        self.mocker.replay()
+
+        main(["resolved", "--retry", "wordpress/0", "db"])
+        yield finished
+
+        self.assertEqual(
+            (yield service_unit1.get_relation_resolved()),
+            {service_relation.internal_relation_id: NO_HOOKS})
+
+        self.assertIn(
+            "Service unit 'wordpress/0' already has relations marked as resol",
+            self.output.getvalue())
+
+    @inlineCallbacks
+    def test_resolved_unknown_service(self):
+        """
+        'ensemble resolved <unit_name>' will report if a service is
+        invalid.
+        """
+        self.setup_exit(0)
+        finished = self.setup_cli_reactor()
+        self.mocker.replay()
+        main(["resolved", "zebra/0"])
+        yield finished
+        self.assertIn("Service 'zebra' was not found", self.stderr.getvalue())
+
+    @inlineCallbacks
+    def test_resolved_unknown_unit(self):
+        """
+        'ensemble resolved <unit_name>' will report if a unit is
+        invalid.
+        """
+        self.setup_exit(0)
+        finished = self.setup_cli_reactor()
+        self.mocker.replay()
+        main(["resolved", "mysql/5"])
+        yield finished
+        self.assertIn(
+            "Service unit 'mysql/5' was not found", self.output.getvalue())
+
+    @inlineCallbacks
+    def test_resolved_unknown_unit_relation(self):
+        """
+        'ensemble resolved <unit_name>' will report if a relation is
+        invalid.
+        """
+        self.setup_exit(0)
+        finished = self.setup_cli_reactor()
+        self.mocker.replay()
+
+        self.assertEqual(
+            (yield self.service_unit1.get_resolved()), None)
+
+        main(["resolved", "mysql/0", "magic"])
+        yield finished
+
+        self.assertIn("Relation not found", self.output.getvalue())
+
+    @inlineCallbacks
+    def test_resolved_already_running(self):
+        """
+        'ensemble resolved <unit_name>' will report if
+        the unit is already running.
+        """
+        # Just verify we don't accidentally mark up another unit of the service
+        unit2_workflow = UnitWorkflowState(
+            self.client, self.service_unit2, None, self.makeDir())
+        unit2_workflow.set_state("start_error")
+
+        self.setup_exit(0)
+        finished = self.setup_cli_reactor()
+        self.mocker.replay()
+
+        main(["resolved", "mysql/0"])
+        yield finished
+
+        self.assertEqual(
+            (yield self.service_unit2.get_resolved()), None)
+        self.assertEqual(
+            (yield self.service_unit1.get_resolved()), None)
+
+        self.assertNotIn(
+            "Unit 'mysql/0 already running: started",
+            self.output.getvalue())
+
+    @inlineCallbacks
+    def test_resolved_already_resolved(self):
+        """
+        'ensemble resolved <unit_name>' will report if
+        the unit is already resolved.
+        """
+        # Mark the unit as resolved and as in an error state.
+        yield self.service_unit1.set_resolved(RETRY_HOOKS)
+        yield self.unit1_workflow.set_state("start_error")
+
+        unit2_workflow = UnitWorkflowState(
+            self.client, self.service_unit1, None, self.makeDir())
+        unit2_workflow.set_state("start_error")
+
+        self.assertEqual(
+            (yield self.service_unit2.get_resolved()), None)
+
+        self.setup_exit(0)
+        finished = self.setup_cli_reactor()
+        self.mocker.replay()
+
+        main(["resolved", "mysql/0"])
+        yield finished
+
+        self.assertEqual(
+            (yield self.service_unit1.get_resolved()),
+            {"retry": RETRY_HOOKS})
+        self.assertNotIn(
+            "Marked unit 'mysql/0' as resolved",
+            self.output.getvalue())
+        self.assertIn(
+            "Service unit 'mysql/0' is already marked as resolved.",
+            self.stderr.getvalue(), "")
+
+    @inlineCallbacks
+    def test_resolved_relation_already_running(self):
+        """
+        'ensemble resolved <unit_name> <rel_name>' will report
+        if the relation is already running.
+        """
+        service2 = yield self.service_state_manager.get_service_state(
+            "wordpress")
+        service_unit1 = yield service2.add_unit_state()
+
+        service_relation = yield self.get_named_service_relation(
+            service2, "db")
+        yield self.setup_unit_relations(
+            service_relation, (service_unit1, "up"))
+
+        self.setup_exit(0)
+        finished = self.setup_cli_reactor()
+        self.mocker.replay()
+
+        main(["resolved", "wordpress/0", "db"])
+        yield finished
+
+        self.assertIn("Matched relations are all running",
+                      self.output.getvalue())
+        self.assertEqual(
+            (yield service_unit1.get_relation_resolved()),
+            None)
 
=== modified file 'ensemble/lib/statemachine.py'
--- ensemble/lib/statemachine.py	2011-04-15 20:20:57 +0000
+++ ensemble/lib/statemachine.py	2011-05-02 21:57:29 +0000
@@ -61,7 +61,7 @@
         returnValue(self._workflow.get_transitions(state_id))
 
     @inlineCallbacks
-    def fire_transition_alias(self, transition_alias):
+    def fire_transition_alias(self, transition_alias, **transition_variables):
         """Fire a transition with the matching alias.
 
         A transition from the current state with the given alias will
@@ -81,6 +81,9 @@
 
         Ambigious (multiple) or no matching transitions cause an exception
         InvalidTransition to be raised.
+
+        Keyword args are used as transition variables and given to the
+        associated transtion action.
         """
 
         found = []
@@ -100,7 +103,8 @@
                 "No transition found for alias:%s state:%s" % (
                     transition_alias, current_state))
 
-        value = yield self.fire_transition(found[0].transition_id)
+        value = yield self.fire_transition(
+            found[0].transition_id, **transition_variables)
         returnValue(value)
 
     @inlineCallbacks
@@ -135,7 +139,8 @@
         returnValue(False)
 
     @inlineCallbacks
-    def fire_transition(self, transition_id, **state_variables):
+    def fire_transition(
+        self, transition_id, **transition_variables):
         """Fire a transition with given id.
 
         Invokes any transition actions, saves state and state variables, and
@@ -156,10 +161,9 @@
                   transition_id,
                   transition.source,
                   transition.destination,
-                  state_variables)
+                  transition_variables)
 
         # Execute any per transition action.
-        state_variables = state_variables
         action_id = "do_%s" % transition_id
         action = getattr(self, action_id, None)
 
@@ -167,9 +171,9 @@
             try:
                 log.debug("%s:  execute action %s",
                           class_name(self), action.__name__)
-                variables = yield action()
+                variables = yield action(**transition_variables)
                 if isinstance(variables, dict):
-                    state_variables.update(variables)
+                    transition_variables.update(variables)
             except TransitionError, e:
                 # If an error happens during the transition, allow for
                 # executing an error transition.
@@ -187,14 +191,16 @@
                 returnValue(False)
 
         # Set the state with state variables
-        yield self.set_state(transition.destination, **state_variables)
+        yield self.set_state(transition.destination, **transition_variables)
         log.debug("%s: transition complete %s (state %s) %r",
                   class_name(self), transition_id,
-                  transition.destination, state_variables)
+                  transition.destination, transition_variables)
         if transition.success_transition_id:
             log.debug("%s: initiating success transition: %s",
                       class_name(self), transition.success_transition_id)
-            yield self.fire_transition(transition.success_transition_id)
+            yield self.fire_transition(
+                transition.success_transition_id,
+                **transition_variables)
         returnValue(True)
 
     @inlineCallbacks
 
=== modified file 'ensemble/lib/tests/test_statemachine.py'
--- ensemble/lib/tests/test_statemachine.py	2011-04-15 20:20:57 +0000
+++ ensemble/lib/tests/test_statemachine.py	2011-05-02 21:57:29 +0000
@@ -295,6 +295,11 @@
             Transition("continue", "", "next-state", "final-state"))
 
         workflow_state = AttributeWorkflowState(workflow)
+        results = []
+
+        def do_begin(*args, **kw):
+            results.append(kw)
+        workflow_state.do_begin = do_begin
 
         yield workflow_state.fire_transition(
             "begin", rabbit="moon", hello=True)
@@ -303,6 +308,10 @@
         variables = yield workflow_state.get_state_variables()
         self.assertEqual({"rabbit": "moon", "hello": True}, variables)
 
+        # Ensure the variables made it through to the transition action
+        self.assertEqual(
+            {"rabbit": "moon", "hello": True}, results[0])
+
         yield workflow_state.fire_transition("continue")
         current_state = yield workflow_state.get_state()
         self.assertEqual(current_state, "final-state")
 
=== modified file 'ensemble/state/errors.py'
--- ensemble/state/errors.py	2011-04-28 17:52:55 +0000
+++ ensemble/state/errors.py	2011-05-02 21:57:29 +0000
@@ -151,6 +151,30 @@
             self.unit_name
 
 
+class ServiceUnitResolvedAlreadyEnabled(StateError):
+    """The unit has already been marked resolved.
+    """
+
+    def __init__(self, unit_name):
+        self.unit_name = unit_name
+
+    def __str__(self):
+        return "Service unit %r is already marked as resolved." % (
+            self.unit_name)
+
+
+class ServiceUnitRelationResolvedAlreadyEnabled(StateError):
+    """The unit has already been marked resolved.
+    """
+
+    def __init__(self, unit_name):
+        self.unit_name = unit_name
+
+    def __str__(self):
+        return "Service unit %r already has relations marked as resolved." % (
+            self.unit_name)
+
+
 class RelationAlreadyExists(StateError):
 
     def __init__(self, *endpoints):
 
=== modified file 'ensemble/state/service.py'
--- ensemble/state/service.py	2011-04-27 16:23:16 +0000
+++ ensemble/state/service.py	2011-05-02 21:57:29 +0000
@@ -15,11 +15,16 @@
     StateChanged, ServiceStateNotFound, ServiceUnitStateNotFound,
     ServiceUnitStateMachineAlreadyAssigned, ServiceStateNameInUse,
     BadDescriptor, BadServiceStateName, NoUnusedMachines,
-    ServiceUnitDebugAlreadyEnabled)
+    ServiceUnitDebugAlreadyEnabled, ServiceUnitResolvedAlreadyEnabled,
+    ServiceUnitRelationResolvedAlreadyEnabled, StopWatcher)
 from ensemble.state.formula import FormulaStateManager
 from ensemble.state.relation import ServiceRelationState, RelationStateManager
 from ensemble.state.machine import _public_machine_id, MachineState
-from ensemble.state.utils import remove_tree
+from ensemble.state.utils import remove_tree, dict_merge
+
+
+RETRY_HOOKS = 1000
+NO_HOOKS = 1001
 
 
 class ServiceStateManager(StateBase):
@@ -855,6 +860,185 @@
         callback_d.addCallback(
             lambda x: watch_d.addCallback(watcher) and x)
 
+    @property
+    def _unit_resolve_path(self):
+        return "/units/%s/resolved" % self.internal_id
+
+    @inlineCallbacks
+    def set_resolved(self, retry):
+        """Mark the unit as in need of being resolved.
+
+        :param retry: A boolean denoting if hooks should fire as a result
+        of the retry.
+
+        The resolved setting is set by the command line to inform
+        a unit to attempt an retry transition from an error state.
+        """
+
+        if not retry in (RETRY_HOOKS, NO_HOOKS):
+            raise ValueError("invalid retry value %r" % retry)
+
+        try:
+            yield self._client.create(
+                self._unit_resolve_path, yaml.safe_dump({"retry": retry}))
+        except zookeeper.NodeExistsException:
+            raise ServiceUnitResolvedAlreadyEnabled(self.unit_name)
+
+    @inlineCallbacks
+    def get_resolved(self):
+        """Get the value of the resolved setting if any.
+
+        The resolved setting is retrieved by the unit agent and if
+        found instructs it to attempt an retry transition from an
+        error state.
+        """
+        try:
+            content, stat = yield self._client.get(self._unit_resolve_path)
+        except zookeeper.NoNodeException:
+            # Return a default value.
+            returnValue(None)
+        returnValue(yaml.load(content))
+
+    @inlineCallbacks
+    def clear_resolved(self):
+        """Remove any resolved setting on the unit."""
+        try:
+            yield self._client.delete(self._unit_resolve_path)
+        except zookeeper.NoNodeException:
+            # We get to the same end state.
+            pass
+
+    @inlineCallbacks
+    def watch_resolved(self, callback):
+        """Set a callback to be invoked when an unit is marked resolved.
+
+        :param callback: The callback recieves a single parameter, the
+               change event. The watcher always recieve an initial
+               boolean value invocation denoting the existence of the
+               resolved setting. Subsequent invocations will be with change
+               events.
+        """
+        @inlineCallbacks
+        def watcher(change_event):
+            if not self._client.connected:
+                returnValue(None)
+
+            exists_d, watch_d = self._client.exists_and_watch(
+                self._unit_resolve_path)
+            try:
+                yield callback(change_event)
+            except StopWatcher:
+                returnValue(None)
+            watch_d.addCallback(watcher)
+
+        exists_d, watch_d = self._client.exists_and_watch(
+            self._unit_resolve_path)
+        exists = yield exists_d
+
+        # Setup the watch deferred callback after the user defined callback
+        # has returned successfully from the existence invocation.
+        callback_d = maybeDeferred(callback, bool(exists))
+        callback_d.addCallback(
+            lambda x: watch_d.addCallback(watcher) and x)
+        callback_d.addErrback(
+            lambda failure: failure.trap(StopWatcher))
+
+    @property
+    def _relation_resolved_path(self):
+        return "/units/%s/relation-resolved" % self.internal_id
+
+    @inlineCallbacks
+    def set_relation_resolved(self, relation_map):
+        """Mark a unit's relations as in need of being resolved.
+
+        :param relation_map: A map of internal relation ids, to retry hook
+              values either ensemble.state.service.NO_HOOKS or
+              RETRY_HOOKS.
+        """
+        if not isinstance(relation_map, dict):
+            raise ValueError(
+                "Relation map must be a dictionary %r" % relation_map)
+
+        if [v for v in relation_map.values() if v not in (
+            RETRY_HOOKS, NO_HOOKS)]:
+
+            print relation_map
+            raise ValueError("Invalid setting for retry hook")
+
+        def update_relation_resolved(content, stat):
+            if not content:
+                return yaml.safe_dump(relation_map)
+
+            content = yaml.safe_dump(
+                dict_merge(yaml.load(content), relation_map))
+            return content
+
+        try:
+            yield retry_change(
+                self._client,
+                self._relation_resolved_path,
+                update_relation_resolved)
+        except StateChanged:
+            raise ServiceUnitRelationResolvedAlreadyEnabled(self.unit_name)
+        returnValue(True)
+
+    @inlineCallbacks
+    def get_relation_resolved(self):
+        """Retrieve any resolved flags set for this unit's relations.
+        """
+        try:
+            content, stat = yield self._client.get(
+                self._relation_resolved_path)
+        except zookeeper.NoNodeException:
+            returnValue(None)
+        returnValue(yaml.load(content))
+
+    @inlineCallbacks
+    def clear_relation_resolved(self):
+        """ Clear the relation resolved setting.
+        """
+        try:
+            yield self._client.delete(self._relation_resolved_path)
+        except zookeeper.NoNodeException:
+            # We get to the same end state.
+            pass
+
+    @inlineCallbacks
+    def watch_relation_resolved(self, callback):
+        """Set a callback to be invoked when a unit's relations are resolved.
+
+        :param callback: The callback recieves a single parameter, the
+               change event. The watcher always recieve an initial
+               boolean value invocation denoting the existence of the
+               resolved setting. Subsequent invocations will be with change
+               events.
+        """
+        @inlineCallbacks
+        def watcher(change_event):
+            if not self._client.connected:
+                returnValue(None)
+            exists_d, watch_d = self._client.exists_and_watch(
+                self._relation_resolved_path)
+            try:
+                yield callback(change_event)
+            except StopWatcher:
+                returnValue(None)
+
+            watch_d.addCallback(watcher)
+
+        exists_d, watch_d = self._client.exists_and_watch(
+            self._relation_resolved_path)
+
+        exists = yield exists_d
+
+        # Setup the watch deferred callback after the user defined callback
+        # has returned successfully from the existence invocation.
+        callback_d = maybeDeferred(callback, bool(exists))
+        callback_d.addCallback(
+            lambda x: watch_d.addCallback(watcher) and x)
+        callback_d.addErrback(
+            lambda failure: failure.trap(StopWatcher))
+
 
 def _parse_unit_name(unit_name):
     """Parse a unit's name into the service name and its sequence.
 
=== modified file 'ensemble/state/tests/test_errors.py'
--- ensemble/state/tests/test_errors.py	2011-02-23 17:47:46 +0000
+++ ensemble/state/tests/test_errors.py	2011-05-02 21:57:29 +0000
@@ -11,7 +11,9 @@
     UnitRelationStateAlreadyAssigned, UnknownRelationRole,
     BadDescriptor, DuplicateEndpoints, IncompatibleEndpoints,
     NoMatchingEndpoints, AmbiguousEndpoints,
-    ServiceUnitStateMachineNotAssigned, ServiceUnitDebugAlreadyEnabled)
+    ServiceUnitStateMachineNotAssigned, ServiceUnitDebugAlreadyEnabled,
+    ServiceUnitResolvedAlreadyEnabled,
+    ServiceUnitRelationResolvedAlreadyEnabled)
 
 
 class StateErrorsTest(TestCase):
@@ -89,6 +91,23 @@
             str(error),
             "Service unit 'wordpress/0' is already in debug mode.")
 
+    def test_unit_already_in_resolved_mode(self):
+        error = ServiceUnitResolvedAlreadyEnabled("wordpress/0")
+        self.assertIsStateError(error)
+        self.assertEquals(error.unit_name, "wordpress/0")
+        self.assertEquals(
+            str(error),
+            "Service unit 'wordpress/0' is already marked as resolved.")
+
+    def test_unit_already_in_relation_resolved_mode(self):
+        error = ServiceUnitRelationResolvedAlreadyEnabled("wordpress/0")
+        self.assertIsStateError(error)
+        self.assertEquals(error.unit_name, "wordpress/0")
+        self.assertEquals(
+            str(error),
+            "Service unit %r already has relations marked as resolved." % (
+                "wordpress/0"))
+
     def test_service_name_in_use(self):
         error = ServiceStateNameInUse("wordpress")
         self.assertIsStateError(error)
 
=== modified file 'ensemble/state/tests/test_service.py'
--- ensemble/state/tests/test_service.py	2011-04-23 01:13:27 +0000
+++ ensemble/state/tests/test_service.py	2011-05-02 21:57:29 +0000
@@ -9,14 +9,15 @@
 from ensemble.formula.tests.test_metadata import test_repository_path
 from ensemble.state.endpoint import RelationEndpoint
 from ensemble.state.formula import FormulaStateManager
-from ensemble.state.service import ServiceStateManager
+from ensemble.state.service import ServiceStateManager, NO_HOOKS, RETRY_HOOKS
 from ensemble.state.machine import MachineStateManager
 from ensemble.state.relation import RelationStateManager
 from ensemble.state.errors import (
     StateChanged, ServiceStateNotFound, ServiceUnitStateNotFound,
     ServiceUnitStateMachineAlreadyAssigned, ServiceStateNameInUse,
     BadDescriptor, BadServiceStateName, ServiceUnitDebugAlreadyEnabled,
-    MachineStateNotFound, NoUnusedMachines)
+    MachineStateNotFound, NoUnusedMachines, ServiceUnitResolvedAlreadyEnabled,
+    ServiceUnitRelationResolvedAlreadyEnabled, StopWatcher)
 
 
 from ensemble.state.tests.common import StateTestBase
@@ -738,6 +739,283 @@
             None)
 
     @inlineCallbacks
+    def test_get_set_clear_resolved(self):
+        """The a unit can be set to resolved to mark a future transition, with
+        an optional retry flag."""
+
+        unit_state = yield self.get_unit_state()
+
+        self.assertIdentical((yield unit_state.get_resolved()), None)
+        yield unit_state.set_resolved(NO_HOOKS)
+
+        yield self.assertFailure(
+            unit_state.set_resolved(NO_HOOKS),
+            ServiceUnitResolvedAlreadyEnabled)
+        yield self.assertEqual(
+
+                (yield unit_state.get_resolved()), {"retry": NO_HOOKS})
+
+        yield unit_state.clear_resolved()
+        self.assertIdentical((yield unit_state.get_resolved()), None)
+        yield unit_state.clear_resolved()
+
+        yield self.assertFailure(unit_state.set_resolved(None), ValueError)
+
+    @inlineCallbacks
+    def test_watch_resolved(self):
+        """A unit resolved watch can be instituted on a permanent basis."""
+        unit_state = yield self.get_unit_state()
+
+        results = []
+
+        def callback(value):
+            results.append(value)
+
+        unit_state.watch_resolved(callback)
+        yield unit_state.set_resolved(RETRY_HOOKS)
+        yield unit_state.clear_resolved()
+        yield unit_state.set_resolved(NO_HOOKS)
+
+        yield self.poke_zk()
+
+        self.assertEqual(len(results), 4)
+        self.assertIdentical(results.pop(0), False)
+        self.assertEqual(results.pop(0).type_name, "created")
+        self.assertEqual(results.pop(0).type_name, "deleted")
+        self.assertEqual(results.pop(0).type_name, "created")
+
+        self.assertEqual(
+            (yield unit_state.get_resolved()),
+            {"retry": NO_HOOKS})
+
+    @inlineCallbacks
+    def test_stop_watch_resolved(self):
+        """A unit resolved watch can be instituted on a permanent basis.
+
+        However the callback can raise StopWatcher at anytime to stop the watch
+        """
+        unit_state = yield self.get_unit_state()
+
+        results = []
+
+        def callback(value):
+            results.append(value)
+            if len(results) == 1:
+                raise StopWatcher()
+            if len(results) == 3:
+                raise StopWatcher()
+
+        unit_state.watch_resolved(callback)
+        yield unit_state.set_resolved(RETRY_HOOKS)
+        yield unit_state.clear_resolved()
+        yield self.poke_zk()
+
+        unit_state.watch_resolved(callback)
+        yield unit_state.set_resolved(NO_HOOKS)
+        yield unit_state.clear_resolved()
+
+        yield self.poke_zk()
+
+        self.assertEqual(len(results), 3)
+        self.assertIdentical(results.pop(0), False)
+        self.assertIdentical(results.pop(0), False)
+        self.assertEqual(results.pop(0).type_name, "created")
+
+        self.assertEqual(
+            (yield unit_state.get_resolved()), None)
+
+    @inlineCallbacks
+    def test_get_set_clear_relation_resolved(self):
+        """The a unit's realtions can be set to resolved to mark a
+        future transition, with an optional retry flag."""
+
+        unit_state = yield self.get_unit_state()
+
+        self.assertIdentical((yield unit_state.get_relation_resolved()), None)
+        yield unit_state.set_relation_resolved({"0": RETRY_HOOKS})
+
+        # Trying to set a conflicting raises an error
+        yield self.assertFailure(
+            unit_state.set_relation_resolved({"0": NO_HOOKS}),
+            ServiceUnitRelationResolvedAlreadyEnabled)
+
+        # Doing the same thing is fine
+        yield unit_state.set_relation_resolved({"0": RETRY_HOOKS}),
+
+        # Its fine to put in new values
+        yield unit_state.set_relation_resolved({"21": RETRY_HOOKS})
+        yield self.assertEqual(
+            (yield unit_state.get_relation_resolved()),
+            {"0": RETRY_HOOKS, "21": RETRY_HOOKS})
+
+        yield unit_state.clear_relation_resolved()
+        self.assertIdentical((yield unit_state.get_relation_resolved()), None)
+        yield unit_state.clear_relation_resolved()
+
+        yield self.assertFailure(
+            unit_state.set_relation_resolved(True), ValueError)
+        yield self.assertFailure(
+            unit_state.set_relation_resolved(None), ValueError)
+
+    @inlineCallbacks
+    def test_watch_relation_resolved(self):
+        """A unit resolved watch can be instituted on a permanent basis."""
+        unit_state = yield self.get_unit_state()
+
+        results = []
+
+        def callback(value):
+            results.append(value)
+
+        unit_state.watch_relation_resolved(callback)
+        yield unit_state.set_relation_resolved({"0": RETRY_HOOKS})
+        yield unit_state.clear_relation_resolved()
+        yield unit_state.set_relation_resolved({"0": NO_HOOKS})
+
+        yield self.poke_zk()
+
+        self.assertEqual(len(results), 4)
+        self.assertIdentical(results.pop(0), False)
+        self.assertEqual(results.pop(0).type_name, "created")
+        self.assertEqual(results.pop(0).type_name, "deleted")
+        self.assertEqual(results.pop(0).type_name, "created")
+
+        self.assertEqual(
+            (yield unit_state.get_relation_resolved()),
+            {"0": NO_HOOKS})
+
+    @inlineCallbacks
+    def test_stop_watch_relation_resolved(self):
+        """A unit resolved watch can be instituted on a permanent basis."""
+        unit_state = yield self.get_unit_state()
+
+        results = []
+
+        def callback(value):
+            results.append(value)
+
+            if len(results) == 1:
+                raise StopWatcher()
+
+            if len(results) == 3:
+                raise StopWatcher()
+
+        unit_state.watch_relation_resolved(callback)
+        yield unit_state.set_relation_resolved({"0": RETRY_HOOKS})
+        yield unit_state.clear_relation_resolved()
+        yield self.poke_zk()
+        self.assertEqual(len(results), 1)
+
+        unit_state.watch_relation_resolved(callback)
+        yield unit_state.set_relation_resolved({"0": RETRY_HOOKS})
+        yield unit_state.clear_relation_resolved()
+        yield self.poke_zk()
+        self.assertEqual(len(results), 3)
+        self.assertIdentical(results.pop(0), False)
+        self.assertIdentical(results.pop(0), False)
+        self.assertEqual(results.pop(0).type_name, "created")
+
+        self.assertEqual(
+            (yield unit_state.get_relation_resolved()), None)
+
+    @inlineCallbacks
+    def test_watch_resolved_slow_callback(self):
+        """A slow watch callback is still invoked serially."""
+        unit_state = yield self.get_unit_state()
+
+        callbacks = [Deferred() for i in range(5)]
+        results = []
+        contents = []
+
+        @inlineCallbacks
+        def watch(value):
+            results.append(value)
+            yield callbacks[len(results) - 1]
+            contents.append((yield unit_state.get_resolved()))
+
+        yield unit_state.watch_resolved(watch)
+
+        # These get collapsed into a single event
+        yield unit_state.set_resolved(RETRY_HOOKS)
+        yield unit_state.clear_resolved()
+        yield self.poke_zk()
+
+        # Verify the callback hasn't completed
+        self.assertEqual(len(results), 1)
+        self.assertEqual(len(contents), 0)
+
+        # Let it finish
+        callbacks[0].callback(True)
+        yield self.poke_zk()
+
+        # Verify result counts
+        self.assertEqual(len(results), 2)
+        self.assertEqual(len(contents), 1)
+
+        # Verify result values. Even though we have created event, the
+        # setting retrieved shows the hook is not enabled.
+        self.assertEqual(results[-1].type_name, "created")
+        self.assertEqual(contents[-1], None)
+
+        yield unit_state.set_resolved(NO_HOOKS)
+        callbacks[1].callback(True)
+        yield self.poke_zk()
+
+        self.assertEqual(len(results), 3)
+        self.assertEqual(contents[-1], {"retry": NO_HOOKS})
+
+        # Clear out any pending activity.
+        yield self.poke_zk()
+
+    @inlineCallbacks
+    def test_watch_relation_resolved_slow_callback(self):
+        """A slow watch callback is still invoked serially."""
+        unit_state = yield self.get_unit_state()
+
+        callbacks = [Deferred() for i in range(5)]
+        results = []
+        contents = []
+
+        @inlineCallbacks
+        def watch(value):
+            results.append(value)
+            yield callbacks[len(results) - 1]
+            contents.append((yield unit_state.get_relation_resolved()))
+
+        yield unit_state.watch_relation_resolved(watch)
+
+        # These get collapsed into a single event
+        yield unit_state.set_relation_resolved({"0": RETRY_HOOKS})
+        yield unit_state.clear_relation_resolved()
+        yield self.poke_zk()
+
+        # Verify the callback hasn't completed
+        self.assertEqual(len(results), 1)
+        self.assertEqual(len(contents), 0)
+
+        # Let it finish
+        callbacks[0].callback(True)
+        yield self.poke_zk()
+
+        # Verify result counts
+        self.assertEqual(len(results), 2)
+        self.assertEqual(len(contents), 1)
+
+        # Verify result values. Even though we have created event, the
+        # setting retrieved shows the hook is not enabled.
+        self.assertEqual(results[-1].type_name, "created")
+        self.assertEqual(contents[-1], None)
+
+        yield unit_state.set_relation_resolved({"0": RETRY_HOOKS})
+        callbacks[1].callback(True)
+        yield self.poke_zk()
+
+        self.assertEqual(len(results), 3)
+        self.assertEqual(contents[-1], {"0": RETRY_HOOKS})
+        # Clear out any pending activity.
+        yield self.poke_zk()
+
+    @inlineCallbacks
     def test_set_and_clear_upgrade_flag(self):
         """An upgrade flag can be set on a unit."""
 
@@ -870,6 +1148,9 @@
         self.assertEqual(results[-1].type_name, "created")
         self.assertEqual(contents[-1], True)
 
+        # Clear out any pending activity.
+        yield self.poke_zk()
+
     @inlineCallbacks
     def test_enable_debug_hook(self):
         """Unit hook debugging can be enabled on the unit state."""
@@ -1061,6 +1342,9 @@
         self.assertEqual(results[-1].type_name, "created")
         self.assertEqual(contents[-1], {"debug_hooks": ["*"]})
 
+        # Clear out any pending activity.
+        yield self.poke_zk()
+
     @inlineCallbacks
     def test_service_unit_agent(self):
         """A service unit state has an associated unit agent."""
 
=== modified file 'ensemble/state/tests/test_utils.py'
--- ensemble/state/tests/test_utils.py	2011-04-28 17:52:55 +0000
+++ ensemble/state/tests/test_utils.py	2011-05-02 21:57:29 +0000
@@ -12,9 +12,10 @@
 import yaml
 
 from ensemble.lib.testing import TestCase
-from ensemble.state.errors import StateNotFound
-from ensemble.state.utils import (PortWatcher, remove_tree,
+from ensemble.state.errors import StateChanged, StateNotFound
+from ensemble.state.utils import (PortWatcher, remove_tree, dict_merge,
                                   get_open_port, YAMLState)
+
 from ensemble.tests.common import get_test_zookeeper_address
 
 
@@ -206,6 +207,26 @@
         self.assertNotIn("zoo", children)
 
 
+class DictMergeTest(TestCase):
+
+    def test_merge_no_match(self):
+        self.assertEqual(
+            dict_merge(dict(a=1), dict(b=2)),
+            dict(a=1, b=2))
+
+    def test_merge_matching_keys_same_value(self):
+        self.assertEqual(
+            dict_merge(dict(a=1, b=2), dict(b=2, c=1)),
+            dict(a=1, b=2, c=1))
+
+    def test_merge_conflict(self):
+        self.assertRaises(
+            StateChanged,
+            dict_merge,
+            dict(a=1, b=3),
+            dict(b=2, c=1))
+
+
 class OpenPortTest(TestCase):
 
     def test_get_open_port(self):
@@ -301,7 +322,6 @@
         zk_data = yaml.load(zk_data)
         self.assertEqual(zk_data, options)
 
-
     @inlineCallbacks
     def test_conflict_on_set(self):
         """Version conflict error tests.
@@ -357,8 +377,8 @@
         yield node.read()
 
         options = dict(alpha="beta", one=1)
-        node["alpha"] =  "beta"
-        node["one"] =  1
+        node["alpha"] = "beta"
+        node["one"] = 1
         yield node.write()
 
         # a local get should reflect proper data
@@ -369,7 +389,6 @@
         zk_data = yaml.load(zk_data)
         self.assertEqual(zk_data, options)
 
-
     @inlineCallbacks
     def test_multiple_reads(self):
         """Calling read resets state to ZK after multiple round-trips."""
@@ -415,7 +434,7 @@
 
         result = node.pop("foo")
         self.assertEqual(result, "bar")
-        self.assertEqual(node, {"alpha": "beta",})
+        self.assertEqual(node, {"alpha": "beta"})
 
         node["delta"] = "gamma"
         self.assertEqual(set(node.keys()), set(("alpha", "delta")))
@@ -424,7 +443,6 @@
         self.assertIn(("alpha", "beta"), result)
         self.assertIn(("delta", "gamma"), result)
 
-
     @inlineCallbacks
     def test_del_empties_state(self):
         d = YAMLState(self.client, self.path)
@@ -500,12 +518,8 @@
         yield d1.read()
         self.assertEquals(d1, d2)
 
-
     @inlineCallbacks
     def test_read_requires_node(self):
         """Validate that read raises when required=True."""
         d1 = YAMLState(self.client, self.path)
         yield self.assertFailure(d1.read(True), StateNotFound)
-
-
-
 
=== modified file 'ensemble/state/utils.py'
--- ensemble/state/utils.py	2011-04-28 17:52:55 +0000
+++ ensemble/state/utils.py	2011-05-02 21:57:29 +0000
@@ -9,8 +9,10 @@
 import yaml
 import zookeeper
 
+from ensemble.state.errors import StateChanged
 from ensemble.state.errors import StateNotFound
 
+
 class PortWatcher(object):
 
     def __init__(self, host, port, timeout, listen=False):
@@ -91,6 +93,22 @@
     return port
 
 
+def dict_merge(d1, d2):
+    """Return a union of dicts if they have no conflicting values.
+
+    Else raise a StateChanged error.
+    """
+    must_match = set(d1).intersection(d2)
+    for k in must_match:
+        if not d1[k] == d2[k]:
+            raise StateChanged()
+
+    d = {}
+    d.update(d1)
+    d.update(d2)
+    return d
+
+
 class YAMLState(DictMixin):
     """Provides a dict like interface around a Zookeeper node
     containing serialised YAML data. The dict provided represents the
@@ -144,7 +162,6 @@
             if required:
                 raise StateNotFound(self._path)
 
-
     def _check(self):
         """Verify that sync was called for operations which expect it."""
         if self._pristine_cache is None:
@@ -164,7 +181,6 @@
         self._check()
         self._cache[key] = value
 
-
     def __delitem__(self, key):
         self._check()
         del self._cache[key]
 
=== modified file 'ensemble/unit/lifecycle.py'
--- ensemble/unit/lifecycle.py	2011-04-15 18:53:29 +0000
+++ ensemble/unit/lifecycle.py	2011-05-02 21:57:29 +0000
@@ -2,14 +2,14 @@
 import logging
 
 from twisted.internet.defer import (
-    inlineCallbacks, DeferredLock)
+    inlineCallbacks, DeferredLock, returnValue)
 
 from ensemble.hooks.invoker import Invoker
 from ensemble.hooks.scheduler import HookScheduler
 from ensemble.state.hook import RelationChange
 from ensemble.state.errors import StopWatcher, UnitRelationStateNotFound
 
-from ensemble.unit.workflow import RelationWorkflowState, RelationWorkflow
+from ensemble.unit.workflow import RelationWorkflowState
 
 
 HOOK_SOCKET_FILE = ".ensemble.hookcli.sock"
@@ -32,7 +32,8 @@
         self._unit_path = unit_path
         self._relations = {}
         self._running = False
-        self._watching = False
+        self._watching_relation_memberships = False
+        self._watching_relation_resolved = False
         self._run_lock = DeferredLock()
         self._log = logging.getLogger("unit.lifecycle")
 
@@ -45,21 +46,23 @@
         return self._relations[relation_id]
 
     @inlineCallbacks
-    def install(self):
+    def install(self, fire_hooks=True):
         """Invoke the unit's install hook.
         """
-        yield self._execute_hook("install")
+        if fire_hooks:
+            yield self._execute_hook("install")
 
     @inlineCallbacks
-    def upgrade_formula(self):
+    def upgrade_formula(self, fire_hooks=True):
         """Invoke the unit's upgrade-formula hook.
         """
-        yield self._execute_hook("upgrade-formula", now=True)
+        if fire_hooks:
+            yield self._execute_hook("upgrade-formula", now=True)
         # Restart hook queued hook execution.
         self._executor.start()
 
     @inlineCallbacks
-    def start(self):
+    def start(self, fire_hooks=True):
         """Invoke the start hook, and setup relation watching.
         """
         self._log.debug("pre-start acquire, running:%s", self._running)
@@ -70,22 +73,29 @@
             assert not self._running, "Already started"
 
             # Execute the start hook
-            yield self._execute_hook("start")
+            if fire_hooks:
+                yield self._execute_hook("start")
 
             # If we have any existing relations in memory, start them.
             if self._relations:
                 self._log.debug("starting relation lifecycles")
 
             for workflow in self._relations.values():
-                # should not transition an
                 yield workflow.transition_state("up")
 
             # Establish a watch on the existing relations.
-            if not self._watching:
+            if not self._watching_relation_memberships:
                 self._log.debug("starting service relation watch")
                 yield self._service.watch_relation_states(
                     self._on_service_relation_changes)
-                self._watching = True
+                self._watching_relation_memberships = True
+
+            # Establish a watch for resolved relations
+            if not self._watching_relation_resolved:
+                self._log.debug("starting unit relation resolved watch")
+                yield self._unit.watch_relation_resolved(
+                    self._on_relation_resolved_changes)
+                self._watching_relation_resolved = True
 
             # Set current status
             self._running = True
@@ -94,7 +104,7 @@
         self._log.debug("started unit lifecycle")
 
     @inlineCallbacks
-    def stop(self):
+    def stop(self, fire_hooks=True):
         """Stop the unit, executes the stop hook, and stops relation watching.
         """
         self._log.debug("pre-stop acquire, running:%s", self._running)
@@ -106,10 +116,12 @@
             # Stop relation lifecycles
             if self._relations:
                 self._log.debug("stopping relation lifecycles")
+
             for workflow in self._relations.values():
                 yield workflow.transition_state("down")
 
-            yield self._execute_hook("stop")
+            if fire_hooks:
+                yield self._execute_hook("stop")
 
             # Set current status
             self._running = False
@@ -118,6 +130,49 @@
         self._log.debug("stopped unit lifecycle")
 
     @inlineCallbacks
+    def _on_relation_resolved_changes(self, event):
+        """Callback for unit relation resolved watching.
+
+        The callback is invoked whenever the relation resolved
+        settings change.
+        """
+        self._log.debug("relation resolved changed")
+        # Acquire the run lock, and process the changes.
+        yield self._run_lock.acquire()
+
+        try:
+            # If the unit lifecycle isn't running we shouldn't process
+            # any relation resolutions.
+            if not self._running:
+                self._log.debug("stop watch relation resolved changes")
+                self._watching_relation_resolved = False
+                raise StopWatcher()
+
+            self._log.info("processing relation resolved changed")
+            if self._client.connected:
+                yield self._process_relation_resolved_changes()
+        finally:
+            yield self._run_lock.release()
+
+    @inlineCallbacks
+    def _process_relation_resolved_changes(self):
+        """Invoke retry transitions on relations if their not running.
+        """
+        relation_resolved = yield self._unit.get_relation_resolved()
+        if relation_resolved is None:
+            returnValue(None)
+        else:
+            yield self._unit.clear_relation_resolved()
+
+        keys = set(relation_resolved).intersection(self._relations)
+        for rel_id in keys:
+            relation_workflow = self._relations[rel_id]
+            relation_state = yield relation_workflow.get_state()
+            if relation_state == "up":
+                continue
+            yield relation_workflow.transition_state("up")
+
+    @inlineCallbacks
     def _on_service_relation_changes(self, old_relations, new_relations):
         """Callback for service relation watching.
 
@@ -137,7 +192,7 @@
             # If the lifecycle is not running, then stop the watcher
             if not self._running:
                 self._log.debug("stop service-rel watcher, discarding changes")
-                self._watching = False
+                self._watching_relation_memberships = False
                 raise StopWatcher()
 
             self._log.debug("processing relations changed")
@@ -147,9 +202,9 @@
 
     @inlineCallbacks
     def _process_service_changes(self, old_relations, new_relations):
-        """Add and remove unit lifecycles per the service relations changes.
+        """Add and remove unit lifecycles per the service relations Determine.
         """
-        # Determine relation delta of global zk state with our memory state.
+        # changes relation delta of global zk state with our memory state.
         new_relations = dict([(service_relation.internal_relation_id,
                                service_relation) for
                               service_relation in new_relations])
@@ -336,8 +391,11 @@
         self._error_handler = handler
 
     @inlineCallbacks
-    def start(self):
+    def start(self, watches=True):
         """Start watching related units and executing change hooks.
+
+        @param watches: boolean parameter denoting if relation watches
+               should be started.
         """
         yield self._run_lock.acquire()
         try:
@@ -348,19 +406,24 @@
                 self._watcher = yield self._unit_relation.watch_related_units(
                     self._scheduler.notify_change)
             # And start the watcher.
-            yield self._watcher.start()
+            if watches:
+                yield self._watcher.start()
         finally:
             self._run_lock.release()
         self._log.debug(
             "started relation:%s lifecycle", self._relation_name)
 
     @inlineCallbacks
-    def stop(self):
+    def stop(self, watches=True):
         """Stop watching changes and stop executing relation change hooks.
+
+        @param watches: boolean parameter denoting if relation watches
+               should be stopped.
         """
         yield self._run_lock.acquire()
         try:
-            self._watcher.stop()
+            if watches and self._watcher:
+                self._watcher.stop()
             self._scheduler.stop()
         finally:
             yield self._run_lock.release()
 
=== modified file 'ensemble/unit/tests/test_lifecycle.py'
--- ensemble/unit/tests/test_lifecycle.py	2011-04-28 17:55:11 +0000
+++ ensemble/unit/tests/test_lifecycle.py	2011-05-02 21:57:29 +0000
@@ -12,6 +12,8 @@
 from ensemble.unit.lifecycle import (
     UnitLifecycle, UnitRelationLifecycle, RelationInvoker)
 
+from ensemble.unit.workflow import RelationWorkflowState
+
 from ensemble.hooks.invoker import Invoker
 from ensemble.hooks.executor import HookExecutor
 
@@ -19,9 +21,11 @@
 
 from ensemble.state.endpoint import RelationEndpoint
 from ensemble.state.relation import ClientServerUnitWatcher
+from ensemble.state.service import NO_HOOKS
 from ensemble.state.tests.test_relation import RelationTestBase
 from ensemble.state.hook import RelationChange
 
+
 from ensemble.lib.testing import TestCase
 from ensemble.lib.mocker import MATCH
 
@@ -145,6 +149,182 @@
         return output
 
 
+class LifecycleResolvedTest(LifecycleTestBase):
+
+    @inlineCallbacks
+    def setUp(self):
+        yield super(LifecycleResolvedTest, self).setUp()
+        yield self.setup_default_test_relation()
+        self.lifecycle = UnitLifecycle(
+            self.client, self.states["unit"], self.states["service"],
+            self.unit_directory, self.executor)
+
+    def get_unit_relation_workflow(self, states):
+        state_dir = os.path.join(self.ensemble_directory, "state")
+        lifecycle = UnitRelationLifecycle(
+            self.client,
+            states["unit_relation"],
+            states["service_relation"].relation_name,
+            self.unit_directory,
+            self.executor)
+
+        workflow = RelationWorkflowState(
+            self.client,
+            states["unit_relation"],
+            lifecycle,
+            state_dir)
+
+        return (workflow, lifecycle)
+
+    @inlineCallbacks
+    def test_resolved_relation_watch_unit_lifecycle_not_running(self):
+        """If the unit is not running then no relation resolving is performed.
+        However the resolution value remains the same.
+        """
+        # Start the unit.
+        yield self.lifecycle.start()
+
+        # Wait for the relation to be started.... TODO: async background work
+        yield self.sleep(0.1)
+
+        # Simulate relation down on an individual unit relation
+        workflow = self.lifecycle._relations.get(
+            self.states["unit_relation"].internal_relation_id)
+        self.assertEqual("up", (yield workflow.get_state()))
+
+        yield workflow.transition_state("down")
+        resolved = self.wait_on_state(workflow, "up")
+
+        # Stop the unit lifecycle
+        yield self.lifecycle.stop()
+
+        # Set the relation to resolved
+        yield self.states["unit"].set_relation_resolved(
+            {self.states["unit_relation"].internal_relation_id: NO_HOOKS})
+
+        # Give a moment for the watch to fire erroneously
+        yield self.sleep(0.2)
+
+        # Ensure we didn't attempt a transition.
+        self.assertFalse(resolved.called)
+        self.assertEqual(
+            {self.states["unit_relation"].internal_relation_id: NO_HOOKS},
+            (yield self.states["unit"].get_relation_resolved()))
+
+        # If the unit is restarted start, we currently have the
+        # behavior that the unit relation workflow will automatically
+        # be transitioned back to running, as part of the normal state
+        # transition. Sigh.. we should have a separate error
+        # state for relation hooks then down with state variable usage.
+
+    @inlineCallbacks
+    def test_resolved_relation_watch_relation_up(self):
+        """If a relation marked as to be resolved is already running,
+        then no work is performed.
+        """
+        # Start the unit.
+        yield self.lifecycle.start()
+
+        # Wait for the relation to be started.... TODO: async background work
+        yield self.sleep(0.1)
+
+        # get a hold of the unit relation and verify state
+        workflow = self.lifecycle._relations.get(
+            self.states["unit_relation"].internal_relation_id)
+        self.assertEqual("up", (yield workflow.get_state()))
+
+        # Set the relation to resolved
+        yield self.states["unit"].set_relation_resolved(
+            {self.states["unit_relation"].internal_relation_id: NO_HOOKS})
+
+        # Give a moment for the async background work.
+        yield self.sleep(0.1)
+
+        # Ensure we're still up and the relation resolved setting has been
+        # cleared.
+        self.assertEqual(
+            None, (yield self.states["unit"].get_relation_resolved()))
+        self.assertEqual("up", (yield workflow.get_state()))
+
+    @inlineCallbacks
+    def test_resolved_relation_watch_from_error(self):
+        """Unit lifecycle's will process a unit relation resolved
+        setting, and transition a down relation back to a running
+        state.
+        """
+        log_output = self.capture_logging(
+            "unit.lifecycle", level=logging.DEBUG)
+
+        # Start the unit.
+        yield self.lifecycle.start()
+
+        # Wait for the relation to be started... TODO: async background work
+        yield self.sleep(0.1)
+
+        # Simulate an error condition
+        workflow = self.lifecycle._relations.get(
+            self.states["unit_relation"].internal_relation_id)
+        self.assertEqual("up", (yield workflow.get_state()))
+        yield workflow.fire_transition("error")
+
+        resolved = self.wait_on_state(workflow, "up")
+
+        # Set the relation to resolved
+        yield self.states["unit"].set_relation_resolved(
+            {self.states["unit_relation"].internal_relation_id: NO_HOOKS})
+
+        # Wait for the relation to come back up
+        value = yield self.states["unit"].get_relation_resolved()
+
+        yield resolved
+
+        # Verify state
+        value = yield workflow.get_state()
+        self.assertEqual(value, "up")
+
+        self.assertIn(
+            "processing relation resolved changed", log_output.getvalue())
+
+    @inlineCallbacks
+    def test_resolved_relation_watch(self):
+        """Unit lifecycle's will process a unit relation resolved
+        setting, and transition a down relation back to a running
+        state.
+        """
+        log_output = self.capture_logging(
+            "unit.lifecycle", level=logging.DEBUG)
+
+        # Start the unit.
+        yield self.lifecycle.start()
+
+        # Wait for the relation to be started... TODO: async background work
+        yield self.sleep(0.1)
+
+        # Simulate an error condition
+        workflow = self.lifecycle._relations.get(
+            self.states["unit_relation"].internal_relation_id)
+        self.assertEqual("up", (yield workflow.get_state()))
+        yield workflow.transition_state("down")
+
+        resolved = self.wait_on_state(workflow, "up")
+
+        # Set the relation to resolved
+        yield self.states["unit"].set_relation_resolved(
+            {self.states["unit_relation"].internal_relation_id: NO_HOOKS})
+
+        # Wait for the relation to come back up
+        value = yield self.states["unit"].get_relation_resolved()
+
+        yield resolved
+
+        # Verify state
+        value = yield workflow.get_state()
+        self.assertEqual(value, "up")
+
+        self.assertIn(
+            "processing relation resolved changed", log_output.getvalue())
+
+
 class UnitLifecycleTest(LifecycleTestBase):
 
     @inlineCallbacks
@@ -187,6 +367,45 @@
         # verify the sockets are cleaned up.
         self.assertEqual(os.listdir(self.unit_directory), ["formula"])
 
+    @inlineCallbacks
+    def test_start_sans_hook(self):
+        """The lifecycle start can be invoked without firing hooks."""
+        self.write_hook("start", "#!/bin/sh\n exit 1")
+        start_executed = self.wait_on_hook("start")
+        yield self.lifecycle.start(fire_hooks=False)
+        # Wait for unit relation background processing....
+        yield self.sleep(0.1)
+        self.assertFalse(start_executed.called)
+
+    @inlineCallbacks
+    def test_stop_sans_hook(self):
+        """The lifecycle stop can be invoked without firing hooks."""
+        self.write_hook("stop", "#!/bin/sh\n exit 1")
+        stop_executed = self.wait_on_hook("stop")
+        yield self.lifecycle.start()
+        yield self.lifecycle.stop(fire_hooks=False)
+        # Wait for unit relation background processing....
+        yield self.sleep(0.1)
+        self.assertFalse(stop_executed.called)
+
+    @inlineCallbacks
+    def test_install_sans_hook(self):
+        """The lifecycle install can be invoked without firing hooks."""
+        self.write_hook("install", "#!/bin/sh\n exit 1")
+        install_executed = self.wait_on_hook("install")
+        yield self.lifecycle.install(fire_hooks=False)
+        self.assertFalse(install_executed.called)
+
+    @inlineCallbacks
+    def test_upgrade_sans_hook(self):
+        """The lifecycle upgrade can be invoked without firing hooks."""
+        self.executor.stop()
+        self.write_hook("upgrade-formula", "#!/bin/sh\n exit 1")
+        upgrade_executed = self.wait_on_hook("upgrade-formula")
+        yield self.lifecycle.upgrade_formula(fire_hooks=False)
+        self.assertFalse(upgrade_executed.called)
+        self.assertTrue(self.executor.running)
+
     def test_hook_error(self):
         """Verify hook execution error, raises an exception."""
         self.write_hook("install", '#!/bin/sh\n exit 1')
@@ -196,14 +415,12 @@
     def test_hook_not_executable(self):
         """A hook not executable, raises an exception."""
         self.write_hook("install", '#!/bin/sh\n exit 0', no_exec=True)
-        # It would be preferrable if this was also a formulainvocation error.
         return self.failUnlessFailure(
             self.lifecycle.install(), FormulaError)
 
     def test_hook_not_formatted_correctly(self):
         """Hook execution error, raises an exception."""
         self.write_hook("install", '!/bin/sh\n exit 0')
-        # It would be preferrable if this was also a formulainvocation error.
         return self.failUnlessFailure(
             self.lifecycle.install(), FormulaInvocationError)
 
@@ -510,7 +727,7 @@
     @inlineCallbacks
     def test_initial_start_lifecycle_no_related_no_exec(self):
         """
-        If there are no related units on startup, the relation changed hook
+        If there are no related units on startup, the relation joined hook
         is not invoked.
         """
         file_path = self.makeFile()
@@ -523,6 +740,26 @@
         self.assertFalse(os.path.exists(file_path))
 
     @inlineCallbacks
+    def test_stop_can_continue_watching(self):
+        """
+        """
+        file_path = self.makeFile()
+        self.write_hook(
+            "%s-relation-changed" % self.relation_name,
+            ("#!/bin/bash\n" "echo executed >> %s\n" % file_path))
+        rel_states = yield self.add_opposite_service_unit(self.states)
+        yield self.lifecycle.start()
+        yield self.wait_on_hook(
+            sequence=["app-relation-joined", "app-relation-changed"])
+        changed_executed = self.wait_on_hook("app-relation-changed")
+        yield self.lifecycle.stop(watches=False)
+        rel_states["unit_relation"].set_data(yaml.dump(dict(hello="world")))
+        yield self.sleep(0.1)
+        self.assertFalse(changed_executed.called)
+        yield self.lifecycle.start(watches=False)
+        yield changed_executed
+
+    @inlineCallbacks
     def test_initial_start_lifecycle_with_related(self):
         """
         If there are related units on startup, the relation changed hook
 
=== modified file 'ensemble/unit/tests/test_workflow.py'
--- ensemble/unit/tests/test_workflow.py	2011-04-28 17:55:11 +0000
+++ ensemble/unit/tests/test_workflow.py	2011-05-02 21:57:29 +0000
@@ -10,7 +10,7 @@
 
 from ensemble.unit.workflow import (
     UnitWorkflowState, RelationWorkflowState, WorkflowStateClient,
-    is_unit_running)
+    is_unit_running, is_relation_running)
 
 
 class WorkflowTestBase(LifecycleTestBase):
@@ -321,6 +321,26 @@
             self.state_directory)
 
     @inlineCallbacks
+    def test_is_relation_running(self):
+        """The unit relation's workflow state can be categorized as a
+        boolean.
+        """
+        running, state = yield is_relation_running(
+            self.client, self.states["unit_relation"])
+        self.assertIdentical(running, False)
+        self.assertIdentical(state, None)
+        yield self.workflow.fire_transition("start")
+        running, state = yield is_relation_running(
+            self.client, self.states["unit_relation"])
+        self.assertIdentical(running, True)
+        self.assertEqual(state, "up")
+        yield self.workflow.fire_transition("stop")
+        running, state = yield is_relation_running(
+            self.client, self.states["unit_relation"])
+        self.assertIdentical(running, False)
+        self.assertEqual(state, "down")
+
+    @inlineCallbacks
     def test_up_down_cycle(self):
         """The workflow can be transition from up to down, and back.
         """
@@ -381,7 +401,7 @@
         # Add a new unit, and wait for the broken hook to result in
         # the transition to the down state.
         yield self.add_opposite_service_unit(self.states)
-        yield self.wait_on_state(self.workflow, "down")
+        yield self.wait_on_state(self.workflow, "error")
 
         f_state, history, zk_state = yield self.read_persistent_state(
             history_id=self.workflow.zk_state_id)
@@ -392,7 +412,7 @@
                          "formula", "hooks", "app-relation-changed"))
 
         self.assertEqual(f_state,
-                         {"state": "down",
+                         {"state": "error",
                           "state_variables": {
                               "change_type": "joined",
                               "error_message": error}})
 
=== modified file 'ensemble/unit/workflow.py'
--- ensemble/unit/workflow.py	2011-04-19 18:55:06 +0000
+++ ensemble/unit/workflow.py	2011-05-02 21:57:29 +0000
@@ -17,15 +17,19 @@
     Transition("install", "Install", None, "installed",
                error_transition_id="error_install"),
     Transition("error_install", "Install Error", None, "install_error"),
-    Transition("retry_install", "Retry Install", "install_error", "installed"),
+    Transition("retry_install", "Retry Install", "install_error", "installed",
+               alias="retry"),
     Transition("start", "Start", "installed", "started",
                error_transition_id="error_start"),
     Transition("error_start", "Start Error", "installed", "start_error"),
-    Transition("retry_start", "Retry Start", "start_error", "started"),
+    Transition("retry_start", "Retry Start", "start_error", "started",
+               alias="retry"),
     Transition("stop", "Stop", "started", "stopped",
                error_transition_id="error_stop"),
     Transition("error_stop", "Stop Error", "started", "stop_error"),
-    Transition("retry_stop", "Retry Stop", "stop_error", "stopped"),
+    Transition("retry_stop", "Retry Stop", "stop_error", "stopped",
+               alias="retry"),
+    Transition("restart", "Restart", "stop", "start", alias="retry"),
 
     # Upgrade Transitions (stay in state, with success transitition)
     Transition(
@@ -36,27 +40,47 @@
         "started", "formula_upgrade_error"),
     Transition(
         "retry_upgrade_formula", "Upgrade from stop error",
-        "formula_upgrade_error", "started")
+        "formula_upgrade_error", "started", alias="retry")
     )
 
 
-# There's been some discussion, if we should have per change type error states
-# here, corresponding to the different changes that the relation-changed hook
-# is invoked for. The important aspects to capture are both observability of
-# error type locally and globally (zk), and per error type and instance
-# recovery of the same. To provide for this functionality without additional
-# states, the error information (change type, and error message) are captured
-# in state variables which are locally and globally observable. Future
-# extension of the restart transition action, will allow for customized
-# recovery based on the change type state variable. Effectively this
-# differs from the unit definition, in that it collapses three possible error
-# states, into a behavior off switch. A separate state will be needed to
-# denote departing.
+# Unit relation error states
+#
+# There's been some discussion, if we should have per change type
+# error states here, corresponding to the different changes that the
+# relation-changed hook is invoked for. The important aspects to
+# capture are both observability of error type locally and globally
+# (zk), and per error type and instance recovery of the same. To
+# provide for this functionality without additional states, the error
+# information (change type, and error message) are captured in state
+# variables which are locally and globally observable. Future
+# extension of the restart transition action, will allow for
+# customized recovery based on the change type state
+# variable. Effectively this differs from the unit definition, in that
+# it collapses three possible error states, into a behavior off
+# switch. A separate state will be needed to denote departing.
+
+
+# Process recovery using on disk workflow state
+#
+# Another interesting issue, process recovery using the on disk state,
+# is complicated by consistency to the the in memory state, which
+# won't be directly recoverable anymore without some state specific
+# semantics to recovering from on disk state, ie a restarted unit
+# agent, with a relation in an error state would require special
+# semantics around loading from disk to ensure that the in-memory
+# process state (watching and scheduling but not executing) matches
+# the recovery transition actions (which just restart hook execution,
+# but assume the watch continues).. this functionality added to better
+# allow for the behavior that while down due to a hook error, the
+# relation would continues to schedule pending hooks
 
 RelationWorkflow = Workflow(
     Transition("start", "Start", None, "up"),
     Transition("stop", "Stop", "up", "down"),
-    Transition("restart", "Restart", "down", "up"),
+    Transition("restart", "Restart", "down", "up", alias="retry"),
+    Transition("error", "Relation hook error", "up", "error"),
+    Transition("reset", "Recover from hook error", "error", "up"),
     Transition("depart", "Relation broken", "up", "departed"),
     Transition("down_depart", "Relation broken", "down", "departed"),
     )
@@ -67,12 +91,26 @@
     """Is the service unit in a running state.
 
     Returns a boolean which is true if the unit is running, and
-    the unit state in two element tuple.
+    the unit workflow state in a two element tuple.
     """
     workflow_state = yield WorkflowStateClient(client, unit).get_state()
     if not workflow_state:
         returnValue((False, None))
-    running = workflow_state in ("started",)
+    running = workflow_state == "started"
+    returnValue((running, workflow_state))
+
+
+@inlineCallbacks
+def is_relation_running(client, relation):
+    """Is the unit relation in a running state.
+
+    Returns a boolean which is true if the relation is running, and
+    the unit relation workflow state in a two element tuple.
+    """
+    workflow_state = yield WorkflowStateClient(client, relation).get_state()
+    if not workflow_state:
+        returnValue((False, None))
+    running = workflow_state == "up"
     returnValue((running, workflow_state))
 
 
@@ -198,7 +236,6 @@
         row per entry with CSV escaping.
         """
         state_serialized = yaml.safe_dump(state_dict)
-
         # State File
         with open(self.state_file_path, "w") as handle:
             handle.write(state_serialized)
@@ -218,6 +255,7 @@
             return {"state": None}
         with open(self.state_file_path, "r") as handle:
             content = handle.read()
+
         return yaml.load(content)
 
 
@@ -257,9 +295,9 @@
         self._lifecycle = lifecycle
 
     @inlineCallbacks
-    def _invoke_lifecycle(self, method):
+    def _invoke_lifecycle(self, method, *args, **kw):
         try:
-            result = yield method()
+            result = yield method(*args, **kw)
         except (FileNotFound, FormulaError, FormulaInvocationError), e:
             raise TransitionError(e)
         returnValue(result)
@@ -274,21 +312,25 @@
     def do_stop(self):
         return self._invoke_lifecycle(self._lifecycle.stop)
 
-    def do_retry_start(self):
-        return self._invoke_lifecycle(self._lifecycle.start)
-
-    def do_retry_stop(self):
-        self._invoke_lifecycle(self._lifecycle.stop)
-
-    def do_retry_install(self):
-        return self._invoke_lifecycle(self._lifecycle.install)
-
-    def do_retry_upgrade_formula(self):
-        return self._invoke_lifecycle(self._lifecycle.upgrade_formula)
-
     def do_upgrade_formula(self):
         return self._invoke_lifecycle(self._lifecycle.upgrade_formula)
 
+    def do_retry_start(self, fire_hooks=True):
+        return self._invoke_lifecycle(
+            self._lifecycle.start, fire_hooks=fire_hooks)
+
+    def do_retry_stop(self, fire_hooks=True):
+        self._invoke_lifecycle(
+            self._lifecycle.stop, fire_hooks=fire_hooks)
+
+    def do_retry_install(self, fire_hooks=True):
+        return self._invoke_lifecycle(
+            self._lifecycle.install, fire_hooks=fire_hooks)
+
+    def do_retry_upgrade_formula(self, fire_hooks=True):
+        return self._invoke_lifecycle(
+            self._lifecycle.upgrade_formula, fire_hooks=fire_hooks)
+
 
 class RelationWorkflowState(DiskWorkflowState):
 
@@ -321,7 +363,7 @@
 
         @param: error: The error from hook invocation.
         """
-        yield self.fire_transition("stop",
+        yield self.fire_transition("error",
                                    change_type=relation_change.change_type,
                                    error_message=str(error))
 
@@ -330,12 +372,30 @@
         """Transition the workflow to the 'down' state.
 
         Turns off the unit-relation lifecycle monitoring and hook execution.
+
+        :param error_info: If called on relation hook error, contains
+        error variables.
         """
         yield self._lifecycle.stop()
 
     @inlineCallbacks
+    def do_reset(self):
+        """Transition the workflow to the 'up' state from an error state.
+
+        Turns on the unit-relation lifecycle monitoring and hook execution.
+        """
+        yield self._lifecycle.start(watches=False)
+
+    @inlineCallbacks
+    def do_error(self, **error_info):
+        """A relation hook error, stops further execution hooks but
+        continues to watch for changes.
+        """
+        yield self._lifecycle.stop(watches=False)
+
+    @inlineCallbacks
     def do_restart(self):
-        """Transition the workflow to the 'up' state.
+        """Transition the workflow to the 'up' state from the down state.
 
         Turns on the unit-relation lifecycle monitoring and hook execution.
         """