Merge lp:~jimbaker/juju-jitsu/watch-subcommand into lp:juju-jitsu

[Jim Baker]

Adds a new jitsu subcommand, watch. It directly uses the Juju API where possible, otherwise ZK, to wait for a set of conditions about a Juju environment to become true. It does not timeout, instead use the core timeout command if that is desired.

The help gives the full flavor of what can be done:

usage: watch [-h] [-e ENVIRONMENT]
           [--loglevel CRITICAL|ERROR|WARNING|INFO|DEBUG] [--verbose]
           [--any] [--number]
           CONDITION [CONDITION ...]

Watch Juju environment for condition(s) to become true, waiting as necessary.

Each condition is specified as follows:

  INSTANCE Either service, relation, or service unit. For a
                        relation, either its relation id (example: db:0) or
                        descriptors can be used (examples: "mysql wordpress"
                        or "mysql:db wordpress:db").

  -n NUM, --num-units NUM
                        Number of units; only applies to services
  -r RELATION, --relation RELATION
                        Specify relation for unit or service
  --setting SETTING Relation setting exists
  --state STATE State of unit or for at least all NUM units of service
  --x-state STATE Cannot be in this state
  [NAME=SETTING [NAME=SETTING ... ]]
                        These settings must be as specified for the unit
                        or for at least all NUM units of a service

optional arguments:
  -h, --help show this help message and exit
  -e ENVIRONMENT, --environment ENVIRONMENT
                        Environment to act upon (otherwise uses default)
  --loglevel CRITICAL|ERROR|WARNING|INFO|DEBUG
                        Log level
  --verbose, -v Enable verbose logging
  --any Any of the conditions may be true
  --number Number output by the corresponding condition

Examples of watches with single conditions:

$ relation_id=`jitsu watch "wordpress mysql"` # waits until relation available and prints id; must be quoted
$ relation_id=`jitsu watch "wordpress:db mysql:app"` # like above, but with fully specified descriptors
$ jitsu watch mysql # service is deployed
$ jitsu watch mysql/0 --state=started # unit is started
$ timeout 60s jitsu watch mysql/0 --state=started # watch up to 60s for mysql/0 to be running, using core timeout command
$ jitsu watch -r db:0 mysql/0 foo=bar # watch for foo to be set to bar on mysql/0 on relation id of db:0
$ jitsu watch mysql/0 -r "wordpress mysql" foo=bar # watch for wordpress<->mysql, then watch for this setting
$ jitsu watch mysql/0 -r db:0 foo= # watch for foo to be unset
$ jitsu watch mysql/0 -r db:0 --setting=foo # watch for foo to be set to *some* value

Multiple conditions can be combined:

$ jitsu watch mysql/0 -r "wordpress mysql" foo= mysql/1 -r "wordpress mysql" foo=bar # all conditions must apply
$ jitsu --any watch ... # any of the conditions may apply

[Mark Mims (mark-mims)]

looks great... does exactly what I need for testing

review: Approve

