This is the groundwork for the upcoming distributed scheduler changes. Nothing is actually wired up here, so it shouldn't break any existing code (and all tests pass).

The goals were to:

1. Define the basic distributed scheduler communication mechanism:
   a. call_zone_method - how each zone can communicate with its children

   b. encrypted child-blobs - how child zones an securely and statelessly report back weight and build-plan info

2. Put in hooks for advanced-filtering (hard-requirements, capabilities) as well as preferences (least-cost-scheduling)

3. Create a base set of dist-scheduler tests that we can extend as we add more functionality.

Next up will be to:

1. Add in a filtering driver

2. Add in a cost-scheduler driver

11 files changed, 531 insertions, 5 deletions

diff --git a/nova/api/openstack/__init__.py b/nova/api/openstack/__init__.py
index 348b70d5b..5b7f080ad 100644
--- a/nova/api/openstack/__init__.py
+++ b/nova/api/openstack/__init__.py
@@ -98,7 +98,8 @@ class APIRouter(wsgi.Router):
             server_members['inject_network_info'] = 'POST'
 
             mapper.resource("zone", "zones", controller=zones.Controller(),
-                        collection={'detail': 'GET', 'info': 'GET'}),
+                        collection={'detail': 'GET', 'info': 'GET',
+                                    'select': 'GET'})
 
             mapper.resource("user", "users", controller=users.Controller(),
                         collection={'detail': 'GET'})
diff --git a/nova/api/openstack/zones.py b/nova/api/openstack/zones.py
index 227ffecdc..af73d8f6d 100644
--- a/nova/api/openstack/zones.py
+++ b/nova/api/openstack/zones.py
@@ -13,7 +13,12 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+import json
+import urlparse
+
+from nova import crypto
 from nova import db
+from nova import exception
 from nova import flags
 from nova import log as logging
 from nova.api.openstack import common
@@ -21,6 +26,12 @@ from nova.scheduler import api
 
 
 FLAGS = flags.FLAGS
+flags.DEFINE_string('build_plan_encryption_key',
+        None,
+        '128bit (hex) encryption key for scheduler build plans.')
+
+
+LOG = logging.getLogger('nova.api.openstack.zones')
 
 
 def _filter_keys(item, keys):
@@ -97,3 +108,35 @@ class Controller(common.OpenstackController):
         zone_id = int(id)
         zone = api.zone_update(context, zone_id, env["zone"])
         return dict(zone=_scrub_zone(zone))
+
+    def select(self, req):
+        """Returns a weighted list of costs to create instances
+           of desired capabilities."""
+        ctx = req.environ['nova.context']
+        qs = req.environ['QUERY_STRING']
+        param_dict = urlparse.parse_qs(qs)
+        param_dict.pop("fresh", None)
+        # parse_qs returns a dict where the values are lists,
+        # since query strings can have multiple values for the
+        # same key. We need to convert that to single values.
+        for key in param_dict:
+            param_dict[key] = param_dict[key][0]
+        build_plan = api.select(ctx, specs=param_dict)
+        cooked = self._scrub_build_plan(build_plan)
+        return {"weights": cooked}
+
+    def _scrub_build_plan(self, build_plan):
+        """Remove all the confidential data and return a sanitized
+        version of the build plan. Include an encrypted full version
+        of the weighting entry so we can get back to it later."""
+        if not FLAGS.build_plan_encryption_key:
+            raise exception.FlagNotSet(flag='build_plan_encryption_key')
+
+        encryptor = crypto.encryptor(FLAGS.build_plan_encryption_key)
+        cooked = []
+        for entry in build_plan:
+            json_entry = json.dumps(entry)
+            cipher_text = encryptor(json_entry)
+            cooked.append(dict(weight=entry['weight'],
+                blob=cipher_text))
+        return cooked
diff --git a/nova/compute/api.py b/nova/compute/api.py
index a12f8d515..7e2494781 100644
--- a/nova/compute/api.py
+++ b/nova/compute/api.py
@@ -251,11 +251,18 @@ class API(base.Base):
             uid = context.user_id
             LOG.debug(_("Casting to scheduler for %(pid)s/%(uid)s's"
                     " instance %(instance_id)s") % locals())
