Merge lp:~hazmat/pyjuju/resolved-state-api into lp:pyjuju

[Kapil Thangavelu]

I'm not particularly thrilled with the method names, but this branch implements a get/set/del/watch api for unit resolved and unit relation resolved, to enable sending resolve requests to the unit agent.

[Gustavo Niemeyer (niemeyer)]

Some comments:

[1]

+class ServiceUnitRelationResolvedAlreadyEnabled(StateError):
+ """The unit has already been marked resolved.

Documentation needs updating.

[2]

+ def get_resolved(self):
+ """Get the value of the resolved flag if any.

The result of this method isn't a flag.

[3]

+ def get_resolved(self):
(...)
+ except zookeeper.NoNodeException:
+ # We get to the same end state.
+ returnValue(None)

This is a bit misleading. We're just reading, so there's no "end state" to be
achieved. The existence of the file is actually what determines if resolved
was requested or not.

[4]

+ unit_resolve_path = "/units/%s/resolved" % self.internal_id

Given that this was used 5 times, it would be handy to have something like:

    @property
    def _unit_resolved_path
        return "/units/%s/resolved" % self.internal_id

Then you can use self._unit_resolved_path everywhere.

[5]

+ rel_resolved_path = "/units/%s/relation_resolved" % self._internal_id

Our file paths so far are all dashed. It'd be good to keep the convention
inside ZooKeeper too.

May also be factored out into a property as per the point above.

[6]

+ def set_relation_resolved(self, relation_map):
+ """Mark a unit's relations as in need of being resolved.

Doesn't this actually mean something like:

    "Mark the problem in the unit's relation as being resolved."

IOW, it's not requesting for the problem to be resolved, but stating
that it has been.

[7]

+ :param relation_map: A map of internal relation ids, to retry booleans.

This feels like a non-ideal API for a couple of reasons:

a) It's exposing internal ids rather than using the relation names as we have
   elsewhere.

b) set_relation_resolved({"db": True, "cache": False}) feels like marking db as
   resolved, and cache as unresolved.

It's a bit hard to suggest something else without knowing exactly how you
intend to use. Let's talk about it.

[8]

+ d1_keys = set(d1.keys())
+ d2_keys = set(d2.keys())
+
+ must_match = d1_keys.intersection(d2_keys)

As a hint, this could be spelled as:

must_match = set(d1).intersection(d2):

[9]

+ running = workflow_state in ("up",)

FWIW, this pattern feels slightly weird when there's a single value.
workflow_state == "up" would be more straightforward.

[10]

+ self.assertIdentical(results.pop(0).type_name, "created")

assertIdentical is being overused a bit. There's no reason why these should be
allocated in the same memory location.

review: Needs Fixing

[Kapil Thangavelu (hazmat)]

Excerpts from Gustavo Niemeyer's message of Tue Apr 26 18:04:27 UTC 2011:
> Review: Needs Fixing
> Some comments:
>
>
> [1]
>
> +class ServiceUnitRelationResolvedAlreadyEnabled(StateError):
> + """The unit has already been marked resolved.
>
> Documentation needs updating.

how so?

>
> [2]
>
> + def get_resolved(self):
> + """Get the value of the resolved flag if any.
>
> The result of this method isn't a flag.
>

fixed.

> [3]
>
> + def get_resolved(self):
> (...)
> + except zookeeper.NoNodeException:
> + # We get to the same end state.
> + returnValue(None)
>
> This is a bit misleading. We're just reading, so there's no "end state" to be
> achieved. The existence of the file is actually what determines if resolved
> was requested or not.
>

fixed.

> [4]
>
> + unit_resolve_path = "/units/%s/resolved" % self.internal_id
>
> Given that this was used 5 times, it would be handy to have something like:
>
> @property
> def _unit_resolved_path
> return "/units/%s/resolved" % self.internal_id
>
> Then you can use self._unit_resolved_path everywhere.
>

done.

> [5]
>
> + rel_resolved_path = "/units/%s/relation_resolved" % self._internal_id
>
> Our file paths so far are all dashed. It'd be good to keep the convention
> inside ZooKeeper too.
>
> May also be factored out into a property as per the point above.

done
>
> [6]
>
> + def set_relation_resolved(self, relation_map):
> + """Mark a unit's relations as in need of being resolved.
>
> Doesn't this actually mean something like:
>
> "Mark the problem in the unit's relation as being resolved."
>
> IOW, it's not requesting for the problem to be resolved, but stating
> that it has been.
>

