Merge lp:~hazmat/juju-deployer/refactor-placement-and-validate-feedback into lp:juju-deployer

[Kapil Thangavelu]

refactor of unit placement and deployment validation/feedback.

[Francesco Banconi (frankban)]

The following summarizes what discussed on IRC.

Currently the guiserver starts the deployer in a separate process using a concurrent.futures.ProcessPoolExecutor. The executor calls the validate and import_bundle functions in deployer.guiserver. This means that, as the branch is now, the log messages will appear in the guiserver logs, and that's good. But what we'd also need is something that the future returned by the executor can receive, i.e.
1) a picklable return value or 2) an picklable exception (that is what we do now).
Long story short, it would be nice to have that kind of feedback implemented as an exception itself, or returned by the functions we call (validate/import_bundle).

Some background:
From the guiserver perspective, in case of errors, we need some error info to sent back to the user, and this can be stored in future.result() or future.exception() (right now the latter is used, i.e. validate() and import_bundle() just return None or raise an exception_.
Currently sometimes the code path started by those functions just raises exceptions without messages. In those cases, we don't have feedback to send back to the guiserver client (usually the GUI).

[Kapil Thangavelu (hazmat)]

just to be clear the branch atm is still backwards compatible with the previous gui (and other) usages, validate and others still raise ErrorExit, just that it now provides a place to cleanly override the handling of feedback handling on deployment._handle_feedback while retaining the defaults (log and error) and that it collects whole sets of errors instead of erroring on the first during validation.

[Francesco Banconi (frankban)]

This branch looks good Kapil, thank you!
I'll just add some minors/comments below.

105 + def get_unit_placement(self, svc, status):
106 + if isinstance(svc, (str, unicode)):

This can also be written as isinstance(svc, basestring) <shrug>.

229 + def _handle_feedback(self, feedback):

So, this can be a plan:
after this branch is merged, I'll start working on another
one that modifies the guiserver module in the following ways:
- _validate raises an error if if any units is in error (the deployer will
  refuse to proceed anyway, and there is no reason for the
  guiserver to wait for that);
- import_bundle instantiates a Deployment subclass overriding
  _handle_feedback so that an error is raised if the feedback
  has errors.
How does it sound?

336 + if p.isdigit() and p == '0':
337 + continue

This could just be "if p == '0'" <shrug>.

372 + if placement.isdigit() and placement == "0":
373 + return self._format_placement(placement, container)

The same here.

382 + # Prefer continuing deployment with a new machine rather
383 + # than an in-progress abort.
384 + return None

Nice.

review: Approve

=== modified file 'deployer/action/importer.py'
--- deployer/action/importer.py	2013-11-12 04:54:58 +0000
+++ deployer/action/importer.py	2013-11-20 05:09:11 +0000
@@ -48,10 +48,9 @@
                     env_status = self.env.status()
                     reloaded = True
 
+                placement = self.deployment.get_unit_placement(svc, env_status)
                 for mid in range(cur_units, svc.num_units):
-                    mspec = self.deployment.get_unit_placement(
-                        svc, mid, env_status)
-                    self.env.add_unit(svc.name, mspec)
+                    self.env.add_unit(svc.name, placement.get(mid))
             else:
                 self.env.add_units(svc.name, abs(delta))
 
@@ -94,6 +93,8 @@
             else:
                 num_units = svc.num_units
 
+            placement = self.deployment.get_unit_placement(svc, env_status)
+
             self.env.deploy(
                 svc.name,
                 charm.charm_url,
@@ -101,7 +102,7 @@
                 svc.config,
                 svc.constraints,
                 num_units,
-                self.deployment.get_unit_placement(svc, 0, env_status))
+                placement.get(0))
 
             if svc.annotations:
                 self.log.debug(" Setting annotations")
 
=== modified file 'deployer/deployment.py'
--- deployer/deployment.py	2013-10-31 19:19:36 +0000
+++ deployer/deployment.py	2013-11-20 05:09:11 +0000
@@ -6,7 +6,8 @@
 import yaml
 
 from .charm import Charm
-from .service import Service
+from .feedback import Feedback
+from .service import Service, ServiceUnitPlacement
 from .relation import Endpoint
 from .utils import path_join, yaml_dump, ErrorExit, resolve_include
 
@@ -56,58 +57,10 @@
             return -1
         return cmp(svc_a.name, svc_b.name)
 
