Arnike/watcher
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge "Fixed Tempest test due to notification issues"

Browse Source
This commit is contained in:
Jenkins
2016-09-15 09:55:33 +00:00
committed by Gerrit Code Review
parent 0873c26b17 72e6564549
commit 0a0f482f2d
27 changed files with 965 additions and 514 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
watcher/decision_engine/model/collector/nova.py
View File

@@ -83,7 +83,7 @@ class NovaClusterDataModelCollector(base.BaseClusterDataModelCollector):
for n in nodes:
service = self.wrapper.nova.services.find(id=n.service['id'])
# create node in cluster_model_collector
node = element.ComputeNode()
node = element.ComputeNode(n.id)
node.uuid = service.host
node.hostname = n.hypervisor_hostname
# set capacity
@@ -105,7 +105,10 @@ class NovaClusterDataModelCollector(base.BaseClusterDataModelCollector):
# set capacity
self.wrapper.get_flavor_instance(v, flavor_cache)
mem.set_capacity(instance, v.flavor['ram'])
# FIXME: update all strategies to use disk_capacity
# for instances instead of disk
disk.set_capacity(instance, v.flavor['disk'])
disk_capacity.set_capacity(instance, v.flavor['disk'])
num_cores.set_capacity(instance, v.flavor['vcpus'])
model.map_instance(instance, node)

3
watcher/decision_engine/model/element/node.py
View File

@@ -28,8 +28,9 @@ class ServiceState(enum.Enum):
class ComputeNode(compute_resource.ComputeResource):
def __init__(self):
def __init__(self, id):
super(ComputeNode, self).__init__()
self.id = id
self._state = ServiceState.ONLINE.value
self._status = ServiceState.ENABLED.value

14
watcher/decision_engine/model/element/resource.py
View File

@@ -16,6 +16,8 @@
import enum
from watcher.common import exception
class ResourceType(enum.Enum):
cpu_cores = 'num_cores'
@@ -50,12 +52,12 @@ class Resource(object):
def unset_capacity(self, element):
del self.mapping[element.uuid]
def get_capacity_from_id(self, uuid):
if str(uuid) in self.mapping.keys():
def get_capacity_by_uuid(self, uuid):
try:
return self.mapping[str(uuid)]
else:
# TODO(jed) throw exception
return None
except KeyError:
raise exception.CapacityNotDefined(
capacity=self.name.value, resource=str(uuid))
def get_capacity(self, element):
return self.get_capacity_from_id(element.uuid)
return self.get_capacity_by_uuid(element.uuid)

14
watcher/decision_engine/model/mapping.py
View File