It is actually making a request for the problem to be solved, not solving the
problem directly, as there are behavioral aspects to transitions that need be
done inside of the unit agent.

> [7]
>
> + :param relation_map: A map of internal relation ids, to retry booleans.
>
> This feels like a non-ideal API for a couple of reasons:
>
> a) It's exposing internal ids rather than using the relation names as we have
> elsewhere.
>
> b) set_relation_resolved({"db": True, "cache": False}) feels like marking db as
> resolved, and cache as unresolved.
>
> It's a bit hard to suggest something else without knowing exactly how you
> intend to use. Let's talk about it.
>

As per irc discussion, we've deferred a) to a latter time, there's a commit in the branch
which does switch the syntax to

set_relation_resolved(db, retry_hooks=True)

but its then inconsistent with its get_resolved api which retrieves a dictionary of
relation ids : retry hook boolean

> [8]
>
> + d1_keys = set(d1.keys())
> + d2_keys = set(d2.keys())
> +
> + must_match = d1_keys.intersection(d2_keys)
>
> As a hint, this could be spelled as:
>
> must_match = set(d1).intersection(d2):

fixed.

>
> [9]
>
> + running = workflow_state in ("up",)
>
> FWIW, this pattern feels slightly weird when there's a single value.
> workflow_state == "up" would be more straightforward.
>
> [10]
>
> + self.assert...

[lp:~hazmat/pyjuju/resolved-state-api]

202. By Kapil Thangavelu on 2011-05-02

merge trunk and pep8ify it.

[Gustavo Niemeyer (niemeyer)]

Looks good, +1. Just two follow ups on your questioning above:

> > [1]
> >
> > +class ServiceUnitRelationResolvedAlreadyEnabled(StateError):
> > + """The unit has already been marked resolved.
> >
> > Documentation needs updating.
>
> how so?

s/unit/relation/

> > [6]
> >
> > + def set_relation_resolved(self, relation_map):
> > + """Mark a unit's relations as in need of being resolved.
> >
> > Doesn't this actually mean something like:
> >
> > "Mark the problem in the unit's relation as being resolved."
> >
> > IOW, it's not requesting for the problem to be resolved, but stating
> > that it has been.
> >
>
> It is actually making a request for the problem to be solved, not solving the
> problem directly, as there are behavioral aspects to transitions that need be
> done inside of the unit agent.

By the time someone calls "resolved", the *problem* which caused the hook to fail
must already have been fixed (hence the 'd' in "resolved"). We're not fixing the
problem with that action, we're simply stating that it's fine to go ahead because
someone else hopefully already fixed it.

review: Approve

[lp:~hazmat/pyjuju/resolved-state-api]

203. By Kapil Thangavelu on 2011-05-04

merge trunk conflict

204. By Kapil Thangavelu on 2011-05-05

merge trunk

205. By Kapil Thangavelu on 2011-05-05

document set_relation_resolved api todo discussion

206. By Kapil Thangavelu on 2011-05-05

doc string cleanups per review.

=== modified file 'ensemble/state/errors.py'
--- ensemble/state/errors.py	2011-05-04 07:40:25 +0000
+++ ensemble/state/errors.py	2011-05-05 16:20:19 +0000
@@ -21,6 +21,7 @@
     """Exception value to denote watching should stop.
     """
 