-    @staticmethod
-    def _format_placement(machine, container=None):
-        if container:
-            return "%s:%s" % (container, machine)
-        else:
-            return machine
-
-    def get_unit_placement(self, svc, unit_number, status):
-        unit_mapping = svc.unit_placement
-        if not unit_mapping:
-            return None
-        if len(unit_mapping) <= unit_number:
-            return None
-
-        unit_placement = placement = str(unit_mapping[unit_number])
-        container = None
-        u_idx = unit_number
-
-        if ':' in unit_placement:
-            container, placement = unit_placement.split(":")
-        if '=' in placement:
-            placement, u_idx = placement.split("=")
-
-        if placement.isdigit() and placement == "0":
-            return self._format_placement(placement, container)
-
-        with_service = status['services'].get(placement)
-        if with_service is None:
-            # Should be caught in validate relations but sanity check
-            # for concurrency.
-            self.log.error(
-                "Service %s to be deployed with non existant service %s",
-                svc.name, placement)
-            # Prefer continuing deployment with a new machine rather
-            # than an in-progress abort.
-            return None
-
-        svc_units = with_service['units']
-        if len(svc_units) <= unit_number:
-            self.log.warning(
-                "Service:%s deploy-with Service:%s, but no with unit found",
-                svc.name, placement)
-            return None
-        unit_names = svc_units.keys()
-        unit_names.sort()
-        machine = svc_units[unit_names[int(u_idx)]].get('machine')
-        if not machine:
-            self.log.warning(
-                "Service:%s deploy-with unit missing machine %s",
-                svc.name, unit_names[unit_number])
-            return None
-        return self._format_placement(machine, container)
+    def get_unit_placement(self, svc, status):
+        if isinstance(svc, (str, unicode)):
+            svc = self.get_service(svc)
+        return ServiceUnitPlacement(svc, self, status)
 
     def get_relations(self):
         if 'relations' not in self.data:
@@ -213,6 +166,7 @@
         self.log.debug("Resolving configuration")
         # XXX TODO, rename resolve, validate relations
         # against defined services
+        feedback = Feedback()
         for svc_name, svc_data in self.data.get('services', {}).items():
             if not 'options' in svc_data:
                 continue
@@ -222,16 +176,21 @@
 
             for k, v in svc_data['options'].items():
                 if not k in config:
-                    self.log.error(
-                        "Invalid config charm %s %s=%s", charm.name, k, v)
-                    raise ErrorExit()
+                    feedback.error(
+                        "Invalid config charm %s %s=%s" % (charm.name, k, v))
+                    continue
                 iv = self._resolve_include(svc_name, k, v)
+                if isinstance(iv, Feedback):
+                    feedback.extend(iv)
+                    continue
                 if iv is not None:
                     v = iv
                 options[k] = v
             svc_data['options'] = options
+        self._handle_feedback(feedback)
 
     def _resolve_include(self, svc_name, k, v):
