Aggregates capabilities from Compute, Network, Volume to the ZoneManager in Scheduler.

17 files changed, 290 insertions, 69 deletions

diff --git a/nova/api/openstack/zones.py b/nova/api/openstack/zones.py
index 8fe84275a..d4a59993b 100644
--- a/nova/api/openstack/zones.py
+++ b/nova/api/openstack/zones.py
@@ -15,9 +15,9 @@
 
 import common
 
+from nova import db
 from nova import flags
 from nova import wsgi
-from nova import db
 from nova.scheduler import api
 
 
@@ -52,7 +52,7 @@ class Controller(wsgi.Controller):
         """Return all zones in brief"""
         # Ask the ZoneManager in the Scheduler for most recent data,
         # or fall-back to the database ...
-        items = api.API().get_zone_list(req.environ['nova.context'])
+        items = api.get_zone_list(req.environ['nova.context'])
         if not items:
             items = db.zone_get_all(req.environ['nova.context'])
 
@@ -67,8 +67,16 @@ class Controller(wsgi.Controller):
 
     def info(self, req):
         """Return name and capabilities for this zone."""
-        return dict(zone=dict(name=FLAGS.zone_name,
-                    capabilities=FLAGS.zone_capabilities))
+        items = api.get_zone_capabilities(req.environ['nova.context'])
+
+        zone = dict(name=FLAGS.zone_name)
+        caps = FLAGS.zone_capabilities
+        for cap in caps:
+            key, value = cap.split('=')
+            zone[key] = value
+        for item, (min_value, max_value) in items.iteritems():
+            zone[item] = "%s,%s" % (min_value, max_value)
+        return dict(zone=zone)
 
     def show(self, req, id):
         """Return data about the given zone id"""
diff --git a/nova/compute/manager.py b/nova/compute/manager.py
index 7316d1510..468771f46 100644
--- a/nova/compute/manager.py
+++ b/nova/compute/manager.py
@@ -111,7 +111,7 @@ def checks_instance_lock(function):
     return decorated_function
 
 
-class ComputeManager(manager.Manager):
+class ComputeManager(manager.SchedulerDependentManager):
 
     """Manages the running instances from creation to destruction."""
 
@@ -132,7 +132,8 @@ class ComputeManager(manager.Manager):
 
         self.network_manager = utils.import_object(FLAGS.network_manager)
         self.volume_manager = utils.import_object(FLAGS.volume_manager)
-        super(ComputeManager, self).__init__(*args, **kwargs)
+        super(ComputeManager, self).__init__(service_name="compute",
+                                             *args, **kwargs)
 
     def init_host(self):
         """Do any initialization that needs to be run if this is a
diff --git a/nova/flags.py b/nova/flags.py
index 9123e9ac7..bf83b8e0f 100644
--- a/nova/flags.py
+++ b/nova/flags.py
@@ -358,5 +358,6 @@ DEFINE_string('node_availability_zone', 'nova',
               'availability zone of this node')
 
 DEFINE_string('zone_name', 'nova', 'name of this zone')
-DEFINE_string('zone_capabilities', 'kypervisor:xenserver;os:linux',
-              'Key/Value tags which represent capabilities of this zone')
+DEFINE_list('zone_capabilities',
+                ['hypervisor=xenserver;kvm', 'os=linux;windows'],
+                 'Key/Multi-value list representng capabilities of this zone')
diff --git a/nova/manager.py b/nova/manager.py
index 3d38504bd..804a50479 100644
--- a/nova/manager.py
+++ b/nova/manager.py
@@ -53,11 +53,14 @@ This module provides Manager, a base class for managers.
 
 from nova import utils
 from nova import flags
+from nova import log as logging
 from nova.db import base
-
+from nova.scheduler import api
 
 FLAGS = flags.FLAGS
 
+LOG = logging.getLogger('nova.manager')
+
 
 class Manager(base.Base):
     def __init__(self, host=None, db_driver=None):
@@ -74,3 +77,29 @@ class Manager(base.Base):
         """Do any initialization that needs to be run if this is a standalone
         service. Child classes should override this method."""
         pass
+
+
+class SchedulerDependentManager(Manager):
+    """Periodically send capability updates to the Scheduler services.
+       Services that need to update the Scheduler of their capabilities
+       should derive from this class. Otherwise they can derive from
+       manager.Manager directly. Updates are only sent after
+       update_service_capabilities is called with non-None values."""
+
+    def __init__(self, host=None, db_driver=None, service_name="undefined"):
+        self.last_capabilities = None
+        self.service_name = service_name
+        super(SchedulerDependentManager, self).__init__(host, db_driver)
+
+    def update_service_capabilities(self, capabilities):
+        """Remember these capabilities to send on next periodic update."""
+        self.last_capabilities = capabilities
+
+    def periodic_tasks(self, context=None):
+        """Pass data back to the scheduler at a periodic interval"""
+        if self.last_capabilities:
+            LOG.debug(_("Notifying Schedulers of capabilities ..."))
+            api.update_service_capabilities(context, self.service_name,
+                                self.host, self.last_capabilities)
+
+        super(SchedulerDependentManager, self).periodic_tasks(context)
diff --git a/nova/network/manager.py b/nova/network/manager.py
index 34fc042e4..d994f7dc8 100644
--- a/nova/network/manager.py
+++ b/nova/network/manager.py
@@ -105,7 +105,7 @@ class AddressAlreadyAllocated(exception.Error):
     pass
 
 
-class NetworkManager(manager.Manager):
+class NetworkManager(manager.SchedulerDependentManager):
     """Implements common network manager functionality.
 
     This class must be subclassed to support specific topologies.
