Merge lp:~sandy-walsh/nova/zones3 into lp:~hudson-openstack/nova/trunk

[Sandy Walsh]

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

This branch adds a Fanout (broadcast) Queue to all services. This lets anyone talk to all services of a particular type without having to iterate through each.

Compute, Volume and Network now derive from nova.SchedulerDependentManager, which gives them the ability to update all the Scheduler nodes of their capabilities (via the fanout queue).

These capabilities are stored in each Scheduler Zone Manager and available via the Scheduler.API

The OS API '/zones/info' call now returns the aggregated (min, max) values of each of the service capabilities: http://paste.openstack.org/show/815/

To integration test this, simply fire up another scheduler (or more), add some capabilities to Compute, Network or Volume via the services update_service_capabilities() call and watch the events appear in each Scheduler node.

[Todd Willey (xtoddx)]

I think this looks pretty good.

Is there any information (RST docs) regarding what meaningful values for capabilities are?

Also, in some flags we are using x=y,a=b syntax for multi-valued flags (defualt_log_levels, etc). Perhaps we can keep the same format? Long term perhaps we could make a new DEFINE_dict or some-such that would parse those into a dict.

It looks like the only information that is being fanned out is the service capabilities, but would it make sense to keep abilities in the database pointing to the services themselves? In other words, why are service capabilities ephemeral with the services starting and stopping? If we keep them in the db we can modify them while running and modify the capabilities with nova-manage. This of course doesn't make sense for things like changing the hypervisor for compute since that would always require a restart, but would it be practical for other capabilities?

review: Needs Information

[Sandy Walsh (sandy-walsh)]

Todd, thanks for the feedback.

I think the meaningful values for caps will come from the distributed scheduler effort. That branch really needs to drive what is required from the services.

RE flags: do you mean "x=1,x=2,x=3,y=9,y=8,y=7" vs. "x:1,2,3;y:9,8,7;" ?

RE storing the capabilities: yes, storing it in the db is a possibility. Since many capabilities are not static (disk remaining, bandwidth usage, cpu, etc) I had concerns about the frequency of the updates and the added load on the db. But this could use the same periodic storage scheme as the fanout mechanism.

[Sandy Walsh (sandy-walsh)]

Oh, perhaps the multi value keys where "x:1;2;3, y;9;8;7" ? That would be an easy change. Perhaps move the parser into utils.py as well?

[Todd Willey (xtoddx)]

From nova/flags.py:

flags.DEFINE_list('default_log_levels',
                  ['amqplib=WARN',
                   'sqlalchemy=WARN',
                   'boto=WARN',
                   'eventlet.wsgi.server=WARN'],
                  'list of logger=LEVEL pairs')

I think that is a good pattern to follow. Your argument ends up looking like

--zone_capabilities=hypervisor=xenserver,os=linux

You'll end up with the list

['hypervisor=xenserver', 'os=linux']

You're definitely treating it as a list by splitting and iterating over it, so DEFINE_list seems reasonable. Based on your comments it looks like you might specify multiple values for any individual capability, in which case using a comma would break the flag list, maybe using a colon as the item separator would be reasonable?

[Eric Day (eday)]

8: Ahh! How did nova.db get into nova.api? I must have missed this in the last branch. It would be nice to abstract this in nova/scheduler or a nova/zones/ package.

Should SchedulerDependentManager just be in nova/manager.py?

