Scheduler notifications added.

Re-added lost fake virt hypervisor info (required for local dev).
Added multiple exchange topic support to RabbitNotifier.

Change-Id: I33cc9076ee35061c21852cabba3275006fc87b86

6 files changed, 80 insertions, 8 deletions

diff --git a/nova/notifier/rabbit_notifier.py b/nova/notifier/rabbit_notifier.py
index 864e8004d..639fe6784 100644
--- a/nova/notifier/rabbit_notifier.py
+++ b/nova/notifier/rabbit_notifier.py
@@ -21,9 +21,9 @@ from nova.openstack.common import cfg
 from nova import rpc
 
 
-notification_topic_opt = cfg.StrOpt('notification_topic',
-        default='notifications',
-        help='RabbitMQ topic used for Nova notifications')
+notification_topic_opt = cfg.ListOpt('notification_topics',
+        default=['notifications', ],
+        help='AMQP topic used for Nova notifications')
 
 FLAGS = flags.FLAGS
 FLAGS.register_opt(notification_topic_opt)
@@ -35,5 +35,10 @@ def notify(message):
     priority = message.get('priority',
                            FLAGS.default_notification_level)
     priority = priority.lower()
-    topic = '%s.%s' % (FLAGS.notification_topic, priority)
-    rpc.notify(context, topic, message)
+    for topic in FLAGS.notification_topics:
+        topic = '%s.%s' % (topic, priority)
+        try:
+            rpc.notify(context, topic, message)
+        except Exception, e:
+            LOG.exception(_("Could not send notification to %(topic)s. "
+                            "Payload=%(message)s" % locals()))
diff --git a/nova/scheduler/distributed_scheduler.py b/nova/scheduler/distributed_scheduler.py
index d11e2318f..debd6c0db 100644
--- a/nova/scheduler/distributed_scheduler.py
+++ b/nova/scheduler/distributed_scheduler.py
@@ -24,6 +24,7 @@ import operator
 from nova import exception
 from nova import flags
 from nova import log as logging
+from nova.notifier import api as notifier
 from nova.scheduler import driver
 from nova.scheduler import least_cost
 from nova.scheduler import scheduler_options
@@ -63,6 +64,10 @@ class DistributedScheduler(driver.Scheduler):
         LOG.debug(_("Attempting to build %(num_instances)d instance(s)") %
                 locals())
 
+        payload = dict(request_spec=request_spec)
+        notifier.notify(notifier.publisher_id("scheduler"),
+                        'scheduler.run_instance.start', notifier.INFO, payload)
+
         weighted_hosts = self._schedule(context, "compute", request_spec,
                                         *args, **kwargs)
 
@@ -85,6 +90,9 @@ class DistributedScheduler(driver.Scheduler):
             if instance:
                 instances.append(instance)
 
+        notifier.notify(notifier.publisher_id("scheduler"),
+                        'scheduler.run_instance.end', notifier.INFO, payload)
+
         return instances
 
     def schedule_prep_resize(self, context, request_spec, *args, **kwargs):
@@ -112,6 +120,14 @@ class DistributedScheduler(driver.Scheduler):
             kwargs):
         """Create the requested resource in this Zone."""
         instance = self.create_instance_db_entry(context, request_spec)
+
+        payload = dict(request_spec=request_spec,
+                       weighted_host=weighted_host.to_dict(),
+                       instance_id=instance['uuid'])
+        notifier.notify(notifier.publisher_id("scheduler"),
+                        'scheduler.run_instance.scheduled', notifier.INFO,
+                        payload)
+
         driver.cast_to_compute_host(context, weighted_host.host_state.host,
                 'run_instance', instance_uuid=instance['uuid'], **kwargs)
         inst = driver.encode_instance(instance, local=True)
@@ -163,6 +179,10 @@ class DistributedScheduler(driver.Scheduler):
         # unfiltered_hosts_dict is {host : ZoneManager.HostInfo()}
         unfiltered_hosts_dict = self.host_manager.get_all_host_states(
                 elevated, topic)
+
+        # Note: remember, we are using an iterator here. So only
+        # traverse this list once. This can bite you if the hosts
+        # are being scanned in a filter or weighing function.
         hosts = unfiltered_hosts_dict.itervalues()
 
         num_instances = request_spec.get('num_instances', 1)