=== added file 'sub-commands/aiki/cli.py'
--- sub-commands/aiki/cli.py	1970-01-01 00:00:00 +0000
+++ sub-commands/aiki/cli.py	2012-06-08 21:25:21 +0000
@@ -0,0 +1,108 @@
+import argparse
+import logging
+import sys
+import traceback
+
+
+from twisted.internet import reactor
+from twisted.internet.defer import inlineCallbacks
+import yaml
+import zookeeper
+
+from aiki.introspect import get_zk_client_connector
+
+
+LOG_LEVELS = dict(
+    CRITICAL=logging.CRITICAL,
+    ERROR=logging.ERROR,
+    WARNING=logging.WARNING,
+    INFO=logging.INFO,
+    DEBUG=logging.DEBUG)
+
+
+def make_arg_parser(*args, **kwargs):
+    parser = argparse.ArgumentParser(*args, **kwargs)
+    parser.add_argument(
+        "-e", "--environment", default=None, help="Environment to act upon (otherwise uses default)", metavar="ENVIRONMENT")
+    parser.add_argument(
+        "--loglevel", default=None, choices=LOG_LEVELS, help="Log level",
+        metavar="CRITICAL|ERROR|WARNING|INFO|DEBUG")
+    parser.add_argument("--verbose", "-v", default=False, action="store_true", help="Enable verbose logging")
+    return parser
+
+
+def setup_logging(options):
+    options.loglevel = LOG_LEVELS.get(options.loglevel)
+    if options.verbose:
+        if options.loglevel is None:
+            options.loglevel = logging.DEBUG
+    else:
+        # Avoid potentially voluminous ZK debug logging 
+        zookeeper.set_debug_level(0)
+        if options.loglevel is None:
+            options.loglevel = logging.INFO
+
+    log_options = {
+        "level": options.loglevel,
+        "format": "%(asctime)s %(name)s:%(levelname)s %(message)s"
+        }
+    logging.basicConfig(**log_options)
+
+
+def run_command(command, options):
+    """Connect to ZK, either from admin machine or on a Juju machine, then run command"""
+    # Introspect to find the right client connector; this will be
+    # called, then yielded upon, to get a ZK client once async code is
+    # entered
+    try:
+        connector = get_zk_client_connector(options)
+    except Exception, e:
+        print >> sys.stderr, e
+        sys.exit(1)
+
+    exit_code = [0]
+    reactor.callWhenRunning(_run_command_in_reactor, command, exit_code, connector, options)
+    reactor.run()
+    sys.exit(exit_code[0])
+
+
+@inlineCallbacks
+def _run_command_in_reactor(command, exit_code, connector, options):
+    try:
+        yield _wrapped_command(command, exit_code, connector, options)
+    except Exception, e:
+        exit_code[0] = 1
+        if options.verbose:
+            traceback.print_exc()  # Writes to stderr
+        else:
+            print >> sys.stderr, e
+    finally:
+        reactor.stop()
+
+
+@inlineCallbacks
+def _wrapped_command(command, result, connector, options):
+    client = yield connector()
+    try:
+        yield command(result, client, options)
+    finally:
+        yield client.close()
+
+
+def parse_relation_settings(args):
+    """Given args, return a list of the remaining args along with settings"""
+    settings = {}
+    kv_iter = iter(args)
+    remaining_args = []
+    for kv in kv_iter:
+        if "=" not in kv:
+            remaining_args = [kv]
+            break
+        k, v = kv.split("=", 1)
+        if v.startswith("@"):
+            filename = v[1:]
+            with open(filename, "r") as f:
+                v = f.read()
+        settings[k] = yaml.safe_load(v)
+    remaining_args.extend(kv_iter)
+    return settings, remaining_args

=== added file 'sub-commands/aiki/twistutils.py'
--- sub-commands/aiki/twistutils.py	1970-01-01 00:00:00 +0000
+++ sub-commands/aiki/twistutils.py	2012-06-08 21:25:21 +0000
@@ -0,0 +1,20 @@
+from twisted.internet.defer import Deferred
+
+
+class CountDownLatch(object):
+    # NOTE For now, we only support successful callbacks, not errbacks
+    def __init__(self, count, deferred_list=()):
+        self.count = count
+        self.completed = Deferred()
+        for d in deferred_list:
+            self.add(d)
+
+    def add(self, deferred):
+        deferred.addCallback(self.count_down)
+
+    def count_down(self, *ignored):
+        if self.count > 0:
+            self.count -= 1
+        if self.count == 0:
+            if not self.completed.called:
+                self.completed.callback(None) # Maybe collect together these results?