As for using a DB to store this data (Todd's comment), we need to move away from compute/network/volume hosts writing to the DB directly for security concerns. We should be using the msg queue for all communications/data and let the scheduler verify the source once we have a mechanism to do so.

[Sandy Walsh (sandy-walsh)]

eric, agree on moving SchedulerDependentManager into nova/manager.py and abstracting out the db call into the scheduler api. Will do.

Todd, colon is bad for URI's. I'll mess around the the List option.

[Sandy Walsh (sandy-walsh)]

Todd, how about x=1;2;3, y=9;8;7 .. that way we can use DEFINE_list and have multi-values?

[Sandy Walsh (sandy-walsh)]

All fixed up as requested ... thanks again for the feedback!

[Todd Willey (xtoddx)]

I'm on board with that style of flag.

review: Approve

[Joshua McKenty (joshua-mckenty)]

Typo on 79 (missing i).
Seems like service_name should be a class property rather than an instance property, any reason we can't do it this way? Then the super __init__ call doesn't need an extra arg.
I think TopicConsumer was the last of my code from the first nova-hacking weekend - I'm glad to see it go.

Log message on 231 seems wrong - is it actually "writing" at that point, or just initing?
I like how you've dropped _call_scheduler out of the API class - but do we even NEED that class any more? The rest of the class methods could be dropped down to functions as well.

467 - can we call this zone_capabilities? Caps seems like a pseudonym for "limits".
Looks like we need a simple test of the fanout in test_rpc.py.

Doesn't seem like there's any way to remove a host from service capabilities once it's been added - is this necessary?

review: Needs Information

[Sandy Walsh (sandy-walsh)]

Joshua, great feedback ... thanks for that. All good suggestions. I'll get on them.

Hmm, I was going to add something to decay the hosts after time. I may have forgotten that, lemme see.

If not, I can easily add it to Zones4 (coming soon).

If you'd like to see the direction this is heading have a look here (WIP):
https://code.launchpad.net/~sandy-walsh/nova/zones4/+merge/53726

Cheers!

[Sandy Walsh (sandy-walsh)]

Changes made ... the fanout test should be an integration test vs. a unit test. We'll need to revisit that once we get the integration test framework ironed out.

[Rick Harris (rconradharris)]

Overall this patch looks really good Sandy. Nice tests, and some really
helpful comments in there (as usual).

Generally speaking, I think a few whitespaces cleanups would make a make this
patch even better (in particular lining up arg-lists). Not a huge deal,
obviously, but worth mentioning.

In terms of design and implementation, I think it looks good. This matches up
with what we whiteboarded (AFAICT), so I'm all for it.

Specifics
=========

> 10 +from nova import log as logging

Imported but not used.

> 35 + for cap in caps:
> 36 + key_values = cap.split('=')
> 37 + zone[key_values[0]] = key_values[1]

Might be slightly easier to read with tuple-unpacking (also, passing 1 to
split in case the value has a '=' in it):

    for cap in caps:
        key, value = cap.split('=', 1)
        zone[key] = value

> 62 + super(ComputeManager, self).__init__(service_name="compute",
> 63 + *args, **kwargs)

Might be better if args aligned:

    super(ComputeManager, self).__init__(service_name="compute",
                                         *args, **kwargs)
> 77 +DEFINE_list('zone_capabilities',
> 78 + ['hypervisor=xenserver;kvm', 'os=linux;windows'],
> 79 + 'Key/Multi-value list representng capabilities of this zone')

Same as above:

    DEFINE_list('zone_capabilities',
                ['hypervisor=xenserver;kvm', 'os=linux;windows'],
                'Key/Multi-value list representng capabilities of this zone')

> 88 +from nova import log as logging

Might be worth using the LOG pattern here to get more specific logging:

    LOG = logging.getLogger('nova.manager')

> 106 + update_service_capabilities is called with non-None values."""
> 107 + def __init__(self, host=None, db_driver=None, service_name="undefined"):

Spacing. Might be better as:

       update_service_capabilities is called with non-None values.
    """

    def __init__(self, host=None, db_driver=None, service_name="undefined"):

> 313 + params=dict(service=service))

Couple spaces ---> thataway (to line up with open paren +1)

> 209 + LOG.info(_("Created '%(exchange)s' fanout exchange "
> 210 + "with '%(key)s' routing key"),
> 211 + dict(exchange=self.exchange, key=self.routing_key))

Could use a bit of formatting work:

    LOG.info(_("Created '%(exchange)s' fanout exchange "
               "with '%(key)s' routing key"),
             dict(exchange=self.exchange, key=self.routing_key))

Although, this might be personal preference on my part, so, if that change
doesn't look good to you, please feel free to ignore! :-)

> 638 + svc10_a=(99, 99), svc10_b=(99, 99)))

Spacing, a couple of spaces <---- thattaway.

review: Needs Fixing

[Matt Dietz (cerberus)]

I don't see anything glaring ;-)

lgtm. Just waiting on Rick

review: Approve

[Sandy Walsh (sandy-walsh)]

Thanks for the feedback Rick ... I agree with your suggestions. Doing the code reviews on glance and seeing the work you and jay have been doing might make me change my style :)

Stay tuned.

[Rick Harris (rconradharris)]

Looks great, thanks for the fix-ups, Sandy!

review: Approve

[OpenStack Infra (hudson-openstack)]

No proposals found for merge of lp:~sandy-walsh/nova/zones2 into lp:nova.

Unmerged revisions

=== modified file 'nova/api/openstack/zones.py'
--- nova/api/openstack/zones.py	2011-03-11 19:49:32 +0000
+++ nova/api/openstack/zones.py	2011-03-23 19:38:07 +0000
@@ -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 @@
         """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 @@
 
     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"""
 
=== modified file 'nova/compute/manager.py'
--- nova/compute/manager.py	2011-03-21 15:41:03 +0000
+++ nova/compute/manager.py	2011-03-23 19:38:07 +0000
@@ -105,7 +105,7 @@
     return decorated_function
 
 
-class ComputeManager(manager.Manager):
+class ComputeManager(manager.SchedulerDependentManager):
 
     """Manages the running instances from creation to destruction."""
 
@@ -124,7 +124,8 @@
 
         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
 
=== modified file 'nova/flags.py'
--- nova/flags.py	2011-03-18 11:35:00 +0000
+++ nova/flags.py	2011-03-23 19:38:07 +0000
@@ -358,5 +358,6 @@
               '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')
 
