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

[Jim Baker]

Adds support for --failfast option, which immediately terminates the watch with exit code 1 if any condition is watching for a service unit (or as implied with a service using --state/--x-state). Also this branch refactors the watch script to use an instance of WatchManager to manage various parameters instead of passing through a bunch of functions.

$ jitsu watch -e testbed --failfast mysql --state=started foobar --state=started
2012-08-08 15:35:38,824 juju.common:INFO Connecting to environment...
2012-08-08 15:35:42,436 juju.common:INFO Connected to environment.
2012-08-08 15:35:42,436 watch:INFO Waiting for mysql service...
2012-08-08 15:35:42,436 watch:INFO Waiting for foobar service...
2012-08-08 15:35:42,735 watch:INFO Completed waiting for mysql service
2012-08-08 15:35:43,054 watch:INFO Waiting for unit mysql/1...
2012-08-08 15:35:43,409 watch:INFO Waiting for unit mysql/1 to be in [error, install-error, start-error, started, stop-error] and not in []
2012-08-08 15:35:43,955 watch:INFO Completed waiting for unit mysql/1 state: start-error
2012-08-08 15:35:43,955 watch:ERROR Failed fast: Unit mysql/1 in error state: start-error
Unit mysql/1 in error state: start-error
2012-08-08 15:35:43,956 watch:INFO Completed waiting for unit mysql/1
mysql

[Mark Mims (mark-mims)]

lgtm

review: Approve

=== modified file 'sub-commands/aiki/cli.py'
--- sub-commands/aiki/cli.py	2012-07-05 17:30:37 +0000
+++ sub-commands/aiki/cli.py	2012-08-08 21:42:19 +0000
@@ -193,3 +193,18 @@
     subparser.add_parser(
         cmd, description=cmd,
         help=help, usage="".join(usage), formatter_class=argparse.RawDescriptionHelpFormatter)
+
+
+def format_states(states):
+    return "[" + ", ".join(sorted(states)) + "]"
+
+
+def format_port_pairs(port_pairs):
+    return "[" + ", ".join("%s/%s" % (port, proto) for port, proto in sorted(port_pairs)) + "]"
+
+
+def format_descriptors(*descriptors):
+    if len(descriptors) == 1:
+        return "peer relation %s" % descriptors[0]
+    else:
+        return "relation between " + " and ".join(sorted(descriptors))

=== modified file 'sub-commands/watch'
--- sub-commands/watch	2012-06-22 05:24:51 +0000
+++ sub-commands/watch	2012-08-08 21:42:19 +0000
@@ -6,20 +6,23 @@
 
 from twisted.internet.defer import inlineCallbacks, returnValue, succeed
 
-from aiki.cli import make_arg_parser, setup_logging, run_command, parse_relation_settings
+from aiki.cli import (
+    make_arg_parser, setup_logging, run_command, parse_relation_settings,
+    format_states, format_descriptors, format_port_pairs)
 from aiki.twistutils import CountDownLatch, wait_for_results
 from aiki.watcher import Watcher, ANY_SETTING
 from juju.hooks.cli import parse_port_protocol
 
 
 log = logging.getLogger("watch")
+ERROR_STATES = set(["error", "install-error", "start-error", "stop-error"])
 
 
 def main():
     parser = make_parser()
     options = parse_options(parser)
     setup_logging(options)
-    run_command(watch, options)
+    run_command(WatchManager(), options)
 
 
 def make_parser(root_parser=None):
@@ -76,6 +79,7 @@
         $ 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("--failfast", default=False, action="store_true", help="Immediately fail if any conditions are in error state")
     main_parser.add_argument("--number", default=False, action="store_true", help="Number output by the corresponding condition")
     return main_parser
 
@@ -123,151 +127,153 @@
     return options
 
 