@@ -116,7 +116,8 @@ class NetworkManager(manager.Manager):
         if not network_driver:
             network_driver = FLAGS.network_driver
         self.driver = utils.import_object(network_driver)
-        super(NetworkManager, self).__init__(*args, **kwargs)
+        super(NetworkManager, self).__init__(service_name='network',
+                                                *args, **kwargs)
 
     def init_host(self):
         """Do any initialization that needs to be run if this is a
diff --git a/nova/rpc.py b/nova/rpc.py
index 5935e1fb3..388f78d69 100644
--- a/nova/rpc.py
+++ b/nova/rpc.py
@@ -137,24 +137,7 @@ class Consumer(messaging.Consumer):
         return timer
 
 
-class Publisher(messaging.Publisher):
-    """Publisher base class"""
-    pass
-
-
-class TopicConsumer(Consumer):
-    """Consumes messages on a specific topic"""
-    exchange_type = "topic"
-
-    def __init__(self, connection=None, topic="broadcast"):
-        self.queue = topic
-        self.routing_key = topic
-        self.exchange = FLAGS.control_exchange
-        self.durable = False
-        super(TopicConsumer, self).__init__(connection=connection)
-
-
-class AdapterConsumer(TopicConsumer):
+class AdapterConsumer(Consumer):
     """Calls methods on a proxy object based on method and args"""
     def __init__(self, connection=None, topic="broadcast", proxy=None):
         LOG.debug(_('Initing the Adapter Consumer for %s') % topic)
@@ -207,6 +190,41 @@ class AdapterConsumer(TopicConsumer):
         return
 
 
+class Publisher(messaging.Publisher):
+    """Publisher base class"""
+    pass
+
+
+class TopicAdapterConsumer(AdapterConsumer):
+    """Consumes messages on a specific topic"""
+    exchange_type = "topic"
+
+    def __init__(self, connection=None, topic="broadcast", proxy=None):
+        self.queue = topic
+        self.routing_key = topic
+        self.exchange = FLAGS.control_exchange
+        self.durable = False
+        super(TopicAdapterConsumer, self).__init__(connection=connection,
+                                    topic=topic, proxy=proxy)
+
+
+class FanoutAdapterConsumer(AdapterConsumer):
+    """Consumes messages from a fanout exchange"""
+    exchange_type = "fanout"
+
+    def __init__(self, connection=None, topic="broadcast", proxy=None):
+        self.exchange = "%s_fanout" % topic
+        self.routing_key = topic
+        unique = uuid.uuid4().hex
+        self.queue = "%s_fanout_%s" % (topic, unique)
+        self.durable = False
+        LOG.info(_("Created '%(exchange)s' fanout exchange "
+                   "with '%(key)s' routing key"),
+                dict(exchange=self.exchange, key=self.routing_key))
+        super(FanoutAdapterConsumer, self).__init__(connection=connection,
+                                    topic=topic, proxy=proxy)
+
+
 class TopicPublisher(Publisher):
     """Publishes messages on a specific topic"""
     exchange_type = "topic"
@@ -218,6 +236,19 @@ class TopicPublisher(Publisher):
         super(TopicPublisher, self).__init__(connection=connection)
 
 
+class FanoutPublisher(Publisher):
+    """Publishes messages to a fanout exchange."""
+    exchange_type = "fanout"
+
+    def __init__(self, topic, connection=None):
+        self.exchange = "%s_fanout" % topic
+        self.queue = "%s_fanout" % topic
+        self.durable = False
+        LOG.info(_("Creating '%(exchange)s' fanout exchange"),
+                            dict(exchange=self.exchange))
+        super(FanoutPublisher, self).__init__(connection=connection)
+
+
 class DirectConsumer(Consumer):
     """Consumes messages directly on a channel specified by msg_id"""
     exchange_type = "direct"
@@ -360,6 +391,16 @@ def cast(context, topic, msg):
     publisher.close()
 
 
+def fanout_cast(context, topic, msg):
+    """Sends a message on a fanout exchange without waiting for a response"""
+    LOG.debug(_("Making asynchronous fanout cast..."))
+    _pack_context(msg, context)
+    conn = Connection.instance()
+    publisher = FanoutPublisher(topic, connection=conn)
+    publisher.send(msg)
+    publisher.close()
+
+
 def generic_response(message_data, message):
     """Logs a result and exits"""
     LOG.debug(_('response %s'), message_data)
diff --git a/nova/scheduler/api.py b/nova/scheduler/api.py
index 2405f1343..19a05b716 100644
--- a/nova/scheduler/api.py
+++ b/nova/scheduler/api.py
@@ -25,25 +25,40 @@ FLAGS = flags.FLAGS
 LOG = logging.getLogger('nova.scheduler.api')
 
 
-class API(object):
-    """API for interacting with the scheduler."""
-
-    def _call_scheduler(self, method, context, params=None):
-        """Generic handler for RPC calls to the scheduler.
-
-        :param params: Optional dictionary of arguments to be passed to the
-                       scheduler worker
-
-        :retval: Result returned by scheduler worker
-        """
-        if not params:
-            params = {}
-        queue = FLAGS.scheduler_topic
-        kwargs = {'method': method, 'args': params}
-        return rpc.call(context, queue, kwargs)
-
-    def get_zone_list(self, context):
-        items = self._call_scheduler('get_zone_list', context)
-        for item in items:
-            item['api_url'] = item['api_url'].replace('\\/', '/')
-        return items
+def _call_scheduler(method, context, params=None):
+    """Generic handler for RPC calls to the scheduler.
+
+    :param params: Optional dictionary of arguments to be passed to the
+                   scheduler worker
+
+    :retval: Result returned by scheduler worker
+    """
+    if not params:
+        params = {}
+    queue = FLAGS.scheduler_topic
+    kwargs = {'method': method, 'args': params}
+    return rpc.call(context, queue, kwargs)
+
+
+def get_zone_list(context):
+    """Return a list of zones assoicated with this zone."""
+    items = _call_scheduler('get_zone_list', context)
+    for item in items:
+        item['api_url'] = item['api_url'].replace('\\/', '/')
+    return items
+
+
+def get_zone_capabilities(context, service=None):
+    """Returns a dict of key, value capabilities for this zone,
+       or for a particular class of services running in this zone."""
+    return _call_scheduler('get_zone_capabilities', context=context,
+                          params=dict(service=service))
+
+
+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."""
+    kwargs = dict(method='update_service_capabilities',
+                  args=dict(service_name=service_name, host=host,
+                            capabilities=capabilities))
+    return rpc.fanout_cast(context, 'scheduler', kwargs)
diff --git a/nova/scheduler/driver.py b/nova/scheduler/driver.py
index ed3dfe1c0..ce05d9f6a 100644
--- a/nova/scheduler/driver.py
+++ b/nova/scheduler/driver.py
@@ -49,6 +49,13 @@ class WillNotSchedule(exception.Error):
 class Scheduler(object):
     """The base class that all Scheduler clases should inherit from."""
 