=== added file 'sub-commands/aiki/watcher.py'
--- sub-commands/aiki/watcher.py	1970-01-01 00:00:00 +0000
+++ sub-commands/aiki/watcher.py	2012-06-08 21:25:21 +0000
@@ -0,0 +1,277 @@
+import logging
+
+from twisted.internet.defer import Deferred, inlineCallbacks, returnValue
+from zookeeper import NoNodeException
+import yaml
+
+from juju.state.base import StateBase
+from juju.state.relation import RelationStateManager, ServiceRelationState
+from juju.state.service import ServiceStateManager, parse_service_name
+from juju.state.errors import (
+    RelationStateNotFound, ServiceStateNotFound, ServiceUnitStateNotFound, InvalidRelationIdentity,
+    StopWatcher, NoMatchingEndpoints, AmbiguousRelation)
+from juju.unit.workflow import WorkflowStateClient
+
+
+class AnySetting(object):
+
+    def __str__(self):
+        return "<Any setting>"
+
+    __repr__ = __str__
+
+ANY_SETTING = AnySetting()
+
+
+class Watcher(StateBase):
+
+    def __init__(self, client):
+        super(Watcher, self).__init__(client)
+        self.log = logging.getLogger("aiki.watcher")
+        self.relation_state_manager = RelationStateManager(self._client)
+        self.service_state_manager = ServiceStateManager(self._client)
+
+    @inlineCallbacks
+    def wait_for_relation(self, *descriptors):
+        """Wait for a relation described by `descriptors` to exist in the topology"""
+        done = Deferred()
+
+        @inlineCallbacks
+        def expected(*ignored):
+            try:
+                relation_id = yield self.get_relation_id_from_descriptors(*descriptors)
+                relation_ident = yield self.get_relation_ident(descriptors[0], relation_id)
+                done.callback(relation_ident)
+                raise StopWatcher()
+            except (ServiceStateNotFound, RelationStateNotFound):
+                self.log.debug("Topology changed but continuing to wait for relation: %s" % (descriptors,))
+
+        yield self._watch_topology(expected)
+        returnValue((yield done))
+
+    @inlineCallbacks
+    def wait_for_service(self, service_name):
+        """Wait for `service_name` to exist in the topology"""
+        done = Deferred()
+
+        @inlineCallbacks
+        def expected(*ignored):
+            try:
+                yield self.service_state_manager.get_service_state(service_name)
+                done.callback(service_name)
+                raise StopWatcher()
+            except ServiceStateNotFound:
+                self.log.debug("Topology changed, still waiting for existence of service: %s" % service_name)
+
+        yield self._watch_topology(expected)
+        returnValue((yield done))
+
+    @inlineCallbacks
+    def wait_for_unit(self, unit_name):
+        """Wait for `unit_name` to exist in the topology"""
+        done = Deferred()
+
+        @inlineCallbacks
+        def expected(*ignored):
+            try:
+                service_name = parse_service_name(unit_name)
+                service = yield self.service_state_manager.get_service_state(service_name)
+                yield service.get_unit_state(unit_name)
+                done.callback(service_name)
+                raise StopWatcher()
+            except (ServiceUnitStateNotFound, ServiceStateNotFound):
+                self.log.debug("Topology changed, still waiting for existence of unit: %s" % unit_name)
+
+        yield self._watch_topology(expected)
+        returnValue((yield done))
+
+    @inlineCallbacks
+    def watch_new_service_units(self, service_name, cb):
+        """As new service units are added to service_name, calls the `cb` function for each unit."""
+
+        service = yield self.service_state_manager.get_service_state(service_name)
+        watched_units = set()
+        
+        @inlineCallbacks
+        def check(*ignored):
+            try:
+                units = yield service.get_all_unit_states()
+                for unit in units:
+                    # NOTE ignore units that have gone missing for the
+                    # time being; from actual testing, this seems
+                    # unlikely, but we might need to do more elaborate
+                    # canceling (StopWatcher). Regardless, that's what
+                    # would do with juju, but this is jitsu.
+                    if unit.unit_name not in watched_units:
+                        watched_units.add(unit.unit_name)
+
+                        from twisted.internet import reactor
+                        reactor.callLater(0, cb, unit.unit_name)
+            except ServiceStateNotFound:
+                self.log.warn("Service is no longer in topology: %s" % service_name)
+                raise StopWatcher()
+
+        yield self._watch_topology(check)
+
+    def setup_zk_watch(self, path, callback):
+        """Returns a Deferred"""
+
+        @inlineCallbacks
+        def manage_callback(*ignored):
+            # Need to guard on the client being connected in the case
+            # 1) a watch is waiting to run (in the reactor);
+            # 2) and the connection is closed.
+            #
+            # It remains the reponsibility of `callback` to raise
+            # `StopWatcher`, per above.
+            if not self._client.connected:
+                returnValue(None)
+            exists_d, watch_d = self._client.exists_and_watch(path)
+            stat = yield exists_d
+            exists = bool(stat)
+            if exists:
+                try:
+                    yield callback(exists)
+                except StopWatcher:
+                    returnValue(None)
+                except NoNodeException, e:
+                    # This may occur if the callback is trying to process
+                    # data related to this path, so just ignore
+                    self.log.debug("Ignoring no node exception when watching %s: %s", path, e)
+
+            watch_d.addCallback(manage_callback)
+
+        return manage_callback()
+
+    def get_relation_id(self, relation_ident):
+        """Return the (internal) relation id for `relation_ident`."""
+        # NOTE may want to refactor by adding to juju.state.relation
+        parts = relation_ident.split(":")
+        if len(parts) != 2 or not parts[1].isdigit():
+            raise InvalidRelationIdentity(relation_ident)
+        relation_name, normalized_id = parts
+        relation_id = "%s-%s" % ("relation", normalized_id.zfill(10))
+        if not self.topology.has_relation(relation_id):
+            raise RelationStateNotFound()
+        return relation_id
+
+    @inlineCallbacks
+    def get_relations(self, service_name):
+        """Get the relations associated to `service_name`."""
+        relations = []
+        service_state_manager = ServiceStateManager(self._client)
+        self.topology = yield self._read_topology()
+        service = yield service_state_manager.get_service_state(service_name)
+        internal_service_id = service.internal_id
+        for info in self.topology.get_relations_for_service(
+                internal_service_id):
+            service_info = info["service"]
+            relations.append(
+                ServiceRelationState(
+                    self._client,
+                    internal_service_id,
+                    info["relation_id"],
+                    info["scope"],
+                    **service_info))
+        returnValue(relations)
+
+    @inlineCallbacks
+    def get_relation_ident(self, descriptor, relation_id):
+        parts = descriptor.split(":", 1)
+        service_name = parts[0]
+        relations = yield self.get_relations(service_name)
+        relation_ident = [r.relation_ident for r in relations if r.internal_relation_id == relation_id][0]
+        returnValue(relation_ident)
+
+    @inlineCallbacks
+    def get_relation_id_from_descriptors(self, *descriptors):
+        endpoint_pairs = yield self.service_state_manager.join_descriptors(*descriptors)
+        if len(endpoint_pairs) == 0:
+            raise NoMatchingEndpoints()
+        elif len(endpoint_pairs) > 1:
+            for pair in endpoint_pairs[1:]:
+                if not (pair[0].relation_name.startswith("juju-") or
+                        pair[1].relation_name.startswith("juju-")):
+                    raise AmbiguousRelation(descriptors, endpoint_pairs)
+        endpoints = endpoint_pairs[0]
+        if endpoints[0] == endpoints[1]:
+            endpoints = endpoints[0:1]
+
+        # so this needs to be put in the context of a continual watch until the topology is in effect
+        relation_state = yield self.relation_state_manager.get_relation_state(*endpoints)
+        returnValue(relation_state.internal_id)
+
+    @inlineCallbacks
+    def watch_unit_settings(self, unit_name, relation_ident, expected_settings):
+        """Finally returns when unit_name has the expected settings, if ever"""
+        # NOTE may want to refactor by adding to juju.state.relation
+        # the generic settings path logic
+        self.topology = topology = yield self._read_topology()  # TODO refactor one of these assignments out
+        relation_id = self.get_relation_id(relation_ident)
+        unit_id = topology.get_service_unit_id_from_name(unit_name)
+        container = topology.get_service_unit_container(unit_id)
+        container_info = "%s/" % container[-1] if container else ""
+        settings_path = "/relations/%s/%ssettings/%s" % (relation_id, container_info, unit_id)
+        done = Deferred()
+
+        @inlineCallbacks
+        def expected(*ignored):
+            content_yaml, stat = yield self._client.get(settings_path)
+            content = yaml.load(content_yaml)
+            for K, V in expected_settings.iteritems():
+                if V is None:
+                    # key should *not* have a setting
+                    if K in content:
+                        break
+                elif V is ANY_SETTING:
+                    if K not in content:
+                        break
+                elif content.get(K) != V:
+                    break
+            else:
+                done.callback(content)
+                raise StopWatcher()
+
+        yield self.setup_zk_watch(settings_path, expected)
+        settings = yield done
+        returnValue(settings)
+
+    @inlineCallbacks
+    def get_unit_agent_state(self, unit):
+        # NOTE extracted from juju.control.status
+        unit_workflow_client = WorkflowStateClient(self._client, unit)
+        workflow_state = yield unit_workflow_client.get_state()
+        if not workflow_state:
+            agent_state = "pending"
+        else:
+            unit_connected = yield unit.has_agent()
+            agent_state = workflow_state.replace("_", "-") if unit_connected else "down"
+        returnValue(agent_state)
+
+    @inlineCallbacks
+    def wait_for_unit_state(self, unit_name, expected_states, excluded_states):
+        service_name = parse_service_name(unit_name)
+        service = yield self.service_state_manager.get_service_state(service_name)
+        unit = yield service.get_unit_state(unit_name)
+        agent_path = [None]
+        done = Deferred()
+
+        @inlineCallbacks
+        def expected(*ignored):
+            if done.called:
+                raise StopWatcher()
+            agent_state = yield self.get_unit_agent_state(unit)
+            self.log.debug("%s has state: %s", unit_name, agent_state)
+            if agent_state in expected_states or (excluded_states and agent_state not in excluded_states):
+                if not done.called:
+                    done.callback(agent_state)
+                raise StopWatcher()
+
+        # Watch both the agent ephemeral plus the workflow state - the
+        # agent state is a composite of the two
+        yield self.setup_zk_watch(unit._get_agent_path(), expected)
+        unit_workflow_client = WorkflowStateClient(self._client, unit)
+        yield self.setup_zk_watch(unit_workflow_client.zk_state_path, expected)
+
+        agent_state = yield done
+        returnValue(agent_state)

