Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Property collector cache #132

Open
wants to merge 9 commits into
base: stable/queens-m3
Choose a base branch
from
Open

Property collector cache #132

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
b73edea
Add upstream propertyCollector cache
mmidolesov Oct 22, 2019
b008683
Change loop for check events in VC
mmidolesov Nov 13, 2019
e53fac8
Remove LOG update.changeSet
mmidolesov Nov 13, 2019
a1c8e03
Add try catch to property collector thread
mmidolesov Nov 14, 2019
0e387f3
Remove polling task for property collector
mmidolesov Jan 14, 2020
d52fc26
Fix index out of range in updateSet inside handleModifyEvent
mmidolesov Jan 23, 2020
97c35fd
Remove log for updateSet because logs are too big
mmidolesov Jan 23, 2020
be9bd0b
Check if the second element in update set has 'val' attr
mmidolesov Jan 24, 2020
5cf33bb
Remove runtime.powerState from power_on/off_instance methods in vm_util
mmidolesov Jan 31, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 4 additions & 1 deletion nova/virt/vmwareapi/vm_util.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1661,6 +1661,8 @@ def power_on_instance(session, instance, vm_ref=None):

if vm_ref is None:
vm_ref = get_vm_ref(session, instance)
if vm_ref.value in _VM_VALUE_CACHE:
_VM_VALUE_CACHE[vm_ref.value].pop("runtime.powerState", None)

LOG.debug("Powering on the VM", instance=instance)
try:
Expand Down Expand Up @@ -1721,7 +1723,8 @@ def power_off_instance(session, instance, vm_ref=None):

if vm_ref is None:
vm_ref = get_vm_ref(session, instance)