+
 class StateNotFound(StateError):
     """State not found.
 
@@ -151,6 +152,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 relation 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-05-05 09:41:11 +0000
+++ ensemble/state/service.py	2011-05-05 16:20:19 +0000
@@ -15,11 +15,16 @@
     StateChanged, ServiceStateNotFound, ServiceUnitStateNotFound,
     ServiceUnitStateMachineAlreadyAssigned, ServiceStateNameInUse,
     BadDescriptor, BadServiceStateName, NoUnusedMachines,
-    ServiceUnitDebugAlreadyEnabled)
+    ServiceUnitDebugAlreadyEnabled, ServiceUnitResolvedAlreadyEnabled,
+    ServiceUnitRelationResolvedAlreadyEnabled)
 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, YAMLState
+
+from ensemble.state.utils import remove_tree, dict_merge, YAMLState
+
+RETRY_HOOKS = 1000
+NO_HOOKS = 1001
 
 
 class ServiceStateManager(StateBase):
@@ -877,6 +882,193 @@
         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)
+            yield callback(change_event)
+            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)
+
+    @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 being resolved.
+
+        The unit agent will watch this setting and unblock the unit,
+        via manipulation of the unit workflow and lifecycle.
+
+        :param relation_map: A map of internal relation ids, to retry hook
+              values either ensemble.state.service.NO_HOOKS or
+              RETRY_HOOKS.
+
+        TODO:
+        The api currently takes internal relation ids, this should be
+        cleaned up with a refactor to state request protocol objects.
+        Only public names should be exposed beyond the state api.
+
+        There's an ongoing discussion on whether this needs to support
+        retries. Currently it doesn't, and this could the arg to this
+        method could just be a list of relations. Supporting retries
+        would mean capturing enough information to retry the hook, and
+        has reconciliation issues wrt to what's current at the time of
+        re-execution. The existing hook scheduler automatically
+        performs merges of redundant events. The retry could execute a
+        relation change hook, for a remote unit that has already
+        departed at the time of re-execution (and for which we have
+        a pending hook execution), which would be inconsistent, wrt
+        to what would be exposed via the hook cli api. With support
+        for on disk persistence and recovery, some of this temporal
+        synchronization would already be in place.
+        """
+        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):
+        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)
+            yield callback(change_event)
+            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)
+
 
 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-05 16:20:19 +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-05-05 09:41:11 +0000
+++ ensemble/state/tests/test_service.py	2011-05-05 16:20:19 +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)
 
 
 from ensemble.state.tests.common import StateTestBase
@@ -739,6 +740,213 @@
             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_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_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."""
 
@@ -871,6 +1079,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."""
@@ -1063,6 +1274,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-05-03 08:54:13 +0000
+++ ensemble/state/tests/test_utils.py	2011-05-05 16:20:19 +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-05-03 08:54:13 +0000
+++ ensemble/state/utils.py	2011-05-05 16:20:19 +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-05-05 09:41:11 +0000
+++ ensemble/unit/lifecycle.py	2011-05-05 16:20:19 +0000
@@ -106,6 +106,7 @@
             # Stop relation lifecycles
             if self._relations:
                 self._log.debug("stopping relation lifecycles")
+
             for workflow in self._relations.values():
                 yield workflow.transition_state("down")
 

=== modified file 'ensemble/unit/tests/test_workflow.py'
--- ensemble/unit/tests/test_workflow.py	2011-05-05 09:41:11 +0000
+++ ensemble/unit/tests/test_workflow.py	2011-05-05 16:20:19 +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):
@@ -367,6 +367,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.
         """

=== modified file 'ensemble/unit/workflow.py'
--- ensemble/unit/workflow.py	2011-05-05 09:41:11 +0000
+++ ensemble/unit/workflow.py	2011-05-05 16:20:19 +0000
@@ -78,12 +78,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))