=== modified file 'nova/manager.py'
--- nova/manager.py	2010-12-15 00:05:39 +0000
+++ nova/manager.py	2011-03-23 19:38:07 +0000
@@ -53,11 +53,14 @@
 
 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 @@
         """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)
 
=== modified file 'nova/network/manager.py'
--- nova/network/manager.py	2011-03-23 05:29:32 +0000
+++ nova/network/manager.py	2011-03-23 19:38:07 +0000
@@ -105,7 +105,7 @@
     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 @@
         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
 
=== modified file 'nova/rpc.py'
--- nova/rpc.py	2011-03-18 13:56:05 +0000
+++ nova/rpc.py	2011-03-23 19:38:07 +0000
@@ -137,24 +137,7 @@
         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 @@
         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 @@
         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 @@
     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)
 
=== modified file 'nova/scheduler/api.py'
--- nova/scheduler/api.py	2011-02-25 21:40:15 +0000
+++ nova/scheduler/api.py	2011-03-23 19:38:07 +0000
@@ -25,25 +25,40 @@
 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)
 
=== modified file 'nova/scheduler/driver.py'
--- nova/scheduler/driver.py	2011-03-10 04:30:52 +0000
+++ nova/scheduler/driver.py	2011-03-23 19:38:07 +0000
@@ -49,6 +49,13 @@
 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."""
 
=== modified file 'nova/scheduler/manager.py'
--- nova/scheduler/manager.py	2011-03-14 17:59:41 +0000
+++ nova/scheduler/manager.py	2011-03-23 19:38:07 +0000
@@ -41,10 +41,11 @@
 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 @@
         """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.
 
 
=== modified file 'nova/scheduler/zone_manager.py'
--- nova/scheduler/zone_manager.py	2011-03-03 14:55:02 +0000
+++ nova/scheduler/zone_manager.py	2011-03-23 19:38:07 +0000
@@ -105,12 +105,36 @@
     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 @@
             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
 
=== modified file 'nova/service.py'
--- nova/service.py	2011-03-18 13:56:05 +0000
+++ nova/service.py	2011-03-23 19:38:07 +0000
@@ -97,18 +97,24 @@
 
         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)
 
=== modified file 'nova/tests/api/openstack/test_zones.py'
--- nova/tests/api/openstack/test_zones.py	2011-03-11 19:49:32 +0000
+++ nova/tests/api/openstack/test_zones.py	2011-03-23 19:38:07 +0000
@@ -75,6 +75,10 @@
     ]
 
 
+def zone_capabilities(method, context, params):
+    return dict()
+
+
 class ZonesTest(test.TestCase):
     def setUp(self):
         super(ZonesTest, self).setUp()
@@ -93,13 +97,18 @@
         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 @@
         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 @@
         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')
 
=== modified file 'nova/tests/test_rpc.py'
--- nova/tests/test_rpc.py	2011-01-19 20:26:09 +0000
+++ nova/tests/test_rpc.py	2011-03-23 19:38:07 +0000
@@ -36,7 +36,7 @@
         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 @@
 
         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()
 
=== modified file 'nova/tests/test_service.py'
--- nova/tests/test_service.py	2011-03-10 06:16:03 +0000
+++ nova/tests/test_service.py	2011-03-23 19:38:07 +0000
@@ -109,20 +109,29 @@
         app = service.Service.create(host=host, binary=binary)
 
         self.mox.StubOutWithMock(rpc,
-                                 'AdapterConsumer',
-                                 use_mock_anything=True)
-        rpc.AdapterConsumer(connection=mox.IgnoreArg(),
+                                 'TopicAdapterConsumer',
+                                 use_mock_anything=True)
+        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.AdapterConsumer.attach_to_eventlet()
-        rpc.AdapterConsumer.attach_to_eventlet()
+                                    rpc.TopicAdapterConsumer)
+
+        rpc.FanoutAdapterConsumer(connection=mox.IgnoreArg(),
+                            topic=topic,
+                            proxy=mox.IsA(service.Service)).AndReturn(
+                                    rpc.FanoutAdapterConsumer)
+
+        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 @@
         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)
 
=== modified file 'nova/tests/test_test.py'
--- nova/tests/test_test.py	2011-02-21 22:55:06 +0000
+++ nova/tests/test_test.py	2011-03-23 19:38:07 +0000
@@ -34,7 +34,7 @@
 
     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()
 
=== modified file 'nova/tests/test_zones.py'
--- nova/tests/test_zones.py	2011-03-03 14:55:02 +0000
+++ nova/tests/test_zones.py	2011-03-23 19:38:07 +0000
@@ -76,6 +76,40 @@
         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()
 
=== modified file 'nova/volume/manager.py'
--- nova/volume/manager.py	2011-03-03 13:54:11 +0000
+++ nova/volume/manager.py	2011-03-23 19:38:07 +0000
@@ -64,14 +64,15 @@
                      '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