+    def __init__(self):
+        self.zone_manager = None
+
+    def set_zone_manager(self, zone_manager):
+        """Called by the Scheduler Service to supply a ZoneManager."""
+        self.zone_manager = zone_manager
+
     @staticmethod
     def service_is_up(service):
         """Check whether a service is up based on last heartbeat."""
diff --git a/nova/scheduler/manager.py b/nova/scheduler/manager.py
index 053a53356..7d62cfc4e 100644
--- a/nova/scheduler/manager.py
+++ b/nova/scheduler/manager.py
@@ -41,10 +41,11 @@ flags.DEFINE_string('scheduler_driver',
 class SchedulerManager(manager.Manager):
     """Chooses a host to run instances on."""
     def __init__(self, scheduler_driver=None, *args, **kwargs):
+        self.zone_manager = zone_manager.ZoneManager()
         if not scheduler_driver:
             scheduler_driver = FLAGS.scheduler_driver
         self.driver = utils.import_object(scheduler_driver)
-        self.zone_manager = zone_manager.ZoneManager()
+        self.driver.set_zone_manager(self.zone_manager)
         super(SchedulerManager, self).__init__(*args, **kwargs)
 
     def __getattr__(self, key):
@@ -59,6 +60,17 @@ class SchedulerManager(manager.Manager):
         """Get a list of zones from the ZoneManager."""
         return self.zone_manager.get_zone_list()
 
+    def get_zone_capabilities(self, context=None, service=None):
+        """Get the normalized set of capabilites for this zone,
+           or for a particular service."""
+        return self.zone_manager.get_zone_capabilities(context, service)
+
+    def update_service_capabilities(self, context=None, service_name=None,
+                                                host=None, capabilities={}):
+        """Process a capability update from a service node."""
+        self.zone_manager.update_service_capabilities(service_name,
+                            host, capabilities)
+
     def _schedule(self, method, context, topic, *args, **kwargs):
         """Tries to call schedule_* method on the driver to retrieve host.
 
diff --git a/nova/scheduler/zone_manager.py b/nova/scheduler/zone_manager.py
index edf9000cc..c1a50dbc3 100644
--- a/nova/scheduler/zone_manager.py
+++ b/nova/scheduler/zone_manager.py
@@ -105,12 +105,36 @@ class ZoneManager(object):
     def __init__(self):
         self.last_zone_db_check = datetime.min
         self.zone_states = {}
+        self.service_states = {}  # { <service> : { <host> : { cap k : v }}}
         self.green_pool = greenpool.GreenPool()
 
     def get_zone_list(self):
         """Return the list of zones we know about."""
         return [zone.to_dict() for zone in self.zone_states.values()]
 
+    def get_zone_capabilities(self, context, service=None):
+        """Roll up all the individual host info to generic 'service'
+           capabilities. Each capability is aggregated into
+           <cap>_min and <cap>_max values."""
+        service_dict = self.service_states
+        if service:
+            service_dict = {service: self.service_states.get(service, {})}
+
+        # TODO(sandy) - be smarter about fabricating this structure.
+        # But it's likely to change once we understand what the Best-Match
+        # code will need better.
+        combined = {}  # { <service>_<cap> : (min, max), ... }
+        for service_name, host_dict in service_dict.iteritems():
+            for host, caps_dict in host_dict.iteritems():
+                for cap, value in caps_dict.iteritems():
+                    key = "%s_%s" % (service_name, cap)
+                    min_value, max_value = combined.get(key, (value, value))
+                    min_value = min(min_value, value)
+                    max_value = max(max_value, value)
+                    combined[key] = (min_value, max_value)
+
+        return combined
+
     def _refresh_from_db(self, context):
         """Make our zone state map match the db."""
         # Add/update existing zones ...
@@ -141,3 +165,11 @@ class ZoneManager(object):
             self.last_zone_db_check = datetime.now()
             self._refresh_from_db(context)
         self._poll_zones(context)
+
+    def update_service_capabilities(self, service_name, host, capabilities):
+        """Update the per-service capabilities based on this notification."""
+        logging.debug(_("Received %(service_name)s service update from "
+                            "%(host)s: %(capabilities)s") % locals())
+        service_caps = self.service_states.get(service_name, {})
+        service_caps[host] = capabilities
+        self.service_states[service_name] = service_caps
diff --git a/nova/service.py b/nova/service.py
index 52bb15ad7..47c0b96c0 100644
--- a/nova/service.py
+++ b/nova/service.py
@@ -97,18 +97,24 @@ class Service(object):
 
         conn1 = rpc.Connection.instance(new=True)
         conn2 = rpc.Connection.instance(new=True)
+        conn3 = rpc.Connection.instance(new=True)
         if self.report_interval:
-            consumer_all = rpc.AdapterConsumer(
+            consumer_all = rpc.TopicAdapterConsumer(
                     connection=conn1,
                     topic=self.topic,
                     proxy=self)
-            consumer_node = rpc.AdapterConsumer(
+            consumer_node = rpc.TopicAdapterConsumer(
                     connection=conn2,
                     topic='%s.%s' % (self.topic, self.host),
                     proxy=self)
+            fanout = rpc.FanoutAdapterConsumer(
+                    connection=conn3,
+                    topic=self.topic,
+                    proxy=self)
 
             self.timers.append(consumer_all.attach_to_eventlet())
             self.timers.append(consumer_node.attach_to_eventlet())
+            self.timers.append(fanout.attach_to_eventlet())
 
             pulse = utils.LoopingCall(self.report_state)
             pulse.start(interval=self.report_interval, now=False)
diff --git a/nova/tests/api/openstack/test_zones.py b/nova/tests/api/openstack/test_zones.py
index 38399bb3f..a3f191aaa 100644
--- a/nova/tests/api/openstack/test_zones.py
+++ b/nova/tests/api/openstack/test_zones.py
@@ -75,6 +75,10 @@ def zone_get_all_db(context):
     ]
 
 
+def zone_capabilities(method, context, params):
+    return dict()
+
+
 class ZonesTest(test.TestCase):
     def setUp(self):
         super(ZonesTest, self).setUp()
@@ -93,13 +97,18 @@ class ZonesTest(test.TestCase):
         self.stubs.Set(nova.db, 'zone_create', zone_create)
         self.stubs.Set(nova.db, 'zone_delete', zone_delete)
 
+        self.old_zone_name = FLAGS.zone_name
+        self.old_zone_capabilities = FLAGS.zone_capabilities
+
     def tearDown(self):
         self.stubs.UnsetAll()
         FLAGS.allow_admin_api = self.allow_admin
+        FLAGS.zone_name = self.old_zone_name
+        FLAGS.zone_capabilities = self.old_zone_capabilities
         super(ZonesTest, self).tearDown()
 
     def test_get_zone_list_scheduler(self):
-        self.stubs.Set(api.API, '_call_scheduler', zone_get_all_scheduler)
+        self.stubs.Set(api, '_call_scheduler', zone_get_all_scheduler)
         req = webob.Request.blank('/v1.0/zones')
         res = req.get_response(fakes.wsgi_app())
         res_dict = json.loads(res.body)
@@ -108,8 +117,7 @@ class ZonesTest(test.TestCase):
         self.assertEqual(len(res_dict['zones']), 2)
 
     def test_get_zone_list_db(self):
-        self.stubs.Set(api.API, '_call_scheduler',
-                                zone_get_all_scheduler_empty)
+        self.stubs.Set(api, '_call_scheduler', zone_get_all_scheduler_empty)
         self.stubs.Set(nova.db, 'zone_get_all', zone_get_all_db)
         req = webob.Request.blank('/v1.0/zones')
         req.headers["Content-Type"] = "application/json"
@@ -167,3 +175,18 @@ class ZonesTest(test.TestCase):
         self.assertEqual(res_dict['zone']['id'], 1)
         self.assertEqual(res_dict['zone']['api_url'], 'http://example.com')
         self.assertFalse('username' in res_dict['zone'])
+
+    def test_zone_info(self):
+        FLAGS.zone_name = 'darksecret'
+        FLAGS.zone_capabilities = ['cap1=a;b', 'cap2=c;d']
+        self.stubs.Set(api, '_call_scheduler', zone_capabilities)
+
+        body = dict(zone=dict(username='zeb', password='sneaky'))
+        req = webob.Request.blank('/v1.0/zones/info')
+
+        res = req.get_response(fakes.wsgi_app())
+        res_dict = json.loads(res.body)
+        self.assertEqual(res.status_int, 200)
+        self.assertEqual(res_dict['zone']['name'], 'darksecret')
+        self.assertEqual(res_dict['zone']['cap1'], 'a;b')
+        self.assertEqual(res_dict['zone']['cap2'], 'c;d')
diff --git a/nova/tests/test_rpc.py b/nova/tests/test_rpc.py
index 4820e04fb..44d7c91eb 100644
--- a/nova/tests/test_rpc.py
+++ b/nova/tests/test_rpc.py
@@ -36,7 +36,7 @@ class RpcTestCase(test.TestCase):
         super(RpcTestCase, self).setUp()
         self.conn = rpc.Connection.instance(True)
         self.receiver = TestReceiver()
-        self.consumer = rpc.AdapterConsumer(connection=self.conn,
+        self.consumer = rpc.TopicAdapterConsumer(connection=self.conn,
                                             topic='test',
                                             proxy=self.receiver)
         self.consumer.attach_to_eventlet()
@@ -97,7 +97,7 @@ class RpcTestCase(test.TestCase):
 
         nested = Nested()
         conn = rpc.Connection.instance(True)
-        consumer = rpc.AdapterConsumer(connection=conn,
+        consumer = rpc.TopicAdapterConsumer(connection=conn,
                                        topic='nested',
                                        proxy=nested)
         consumer.attach_to_eventlet()
diff --git a/nova/tests/test_service.py b/nova/tests/test_service.py
index 393f9d20b..d48de2057 100644
--- a/nova/tests/test_service.py
+++ b/nova/tests/test_service.py
@@ -109,20 +109,29 @@ class ServiceTestCase(test.TestCase):
         app = service.Service.create(host=host, binary=binary)
 
         self.mox.StubOutWithMock(rpc,
-                                 'AdapterConsumer',
+                                 'TopicAdapterConsumer',
                                  use_mock_anything=True)
-        rpc.AdapterConsumer(connection=mox.IgnoreArg(),
+        self.mox.StubOutWithMock(rpc,
+                                 'FanoutAdapterConsumer',
+                                 use_mock_anything=True)
+        rpc.TopicAdapterConsumer(connection=mox.IgnoreArg(),
                             topic=topic,
                             proxy=mox.IsA(service.Service)).AndReturn(
-                                    rpc.AdapterConsumer)
+                                    rpc.TopicAdapterConsumer)
 
-        rpc.AdapterConsumer(connection=mox.IgnoreArg(),
+        rpc.TopicAdapterConsumer(connection=mox.IgnoreArg(),
                             topic='%s.%s' % (topic, host),
                             proxy=mox.IsA(service.Service)).AndReturn(
-                                    rpc.AdapterConsumer)
+                                    rpc.TopicAdapterConsumer)
+
+        rpc.FanoutAdapterConsumer(connection=mox.IgnoreArg(),
+                            topic=topic,
+                            proxy=mox.IsA(service.Service)).AndReturn(
+                                    rpc.FanoutAdapterConsumer)
 
-        rpc.AdapterConsumer.attach_to_eventlet()
-        rpc.AdapterConsumer.attach_to_eventlet()
+        rpc.TopicAdapterConsumer.attach_to_eventlet()
+        rpc.TopicAdapterConsumer.attach_to_eventlet()
+        rpc.FanoutAdapterConsumer.attach_to_eventlet()
 
         service_create = {'host': host,
                           'binary': binary,
@@ -279,6 +288,7 @@ class ServiceTestCase(test.TestCase):
         self.mox.StubOutWithMock(service.rpc.Connection, 'instance')
         service.rpc.Connection.instance(new=mox.IgnoreArg())
         service.rpc.Connection.instance(new=mox.IgnoreArg())
+        service.rpc.Connection.instance(new=mox.IgnoreArg())
         self.mox.StubOutWithMock(serv.manager.driver,
                                  'update_available_resource')
         serv.manager.driver.update_available_resource(mox.IgnoreArg(), host)
diff --git a/nova/tests/test_test.py b/nova/tests/test_test.py
index e237674e6..35c838065 100644
--- a/nova/tests/test_test.py
+++ b/nova/tests/test_test.py
@@ -34,7 +34,7 @@ class IsolationTestCase(test.TestCase):
 
     def test_rpc_consumer_isolation(self):
         connection = rpc.Connection.instance(new=True)
-        consumer = rpc.TopicConsumer(connection, topic='compute')
+        consumer = rpc.TopicAdapterConsumer(connection, topic='compute')
         consumer.register_callback(
                 lambda x, y: self.fail('I should never be called'))
         consumer.attach_to_eventlet()
diff --git a/nova/tests/test_zones.py b/nova/tests/test_zones.py
index 5a52a0506..688dc704d 100644
--- a/nova/tests/test_zones.py
+++ b/nova/tests/test_zones.py
@@ -76,6 +76,40 @@ class ZoneManagerTestCase(test.TestCase):
         self.assertEquals(len(zm.zone_states), 1)
         self.assertEquals(zm.zone_states[1].username, 'user1')
 
+    def test_service_capabilities(self):
+        zm = zone_manager.ZoneManager()
+        caps = zm.get_zone_capabilities(self, None)
+        self.assertEquals(caps, {})
+
+        zm.update_service_capabilities("svc1", "host1", dict(a=1, b=2))
+        caps = zm.get_zone_capabilities(self, None)
+        self.assertEquals(caps, dict(svc1_a=(1, 1), svc1_b=(2, 2)))
+
+        zm.update_service_capabilities("svc1", "host1", dict(a=2, b=3))
+        caps = zm.get_zone_capabilities(self, None)
+        self.assertEquals(caps, dict(svc1_a=(2, 2), svc1_b=(3, 3)))
+
+        zm.update_service_capabilities("svc1", "host2", dict(a=20, b=30))
+        caps = zm.get_zone_capabilities(self, None)
+        self.assertEquals(caps, dict(svc1_a=(2, 20), svc1_b=(3, 30)))
+
+        zm.update_service_capabilities("svc10", "host1", dict(a=99, b=99))
+        caps = zm.get_zone_capabilities(self, None)
+        self.assertEquals(caps, dict(svc1_a=(2, 20), svc1_b=(3, 30),
+                                     svc10_a=(99, 99), svc10_b=(99, 99)))
+
+        zm.update_service_capabilities("svc1", "host3", dict(c=5))
+        caps = zm.get_zone_capabilities(self, None)
+        self.assertEquals(caps, dict(svc1_a=(2, 20), svc1_b=(3, 30),
+                                     svc1_c=(5, 5), svc10_a=(99, 99),
+                                     svc10_b=(99, 99)))
+
+        caps = zm.get_zone_capabilities(self, 'svc1')
+        self.assertEquals(caps, dict(svc1_a=(2, 20), svc1_b=(3, 30),
+                                     svc1_c=(5, 5)))
+        caps = zm.get_zone_capabilities(self, 'svc10')
+        self.assertEquals(caps, dict(svc10_a=(99, 99), svc10_b=(99, 99)))
+
     def test_refresh_from_db_replace_existing(self):
         zm = zone_manager.ZoneManager()
         zone_state = zone_manager.ZoneState()
diff --git a/nova/volume/manager.py b/nova/volume/manager.py
index 9dea35b35..2178389ce 100644
--- a/nova/volume/manager.py
+++ b/nova/volume/manager.py
@@ -64,14 +64,15 @@ flags.DEFINE_boolean('use_local_volumes', True,
                      'if True, will not discover local volumes')
 
 
-class VolumeManager(manager.Manager):
+class VolumeManager(manager.SchedulerDependentManager):
     """Manages attachable block storage devices."""
     def __init__(self, volume_driver=None, *args, **kwargs):
         """Load the driver from the one specified in args, or from flags."""
         if not volume_driver:
             volume_driver = FLAGS.volume_driver
         self.driver = utils.import_object(volume_driver)
-        super(VolumeManager, self).__init__(*args, **kwargs)
+        super(VolumeManager, self).__init__(service_name='volume',
+                                                    *args, **kwargs)
         # NOTE(vish): Implementation specific db handling is done
         #             by the driver.
         self.driver.db = self.db