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

Workload balance migration strategy implementation

Browse Source 
This is one of the algorithm of Intel thermal POC.
It's based on the VM workloads of hypervisors.

Change-Id: I45ab0cf0f05786e6f68025bdd315f38381900a68
blueprint: workload-balance-migration-strategy
This commit is contained in:
junjie huang
2016-03-13 18:02:27 +00:00
parent dcb5c1f9fc
commit 30bdf29002
5 changed files with 572 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

70
watcher/tests/decision_engine/strategy/strategies/faker_cluster_state.py
View File

@@ -235,3 +235,73 @@ class FakerModelCollector(base.BaseClusterModelCollector):
current_state_cluster.get_vm_from_id("VM_0"))
return current_state_cluster
def generate_scenario_6_with_2_hypervisors(self):
vms = []
root = modelroot.ModelRoot()
# number of nodes
count_node = 2
# define ressouce ( CPU, MEM disk, ... )
mem = resource.Resource(resource.ResourceType.memory)
# 2199.954 Mhz
num_cores = resource.Resource(resource.ResourceType.cpu_cores)
disk = resource.Resource(resource.ResourceType.disk)
root.create_resource(mem)
root.create_resource(num_cores)
root.create_resource(disk)
for i in range(0, count_node):
node_uuid = "Node_{0}".format(i)
node = hypervisor.Hypervisor()
node.uuid = node_uuid
node.hostname = "hostname_{0}".format(i)
mem.set_capacity(node, 132)
disk.set_capacity(node, 250)
num_cores.set_capacity(node, 40)
root.add_hypervisor(node)
vm1 = modelvm.VM()
vm1.uuid = "VM_1"
mem.set_capacity(vm1, 2)
disk.set_capacity(vm1, 20)
num_cores.set_capacity(vm1, 10)
vms.append(vm1)
root.add_vm(vm1)
vm11 = modelvm.VM()
vm11.uuid = "73b09e16-35b7-4922-804e-e8f5d9b740fc"
mem.set_capacity(vm11, 2)
disk.set_capacity(vm11, 20)
num_cores.set_capacity(vm11, 10)
vms.append(vm11)
root.add_vm(vm11)
vm2 = modelvm.VM()
vm2.uuid = "VM_3"
mem.set_capacity(vm2, 2)
disk.set_capacity(vm2, 20)
num_cores.set_capacity(vm2, 10)
vms.append(vm2)
root.add_vm(vm2)
vm21 = modelvm.VM()
vm21.uuid = "VM_4"
mem.set_capacity(vm21, 2)
disk.set_capacity(vm21, 20)
num_cores.set_capacity(vm21, 10)
vms.append(vm21)
root.add_vm(vm21)
root.get_mapping().map(root.get_hypervisor_from_id("Node_0"),
root.get_vm_from_id(str(vm1.uuid)))
root.get_mapping().map(root.get_hypervisor_from_id("Node_0"),
root.get_vm_from_id(str(vm11.uuid)))
root.get_mapping().map(root.get_hypervisor_from_id("Node_1"),
root.get_vm_from_id(str(vm2.uuid)))
root.get_mapping().map(root.get_hypervisor_from_id("Node_1"),
root.get_vm_from_id(str(vm21.uuid)))
return root

28
watcher/tests/decision_engine/strategy/strategies/faker_metrics_collector.py
View File