+        feedback = Feedback()
         for include_type in ["file", "base64"]:
             if (not isinstance(v, basestring)
                 or not v.startswith(
@@ -240,70 +199,47 @@
             include, fname = v.split("://", 1)
             ip = resolve_include(fname, self.include_dirs)
             if ip is None:
-                self.log.warning(
-                    "Invalid config %s.%s include not found %s",
-                    svc_name, k, v)
+                feedback.error(
+                    "Invalid config %s.%s include not found %s" % (
+                    svc_name, k, v))
                 continue
             with open(ip) as fh:
                 v = fh.read()
                 if include_type == "base64":
                     v = b64encode(v)
                 return v
+        if feedback:
+            return feedback
 
     def validate_relations(self):
         # Could extend to do interface matching against charms.
         services = dict([(s.name, s) for s in self.get_services()])
+        feedback = Feedback()
         for e_a, e_b in self.get_relations():
             for ep in [Endpoint(e_a), Endpoint(e_b)]:
                 if not ep.service in services:
-                    self.log.error(
+                    feedback.error(
                         ("Invalid relation in config,"
-                         " service %s not found, rel %s"),
-                        ep.service, "%s <-> %s" % (e_a, e_b))
-                    raise ErrorExit()
+                         " service %s not found, rel %s") % (
+                             ep.service, "%s <-> %s" % (e_a, e_b)))
+                    continue
+        self._handle_feedback(feedback)
 
     def validate_placement(self):
         services = dict([(s.name, s) for s in self.get_services()])
+        feedback = Feedback()
         for name, s in services.items():
-            unit_placement = s.unit_placement
-            if unit_placement is None:
-                continue
-            if not isinstance(unit_placement, list):
-                unit_placement = [unit_placement]
-            unit_placement = map(str, unit_placement)
-            for idx, p in enumerate(unit_placement):
-                if ':' in p:
-                    container, p = p.split(':')
-                    if container not in ('lxc', 'kvm'):
-                        self.log.error(
-                            "Invalid service:%s placement: %s",
-                            name, unit_placement[idx])
-                        raise ErrorExit()
-                if '=' in p:
-                    p, u_idx = p.split("=")
-                    if not u_idx.isdigit():
-                        self.log.error(
-                            "Invalid service:%s placement: %s",
-                            name, unit_placement[idx])
-                        raise ErrorExit()
-                if p.isdigit() and p == '0':
-                    continue
-                elif p.isdigit():
-                    self.log.error(
-                        "Service placement to machine not supported %s to %s",
-                        name, unit_placement[idx])
-                    raise ErrorExit()
-                elif p in services:
-                    if services[p].unit_placement:
-                        self.log.error(
-                            "Nested placement not supported %s -> %s -> %s" % (
-                                name, p, services[p].unit_placement))
-                        raise ErrorExit()
-                else:
-                    self.log.error(
-                        "Invalid service placement %s to %s" % (
-                            name, unit_placement[idx]))
-                    raise ErrorExit()
+            placement = self.get_unit_placement(s, {})
+            feedback.extend(placement.validate())
+        self._handle_feedback(feedback)
+
+    def _handle_feedback(self, feedback):
+        for e in feedback.get_errors():
+            self.log.error(e)
+        for w in feedback.get_warnings():
+            self.log.warning(w)
+        if feedback.has_errors:
+            raise ErrorExit()
 
     def save(self, path):
         with open(path, "w") as fh:
 
=== added file 'deployer/feedback.py'
--- deployer/feedback.py	1970-01-01 00:00:00 +0000
+++ deployer/feedback.py	2013-11-20 05:09:11 +0000
@@ -0,0 +1,35 @@
+
+
+WARN = 3
+ERROR = 7
+
+
+class Feedback(object):
+
+    def __init__(self):
+        self.messages = []
+        self.has_errors = False
+
+    def error(self, msg):
+        self.messages.append((ERROR, msg))
+        self.has_errors = True
+
+    def warn(self, msg):
+        self.messages.append((WARN, msg))
+
+    def __iter__(self):
+        return iter(self.messages)
+
+    def __nonzero__(self):
+        return bool(self.messages)
+
+    def get_errors(self):
+        return [m for (m_kind, m) in self.messages if m_kind == ERROR]
+
+    def get_warnings(self):
+        return [m for (m_kind, m) in self.messages if m_kind == WARN]
+
+    def extend(self, other):
+        self.messages.extend(other.messages)
+        if not self.has_errors and other.has_errors:
+            self.has_errors = True
 
=== modified file 'deployer/service.py'
--- deployer/service.py	2013-11-20 02:28:01 +0000
+++ deployer/service.py	2013-11-20 05:09:11 +0000
@@ -1,3 +1,6 @@
+from feedback import Feedback
+
+
 class Service(object):
 
     def __init__(self, name, svc_data):
@@ -43,3 +46,110 @@
     @property
     def expose(self):
         return self.svc_data.get('expose', False)
+
+
+class ServiceUnitPlacement(object):
+
+    def __init__(self, service, deployment, status):
+        self.service = service
+        self.deployment = deployment
+        self.status = status
+
+    @staticmethod
+    def _format_placement(machine, container=None):
+        if container:
+            return "%s:%s" % (container, machine)
+        else:
+            return machine
+
+    def validate(self):
+        feedback = Feedback()
+
+        unit_placement = self.service.unit_placement
+        if unit_placement is None:
+            return feedback
+
+        if not isinstance(unit_placement, list):
+            unit_placement = [unit_placement]
+        unit_placement = map(str, unit_placement)
+
+        services = dict([(s.name, s) for s in self.deployment.get_services()])
+
+        for idx, p in enumerate(unit_placement):
+            if ':' in p:
+                container, p = p.split(':')
+                if container not in ('lxc', 'kvm'):
+                    feedback.error(
+                        "Invalid service:%s placement: %s" % (
+                            self.service.name, unit_placement[idx]))
+            if '=' in p:
+                p, u_idx = p.split("=")
+                if not u_idx.isdigit():
+                    feedback.error(
+                        "Invalid service:%s placement: %s",
+                        self.service.name, unit_placement[idx])
+            if p.isdigit() and p == '0':
+                continue
+            elif p.isdigit():
+                feedback.error(
+                    "Service placement to machine not supported %s to %s",
+                    self.service.name, unit_placement[idx])
+            elif p in services:
+                if services[p].unit_placement:
+                    feedback.error(
+                        "Nested placement not supported %s -> %s -> %s" % (
+                            self.service.name, p, services[p].unit_placement))
+            else:
+                feedback.error(
+                    "Invalid service placement %s to %s" % (
+                        self.service.name, unit_placement[idx]))
+        return feedback
+
+    def get(self, unit_number):
+        status = self.status
+        svc = self.service
+
+        unit_mapping = svc.unit_placement
+        if not unit_mapping:
+            return None
+        if len(unit_mapping) <= unit_number:
+            return None
+
+        unit_placement = placement = str(unit_mapping[unit_number])
+        container = None
+        u_idx = unit_number
+
+        if ':' in unit_placement:
+            container, placement = unit_placement.split(":")
+        if '=' in placement:
+            placement, u_idx = placement.split("=")
+
+        if placement.isdigit() and placement == "0":
+            return self._format_placement(placement, container)
+
+        with_service = status['services'].get(placement)
+        if with_service is None:
+            # Should be caught in validate relations but sanity check
+            # for concurrency.
+            self.deployment.log.error(
+                "Service %s to be deployed with non existant service %s",
+                svc.name, placement)
+            # Prefer continuing deployment with a new machine rather
+            # than an in-progress abort.
+            return None
+
+        svc_units = with_service['units']
+        if len(svc_units) <= unit_number:
+            self.deployment.log.warning(
+                "Service:%s deploy-with Service:%s, but no with unit found",
+                svc.name, placement)
+            return None
+        unit_names = svc_units.keys()
+        unit_names.sort()
+        machine = svc_units[unit_names[int(u_idx)]].get('machine')
+        if not machine:
+            self.deployment.log.warning(
+                "Service:%s deploy-with unit missing machine %s",
+                svc.name, unit_names[unit_number])
+            return None
+        return self._format_placement(machine, container)
 
=== modified file 'deployer/tests/base.py'
--- deployer/tests/base.py	2013-07-22 18:26:06 +0000
+++ deployer/tests/base.py	2013-11-20 05:09:11 +0000
@@ -7,6 +7,7 @@
 import tempfile
 
 import deployer
+from deployer.config import ConfigStack
 
 
 class Base(unittest.TestCase):
@@ -14,6 +15,13 @@
     test_data_dir = os.path.join(
         os.path.dirname(inspect.getabsfile(deployer)), "tests", "test_data")
 
+    def get_named_deployment(self, file_name, stack_name):
+        """ Get deployment from test_data file.
+        """
+        return ConfigStack(
+            [os.path.join(
+                self.test_data_dir, file_name)]).get(stack_name)
+
     def capture_logging(self, name="", level=logging.INFO,
                         log_file=None, formatter=None):
         if log_file is None:
 
=== modified file 'deployer/tests/test_deployment.py'
--- deployer/tests/test_deployment.py	2013-10-31 19:19:36 +0000
+++ deployer/tests/test_deployment.py	2013-11-20 05:09:11 +0000
@@ -16,11 +16,6 @@
         self.output = setup_logging(
             debug=True, verbose=True, stream=StringIO.StringIO())
 
-    def get_named_deployment(self, file_name, stack_name):
-        return ConfigStack(
-            [os.path.join(
-                self.test_data_dir, file_name)]).get(stack_name)
-
     def test_deployer(self):
         d = ConfigStack(
             [os.path.join(
@@ -97,24 +92,24 @@
                         'nova-compute/2': {'machine': '1'},
                         'nova-compute/3': {'machine': '2'},
                         'nova-compute/4': {'machine': '3'}}}}}