-@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)
-
-
-def _format_states(states):
-    return "[" + ", ".join(sorted(states)) + "]"
-
-
-def _format_port_pairs(port_pairs):
-    return "[" + ", ".join("%s/%s" % (port, proto) for port, proto in sorted(port_pairs)) + "]"
-
-
-@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)
-
-    # Support --state/--x-state
-    if condition.states or condition.excluded_states:
-        log.info(
-            "Waiting for unit %s to be in %s and not in %s", 
-            unit_name, _format_states(condition.states), _format_states(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 state: %s", unit_name, agent_state)
-
-    # Support --open-port/--closed-port
-    if condition.open_ports or condition.closed_ports:
-        open_ports = set(parse_port_protocol(p) for p in condition.open_ports)
-        closed_ports = set(parse_port_protocol(p) for p in condition.closed_ports)
-        log.info(
-            "Waiting for unit %s to have open ports in %s and not in %s", 
-            unit_name, _format_port_pairs(open_ports), _format_port_pairs(closed_ports))
-        open_ports = yield watcher.wait_for_unit_ports(unit_name, open_ports, closed_ports)
-        log.info("Completed waiting for unit %s open ports: %s", unit_name, _format_port_pairs(open_ports))
-
-    # Support relation settings
-    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)
-    log.info("Completed waiting for unit %s", unit_name) 
-    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
+class WatchManager(object):
+    """Manages the watching of all conditions"""
+
+    def __init__(self):
+        self.overall = None
+        self.exception = None
+
+    def failfast(self, e):
+        log.error("Failed fast: %s", e)
+        self.exception = e
+        self.overall.callback(None)
+
+    @inlineCallbacks
+    def __call__(self, result, client, options):
+        self.options = options
+        self.client = client
+        self.watcher = Watcher(self.client)
+        wait_for_conditions = []
+        for condition in self.options.conditions:
+            wait_for_conditions.append(self.parse_and_wait_on_condition(condition))
+        self.overall = wait_for_results(wait_for_conditions, self.options.any)
+        yield self.overall
+        if self.exception:
+            raise self.exception
+
+    def print_result(self, condition, result):
+        """Given conditions complete asynchronously, the user can specify numbering to disambiguate"""
+        if self.options.number:
+            print condition.count,
+        print result
+
+    @inlineCallbacks
+    def parse_and_wait_on_condition(self, condition):
+        """Given a condition description from argparse, creates a watcher for it"""
+
+        # 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 self.wait_for_relation(condition, condition.instance)
+            self.print_result(condition, relation_ident)
+            return
+
+        # 2. Service unit
+        if "/" in condition.instance:
+            yield self.wait_for_unit(condition, condition.instance)
+            self.print_result(condition, condition.instance)
+            return
+
+        # 3. Service
+        yield self.wait_for_service(condition)
+        self.print_result(condition, condition.instance)
+
+    @inlineCallbacks
+    def wait_for_unit(self, condition, unit_name, relation_ident=None):
+        log.info("Waiting for unit %s...", unit_name)
+        yield self.watcher.wait_for_unit(unit_name)
+
+        # Support --state/--x-state
+        if self.options.failfast:
+            condition.states.update(ERROR_STATES)
+
+        if condition.states or condition.excluded_states:
+            log.info(
+                "Waiting for unit %s to be in %s and not in %s", 
+                unit_name, format_states(condition.states), format_states(condition.excluded_states))
+            agent_state = yield self.watcher.wait_for_unit_state(unit_name, condition.states, condition.excluded_states)
+            log.info("Completed waiting for unit %s state: %s", unit_name, agent_state)
+            if self.options.failfast and agent_state in ERROR_STATES:
+                self.failfast(Exception("Unit %s in error state: %s" % (unit_name, agent_state)))
+
+        # Support --open-port/--closed-port
+        if condition.open_ports or condition.closed_ports:
+            open_ports = set(parse_port_protocol(p) for p in condition.open_ports)
+            closed_ports = set(parse_port_protocol(p) for p in condition.closed_ports)
+            log.info(
+                "Waiting for unit %s to have open ports in %s and not in %s", 
+                unit_name, format_port_pairs(open_ports), format_port_pairs(closed_ports))
+            open_ports = yield self.watcher.wait_for_unit_ports(unit_name, open_ports, closed_ports)
+            log.info("Completed waiting for unit %s open ports: %s", unit_name, format_port_pairs(open_ports))
+
+        # Support relation settings
+        if condition.relation:
+            if relation_ident is None:
+                relation_ident = yield self.wait_for_relation(condition)
+            if condition.settings:
+                log.info("Waiting for %s: settings %s", unit_name, condition.settings)
+                settings = yield self.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)
+        log.info("Completed waiting for unit %s", unit_name) 
+        returnValue(unit_name)
+
+    @inlineCallbacks
+    def wait_for_relation(self, 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 self.watcher.wait_for_relation(*descriptors)
+            log.info("Completed waiting for %s", format_descriptors(*descriptors))
             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)
-
-    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 options used in the condition imply service unit watching,
-    # then default num_units to 1
-    if condition.num_units is None and (
-            condition.states or condition.excluded_states or condition.settings or
-            condition.open_ports or condition.closed_ports):
-        condition.num_units = 1
-
-    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))
+
+        raise ValueError("Bad relation: %s" % relation)
+
+    @inlineCallbacks
+    def wait_for_service(self, 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 self.watcher.wait_for_service(condition.instance)
+        log.info("Completed waiting for %s service", condition.instance)
+        if condition.relation:
+            relation_ident = yield self.wait_for_relation(condition)
+        else:
+            relation_ident = None
+
+        # If options used in the condition imply service unit watching,
+        # then default num_units to 1
+        if condition.num_units is None and (
+                condition.states or condition.excluded_states or condition.settings or
+                condition.open_ports or condition.closed_ports):
+            condition.num_units = 1
+
+        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(self.wait_for_unit(condition, unit_name, relation_ident)))
+
+            yield self.watcher.watch_new_service_units(condition.instance, new_unit_cb)
+            yield latch.completed
+        returnValue(condition.instance)
 
 
 if __name__ == '__main__':