@@ -44,6 +44,13 @@ class FakerMetricsCollector(object):
result = self.get_average_outlet_temperature(resource_id)
return result
def mock_get_statistics_wb(self, resource_id, meter_name, period,
aggregate='avg'):
result = 0
if meter_name == "cpu_util":
result = self.get_average_usage_vm_cpu_wb(resource_id)
return result
def get_average_outlet_temperature(self, uuid):
"""The average outlet temperature for host"""
mock = {}
@@ -52,7 +59,6 @@ class FakerMetricsCollector(object):
mock['Node_1'] = 100
if uuid not in mock.keys():
mock[uuid] = 100
return mock[str(uuid)]
def get_usage_node_ram(self, uuid):
@@ -109,6 +115,26 @@ class FakerMetricsCollector(object):
return float(mock[str(uuid)])
def get_average_usage_vm_cpu_wb(self, uuid):
"""The last VM CPU usage values to average
:param uuid:00
:return:
"""
# query influxdb stream
# compute in stream
# Normalize
mock = {}
# node 0
mock['VM_1'] = 80
mock['73b09e16-35b7-4922-804e-e8f5d9b740fc'] = 50
# node 1
mock['VM_3'] = 20
mock['VM_4'] = 10
return float(mock[str(uuid)])
def get_average_usage_vm_cpu(self, uuid):
"""The last VM CPU usage values to average

150
watcher/tests/decision_engine/strategy/strategies/test_workload_balance.py Normal file
View File

@@ -0,0 +1,150 @@
# -*- encoding: utf-8 -*-
# Copyright (c) 2016 Intel Corp
#
# Authors: Junjie-Huang <junjie.huang@intel.com>
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
import collections
import mock
from watcher.applier.actions.loading import default
from watcher.common import exception
from watcher.decision_engine.model import model_root
from watcher.decision_engine.model import resource
from watcher.decision_engine.strategy.strategies import workload_balance
from watcher.tests import base
from watcher.tests.decision_engine.strategy.strategies \
import faker_cluster_state
from watcher.tests.decision_engine.strategy.strategies \
import faker_metrics_collector
class TestWorkloadBalance(base.BaseTestCase):
# fake metrics
fake_metrics = faker_metrics_collector.FakerMetricsCollector()
# fake cluster
fake_cluster = faker_cluster_state.FakerModelCollector()
def test_calc_used_res(self):
model = self.fake_cluster.generate_scenario_6_with_2_hypervisors()
strategy = workload_balance.WorkloadBalance()
hypervisor = model.get_hypervisor_from_id('Node_0')
cap_cores = model.get_resource_from_id(resource.ResourceType.cpu_cores)
cap_mem = model.get_resource_from_id(resource.ResourceType.memory)
cap_disk = model.get_resource_from_id(resource.ResourceType.disk)
cores_used, mem_used, disk_used = strategy.calculate_used_resource(
model, hypervisor, cap_cores, cap_mem, cap_disk)
self.assertEqual((cores_used, mem_used, disk_used), (20, 4, 40))
def test_group_hosts_by_cpu_util(self):
model = self.fake_cluster.generate_scenario_6_with_2_hypervisors()
strategy = workload_balance.WorkloadBalance()
strategy.threshold = 30
strategy.ceilometer = mock.MagicMock(
statistic_aggregation=self.fake_metrics.mock_get_statistics_wb)
h1, h2, avg, w_map = strategy.group_hosts_by_cpu_util(model)
# print h1, h2, avg, w_map
self.assertEqual(h1[0]['hv'].uuid, 'Node_0')
self.assertEqual(h2[0]['hv'].uuid, 'Node_1')
self.assertEqual(avg, 8.0)
def test_choose_vm_to_migrate(self):
model = self.fake_cluster.generate_scenario_6_with_2_hypervisors()
strategy = workload_balance.WorkloadBalance()
strategy.ceilometer = mock.MagicMock(
statistic_aggregation=self.fake_metrics.mock_get_statistics_wb)
h1, h2, avg, w_map = strategy.group_hosts_by_cpu_util(model)
vm_to_mig = strategy.choose_vm_to_migrate(model, h1, avg, w_map)
self.assertEqual(vm_to_mig[0].uuid, 'Node_0')
self.assertEqual(vm_to_mig[1].uuid,
"73b09e16-35b7-4922-804e-e8f5d9b740fc")
def test_choose_vm_notfound(self):
model = self.fake_cluster.generate_scenario_6_with_2_hypervisors()
strategy = workload_balance.WorkloadBalance()
strategy.ceilometer = mock.MagicMock(
statistic_aggregation=self.fake_metrics.mock_get_statistics_wb)
h1, h2, avg, w_map = strategy.group_hosts_by_cpu_util(model)
vms = model.get_all_vms()
vms.clear()
vm_to_mig = strategy.choose_vm_to_migrate(model, h1, avg, w_map)
self.assertEqual(vm_to_mig, None)
def test_filter_destination_hosts(self):
model = self.fake_cluster.generate_scenario_6_with_2_hypervisors()
strategy = workload_balance.WorkloadBalance()
strategy.ceilometer = mock.MagicMock(
statistic_aggregation=self.fake_metrics.mock_get_statistics_wb)
h1, h2, avg, w_map = strategy.group_hosts_by_cpu_util(model)
vm_to_mig = strategy.choose_vm_to_migrate(model, h1, avg, w_map)
dest_hosts = strategy.filter_destination_hosts(model, h2, vm_to_mig[1],
avg, w_map)
self.assertEqual(len(dest_hosts), 1)
self.assertEqual(dest_hosts[0]['hv'].uuid, 'Node_1')
def test_exception_model(self):
strategy = workload_balance.WorkloadBalance()
self.assertRaises(exception.ClusterStateNotDefined, strategy.execute,
None)
def test_exception_cluster_empty(self):
strategy = workload_balance.WorkloadBalance()
model = model_root.ModelRoot()
self.assertRaises(exception.ClusterEmpty, strategy.execute, model)
def test_execute_cluster_empty(self):
strategy = workload_balance.WorkloadBalance()
strategy.ceilometer = mock.MagicMock(
statistic_aggregation=self.fake_metrics.mock_get_statistics_wb)
model = model_root.ModelRoot()
self.assertRaises(exception.ClusterEmpty, strategy.execute, model)
def test_execute_no_workload(self):
strategy = workload_balance.WorkloadBalance()
strategy.ceilometer = mock.MagicMock(
statistic_aggregation=self.fake_metrics.mock_get_statistics_wb)
current_state_cluster = faker_cluster_state.FakerModelCollector()
model = current_state_cluster. \
generate_scenario_4_with_1_hypervisor_no_vm()
solution = strategy.execute(model)
self.assertEqual(solution.actions, [])
def test_execute(self):
strategy = workload_balance.WorkloadBalance()
strategy.ceilometer = mock.MagicMock(
statistic_aggregation=self.fake_metrics.mock_get_statistics_wb)
model = self.fake_cluster.generate_scenario_6_with_2_hypervisors()
solution = strategy.execute(model)
actions_counter = collections.Counter(
[action.get('action_type') for action in solution.actions])
num_migrations = actions_counter.get("migrate", 0)
self.assertEqual(num_migrations, 1)
def test_check_parameters(self):
strategy = workload_balance.WorkloadBalance()
strategy.ceilometer = mock.MagicMock(
statistic_aggregation=self.fake_metrics.mock_get_statistics_wb)
model = self.fake_cluster.generate_scenario_6_with_2_hypervisors()
solution = strategy.execute(model)
loader = default.DefaultActionLoader()
for action in solution.actions:
loaded_action = loader.load(action['action_type'])
loaded_action.input_parameters = action['input_parameters']
loaded_action.validate_parameters()