@@ -58,9 +58,9 @@ class Mapping(object):
:param node: the node
:param instance: the virtual machine or instance
"""
self.unmap_from_id(node.uuid, instance.uuid)
self.unmap_by_uuid(node.uuid, instance.uuid)
def unmap_from_id(self, node_uuid, instance_uuid):
def unmap_by_uuid(self, node_uuid, instance_uuid):
"""Remove the instance (by id) from the node (by id)
:rtype : object
@@ -84,15 +84,15 @@ class Mapping(object):
return self.compute_node_mapping
def get_node_from_instance(self, instance):
return self.get_node_from_instance_id(instance.uuid)
return self.get_node_by_instance_uuid(instance.uuid)
def get_node_from_instance_id(self, instance_uuid):
def get_node_by_instance_uuid(self, instance_uuid):
"""Getting host information from the guest instance
:param instance: the uuid of the instance
:return: node
"""
return self.model.get_node_from_id(
return self.model.get_node_by_uuid(
self.instance_mapping[str(instance_uuid)])
def get_node_instances(self, node):
@@ -101,9 +101,9 @@ class Mapping(object):
:param node:
:return:
"""
return self.get_node_instances_from_id(node.uuid)
return self.get_node_instances_by_uuid(node.uuid)
def get_node_instances_from_id(self, node_uuid):
def get_node_instances_by_uuid(self, node_uuid):
if str(node_uuid) in self.compute_node_mapping.keys():
return self.compute_node_mapping[str(node_uuid)]
else:

41
watcher/decision_engine/model/model_root.py
View File

@@ -77,9 +77,9 @@ class ModelRoot(object):
:type node: str or :py:class:`~.Instance`
"""
if isinstance(instance, six.string_types):
instance = self.get_instance_from_id(instance)
instance = self.get_instance_by_uuid(instance)
if isinstance(node, six.string_types):
node = self.get_node_from_id(node)
node = self.get_node_by_uuid(node)
self.add_instance(instance)
self.mapping.map(node, instance)
@@ -93,17 +93,18 @@ class ModelRoot(object):
:type node: str or :py:class:`~.Instance`
"""
if isinstance(instance, six.string_types):
instance = self.get_instance_from_id(instance)
instance = self.get_instance_by_uuid(instance)
if isinstance(node, six.string_types):
node = self.get_node_from_id(node)
node = self.get_node_by_uuid(node)
self.add_instance(instance)
self.mapping.unmap(node, instance)
def delete_instance(self, instance, node):
self.remove_instance(instance)
def delete_instance(self, instance, node=None):
if node is not None:
self.mapping.unmap(node, instance)
self.mapping.unmap(node, instance)
self.remove_instance(instance)
for resource in self.resource.values():
try:
@@ -130,17 +131,17 @@ class ModelRoot(object):
def get_all_compute_nodes(self):
return self._nodes
def get_node_from_id(self, node_uuid):
def get_node_by_uuid(self, node_uuid):
if str(node_uuid) not in self._nodes:
raise exception.ComputeNodeNotFound(name=node_uuid)
return self._nodes[str(node_uuid)]
def get_instance_from_id(self, uuid):
def get_instance_by_uuid(self, uuid):
if str(uuid) not in self._instances:
raise exception.InstanceNotFound(name=uuid)
return self._instances[str(uuid)]
def get_node_from_instance_id(self, instance_uuid):
def get_node_by_instance_uuid(self, instance_uuid):
"""Getting host information from the guest instance
:param instance_uuid: the uuid of the instance
@@ -148,7 +149,7 @@ class ModelRoot(object):
"""
if str(instance_uuid) not in self.mapping.instance_mapping:
raise exception.InstanceNotFound(name=instance_uuid)
return self.get_node_from_id(
return self.get_node_by_uuid(
self.mapping.instance_mapping[str(instance_uuid)])
def get_all_instances(self):
@@ -160,7 +161,7 @@ class ModelRoot(object):
def create_resource(self, r):
self.resource[str(r.name)] = r
def get_resource_from_id(self, resource_id):
def get_resource_by_uuid(self, resource_id):
return self.resource[str(resource_id)]
def get_node_instances(self, node):
@@ -168,11 +169,12 @@ class ModelRoot(object):
def _build_compute_node_element(self, compute_node):
attrib = collections.OrderedDict(
uuid=compute_node.uuid, human_id=compute_node.human_id,
hostname=compute_node.hostname, state=compute_node.state,
status=compute_node.status)
id=six.text_type(compute_node.id), uuid=compute_node.uuid,
human_id=compute_node.human_id, hostname=compute_node.hostname,
state=compute_node.state, status=compute_node.status)
for resource_name, resource in self.resource.items():
for resource_name, resource in sorted(
self.resource.items(), key=lambda x: x[0]):
res_value = resource.get_capacity(compute_node)
if res_value is not None:
attrib[resource_name] = six.text_type(res_value)
@@ -186,7 +188,8 @@ class ModelRoot(object):
uuid=instance.uuid, human_id=instance.human_id,
hostname=instance.hostname, state=instance.state)
for resource_name, resource in self.resource.items():
for resource_name, resource in sorted(
self.resource.items(), key=lambda x: x[0]):
res_value = resource.get_capacity(instance)
if res_value is not None:
attrib[resource_name] = six.text_type(res_value)
@@ -205,7 +208,7 @@ class ModelRoot(object):
# Build mapped instance tree
node_instance_uuids = self.get_node_instances(cn)
for instance_uuid in sorted(node_instance_uuids):
instance = self.get_instance_from_id(instance_uuid)
instance = self.get_instance_by_uuid(instance_uuid)
instance_el = self._build_instance_element(instance)
compute_node_el.append(instance_el)
@@ -215,7 +218,7 @@ class ModelRoot(object):
for instance in sorted(self.get_all_instances().values(),
key=lambda inst: inst.uuid):
try:
self.get_node_from_instance_id(instance.uuid)
self.get_node_by_instance_uuid(instance.uuid)
except exception.InstanceNotFound:
root.append(self._build_instance_element(instance))

178
watcher/decision_engine/model/notification/nova.py
View File

@@ -20,6 +20,7 @@ from oslo_log import log
from watcher._i18n import _LI
from watcher.common import exception
from watcher.common import nova_helper
from watcher.decision_engine.model import element
from watcher.decision_engine.model.notification import base
from watcher.decision_engine.model.notification import filtering
@@ -29,9 +30,19 @@ LOG = log.getLogger(__name__)
class NovaNotification(base.NotificationEndpoint):
def __init__(self, collector):
super(NovaNotification, self).__init__(collector)
self._nova = None
@property
def nova(self):
if self._nova is None:
self._nova = nova_helper.NovaHelper()
return self._nova
def get_or_create_instance(self, uuid):
try:
instance = self.cluster_data_model.get_instance_from_id(uuid)
instance = self.cluster_data_model.get_instance_by_uuid(uuid)
except exception.InstanceNotFound:
# The instance didn't exist yet so we create a new instance object
LOG.debug("New instance created: %s", uuid)
@@ -59,13 +70,20 @@ class NovaNotification(base.NotificationEndpoint):
element.ResourceType.cpu_cores, instance, num_cores)
self.update_capacity(
element.ResourceType.disk, instance, disk_gb)
self.update_capacity(
element.ResourceType.disk_capacity, instance, disk_gb)
node = self.get_or_create_node(instance_data['host'])
try:
node = self.get_or_create_node(instance_data['host'])
except exception.ComputeNodeNotFound as exc:
LOG.exception(exc)
# If we can't create the node, we consider the instance as unmapped
node = None
self.update_instance_mapping(instance, node)
def update_capacity(self, resource_id, obj, value):
resource = self.cluster_data_model.get_resource_from_id(resource_id)
resource = self.cluster_data_model.get_resource_by_uuid(resource_id)
resource.set_capacity(obj, value)
def legacy_update_instance(self, instance, data):
@@ -82,34 +100,83 @@ class NovaNotification(base.NotificationEndpoint):
element.ResourceType.cpu_cores, instance, num_cores)
self.update_capacity(
element.ResourceType.disk, instance, disk_gb)
self.update_capacity(
element.ResourceType.disk_capacity, instance, disk_gb)
node = self.get_or_create_node(data['host'])
try:
node = self.get_or_create_node(data['host'])
except exception.ComputeNodeNotFound as exc:
LOG.exception(exc)
# If we can't create the node, we consider the instance as unmapped
node = None
self.update_instance_mapping(instance, node)
def update_compute_node(self, node, data):
"""Update the compute node using the notification data."""
node_data = data['nova_object.data']
node.hostname = node_data['host']
node.state = (
element.ServiceState.OFFLINE.value
if node_data['forced_down'] else element.ServiceState.ONLINE.value)
node.status = (
element.ServiceState.DISABLED.value
if node_data['host'] else element.ServiceState.ENABLED.value)
def create_compute_node(self, node_hostname):
"""Update the compute node by querying the Nova API."""
try:
_node = self.nova.get_compute_node_by_hostname(node_hostname)
node = element.ComputeNode(_node.id)
node.uuid = node_hostname
node.hostname = _node.hypervisor_hostname
node.state = _node.state
node.status = _node.status
self.update_capacity(
element.ResourceType.memory, node, _node.memory_mb)
self.update_capacity(
element.ResourceType.cpu_cores, node, _node.vcpus)
self.update_capacity(
element.ResourceType.disk, node, _node.free_disk_gb)
self.update_capacity(
element.ResourceType.disk_capacity, node, _node.local_gb)
return node
except Exception as exc:
LOG.exception(exc)
LOG.debug("Could not refresh the node %s.", node_hostname)
raise exception.ComputeNodeNotFound(name=node_hostname)
return False
def get_or_create_node(self, uuid):
if uuid is None:
LOG.debug("Compute node UUID not provided: skipping")
return
try:
node = self.cluster_data_model.get_node_from_id(uuid)
return self.cluster_data_model.get_node_by_uuid(uuid)
except exception.ComputeNodeNotFound:
# The node didn't exist yet so we create a new node object
node = self.create_compute_node(uuid)
LOG.debug("New compute node created: %s", uuid)
node = element.ComputeNode()
node.uuid = uuid
self.cluster_data_model.add_node(node)
return node
return node
def update_instance_mapping(self, instance, node):
if not node:
if node is None:
self.cluster_data_model.add_instance(instance)
LOG.debug("Instance %s not yet attached to any node: skipping",
instance.uuid)
return
try:
old_node = self.get_or_create_node(node.uuid)
try:
old_node = self.get_or_create_node(node.uuid)
except exception.ComputeNodeNotFound as exc:
LOG.exception(exc)
# If we can't create the node,
# we consider the instance as unmapped
old_node = None
LOG.debug("Mapped node %s found", node.uuid)
if node and node != old_node:
LOG.debug("Unmapping instance %s from %s",
@@ -126,8 +193,7 @@ class NovaNotification(base.NotificationEndpoint):
def delete_instance(self, instance, node):
try:
self.cluster_data_model.delete_instance(instance, node)
except Exception as exc:
LOG.exception(exc)
except Exception:
LOG.info(_LI("Instance %s already deleted"), instance.uuid)
@@ -150,19 +216,18 @@ class ServiceUpdated(VersionnedNotificationEndpoint):
)
def info(self, ctxt, publisher_id, event_type, payload, metadata):
LOG.info(_LI("Event '%(event)s' received from %(publisher)s") %
dict(event=event_type, publisher=publisher_id))
LOG.info(_LI("Event '%(event)s' received from %(publisher)s "
"with metadata %(metadata)s") %
dict(event=event_type,
publisher=publisher_id,
metadata=metadata))
node_data = payload['nova_object.data']
node_uuid = node_data['host']
node = self.get_or_create_node(node_uuid)
node.hostname = node_data['host']
node.state = (
element.ServiceState.OFFLINE.value
if node_data['forced_down'] else element.ServiceState.ONLINE.value)
node.status = (
element.ServiceState.DISABLED.value
if node_data['host'] else element.ServiceState.ENABLED.value)
try:
node = self.get_or_create_node(node_uuid)
self.update_compute_node(node, payload)
except exception.ComputeNodeNotFound as exc:
LOG.exception(exc)
class InstanceCreated(VersionnedNotificationEndpoint):
@@ -192,8 +257,11 @@ class InstanceCreated(VersionnedNotificationEndpoint):
)
def info(self, ctxt, publisher_id, event_type, payload, metadata):
LOG.info(_LI("Event '%(event)s' received from %(publisher)s") %
dict(event=event_type, publisher=publisher_id))
LOG.info(_LI("Event '%(event)s' received from %(publisher)s "
"with metadata %(metadata)s") %
dict(event=event_type,
publisher=publisher_id,
metadata=metadata))
instance_data = payload['nova_object.data']
instance_uuid = instance_data['uuid']
@@ -221,8 +289,11 @@ class InstanceUpdated(VersionnedNotificationEndpoint):
)
def info(self, ctxt, publisher_id, event_type, payload, metadata):
LOG.info(_LI("Event '%(event)s' received from %(publisher)s") %
dict(event=event_type, publisher=publisher_id))
LOG.info(_LI("Event '%(event)s' received from %(publisher)s "
"with metadata %(metadata)s") %
dict(event=event_type,
publisher=publisher_id,
metadata=metadata))
instance_data = payload['nova_object.data']
instance_uuid = instance_data['uuid']
instance = self.get_or_create_instance(instance_uuid)
@@ -241,14 +312,22 @@ class InstanceDeletedEnd(VersionnedNotificationEndpoint):
)
def info(self, ctxt, publisher_id, event_type, payload, metadata):
LOG.info(_LI("Event '%(event)s' received from %(publisher)s") %
dict(event=event_type, publisher=publisher_id))
LOG.info(_LI("Event '%(event)s' received from %(publisher)s "
"with metadata %(metadata)s") %
dict(event=event_type,
publisher=publisher_id,
metadata=metadata))
instance_data = payload['nova_object.data']
instance_uuid = instance_data['uuid']
instance = self.get_or_create_instance(instance_uuid)
node = self.get_or_create_node(instance_data['host'])
try:
node = self.get_or_create_node(instance_data['host'])
except exception.ComputeNodeNotFound as exc:
LOG.exception(exc)
# If we can't create the node, we consider the instance as unmapped
node = None
self.delete_instance(instance, node)
@@ -264,8 +343,11 @@ class LegacyInstanceUpdated(UnversionnedNotificationEndpoint):
)
def info(self, ctxt, publisher_id, event_type, payload, metadata):
LOG.info(_LI("Event '%(event)s' received from %(publisher)s") %
dict(event=event_type, publisher=publisher_id))
LOG.info(_LI("Event '%(event)s' received from %(publisher)s "
"with metadata %(metadata)s") %
dict(event=event_type,
publisher=publisher_id,
metadata=metadata))
instance_uuid = payload['instance_id']
instance = self.get_or_create_instance(instance_uuid)
@@ -284,8 +366,11 @@ class LegacyInstanceCreatedEnd(UnversionnedNotificationEndpoint):
)
def info(self, ctxt, publisher_id, event_type, payload, metadata):
LOG.info(_LI("Event '%(event)s' received from %(publisher)s") %
dict(event=event_type, publisher=publisher_id))
LOG.info(_LI("Event '%(event)s' received from %(publisher)s "
"with metadata %(metadata)s") %
dict(event=event_type,
publisher=publisher_id,
metadata=metadata))
instance_uuid = payload['instance_id']
instance = self.get_or_create_instance(instance_uuid)
@@ -304,12 +389,20 @@ class LegacyInstanceDeletedEnd(UnversionnedNotificationEndpoint):
)
def info(self, ctxt, publisher_id, event_type, payload, metadata):
LOG.info(_LI("Event '%(event)s' received from %(publisher)s") %
dict(event=event_type, publisher=publisher_id))
LOG.info(_LI("Event '%(event)s' received from %(publisher)s "
"with metadata %(metadata)s") %
dict(event=event_type,
publisher=publisher_id,
metadata=metadata))
instance_uuid = payload['instance_id']
instance = self.get_or_create_instance(instance_uuid)
node = self.get_or_create_node(payload['host'])
try:
node = self.get_or_create_node(payload['host'])
except exception.ComputeNodeNotFound as exc:
LOG.exception(exc)
# If we can't create the node, we consider the instance as unmapped
node = None
self.delete_instance(instance, node)
@@ -325,8 +418,11 @@ class LegacyLiveMigratedEnd(UnversionnedNotificationEndpoint):
)
def info(self, ctxt, publisher_id, event_type, payload, metadata):
LOG.info(_LI("Event '%(event)s' received from %(publisher)s") %
dict(event=event_type, publisher=publisher_id))
LOG.info(_LI("Event '%(event)s' received from %(publisher)s "
"with metadata %(metadata)s") %
dict(event=event_type,
publisher=publisher_id,
metadata=metadata))
instance_uuid = payload['instance_id']
instance = self.get_or_create_instance(instance_uuid)

52
watcher/decision_engine/strategy/strategies/basic_consolidation.py
View File

@@ -152,16 +152,16 @@ class BasicConsolidation(base.ServerConsolidationBaseStrategy):
total_cores = 0
total_disk = 0
total_mem = 0
cpu_capacity = self.compute_model.get_resource_from_id(
cpu_capacity = self.compute_model.get_resource_by_uuid(
element.ResourceType.cpu_cores)
disk_capacity = self.compute_model.get_resource_from_id(
disk_capacity = self.compute_model.get_resource_by_uuid(
element.ResourceType.disk)
memory_capacity = self.compute_model.get_resource_from_id(
memory_capacity = self.compute_model.get_resource_by_uuid(
element.ResourceType.memory)
for instance_id in self.compute_model. \
get_mapping().get_node_instances(destination_node):
instance = self.compute_model.get_instance_from_id(instance_id)
for instance_id in self.compute_model.mapping.get_node_instances(
destination_node):
instance = self.compute_model.get_instance_by_uuid(instance_id)
total_cores += cpu_capacity.get_capacity(instance)
total_disk += disk_capacity.get_capacity(instance)
total_mem += memory_capacity.get_capacity(instance)
@@ -188,11 +188,11 @@ class BasicConsolidation(base.ServerConsolidationBaseStrategy):
:param total_mem: total memory used by the virtual machine
:return: True if the threshold is not exceed
"""
cpu_capacity = self.compute_model.get_resource_from_id(
cpu_capacity = self.compute_model.get_resource_by_uuid(
element.ResourceType.cpu_cores).get_capacity(destination_node)
disk_capacity = self.compute_model.get_resource_from_id(
disk_capacity = self.compute_model.get_resource_by_uuid(
element.ResourceType.disk).get_capacity(destination_node)
memory_capacity = self.compute_model.get_resource_from_id(
memory_capacity = self.compute_model.get_resource_by_uuid(
element.ResourceType.memory).get_capacity(destination_node)
return (cpu_capacity >= total_cores * self.threshold_cores and
@@ -219,13 +219,13 @@ class BasicConsolidation(base.ServerConsolidationBaseStrategy):
:param total_memory_used:
:return:
"""
cpu_capacity = self.compute_model.get_resource_from_id(
cpu_capacity = self.compute_model.get_resource_by_uuid(
element.ResourceType.cpu_cores).get_capacity(compute_resource)
disk_capacity = self.compute_model.get_resource_from_id(
disk_capacity = self.compute_model.get_resource_by_uuid(
element.ResourceType.disk).get_capacity(compute_resource)
memory_capacity = self.compute_model.get_resource_from_id(
memory_capacity = self.compute_model.get_resource_by_uuid(
element.ResourceType.memory).get_capacity(compute_resource)
score_cores = (1 - (float(cpu_capacity) - float(total_cores_used)) /
@@ -266,7 +266,7 @@ class BasicConsolidation(base.ServerConsolidationBaseStrategy):
metric_name=self.HOST_CPU_USAGE_METRIC_NAME))
host_avg_cpu_util = 100
cpu_capacity = self.compute_model.get_resource_from_id(
cpu_capacity = self.compute_model.get_resource_by_uuid(
element.ResourceType.cpu_cores).get_capacity(node)
total_cores_used = cpu_capacity * (host_avg_cpu_util / 100.0)
@@ -306,7 +306,7 @@ class BasicConsolidation(base.ServerConsolidationBaseStrategy):
metric_name=self.INSTANCE_CPU_USAGE_METRIC_NAME))
instance_cpu_utilization = 100
cpu_capacity = self.compute_model.get_resource_from_id(
cpu_capacity = self.compute_model.get_resource_by_uuid(
element.ResourceType.cpu_cores).get_capacity(instance)
total_cores_used = cpu_capacity * (instance_cpu_utilization / 100.0)
@@ -334,8 +334,7 @@ class BasicConsolidation(base.ServerConsolidationBaseStrategy):
def score_of_nodes(self, score):
"""Calculate score of nodes based on load by VMs"""
for node in self.compute_model.get_all_compute_nodes().values():
count = self.compute_model.mapping.get_node_instances_from_id(
node.uuid)
count = self.compute_model.mapping.get_node_instances(node)
if len(count) > 0:
result = self.calculate_score_node(node)
else:
@@ -348,13 +347,12 @@ class BasicConsolidation(base.ServerConsolidationBaseStrategy):
def node_and_instance_score(self, sorted_score, score):
"""Get List of VMs from node"""
node_to_release = sorted_score[len(score) - 1][0]
instances_to_migrate = (
self.compute_model.mapping.get_node_instances_from_id(
node_to_release))
instances_to_migrate = self.compute_model.mapping.get_node_instances(
self.compute_model.get_node_by_uuid(node_to_release))
instance_score = []
for instance_id in instances_to_migrate:
instance = self.compute_model.get_instance_from_id(instance_id)
instance = self.compute_model.get_instance_by_uuid(instance_id)
if instance.state == element.InstanceState.ACTIVE.value:
instance_score.append(
(instance_id, self.calculate_score_instance(instance)))
@@ -370,7 +368,7 @@ class BasicConsolidation(base.ServerConsolidationBaseStrategy):
mig_source_node.uuid,
mig_destination_node.uuid)
if len(self.compute_model.get_mapping().get_node_instances(
if len(self.compute_model.mapping.get_node_instances(
mig_source_node)) == 0:
self.add_change_service_state(mig_source_node.
uuid,
@@ -382,11 +380,11 @@ class BasicConsolidation(base.ServerConsolidationBaseStrategy):
number_migrations = 0
for instance in sorted_instances:
for j in range(0, len(sorted_score)):
mig_instance = self.compute_model.get_instance_from_id(
mig_instance = self.compute_model.get_instance_by_uuid(
instance[0])
mig_source_node = self.compute_model.get_node_from_id(
mig_source_node = self.compute_model.get_node_by_uuid(
node_to_release)
mig_destination_node = self.compute_model.get_node_from_id(
mig_destination_node = self.compute_model.get_node_by_uuid(
sorted_score[j][0])
result = self.check_migration(
@@ -411,6 +409,8 @@ class BasicConsolidation(base.ServerConsolidationBaseStrategy):
if not self.compute_model:
raise exception.ClusterStateNotDefined()
LOG.debug(self.compute_model.to_string())
def do_execute(self):
# todo(jed) clone model
self.efficacy = 100
@@ -425,8 +425,8 @@ class BasicConsolidation(base.ServerConsolidationBaseStrategy):
for node_uuid, node in self.compute_model.get_all_compute_nodes(
).items():
node_instances = (self.compute_model.mapping
.get_node_instances_from_id(node_uuid))
node_instances = self.compute_model.mapping.get_node_instances(
node)
if node_instances:
if node.state == element.ServiceState.ENABLED:
self.add_change_service_state(

12
watcher/decision_engine/strategy/strategies/outlet_temp_control.py
View File

@@ -130,7 +130,7 @@ class OutletTempControl(base.ThermalOptimizationBaseStrategy):
disk_gb_used = 0
if len(instances) > 0:
for instance_id in instances:
instance = self.compute_model.get_instance_from_id(instance_id)
instance = self.compute_model.get_instance_by_uuid(instance_id)
vcpus_used += cpu_capacity.get_capacity(instance)
memory_mb_used += memory_capacity.get_capacity(instance)
disk_gb_used += disk_capacity.get_capacity(instance)
@@ -147,7 +147,7 @@ class OutletTempControl(base.ThermalOptimizationBaseStrategy):
hosts_need_release = []
hosts_target = []
for node_id in nodes:
node = self.compute_model.get_node_from_id(
node = self.compute_model.get_node_by_uuid(
node_id)
resource_id = node.uuid
@@ -180,7 +180,7 @@ class OutletTempControl(base.ThermalOptimizationBaseStrategy):
for instance_id in instances_of_src:
try:
# select the first active instance to migrate
instance = self.compute_model.get_instance_from_id(
instance = self.compute_model.get_instance_by_uuid(
instance_id)
if (instance.state !=
element.InstanceState.ACTIVE.value):
@@ -196,11 +196,11 @@ class OutletTempControl(base.ThermalOptimizationBaseStrategy):
def filter_dest_servers(self, hosts, instance_to_migrate):
"""Only return hosts with sufficient available resources"""
cpu_capacity = self.compute_model.get_resource_from_id(
cpu_capacity = self.compute_model.get_resource_by_uuid(
element.ResourceType.cpu_cores)
disk_capacity = self.compute_model.get_resource_from_id(
disk_capacity = self.compute_model.get_resource_by_uuid(
element.ResourceType.disk)
memory_capacity = self.compute_model.get_resource_from_id(
memory_capacity = self.compute_model.get_resource_by_uuid(
element.ResourceType.memory)
required_cores = cpu_capacity.get_capacity(instance_to_migrate)

14
watcher/decision_engine/strategy/strategies/uniform_airflow.py
View File

@@ -144,7 +144,7 @@ class UniformAirflow(base.BaseStrategy):
memory_mb_used = 0
disk_gb_used = 0
for instance_id in instances:
instance = self.compute_model.get_instance_from_id(
instance = self.compute_model.get_instance_by_uuid(
instance_id)
vcpus_used += cap_cores.get_capacity(instance)
memory_mb_used += cap_mem.get_capacity(instance)
@@ -179,7 +179,7 @@ class UniformAirflow(base.BaseStrategy):
for instance_id in source_instances:
try:
instance = (self.compute_model.
get_instance_from_id(instance_id))
get_instance_by_uuid(instance_id))
instances_tobe_migrate.append(instance)
except wexc.InstanceNotFound:
LOG.error(_LE("Instance not found; error: %s"),
@@ -190,7 +190,7 @@ class UniformAirflow(base.BaseStrategy):
for instance_id in source_instances:
try:
instance = (self.compute_model.
get_instance_from_id(instance_id))
get_instance_by_uuid(instance_id))
if (instance.state !=
element.InstanceState.ACTIVE.value):
LOG.info(
@@ -209,11 +209,11 @@ class UniformAirflow(base.BaseStrategy):
def filter_destination_hosts(self, hosts, instances_to_migrate):
"""Find instance and host with sufficient available resources"""
cap_cores = self.compute_model.get_resource_from_id(
cap_cores = self.compute_model.get_resource_by_uuid(
element.ResourceType.cpu_cores)
cap_disk = self.compute_model.get_resource_from_id(
cap_disk = self.compute_model.get_resource_by_uuid(
element.ResourceType.disk)
cap_mem = self.compute_model.get_resource_from_id(
cap_mem = self.compute_model.get_resource_by_uuid(
element.ResourceType.memory)
# large instance go first
instances_to_migrate = sorted(
@@ -265,7 +265,7 @@ class UniformAirflow(base.BaseStrategy):
overload_hosts = []
nonoverload_hosts = []
for node_id in nodes:
node = self.compute_model.get_node_from_id(
node = self.compute_model.get_node_by_uuid(
node_id)
resource_id = node.uuid
airflow = self.ceilometer.statistic_aggregation(

14
watcher/decision_engine/strategy/strategies/vm_workload_consolidation.py
View File

@@ -162,7 +162,7 @@ class VMWorkloadConsolidation(base.ServerConsolidationBaseStrategy):
:param model: model_root object
:return: None
"""
instance = model.get_instance_from_id(instance_uuid)
instance = model.get_instance_by_uuid(instance_uuid)
instance_state_str = self.get_state_str(instance.state)
if instance_state_str != element.InstanceState.ACTIVE.value:
@@ -226,9 +226,9 @@ class VMWorkloadConsolidation(base.ServerConsolidationBaseStrategy):
instance_cpu_util = self.ceilometer.statistic_aggregation(
resource_id=instance_uuid, meter_name=cpu_util_metric,
period=period, aggregate=aggr)
instance_cpu_cores = model.get_resource_from_id(
instance_cpu_cores = model.get_resource_by_uuid(
element.ResourceType.cpu_cores).get_capacity(
model.get_instance_from_id(instance_uuid))
model.get_instance_by_uuid(instance_uuid))
if instance_cpu_util:
total_cpu_utilization = (
@@ -271,7 +271,7 @@ class VMWorkloadConsolidation(base.ServerConsolidationBaseStrategy):
:param aggr: string
:return: dict(cpu(number of cores used), ram(MB used), disk(B used))
"""
node_instances = model.mapping.get_node_instances_from_id(
node_instances = model.mapping.get_node_instances_by_uuid(
node.uuid)
node_ram_util = 0
node_disk_util = 0
@@ -293,13 +293,13 @@ class VMWorkloadConsolidation(base.ServerConsolidationBaseStrategy):
:param model: model_root object
:return: dict(cpu(cores), ram(MB), disk(B))
"""
node_cpu_capacity = model.get_resource_from_id(
node_cpu_capacity = model.get_resource_by_uuid(
element.ResourceType.cpu_cores).get_capacity(node)
node_disk_capacity = model.get_resource_from_id(
node_disk_capacity = model.get_resource_by_uuid(
element.ResourceType.disk_capacity).get_capacity(node)
node_ram_capacity = model.get_resource_from_id(
node_ram_capacity = model.get_resource_by_uuid(
element.ResourceType.memory).get_capacity(node)
return dict(cpu=node_cpu_capacity, ram=node_ram_capacity,
disk=node_disk_capacity)

18
watcher/decision_engine/strategy/strategies/workload_balance.py
View File

@@ -122,7 +122,7 @@ class WorkloadBalance(base.WorkloadStabilizationBaseStrategy):
memory_mb_used = 0
disk_gb_used = 0
for instance_id in instances:
instance = self.compute_model.get_instance_from_id(instance_id)
instance = self.compute_model.get_instance_by_uuid(instance_id)
vcpus_used += cap_cores.get_capacity(instance)
memory_mb_used += cap_mem.get_capacity(instance)
disk_gb_used += cap_disk.get_capacity(instance)
@@ -147,7 +147,7 @@ class WorkloadBalance(base.WorkloadStabilizationBaseStrategy):
for inst_id in source_instances:
try:
# select the first active VM to migrate
instance = self.compute_model.get_instance_from_id(
instance = self.compute_model.get_instance_by_uuid(
inst_id)
if (instance.state !=
element.InstanceState.ACTIVE.value):
@@ -164,7 +164,7 @@ class WorkloadBalance(base.WorkloadStabilizationBaseStrategy):
instance_id)
if instance_id:
return (source_node,
self.compute_model.get_instance_from_id(
self.compute_model.get_instance_by_uuid(
instance_id))
else:
LOG.info(_LI("VM not found from node: %s"),
@@ -174,11 +174,11 @@ class WorkloadBalance(base.WorkloadStabilizationBaseStrategy):
avg_workload, workload_cache):
'''Only return hosts with sufficient available resources'''
cap_cores = self.compute_model.get_resource_from_id(
cap_cores = self.compute_model.get_resource_by_uuid(
element.ResourceType.cpu_cores)
cap_disk = self.compute_model.get_resource_from_id(
cap_disk = self.compute_model.get_resource_by_uuid(
element.ResourceType.disk)
cap_mem = self.compute_model.get_resource_from_id(
cap_mem = self.compute_model.get_resource_by_uuid(
element.ResourceType.memory)
required_cores = cap_cores.get_capacity(instance_to_migrate)
@@ -222,7 +222,7 @@ class WorkloadBalance(base.WorkloadStabilizationBaseStrategy):
if not nodes:
raise wexc.ClusterEmpty()
# get cpu cores capacity of nodes and instances
cap_cores = self.compute_model.get_resource_from_id(
cap_cores = self.compute_model.get_resource_by_uuid(
element.ResourceType.cpu_cores)
overload_hosts = []
nonoverload_hosts = []
@@ -232,12 +232,12 @@ class WorkloadBalance(base.WorkloadStabilizationBaseStrategy):
# use workload_cache to store the workload of VMs for reuse purpose
workload_cache = {}
for node_id in nodes:
node = self.compute_model.get_node_from_id(
node = self.compute_model.get_node_by_uuid(
node_id)
instances = self.compute_model.mapping.get_node_instances(node)
node_workload = 0.0
for instance_id in instances:
instance = self.compute_model.get_instance_from_id(instance_id)
instance = self.compute_model.get_instance_by_uuid(instance_id)
try:
cpu_util = self.ceilometer.statistic_aggregation(
resource_id=instance_id,

30
watcher/decision_engine/strategy/strategies/workload_stabilization.py
View File

@@ -187,9 +187,9 @@ class WorkloadStabilization(base.WorkloadStabilizationBaseStrategy):
:return: dict
"""
LOG.debug('get_instance_load started')
instance_vcpus = self.compute_model.get_resource_from_id(
instance_vcpus = self.compute_model.get_resource_by_uuid(
element.ResourceType.cpu_cores).get_capacity(
self.compute_model.get_instance_from_id(instance_uuid))
self.compute_model.get_instance_by_uuid(instance_uuid))
instance_load = {'uuid': instance_uuid, 'vcpus': instance_vcpus}
for meter in self.metrics:
avg_meter = self.ceilometer.statistic_aggregation(
@@ -208,9 +208,9 @@ class WorkloadStabilization(base.WorkloadStabilizationBaseStrategy):
normalized_hosts = copy.deepcopy(hosts)
for host in normalized_hosts:
if 'memory.resident' in normalized_hosts[host]:
h_memory = self.compute_model.get_resource_from_id(
h_memory = self.compute_model.get_resource_by_uuid(
element.ResourceType.memory).get_capacity(
self.compute_model.get_node_from_id(host))
self.compute_model.get_node_by_uuid(host))
normalized_hosts[host]['memory.resident'] /= float(h_memory)
return normalized_hosts
@@ -220,9 +220,9 @@ class WorkloadStabilization(base.WorkloadStabilizationBaseStrategy):
hosts_load = {}
for node_id in self.compute_model.get_all_compute_nodes():
hosts_load[node_id] = {}
host_vcpus = self.compute_model.get_resource_from_id(
host_vcpus = self.compute_model.get_resource_by_uuid(
element.ResourceType.cpu_cores).get_capacity(
self.compute_model.get_node_from_id(node_id))
self.compute_model.get_node_by_uuid(node_id))
hosts_load[node_id]['vcpus'] = host_vcpus
for metric in self.metrics:
@@ -319,10 +319,10 @@ class WorkloadStabilization(base.WorkloadStabilizationBaseStrategy):
c_nodes.remove(source_hp_id)
node_list = yield_nodes(c_nodes)
instances_id = self.compute_model.get_mapping(). \
get_node_instances_from_id(source_hp_id)
get_node_instances_by_uuid(source_hp_id)
for instance_id in instances_id:
min_sd_case = {'value': len(self.metrics)}
instance = self.compute_model.get_instance_from_id(instance_id)
instance = self.compute_model.get_instance_by_uuid(instance_id)
if instance.state not in [element.InstanceState.ACTIVE.value,
element.InstanceState.PAUSED.value]:
continue
@@ -371,10 +371,10 @@ class WorkloadStabilization(base.WorkloadStabilizationBaseStrategy):
mig_destination_node.uuid)
def migrate(self, instance_uuid, src_host, dst_host):
mig_instance = self.compute_model.get_instance_from_id(instance_uuid)
mig_source_node = self.compute_model.get_node_from_id(
mig_instance = self.compute_model.get_instance_by_uuid(instance_uuid)
mig_source_node = self.compute_model.get_node_by_uuid(
src_host)
mig_destination_node = self.compute_model.get_node_from_id(
mig_destination_node = self.compute_model.get_node_by_uuid(
dst_host)
self.create_migration_instance(mig_instance, mig_source_node,
mig_destination_node)
@@ -403,13 +403,13 @@ class WorkloadStabilization(base.WorkloadStabilizationBaseStrategy):
min_sd = 1
balanced = False
for instance_host in migration:
dst_hp_disk = self.compute_model.get_resource_from_id(
dst_hp_disk = self.compute_model.get_resource_by_uuid(
element.ResourceType.disk).get_capacity(
self.compute_model.get_node_from_id(
self.compute_model.get_node_by_uuid(
instance_host['host']))
instance_disk = self.compute_model.get_resource_from_id(
instance_disk = self.compute_model.get_resource_by_uuid(
element.ResourceType.disk).get_capacity(
self.compute_model.get_instance_from_id(
self.compute_model.get_instance_by_uuid(
instance_host['instance']))
if instance_disk > dst_hp_disk:
continue