-        svc = d.get_service('ceph')
-        self.assertEqual(d.get_unit_placement(svc, 0, status), '1')
-        self.assertEqual(d.get_unit_placement(svc, 1, status), '2')
-        self.assertEqual(d.get_unit_placement(svc, 2, status), None)
-
-        svc = d.get_service('quantum')
-        self.assertEqual(d.get_unit_placement(svc, 0, status), 'lxc:1')
-        self.assertEqual(d.get_unit_placement(svc, 2, status), 'lxc:3')
-        self.assertEqual(d.get_unit_placement(svc, 3, status), None)
-
-        svc = d.get_service('verity')
-        self.assertEqual(d.get_unit_placement(svc, 0, status), 'lxc:3')
-
-        svc = d.get_service('mysql')
-        self.assertEqual(d.get_unit_placement(svc, 0, status), '0')
-
-        svc = d.get_service('semper')
-        self.assertEqual(d.get_unit_placement(svc, 0, status), '3')
+        placement = d.get_unit_placement('ceph', status)
+        self.assertEqual(placement.get(0), '1')
+        self.assertEqual(placement.get(1), '2')
+        self.assertEqual(placement.get(2), None)
+
+        placement = d.get_unit_placement('quantum', status)
+        self.assertEqual(placement.get(0), 'lxc:1')
+        self.assertEqual(placement.get(2), 'lxc:3')
+        self.assertEqual(placement.get(3), None)
+
+        placement = d.get_unit_placement('verity', status)
+        self.assertEqual(placement.get(0), 'lxc:3')
+
+        placement = d.get_unit_placement('mysql', status)
+        self.assertEqual(placement.get(0), '0')
+
+        placement = d.get_unit_placement('semper', status)
+        self.assertEqual(placement.get(0), '3')
 
     def test_multiple_relations_no_weight(self):
         data = {"relations": {"wordpress": {"consumes": ["mysql"]},