openstack/nova
Code Issues Proposed changes

Implements notifications for more instance changes

Browse Source 
Now sends a notification of other changes in the instance besides state
changes. Adds access_ip in notifications as well.

fixes bug 1028559

Change-Id: I664e6b858eda17dd732927f2c9b259212084a8fa
This commit is contained in:
Nikhil Komawar
2012-07-25 15:01:24 -04:00
parent 5110f707df
commit 19b54998bb
5 changed files with 167 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
nova/compute/api.py
View File

@@ -1169,7 +1169,7 @@ class API(base.Base):
notifications.send_update_with_states(context, instance, old_vm_state,
instance["vm_state"], old_task_state, instance["task_state"],
service="api")
service="api", verify_states=True)
properties = {
'instance_uuid': instance_uuid,

1
nova/compute/manager.py
View File

@@ -642,6 +642,7 @@ class ComputeManager(manager.SchedulerDependentManager):
update_info['access_ip_v6'] = ip['address']
if update_info:
self.db.instance_update(context, instance.uuid, update_info)
notifications.send_update(context, instance, instance)
def _check_instance_not_already_created(self, context, instance):
"""Ensure an instance with the same name is not already present."""

2
nova/compute/utils.py
View File

@@ -96,7 +96,7 @@ def notify_about_instance_usage(context, instance, event_suffix,
if not extra_usage_info:
extra_usage_info = {}
usage_info = notifications.usage_from_instance(context, instance,
usage_info = notifications.info_from_instance(context, instance,
network_info, system_metadata, **extra_usage_info)
notifier_api.notify(context, 'compute.%s' % host,

130
nova/notifications.py
View File

@@ -40,56 +40,111 @@ notify_state_opt = cfg.StrOpt('notify_on_state_change', default=None,
'"vm_and_task_state" for notifications on VM and task state '
'changes.')
notify_any_opt = cfg.BoolOpt('notify_on_any_change', default=False,
help='If set, send compute.instance.update notifications on instance '
'state changes. Valid values are False for no notifications, '
'True for notifications on any instance changes.')
FLAGS = flags.FLAGS
FLAGS.register_opt(notify_state_opt)
FLAGS.register_opt(notify_any_opt)
def send_update(context, old_instance, new_instance, service=None, host=None):
"""Send compute.instance.update notification to report changes
in vm state and (optionally) task state
"""Send compute.instance.update notification to report any changes occurred
in that instance
"""
send_update_with_states(context, new_instance, old_instance["vm_state"],
new_instance["vm_state"], old_instance["task_state"],
new_instance["task_state"], service, host)
def send_update_with_states(context, instance, old_vm_state, new_vm_state,
old_task_state, new_task_state, service=None, host=None):
"""Send compute.instance.update notification to report changes
in vm state and (optionally) task state
"""
if not FLAGS.notify_on_state_change:
# skip all this if state updates are disabled
if not FLAGS.notify_on_any_change and not FLAGS.notify_on_state_change:
# skip all this if updates are disabled
return
fire_update = False
update_with_state_change = False
old_vm_state = old_instance["vm_state"]
new_vm_state = new_instance["vm_state"]
old_task_state = old_instance["task_state"],
new_task_state = new_instance["task_state"]
# we should check if we need to send a state change or a regular
# notification
if old_vm_state != new_vm_state:
# yes, the vm state is changing:
fire_update = True
elif (FLAGS.notify_on_state_change.lower() == "vm_and_task_state" and
update_with_state_change = True
elif FLAGS.notify_on_state_change:
if (FLAGS.notify_on_state_change.lower() == "vm_and_task_state" and
old_task_state != new_task_state):
# yes, the task state is changing:
fire_update = True
update_with_state_change = True
if fire_update:
if update_with_state_change:
# send a notification with state changes
# value of verify_states need not be True as the check for states is
# already done here
send_update_with_states(context, new_instance, old_vm_state,
new_vm_state, old_task_state, new_task_state, service, host)
else:
try:
_send_instance_update_notification(context, instance, old_vm_state,
old_task_state, new_vm_state, new_task_state, service,
host)
_send_instance_update_notification(context, new_instance,
service=service, host=host)
except Exception:
LOG.exception(_("Failed to send state update notification"),
instance=instance)
def _send_instance_update_notification(context, instance, old_vm_state,
old_task_state, new_vm_state, new_task_state, service=None, host=None):
def send_update_with_states(context, instance, old_vm_state, new_vm_state,
old_task_state, new_task_state, service="compute", host=None,
verify_states=False):
"""Send compute.instance.update notification to report changes if there
are any, in the instance
"""
if not FLAGS.notify_on_state_change:
# skip all this if updates are disabled
return
fire_update = True
# send update notification by default
if verify_states:
# check whether we need to send notification related to state changes
fire_update = False
# do not send notification if the confitions for vm and(or) task state
# are not satisfied
if old_vm_state != new_vm_state:
# yes, the vm state is changing:
fire_update = True
elif FLAGS.notify_on_state_change:
if (FLAGS.notify_on_state_change.lower() == "vm_and_task_state" and
old_task_state != new_task_state):
# yes, the task state is changing:
fire_update = True
if fire_update:
# send either a state change or a regular notificaion
try:
_send_instance_update_notification(context, instance,
old_vm_state=old_vm_state, old_task_state=old_task_state,
new_vm_state=new_vm_state, new_task_state=new_task_state,
service=service, host=host)
except Exception:
LOG.exception(_("Failed to send state update notification"),
instance=instance)
def _send_instance_update_notification(context, instance, old_vm_state=None,
old_task_state=None, new_vm_state=None, new_task_state=None,
service="compute", host=None):
"""Send 'compute.instance.update' notification to inform observers
about instance state changes"""
payload = usage_from_instance(context, instance, None, None)
payload = info_from_instance(context, instance, None, None)
if not new_vm_state:
new_vm_state = instance["vm_state"]
if not new_task_state:
new_task_state = instance["task_state"]
states_payload = {
"old_state": old_vm_state,
@@ -109,11 +164,6 @@ def _send_instance_update_notification(context, instance, old_vm_state,
bw = bandwidth_usage(instance, audit_start)
payload["bandwidth"] = bw
# if the service name (e.g. api/scheduler/compute) is not provided, default
# to "compute"
if not service:
service = "compute"
publisher_id = notifier_api.publisher_id(service, host)
notifier_api.notify(context, publisher_id, 'compute.instance.update',
@@ -194,9 +244,9 @@ def image_meta(system_metadata):
return image_meta
def usage_from_instance(context, instance_ref, network_info,
def info_from_instance(context, instance_ref, network_info,
system_metadata, **kw):
"""Get usage information for an instance which is common to all
"""Get detailed instance information for an instance which is common to all
notifications.
:param network_info: network_info provided if not None
@@ -220,7 +270,7 @@ def usage_from_instance(context, instance_ref, network_info,
except exception.NotFound:
system_metadata = {}
usage_info = dict(
instance_info = dict(
# Owner properties
tenant_id=instance_ref['project_id'],
user_id=instance_ref['user_id'],
@@ -266,6 +316,10 @@ def usage_from_instance(context, instance_ref, network_info,
# Status properties
state=instance_ref['vm_state'],
state_description=null_safe_str(instance_ref.get('task_state')),
# accessIPs
access_ip_v4=instance_ref['access_ip_v4'],
access_ip_v6=instance_ref['access_ip_v6'],
)
if network_info is not None:
@@ -274,11 +328,11 @@ def usage_from_instance(context, instance_ref, network_info,
for ip in vif.fixed_ips():
ip["label"] = vif["network"]["label"]
fixed_ips.append(ip)
usage_info['fixed_ips'] = fixed_ips
instance_info['fixed_ips'] = fixed_ips
# add image metadata
image_meta_props = image_meta(system_metadata)
usage_info["image_meta"] = image_meta_props
instance_info["image_meta"] = image_meta_props
usage_info.update(kw)
return usage_info
instance_info.update(kw)
return instance_info

78
nova/tests/test_notifications.py
View File

@@ -75,6 +75,9 @@ class NotificationsTestCase(test.TestCase):
inst['instance_type_id'] = type_id
inst['root_gb'] = 0
inst['ephemeral_gb'] = 0
inst['access_ip_v4'] = '1.2.3.4'
inst['access_ip_v6'] = 'feed:5eed'
inst['display_name'] = 'test_instance'
if params:
inst.update(params)
return db.instance_create(self.context, inst)
@@ -83,10 +86,20 @@ class NotificationsTestCase(test.TestCase):
# test config disable of the notifcations
self.flags(notify_on_state_change=None)
self.flags(notify_on_any_change=False)
old = copy.copy(self.instance)
self.instance["vm_state"] = vm_states.ACTIVE
old_vm_state = old['vm_state']
new_vm_state = self.instance["vm_state"]
old_task_state = old['task_state']
new_task_state = self.instance["task_state"]
notifications.send_update_with_states(self.context, self.instance,
old_vm_state, new_vm_state, old_task_state, new_task_state,
verify_states=True)
notifications.send_update(self.context, old, self.instance)
self.assertEquals(0, len(test_notifier.NOTIFICATIONS))
@@ -99,7 +112,15 @@ class NotificationsTestCase(test.TestCase):
old = copy.copy(self.instance)
self.instance["task_state"] = task_states.SPAWNING
notifications.send_update(self.context, old, self.instance)
old_vm_state = old['vm_state']
new_vm_state = self.instance["vm_state"]
old_task_state = old['task_state']
new_task_state = self.instance["task_state"]
notifications.send_update_with_states(self.context, self.instance,
old_vm_state, new_vm_state, old_task_state, new_task_state,
verify_states=True)
self.assertEquals(0, len(test_notifier.NOTIFICATIONS))
# ok now enable task state notifcations and re-try
@@ -111,7 +132,15 @@ class NotificationsTestCase(test.TestCase):
def test_send_no_notif(self):
# test notification on send no initial vm state:
notifications.send_update(self.context, self.instance, self.instance)
old_vm_state = self.instance['vm_state']
new_vm_state = self.instance['vm_state']
old_task_state = self.instance['task_state']
new_task_state = self.instance['task_state']
notifications.send_update_with_states(self.context, self.instance,
old_vm_state, new_vm_state, old_task_state, new_task_state,
service="compute", host=None, verify_states=True)
self.assertEquals(0, len(test_notifier.NOTIFICATIONS))
def test_send_on_vm_change(self):
@@ -140,36 +169,49 @@ class NotificationsTestCase(test.TestCase):
notifications.send_update_with_states(self.context, self.instance,
vm_states.BUILDING, vm_states.BUILDING, task_states.SPAWNING,
task_states.SPAWNING)
task_states.SPAWNING, verify_states=True)
self.assertEquals(0, len(test_notifier.NOTIFICATIONS))
def test_vm_update_with_states(self):
notifications.send_update_with_states(self.context, self.instance,
vm_states.BUILDING, vm_states.ACTIVE, task_states.SPAWNING,
task_states.SPAWNING)
task_states.SPAWNING, verify_states=True)
self.assertEquals(1, len(test_notifier.NOTIFICATIONS))
notif = test_notifier.NOTIFICATIONS[0]
payload = notif["payload"]
access_ip_v4 = self.instance["access_ip_v4"]
access_ip_v6 = self.instance["access_ip_v6"]
display_name = self.instance["display_name"]
self.assertEquals(vm_states.BUILDING, payload["old_state"])
self.assertEquals(vm_states.ACTIVE, payload["state"])
self.assertEquals(task_states.SPAWNING, payload["old_task_state"])
self.assertEquals(task_states.SPAWNING, payload["new_task_state"])
self.assertEquals(payload["access_ip_v4"], access_ip_v4)
self.assertEquals(payload["access_ip_v6"], access_ip_v6)
self.assertEquals(payload["display_name"], display_name)
def test_task_update_with_states(self):
self.flags(notify_on_state_change="vm_and_task_state")
notifications.send_update_with_states(self.context, self.instance,
vm_states.BUILDING, vm_states.BUILDING, task_states.SPAWNING,
None)
None, verify_states=True)
self.assertEquals(1, len(test_notifier.NOTIFICATIONS))
notif = test_notifier.NOTIFICATIONS[0]
payload = notif["payload"]
access_ip_v4 = self.instance["access_ip_v4"]
access_ip_v6 = self.instance["access_ip_v6"]
display_name = self.instance["display_name"]
self.assertEquals(vm_states.BUILDING, payload["old_state"])
self.assertEquals(vm_states.BUILDING, payload["state"])
self.assertEquals(task_states.SPAWNING, payload["old_task_state"])
self.assertEquals(None, payload["new_task_state"])
self.assertEquals(payload["access_ip_v4"], access_ip_v4)
self.assertEquals(payload["access_ip_v6"], access_ip_v6)
self.assertEquals(payload["display_name"], display_name)
def test_update_no_service_name(self):
notifications.send_update_with_states(self.context, self.instance,
@@ -202,7 +244,27 @@ class NotificationsTestCase(test.TestCase):
self.assertEquals('compute.someotherhost', notif['publisher_id'])
def test_payload_has_fixed_ip_labels(self):
usage = notifications.usage_from_instance(self.context, self.instance,
info = notifications.info_from_instance(self.context, self.instance,
self.net_info, None)
self.assertTrue("fixed_ips" in usage)
self.assertEquals(usage["fixed_ips"][0]["label"], "test1")
self.assertTrue("fixed_ips" in info)
self.assertEquals(info["fixed_ips"][0]["label"], "test1")
def test_send_access_ip_update(self):
notifications.send_update(self.context, self.instance, self.instance)
self.assertEquals(1, len(test_notifier.NOTIFICATIONS))
notif = test_notifier.NOTIFICATIONS[0]
payload = notif["payload"]
access_ip_v4 = self.instance["access_ip_v4"]
access_ip_v6 = self.instance["access_ip_v6"]
self.assertEquals(payload["access_ip_v4"], access_ip_v4)
self.assertEquals(payload["access_ip_v6"], access_ip_v6)
def test_send_name_update(self):
notifications.send_update(self.context, self.instance, self.instance)
self.assertEquals(1, len(test_notifier.NOTIFICATIONS))
notif = test_notifier.NOTIFICATIONS[0]
payload = notif["payload"]
display_name = self.instance["display_name"]
self.assertEquals(payload["display_name"], display_name)