diff --git a/nova/scheduler/least_cost.py b/nova/scheduler/least_cost.py
index f54f8fe46..27878ec68 100644
--- a/nova/scheduler/least_cost.py
+++ b/nova/scheduler/least_cost.py
@@ -59,6 +59,14 @@ class WeightedHost(object):
         self.weight = weight
         self.host_state = host_state
 
+    def to_dict(self):
+        x = dict(weight=self.weight)
+        if self.host_state:
+            x['host'] = self.host_state.host
+        if self.zone:
+            x['zone'] = self.zone
+        return x
+
 
 def noop_cost_fn(host_state, weighing_properties):
     """Return a pre-weight cost of 1 for each host"""
diff --git a/nova/scheduler/manager.py b/nova/scheduler/manager.py
index b3391186b..0390f5ced 100644
--- a/nova/scheduler/manager.py
+++ b/nova/scheduler/manager.py
@@ -29,6 +29,7 @@ from nova import exception
 from nova import flags
 from nova import log as logging
 from nova import manager
+from nova.notifier import api as notifier
 from nova.openstack.common import cfg
 from nova import utils
 
@@ -118,6 +119,15 @@ class SchedulerManager(manager.Manager):
             db.instance_update(context, instance_uuid,
                     {'vm_state': vm_states.ERROR})
 
+        payload = dict(request_spec=request_spec,
+                       instance_properties=properties,
+                       instance_id=instance_uuid,
+                       state=vm_states.ERROR,
+                       method=method,
+                       reason=ex)
+        notifier.notify(notifier.publisher_id("scheduler"),
+                        'scheduler.run_instance', notifier.ERROR, payload)
+
     # NOTE (masumotok) : This method should be moved to nova.api.ec2.admin.
     #                    Based on bexar design summit discussion,
     #                    just put this here for bexar release.
diff --git a/nova/tests/test_notifier.py b/nova/tests/test_notifier.py
index be6c483e2..3eeb864cb 100644
--- a/nova/tests/test_notifier.py
+++ b/nova/tests/test_notifier.py
@@ -88,8 +88,8 @@ class NotifierTestCase(test.TestCase):
     def test_rabbit_priority_queue(self):
         self.stubs.Set(nova.flags.FLAGS, 'notification_driver',
                 'nova.notifier.rabbit_notifier')
-        self.stubs.Set(nova.flags.FLAGS, 'notification_topic',
-                'testnotify')
+        self.stubs.Set(nova.flags.FLAGS, 'notification_topics',
+                       ['testnotify', ])
 
         self.test_topic = None
 
diff --git a/nova/virt/fake.py b/nova/virt/fake.py
index 53a1ea4c9..ba2d6dae7 100644
--- a/nova/virt/fake.py
+++ b/nova/virt/fake.py
@@ -242,7 +242,36 @@ class FakeConnection(driver.ComputeDriver):
         pass
 
     def update_available_resource(self, ctxt, host):
-        pass
+        """Updates compute manager resource info on ComputeNode table.
+
+           Since we don't have a real hypervisor, pretend we have lots of
+           disk and ram.
+        """
+
+        try:
+            service_ref = db.service_get_all_compute_by_host(ctxt, host)[0]
+        except exception.NotFound:
+            raise exception.ComputeServiceUnavailable(host=host)
+
+        # Updating host information
+        dic = {'vcpus': 1,
+               'memory_mb': 4096,
+               'local_gb': 1028,
+               'vcpus_used': 0,
+               'memory_mb_used': 0,
+               'local_gb_used': 0,
+               'hypervisor_type': 'fake',
+               'hypervisor_version': '1.0',
+                  'service_id': service_ref['id'],
+                 'cpu_info': '?'}
+
+        compute_node_ref = service_ref['compute_node']
+        if not compute_node_ref:
+            LOG.info(_('Compute_service record created for %s ') % host)
+            db.compute_node_create(ctxt, dic)
+        else:
+            LOG.info(_('Compute_service record updated for %s ') % host)
+            db.compute_node_update(ctxt, compute_node_ref[0]['id'], dic)
 
     def compare_cpu(self, xml):
         """This method is supported only by libvirt."""