if vm_ref.value in _VM_VALUE_CACHE:
_VM_VALUE_CACHE[vm_ref.value].pop("runtime.powerState", None)
LOG.debug("Powering off the VM", instance=instance)
try:
poweroff_task = session._call_method(session.vim,
Expand Down
155 changes: 82 additions & 73 deletions nova/virt/vmwareapi/vmops.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,9 @@
import six
import time

from eventlet import spawn as spawn_background_task
from oslo_concurrency.lockutils import synchronized

import cluster_util
import decorator
from oslo_concurrency import lockutils
Expand Down Expand Up @@ -154,6 +157,7 @@ def __init__(self, session, virtapi, volumeops, cluster=None,
self._imagecache = imagecache.ImageCacheManager(self._session,
self._base_folder)
self._network_api = network.API()
spawn_background_task(self.update_cached_instances)

def _get_base_folder(self):
# Enable more than one compute node to run on the same host
Expand Down Expand Up @@ -1758,7 +1762,6 @@ def power_off(self, instance, timeout=0, retry_interval=0):
return

vm_util.power_off_instance(self._session, instance)
self.update_cached_instances()

def _clean_shutdown(self, instance, timeout, retry_interval):
"""Perform a soft shutdown on the VM.
Expand Down Expand Up @@ -1817,7 +1820,6 @@ def _get_instance_props(self, vm_ref):
"summary.guest.toolsRunningStatus",
]

self.update_cached_instances()
vm_props = vm_util._VM_VALUE_CACHE.get(vm_ref.value, {})

if set(vm_props.keys()).issuperset(lst_properties):
Expand All @@ -1829,7 +1831,6 @@ def _get_instance_props(self, vm_ref):

def power_on(self, instance):
vm_util.power_on_instance(self._session, instance)
self.update_cached_instances()

def _update_instance_progress(self, context, instance, step, total_steps):
"""Update instance progress percent to reflect current step number
Expand Down Expand Up @@ -2078,9 +2079,6 @@ def get_info(self, instance):
"""Return data about the VM instance."""
lst_properties = ["runtime.powerState"]

if not vm_util._VM_VALUE_CACHE:
self.update_cached_instances()

vm_ref = vm_util.get_vm_ref(self._session, instance)
vm_props = vm_util._VM_VALUE_CACHE.get(vm_ref.value, {})
if not vm_props or "runtime.powerState" not in vm_props:
Expand Down Expand Up @@ -2534,7 +2532,6 @@ def list_instances(self):
if not CONF.vmware.use_property_collector:
lst_vm_names = self._list_instances_in_cluster()
else:
self.update_cached_instances()
lst_vm_names = [item["config.instanceUuid"]
for item in vm_util._VM_VALUE_CACHE.values()
if "config.instanceUuid" in item and item.get(
Expand Down Expand Up @@ -2609,76 +2606,93 @@ def _parse_change_set(change_set):
else:
yield change.name, val

@utils.synchronized("vmware.update_cache")
@synchronized('update_cache')
def update_cached_instances(self):
if not CONF.vmware.use_property_collector:
return

vim = self._session.vim
options = self._session.vim.client.factory.create("ns0:WaitOptions")
options.maxWaitSeconds = 0

options = vim.client.factory.create('ns0:WaitOptions')
options.maxWaitSeconds = 30
version = ''
if self._property_collector is None:
pc = vim.service_content.propertyCollector
self._property_collector = self._session._call_method(
self._session.vim,
"CreatePropertyCollector", pc)
self._property_collector = vim.service_content.propertyCollector
vim.CreateFilter(self._property_collector,
spec=self._get_vm_monitor_spec(vim),
partialUpdates=False)

update_set = vim.WaitForUpdatesEx(self._property_collector,
version=self._property_collector_version,
options=options)

while update_set:
self._property_collector_version = update_set.version
if update_set.filterSet and update_set.filterSet[0].objectSet:
for update in update_set.filterSet[0].objectSet:
vm_ref = update.obj
if vm_ref["_type"] != "VirtualMachine":
continue
values = vm_util._VM_VALUE_CACHE[vm_ref.value]
spec=self._get_vm_monitor_spec(vim),
partialUpdates=False)

if update.kind == "leave":
instance_uuid = values.get("config.instanceUuid")
vm_util.vm_ref_cache_delete(instance_uuid)
vm_util.vm_value_cache_delete(vm_ref)
LOG.debug("Removed instance %s (%s) from cache...",
instance_uuid, vm_ref.value)
else:
changes = dict(self._parse_change_set(
update.changeSet))
LOG.debug("%s.%s.%s: %s", vm_ref["_type"], update.kind,
vm_ref.value, changes)
if not (changes.get("config.managedBy") or
values.get("config.managedBy")):
LOG.debug("%s Not managed by nova", vm_ref.value)
continue
while True:
try:
update_set = vim.WaitForUpdatesEx(self._property_collector,
version=version,
options=options)
if update_set is None:
continue

instance_uuid = changes.get("config.instanceUuid")
self.process_update_set(update_set)
version = update_set.version

if update.kind == "enter":
vm_util.vm_ref_cache_update(instance_uuid, vm_ref)
elif update.kind == "modify":
old_instance_uuid = values.get(
"config.instanceUuid")
new_instance_uuid = changes.get(
"config.instanceUuid")
except Exception as e:
LOG.error(e)

if old_instance_uuid != new_instance_uuid:
vm_util.vm_ref_cache_delete(
old_instance_uuid)
vm_util.vm_ref_cache_update(
instance_uuid, vm_ref)

values.update(changes)
LOG.debug("%s.%s.%s -> %s", vm_ref["_type"],
update.kind, vm_ref.value, values)

update_set = vim.WaitForUpdatesEx(self._property_collector,
version=self._property_collector_version,
options=options)
def process_update_set(self, update_set):
if update_set.filterSet and update_set.filterSet[0].objectSet:
for update in update_set.filterSet[0].objectSet:
if update.obj['_type'] == "VirtualMachine":
if update.kind == "leave":
self.handle_leave_event(update)
else:
if update.kind == "enter":
self.handle_enter_event(update)
elif update.kind == "modify":
self.handle_modify_event(update)

def set_val_cache(self, key, name, val):
vm_util._VM_VALUE_CACHE[key][name] = val

def handle_leave_event(self, update):
LOG.info("Removing instance from cache...")
cache_id_to_delete = vm_util._VM_VALUE_CACHE.get(
update.obj.value, {}).get('config.instanceUuid')
vm_util.vm_ref_cache_delete(cache_id_to_delete)
vm_util.vm_value_cache_delete(update.obj.value)

def handle_enter_event(self, update):
for change in update.changeSet:
if change.op == 'assign':
if hasattr(update.changeSet[0], 'val') and \
hasattr(update.changeSet[1], 'val'):
self.set_val_cache(update.obj.value,
change.name,
change.val)
if hasattr(update.changeSet[0], 'val') and \
hasattr(update.changeSet[1], 'val'):
if update.changeSet[1].val.extensionKey == constants.EXTENSION_KEY:
LOG.info("Adding vm to cache...")
vm_util.vm_ref_cache_update(update.changeSet[0].val,
update.obj)

def handle_modify_event(self, update):
if update.changeSet[0].name == "config.instanceUuid":
for change in update.changeSet:
LOG.debug('Value found: %s' % change)

if change['op'] == 'assign':
self.set_val_cache(update.obj.value,
change.name,
change.val)
if len(update.changeSet) > 1:
if hasattr(update.changeSet[0], 'val') and \
hasattr(update.changeSet[1], 'val'):
if update.changeSet[1].val.extensionKey ==\
constants.EXTENSION_KEY:
vm_util.vm_ref_cache_update(update.changeSet[0].val,
update.obj)
else:
for change in update.changeSet:
if change['op'] == 'assign':
self.set_val_cache(update.obj.value,
change.name,
change.val)

def _get_vm_monitor_spec(self, vim):
traversal_spec_vm = vutil.build_traversal_spec(
Expand Down Expand Up @@ -2706,12 +2720,7 @@ def _get_vm_monitor_spec(self, vim):
property_spec_vm = vutil.build_property_spec(
vim.client.factory,
"VirtualMachine",
["config.instanceUuid",
"config.managedBy",
"runtime.powerState",
"summary.guest.toolsStatus",
"summary.guest.toolsRunningStatus",
])
["config.instanceUuid", "config.managedBy", "runtime.powerState"])
property_filter_spec = vutil.build_property_filter_spec(
vim.client.factory,
[property_spec, property_spec_vm],
Expand Down