=== added file 'sub-commands/topodump'
--- sub-commands/topodump	1970-01-01 00:00:00 +0000
+++ sub-commands/topodump	2012-06-08 21:25:21 +0000
@@ -0,0 +1,31 @@
+#!/usr/bin/env python
+
+from twisted.internet.defer import inlineCallbacks
+from aiki.cli import make_arg_parser, setup_logging, run_command
+from juju.state.base import StateBase
+
+
+def main():
+    parser = make_arg_parser()
+    options = parser.parse_args()
+    setup_logging(options)
+    print "options", options
+    run_command(topodump, options)
+
+
+@inlineCallbacks
+def topodump(result, client, options):
+    Dumper = TopoDump(client)
+    yield Dumper.dump()
+
+
+class TopoDump(StateBase):
+
+    @inlineCallbacks
+    def dump(self):
+        topology = yield self._read_topology()
+        print topology.dump()
+
+
+if __name__ == '__main__':
+    main()

=== added file 'sub-commands/watch'
--- sub-commands/watch	1970-01-01 00:00:00 +0000
+++ sub-commands/watch	2012-06-08 21:25:21 +0000
@@ -0,0 +1,242 @@
+#!/usr/bin/env python
+
+import argparse
+import logging
+import textwrap
+
+import yaml
+from twisted.internet.defer import inlineCallbacks, DeferredList, returnValue, succeed
+
+from aiki.cli import make_arg_parser, setup_logging, run_command, parse_relation_settings
+from aiki.twistutils import CountDownLatch
+from aiki.watcher import Watcher, ANY_SETTING
+
+
+log = logging.getLogger("watch")
+
+
+def main():
+    options = make_watch_parser()
+    setup_logging(options)
+    run_command(watch, options)
+
+
+def make_watch_parser():
+    condition_parser = argparse.ArgumentParser(description="Parses each condition")
+    condition_parser.add_argument(
+        "-n", "--num-units", default=None, type=int, metavar="NUM", help="Number of units; only applies to services")
+    condition_parser.add_argument("-r", "--relation", default=None, metavar="RELATION", help="Specify relation")
+    condition_parser.add_argument("--setting", default=[], action="append", metavar="SETTING", help="Relation setting exists")
+    condition_parser.add_argument(
+        "--state", default=[], dest="states", action="append", metavar="STATE", help="State of service or a unit")
+    condition_parser.add_argument(
+        "--x-state", default=[], dest="excluded_states", action="append", metavar="STATE", help="Cannot be in this state")
+    condition_parser.add_argument("rest", nargs=argparse.REMAINDER)
+
+    # If updating condition_parser, it can be useful to do
+    # condition_parser.format_help() to get the usage info for
+    # CONDITION; some manual editing will be required however.
+
+    main_parser = make_arg_parser(
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+        usage=textwrap.dedent("""\
+        watch [-h] [-e ENVIRONMENT]
+                   [--loglevel CRITICAL|ERROR|WARNING|INFO|DEBUG] [--verbose]
+                   [--any] [--number]
+                   CONDITION [CONDITION ...]
+
+        Watch Juju environment for condition(s) to become true, waiting as necessary.
+
+        Each condition is specified as follows:
+
+          INSTANCE              Either service, relation, or service unit. For a 
+                                relation, either its relation id (example: db:0) or
+                                descriptors can be used (examples: "mysql wordpress"
+                                or "mysql:db wordpress:db").
+
+          -n NUM, --num-units NUM
+                                Number of units; only applies to services
+          -r RELATION, --relation RELATION
+                                Specify relation for unit or service
+          --setting SETTING     Relation setting exists
+          --state STATE         State of unit or for at least all NUM units of service
+          --x-state STATE       Cannot be in this state
+          [NAME=SETTING [NAME=SETTING ... ]]
+                                These settings must be as specified for the unit
+                                or for at least all NUM units of a service
+        """),
+        epilog=textwrap.dedent("""\
+        Examples of watches with single conditions:
+
+        $ relation_id=`jitsu watch "wordpress mysql"`        # waits until relation available and prints id; must be quoted
+        $ relation_id=`jitsu watch "wordpress:db mysql:app"` # like above, but with fully specified descriptors
+        $ jitsu watch mysql                                  # service is deployed
+        $ jitsu watch mysql/0 --state=started                # unit is started
+        $ timeout 60s jitsu watch mysql/0 --state=started    # watch up to 60s for mysql/0 to be running, using core timeout command
+        $ jitsu watch -r db:0 mysql/0 foo=bar                # watch for foo to be set to bar on mysql/0 on relation id of db:0
+        $ jitsu watch mysql/0 -r "wordpress mysql" foo=bar   # watch for wordpress<->mysql, then watch for this setting
+        $ jitsu watch mysql/0 -r db:0 foo=                   # watch for foo to be unset
+        $ jitsu watch mysql/0 -r db:0 --setting=foo          # watch for foo to be set to *some* value
+
+        Multiple conditions can be combined:
+
+        $ jitsu watch mysql/0 -r "wordpress mysql" foo= mysql/1 -r "wordpress mysql" foo=bar # all conditions must apply
+        $ jitsu --any watch ...                                                              # any of the conditions may apply
+    """))
+    main_parser.add_argument("--any", default=False, action="store_true", help="Any of the conditions may be true")
+    main_parser.add_argument("--number", default=False, action="store_true", help="Number output by the corresponding condition")
+
+    options, condition_args = main_parser.parse_known_args()
+
+    # Partially parse (using remainder args) multiple times working
+    # left to right to get conditions until all original args are
+    # consumed
+    options.conditions = []
+    count = 0
+    while condition_args:
+        instance = condition_args.pop(0)
+        condition = condition_parser.parse_args(condition_args)
+        condition.instance = instance
+        condition.count = count
+        options.conditions.append(condition)  # Just collect together
+        condition.settings, condition_args = parse_relation_settings(condition.rest)
+        for setting in condition.setting:
+            condition.settings[setting] = ANY_SETTING
+        condition.states = set(condition.states)
+        condition.excluded_states = set(condition.excluded_states)
+        count += 1
+    return options
+
+
+def wait_for_results(deferreds, fireOnOneCallback):
+    d = DeferredList(deferreds, fireOnOneCallback=fireOnOneCallback, fireOnOneErrback=True, consumeErrors=True)
+    d.addCallback(lambda r: [x[1] for x in r])
+
+    def get_errors(f):
+        # Avoid spurious errors seen in closing connections; we don't care!
+        if not d.called:
+            return f.value.subFailure
+
+    d.addErrback(get_errors)
+    return d
+
+
+@inlineCallbacks
+def watch(result, client, options):
+    wait_for_conditions = []
+    for condition in options.conditions:
+        wait_for_conditions.append(parse_and_wait_on_condition(client, options, condition))
+    yield wait_for_results(wait_for_conditions, options.any)
+
+
+def print_result(options, condition, result):
+    """Given conditions complete asynchronously, the user can specify numbering to disambiguate"""
+    if options.number:
+        print condition.count,
+    print result
+
+
+@inlineCallbacks
+def parse_and_wait_on_condition(client, options, condition):
+    """Given a condition description from argparse, creates a watcher for it"""
+    watcher = Watcher(client)
+
+    # Disambiguate the following cases: relation, unit, or service:
+    # 1. Relation
+    if len(condition.instance.split()) >= 2:
+        # Only handle non-peer relations, since peer relations are automatically established anyway;
+        # also handle error case of a relation between more than 2 services
+        relation_ident = yield wait_for_relation(client, watcher, options, condition, condition.instance)
+        print_result(options, condition, relation_ident)
+        return
+
+    # 2. Service unit
+    if "/" in condition.instance:
+        yield wait_for_unit(client, watcher, options, condition, condition.instance)
+        print_result(options, condition, condition.instance)
+        return
+
+    # 3. Service
+    yield wait_for_service(client, watcher, options, condition)
+    print_result(options, condition, condition.instance)
+
+
+@inlineCallbacks
+def wait_for_unit(client, watcher, options, condition, unit_name, relation_ident=None):
+    log.info("Waiting for unit %s...", unit_name)
+    yield watcher.wait_for_unit(unit_name)
+    if condition.states or condition.excluded_states:
+        log.info("Waiting for unit %s to be in %s and not in %s", 
+                 unit_name, sorted(condition.states), sorted(condition.excluded_states))
+        agent_state = yield watcher.wait_for_unit_state(unit_name, condition.states, condition.excluded_states)
+    log.info("Completed waiting for unit %s", unit_name)
+
+    if condition.relation:
+        if relation_ident is None:
+            relation_ident = yield wait_for_relation(client, watcher, options, condition)
+        if condition.settings:
+            log.info("Waiting for %s: settings %s", unit_name, condition.settings)
+            settings = yield watcher.watch_unit_settings(unit_name, relation_ident, condition.settings)
+            log.info("Completed waiting for %s: expected %s, actual %s",
+                     unit_name, condition.settings, settings)
+    returnValue(unit_name)
+
+
+@inlineCallbacks
+def wait_for_relation(client, watcher, options, condition, relation=None):
+    """Return relation ident corresponding to the condition or relation (if specified), waiting as necessary"""
+
+    if relation is None:
+        relation = condition.relation
+
+    if ":" in relation:
+        parts = relation.split(":")
+        if len(parts) == 2 and parts[1].isdigit():
+            relation_ident = relation
+            returnValue(relation_ident)
+    
+    # Otherwise wait for it:
+    descriptors = relation.split()
+    if len(descriptors) == 1 or len(descriptors) == 2:
+        log.info("Waiting for %s...", format_descriptors(*descriptors))
+        relation_ident = yield watcher.wait_for_relation(*descriptors)
+        log.info("Completed waiting for %s", format_descriptors(*descriptors))
+        returnValue(relation_ident)
+
+    # log.error("Bad relation: %s" % relation)
+    raise ValueError("Bad relation: %s" % relation)
+
+
+@inlineCallbacks
+def wait_for_service(client, watcher, options, condition):
+    """Return service name when sucessfully waited"""
+    log.info("Waiting for %s%s service...", 
+             "%d unit(s) of " % condition.num_units if condition.num_units else "",
+             condition.instance)
+    yield watcher.wait_for_service(condition.instance)
+    log.info("Completed waiting for %s service", condition.instance)
+    if condition.relation:
+        relation_ident = yield wait_for_relation(client, watcher, options, condition)
+    else:
+        relation_ident = None
+    if condition.num_units:
+        latch = CountDownLatch(condition.num_units)
+        
+        def new_unit_cb(unit_name):
+            # NOTE Could also presumably cancel any removed units, consider that for a future refactoring
+            return succeed(latch.add(wait_for_unit(client, watcher, options, condition, unit_name, relation_ident)))
+
+        yield watcher.watch_new_service_units(condition.instance, new_unit_cb)
+        yield latch.completed
+    returnValue(condition.instance)
+
+
+def format_descriptors(*descriptors):
+    if len(descriptors) == 1:
+        return "peer relation %s" % descriptors[0]
+    else:
+        return "relation between " + " and ".join(sorted(descriptors))
+
+
+if __name__ == '__main__':
+    main()