+
+            # NOTE(sandy): For now we're just going to pass in the
+            # instance_type record to the scheduler. In a later phase
+            # we'll be ripping this whole for-loop out and deferring the
+            # creation of the Instance record. At that point all this will
+            # change.
             rpc.cast(context,
                      FLAGS.scheduler_topic,
                      {"method": "run_instance",
                       "args": {"topic": FLAGS.compute_topic,
                                "instance_id": instance_id,
+                               "instance_type": instance_type,
                                "availability_zone": availability_zone,
                                "injected_files": injected_files}})
 
diff --git a/nova/crypto.py b/nova/crypto.py
index 14b9cbef6..bdc32482a 100644
--- a/nova/crypto.py
+++ b/nova/crypto.py
@@ -332,6 +332,51 @@ def mkcacert(subject='nova', years=1):
     return cert, pk, pkey
 
 
+def _build_cipher(key, iv, encode=True):
+    """Make a 128bit AES CBC encode/decode Cipher object.
+       Padding is handled internally."""
+    operation = 1 if encode else 0
+    return M2Crypto.EVP.Cipher(alg='aes_128_cbc', key=key, iv=iv, op=operation)
+
+
+def encryptor(key, iv=None):
+    """Simple symmetric key encryption."""
+    key = base64.b64decode(key)
+    if iv is None:
+        iv = '\0' * 16
+    else:
+        iv = base64.b64decode(iv)
+
+    def encrypt(data):
+        cipher = _build_cipher(key, iv, encode=True)
+        v = cipher.update(data)
+        v = v + cipher.final()
+        del cipher
+        v = base64.b64encode(v)
+        return v
+
+    return encrypt
+
+
+def decryptor(key, iv=None):
+    """Simple symmetric key decryption."""
+    key = base64.b64decode(key)
+    if iv is None:
+        iv = '\0' * 16
+    else:
+        iv = base64.b64decode(iv)
+
+    def decrypt(data):
+        data = base64.b64decode(data)
+        cipher = _build_cipher(key, iv, encode=False)
+        v = cipher.update(data)
+        v = v + cipher.final()
+        del cipher
+        return v
+
+    return decrypt
+
+
 # Copyright (c) 2006-2009 Mitch Garnaat http://garnaat.org/
 #
 # Permission is hereby granted, free of charge, to any person obtaining a
diff --git a/nova/exception.py b/nova/exception.py
index cf6069454..56c20d111 100644
--- a/nova/exception.py
+++ b/nova/exception.py
@@ -255,6 +255,10 @@ class NotFound(NovaException):
         super(NotFound, self).__init__(**kwargs)
 
 
+class FlagNotSet(NotFound):
+    message = _("Required flag %(flag)s not set.")
+
+
 class InstanceNotFound(NotFound):
     message = _("Instance %(instance_id)s could not be found.")
 
diff --git a/nova/scheduler/api.py b/nova/scheduler/api.py
index 816ae5513..55f8e0a6d 100644
--- a/nova/scheduler/api.py
+++ b/nova/scheduler/api.py
@@ -81,6 +81,12 @@ def get_zone_capabilities(context):
     return _call_scheduler('get_zone_capabilities', context=context)
 
 
+def select(context, specs=None):
+    """Returns a list of hosts."""
+    return _call_scheduler('select', context=context,
+            params={"specs": specs})
+
+
 def update_service_capabilities(context, service_name, host, capabilities):
     """Send an update to all the scheduler services informing them
        of the capabilities of this service."""
@@ -105,6 +111,45 @@ def _process(func, zone):
     return func(nova, zone)
 
 
+def call_zone_method(context, method, errors_to_ignore=None, *args, **kwargs):
+    """Returns a list of (zone, call_result) objects."""
+    if not isinstance(errors_to_ignore, (list, tuple)):
+        # This will also handle the default None
+        errors_to_ignore = [errors_to_ignore]
+
+    pool = greenpool.GreenPool()
+    results = []
+    for zone in db.zone_get_all(context):
+        try:
+            nova = novaclient.OpenStack(zone.username, zone.password,
+                    zone.api_url)
+            nova.authenticate()
+        except novaclient.exceptions.BadRequest, e:
+            url = zone.api_url
+            LOG.warn(_("Failed request to zone; URL=%(url)s: %(e)s")
+                    % locals())
+            #TODO (dabo) - add logic for failure counts per zone,
+            # with escalation after a given number of failures.
+            continue
+        zone_method = getattr(nova.zones, method)
+
+        def _error_trap(*args, **kwargs):
+            try:
+                return zone_method(*args, **kwargs)
+            except Exception as e:
+                if type(e) in errors_to_ignore:
+                    return None
+                # TODO (dabo) - want to be able to re-raise here.
+                # Returning a string now; raising was causing issues.
+                # raise e
+                return "ERROR", "%s" % e
+
+        res = pool.spawn(_error_trap, *args, **kwargs)
+        results.append((zone, res))
+    pool.waitall()
+    return [(zone.id, res.wait()) for zone, res in results]
+
+
 def child_zone_helper(zone_list, func):
     """Fire off a command to each zone in the list.
     The return is [novaclient return objects] from each child zone.
diff --git a/nova/scheduler/zone_aware_scheduler.py b/nova/scheduler/zone_aware_scheduler.py
new file mode 100644
index 000000000..b3d230bd2
--- /dev/null
+++ b/nova/scheduler/zone_aware_scheduler.py
@@ -0,0 +1,119 @@
+# Copyright (c) 2011 Openstack, LLC.
+# All Rights Reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+"""
+The Zone Aware Scheduler is a base class Scheduler for creating instances
+across zones. There are two expansion points to this class for:
+1. Assigning Weights to hosts for requested instances
+2. Filtering Hosts based on required instance capabilities
+"""
+
+import operator
+
+from nova import log as logging
+from nova.scheduler import api
+from nova.scheduler import driver
+
+LOG = logging.getLogger('nova.scheduler.zone_aware_scheduler')
+
+
+class ZoneAwareScheduler(driver.Scheduler):
+    """Base class for creating Zone Aware Schedulers."""
+
+    def _call_zone_method(self, context, method, specs):
+        """Call novaclient zone method. Broken out for testing."""
+        return api.call_zone_method(context, method, specs=specs)
+
+    def schedule_run_instance(self, context, topic='compute', specs={},
+                                        *args, **kwargs):
+        """This method is called from nova.compute.api to provision
+        an instance. However we need to look at the parameters being
+        passed in to see if this is a request to:
+        1. Create a Build Plan and then provision, or
+        2. Use the Build Plan information in the request parameters
+           to simply create the instance (either in this zone or
+           a child zone)."""
+
+        if 'blob' in specs:
+            return self.provision_instance(context, topic, specs)
+
+        # Create build plan and provision ...
+        build_plan = self.select(context, specs)
+        for item in build_plan:
+            self.provision_instance(context, topic, item)
+
+    def provision_instance(context, topic, item):
+        """Create the requested instance in this Zone or a child zone."""
+        pass
+
+    def select(self, context, *args, **kwargs):
+        """Select returns a list of weights and zone/host information
+        corresponding to the best hosts to service the request. Any
+        child zone information has been encrypted so as not to reveal
+        anything about the children."""
+        return self._schedule(context, "compute", *args, **kwargs)
+
+    def schedule(self, context, topic, *args, **kwargs):
+        """The schedule() contract requires we return the one
+        best-suited host for this request.
+        """
+        res = self._schedule(context, topic, *args, **kwargs)
+        # TODO(sirp): should this be a host object rather than a weight-dict?
+        if not res:
+            raise driver.NoValidHost(_('No hosts were available'))
+        return res[0]
+
+    def _schedule(self, context, topic, *args, **kwargs):
+        """Returns a list of hosts that meet the required specs,
+        ordered by their fitness.
+        """
+
+        #TODO(sandy): extract these from args.
+        num_instances = 1
+        specs = {}
+
+        # Filter local hosts based on requirements ...
+        host_list = self.filter_hosts(num_instances, specs)
+
+        # then weigh the selected hosts.
+        # weighted = [{weight=weight, name=hostname}, ...]
+        weighted = self.weigh_hosts(num_instances, specs, host_list)
+
+        # Next, tack on the best weights from the child zones ...
+        child_results = self._call_zone_method(context, "select",
+                specs=specs)
+        for child_zone, result in child_results:
+            for weighting in result:
+                # Remember the child_zone so we can get back to
+                # it later if needed. This implicitly builds a zone
+                # path structure.
+                host_dict = {
+                        "weight": weighting["weight"],
+                        "child_zone": child_zone,
+                        "child_blob": weighting["blob"]}
+                weighted.append(host_dict)
+
+        weighted.sort(key=operator.itemgetter('weight'))
+        return weighted
+
+    def filter_hosts(self, num, specs):
+        """Derived classes must override this method and return
+           a list of hosts in [(hostname, capability_dict)] format."""
+        raise NotImplemented()
+
+    def weigh_hosts(self, num, specs, hosts):
+        """Derived classes must override this method and return
+           a lists of hosts in [{weight, hostname}] format."""
+        raise NotImplemented()
diff --git a/nova/tests/api/openstack/test_zones.py b/nova/tests/api/openstack/test_zones.py
index 5d5799b59..fa2e05033 100644
--- a/nova/tests/api/openstack/test_zones.py
+++ b/nova/tests/api/openstack/test_zones.py
@@ -20,6 +20,8 @@ import json
 
 import nova.db
 from nova import context
+from nova import crypto
+from nova import exception
 from nova import flags
 from nova import test
 from nova.api.openstack import zones
@@ -79,6 +81,18 @@ def zone_capabilities(method, context):
     return dict()
 
 
+GLOBAL_BUILD_PLAN = [
+        dict(name='host1', weight=10, ip='10.0.0.1', zone='zone1'),
+        dict(name='host2', weight=9, ip='10.0.0.2', zone='zone2'),
+        dict(name='host3', weight=8, ip='10.0.0.3', zone='zone3'),
+        dict(name='host4', weight=7, ip='10.0.0.4', zone='zone4'),
+     ]
+
+
+def zone_select(context, specs):
+    return GLOBAL_BUILD_PLAN
+
+
 class ZonesTest(test.TestCase):
     def setUp(self):
         super(ZonesTest, self).setUp()
@@ -190,3 +204,31 @@ class ZonesTest(test.TestCase):
         self.assertEqual(res_dict['zone']['name'], 'darksecret')
         self.assertEqual(res_dict['zone']['cap1'], 'a;b')
         self.assertEqual(res_dict['zone']['cap2'], 'c;d')
+
+    def test_zone_select(self):
+        FLAGS.build_plan_encryption_key = 'c286696d887c9aa0611bbb3e2025a45a'
+        self.stubs.Set(api, 'select', zone_select)
+
+        req = webob.Request.blank('/v1.0/zones/select')
+
+        res = req.get_response(fakes.wsgi_app())
+        res_dict = json.loads(res.body)
+        self.assertEqual(res.status_int, 200)
+
+        self.assertTrue('weights' in res_dict)
+
+        for item in res_dict['weights']:
+            blob = item['blob']
+            decrypt = crypto.decryptor(FLAGS.build_plan_encryption_key)
+            secret_item = json.loads(decrypt(blob))
+            found = False
+            for original_item in GLOBAL_BUILD_PLAN:
+                if original_item['name'] != secret_item['name']:
+                    continue
+                found = True
+                for key in ('weight', 'ip', 'zone'):
+                    self.assertEqual(secret_item[key], original_item[key])
+
+            self.assertTrue(found)
+            self.assertEqual(len(item), 2)
+            self.assertTrue('weight' in item)
diff --git a/nova/tests/test_crypto.py b/nova/tests/test_crypto.py
new file mode 100644
index 000000000..945d78794
--- /dev/null
+++ b/nova/tests/test_crypto.py
@@ -0,0 +1,48 @@
+# Copyright 2011 OpenStack LLC.
+# All Rights Reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+"""
+Tests for Crypto module.
+"""
+
+from nova import crypto
+from nova import test
+
+
+class SymmetricKeyTestCase(test.TestCase):
+    """Test case for Encrypt/Decrypt"""
+    def test_encrypt_decrypt(self):
+        key = 'c286696d887c9aa0611bbb3e2025a45a'
+        plain_text = "The quick brown fox jumped over the lazy dog."
+
+        # No IV supplied (all 0's)
+        encrypt = crypto.encryptor(key)
+        cipher_text = encrypt(plain_text)
+        self.assertNotEquals(plain_text, cipher_text)
+
+        decrypt = crypto.decryptor(key)
+        plain = decrypt(cipher_text)
+
+        self.assertEquals(plain_text, plain)
+
+        # IV supplied ...
+        iv = '562e17996d093d28ddb3ba695a2e6f58'
+        encrypt = crypto.encryptor(key, iv)
+        cipher_text = encrypt(plain_text)
+        self.assertNotEquals(plain_text, cipher_text)
+
+        decrypt = crypto.decryptor(key, iv)
+        plain = decrypt(cipher_text)
+
+        self.assertEquals(plain_text, plain)
diff --git a/nova/tests/test_scheduler.py b/nova/tests/test_scheduler.py
index 968ef9d6c..54b3f80fb 100644
--- a/nova/tests/test_scheduler.py
+++ b/nova/tests/test_scheduler.py
@@ -912,7 +912,8 @@ class SimpleDriverTestCase(test.TestCase):
 
 
 class FakeZone(object):
-    def __init__(self, api_url, username, password):
+    def __init__(self, id, api_url, username, password):
+        self.id = id
         self.api_url = api_url
         self.username = username
         self.password = password
@@ -920,7 +921,7 @@ class FakeZone(object):
 
 def zone_get_all(context):
     return [
-                FakeZone('http://example.com', 'bob', 'xxx'),
+                FakeZone(1, 'http://example.com', 'bob', 'xxx'),
            ]
 
 
@@ -1037,7 +1038,7 @@ class FakeNovaClient(object):
 
 class DynamicNovaClientTest(test.TestCase):
     def test_issue_novaclient_command_found(self):
-        zone = FakeZone('http://example.com', 'bob', 'xxx')
+        zone = FakeZone(1, 'http://example.com', 'bob', 'xxx')
         self.assertEquals(api._issue_novaclient_command(
                     FakeNovaClient(FakeServerCollection()),
                     zone, "servers", "get", 100).a, 10)
@@ -1051,7 +1052,7 @@ class DynamicNovaClientTest(test.TestCase):
                     zone, "servers", "pause", 100), None)
 
     def test_issue_novaclient_command_not_found(self):
-        zone = FakeZone('http://example.com', 'bob', 'xxx')
+        zone = FakeZone(1, 'http://example.com', 'bob', 'xxx')
         self.assertEquals(api._issue_novaclient_command(
                     FakeNovaClient(FakeEmptyServerCollection()),
                     zone, "servers", "get", 100), None)
@@ -1063,3 +1064,55 @@ class DynamicNovaClientTest(test.TestCase):
         self.assertEquals(api._issue_novaclient_command(
                     FakeNovaClient(FakeEmptyServerCollection()),
                     zone, "servers", "any", "name"), None)
+
+
+class FakeZonesProxy(object):
+    def do_something(*args, **kwargs):
+        return 42
+
+    def raises_exception(*args, **kwargs):
+        raise Exception('testing')
+
+
+class FakeNovaClientOpenStack(object):
+    def __init__(self, *args, **kwargs):
+        self.zones = FakeZonesProxy()
+
+    def authenticate(self):
+        pass
+
+
+class CallZoneMethodTest(test.TestCase):
+    def setUp(self):
+        super(CallZoneMethodTest, self).setUp()
+        self.stubs = stubout.StubOutForTesting()
+        self.stubs.Set(db, 'zone_get_all', zone_get_all)
+        self.stubs.Set(novaclient, 'OpenStack', FakeNovaClientOpenStack)
+
+    def tearDown(self):
+        self.stubs.UnsetAll()
+        super(CallZoneMethodTest, self).tearDown()
+
+    def test_call_zone_method(self):
+        context = {}
+        method = 'do_something'
+        results = api.call_zone_method(context, method)
+        expected = [(1, 42)]
+        self.assertEqual(expected, results)
+
+    def test_call_zone_method_not_present(self):
+        context = {}
+        method = 'not_present'
+        self.assertRaises(AttributeError, api.call_zone_method,
+                          context, method)
+
+    def test_call_zone_method_generates_exception(self):
+        context = {}
+        method = 'raises_exception'
+        results = api.call_zone_method(context, method)
+
+        # FIXME(sirp): for now the _error_trap code is catching errors and
+        # converting them to a ("ERROR", "string") tuples. The code (and this
+        # test) should eventually handle real exceptions.
+        expected = [(1, ('ERROR', 'testing'))]
+        self.assertEqual(expected, results)
diff --git a/nova/tests/test_zone_aware_scheduler.py b/nova/tests/test_zone_aware_scheduler.py
new file mode 100644
index 000000000..fdcde34c9
--- /dev/null
+++ b/nova/tests/test_zone_aware_scheduler.py
@@ -0,0 +1,119 @@
+# Copyright 2011 OpenStack LLC.
+# All Rights Reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+"""
+Tests For Zone Aware Scheduler.
+"""
+
+from nova import test
+from nova.scheduler import driver
+from nova.scheduler import zone_aware_scheduler
+from nova.scheduler import zone_manager
+
+
+class FakeZoneAwareScheduler(zone_aware_scheduler.ZoneAwareScheduler):
+    def filter_hosts(self, num, specs):
+        # NOTE(sirp): this is returning [(hostname, services)]
+        return self.zone_manager.service_states.items()
+
+    def weigh_hosts(self, num, specs, hosts):
+        fake_weight = 99
+        weighted = []
+        for hostname, caps in hosts:
+            weighted.append(dict(weight=fake_weight, name=hostname))
+        return weighted
+
+
+class FakeZoneManager(zone_manager.ZoneManager):
+    def __init__(self):
+        self.service_states = {
+                        'host1': {
+                            'compute': {'ram': 1000}
+                         },
+                         'host2': {
+                            'compute': {'ram': 2000}
+                         },
+                         'host3': {
+                            'compute': {'ram': 3000}
+                         }
+                     }
+
+
+class FakeEmptyZoneManager(zone_manager.ZoneManager):
+    def __init__(self):
+        self.service_states = {}
+
+
+def fake_empty_call_zone_method(context, method, specs):
+    return []
+
+
+def fake_call_zone_method(context, method, specs):
+    return [
+        ('zone1', [
+            dict(weight=1, blob='AAAAAAA'),
+            dict(weight=111, blob='BBBBBBB'),
+            dict(weight=112, blob='CCCCCCC'),
+            dict(weight=113, blob='DDDDDDD'),
+        ]),
+        ('zone2', [
+            dict(weight=120, blob='EEEEEEE'),
+            dict(weight=2, blob='FFFFFFF'),
+            dict(weight=122, blob='GGGGGGG'),
+            dict(weight=123, blob='HHHHHHH'),
+        ]),
+        ('zone3', [
+            dict(weight=130, blob='IIIIIII'),
+            dict(weight=131, blob='JJJJJJJ'),
+            dict(weight=132, blob='KKKKKKK'),
+            dict(weight=3, blob='LLLLLLL'),
+        ]),
+    ]
+
+
+class ZoneAwareSchedulerTestCase(test.TestCase):
+    """Test case for Zone Aware Scheduler."""
+
+    def test_zone_aware_scheduler(self):
+        """
+        Create a nested set of FakeZones, ensure that a select call returns the
+        appropriate build plan.
+        """
+        sched = FakeZoneAwareScheduler()
+        self.stubs.Set(sched, '_call_zone_method', fake_call_zone_method)
+
+        zm = FakeZoneManager()
+        sched.set_zone_manager(zm)
+
+        fake_context = {}
+        build_plan = sched.select(fake_context, {})
+
+        self.assertEqual(15, len(build_plan))
+
+        hostnames = [plan_item['name']
+                     for plan_item in build_plan if 'name' in plan_item]
+        self.assertEqual(3, len(hostnames))
+
+    def test_empty_zone_aware_scheduler(self):
+        """
+        Ensure empty hosts & child_zones result in NoValidHosts exception.
+        """
+        sched = FakeZoneAwareScheduler()
+        self.stubs.Set(sched, '_call_zone_method', fake_empty_call_zone_method)
+
+        zm = FakeEmptyZoneManager()
+        sched.set_zone_manager(zm)
+
+        fake_context = {}
+        self.assertRaises(driver.NoValidHost, sched.schedule, fake_context, {})