x
/
valet
Code Issues Proposed changes

Merge "Moved logger references to file headers"

This commit is contained in:
Jenkins 2017-06-18 21:43:50 +00:00 committed by Gerrit Code Review
parent b679af450b 77d4e1e47f
commit f973b8acfc
20 changed files with 475 additions and 457 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
bin/valet-engine
View File

@ -13,31 +13,33 @@
# 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.
"""Daemon foe Valet Engine."""
import os
import sys
import traceback
from oslo_log import log
from valet.common.conf import get_logger
from valet.engine.optimizer.ostro.ostro import Ostro
from valet.engine.optimizer.ostro_server.configuration import Config
from valet.engine.optimizer.ostro_server.daemon import Daemon
LOG = log.getLogger(__name__)
class OstroDaemon(Daemon):
"""Daemon foe Valet Engine."""
"""Daemon for Valet Engine."""
def run(self):
"""Run the daemon."""
self.logger.info("##### Valet Engine is launched #####")
LOG.info("Valet Engine is launched")
try:
ostro = Ostro(config, self.logger)
ostro = Ostro(config)
except Exception:
self.logger.error(traceback.format_exc())
LOG.error(traceback.format_exc())
if ostro.bootstrap() is False:
self.logger.error("ostro bootstrap failed")
LOG.error("Valet Engine bootstrap failed")
sys.exit(2)
# Write pidfile

44
valet/engine/listener/listener_manager.py
View File

@ -20,11 +20,14 @@ import json
import pika
import threading
import traceback
from valet.common.conf import get_logger
from oslo_log import log
from valet.common.music import Music
from valet.engine.listener.oslo_messages import OsloMessage
import yaml
LOG = log.getLogger(__name__)
class ListenerManager(threading.Thread):
"""Listener Manager Thread Class."""
@ -35,7 +38,6 @@ class ListenerManager(threading.Thread):
self.thread_id = _t_id
self.thread_name = _t_name
self.config = _config
self.listener_logger = get_logger("ostro_listener")
self.MUSIC = None
def run(self):
@ -47,8 +49,8 @@ class ListenerManager(threading.Thread):
credentials = pika.PlainCredentials("guest", "PASSWORD").
"""
try:
self.listener_logger.info("ListenerManager: start " +
self.thread_name + " ......")
LOG.info("ListenerManager: start " +
self.thread_name + " ......")
if self.config.events_listener.store:
@ -58,19 +60,21 @@ class ListenerManager(threading.Thread):
'replication_factor': self.config.music.replication_factor,
'music_server_retries':
self.config.music.music_server_retries,
'logger': self.listener_logger,
'logger': LOG,
}
engine = Music(**kwargs)
engine.create_keyspace(self.config.music.keyspace)
self.MUSIC = {'engine': engine,
'keyspace': self.config.music.keyspace}
self.listener_logger.debug(
'Storing in music on %s, keyspace %s',
self.config.music.hosts, self.config.music.keyspace)
self.listener_logger.debug('Connecting to %s, with %s',
self.config.messaging.host,
self.config.messaging.username)
LOG.debug(
'Storing in music on %s, keyspace %s' %
(self.config.music.host, self.config.music.keyspace))
LOG.debug('Connecting to %s, with %s' %
(self.config.messaging.host,
self.config.messaging.username))
credentials = pika.PlainCredentials(self.config.messaging.username,
self.config.messaging.password)
parameters = pika.ConnectionParameters(self.config.messaging.host,
@ -103,15 +107,16 @@ class ListenerManager(threading.Thread):
# Bind the queue to the selected exchange
channel.queue_bind(exchange=exchange_name, queue=queue_name,
routing_key=binding_key)
self.listener_logger.info('Channel is bound,listening on %s '
'exchange %s',
self.config.messaging.host,
self.config.events_listener.exchange)
LOG.info('Channel is bound,listening on %s exchange %s',
self.config.messaging.host,
self.config.events_listener.exchange)
# Start consuming messages
channel.basic_consume(self.on_message, queue_name)
except Exception:
self.listener_logger.error(traceback.format_exc())
LOG.error("Failed to start ListenerManager thread: %s",
traceback.format_exc())
return
try:
@ -136,13 +141,14 @@ class ListenerManager(threading.Thread):
else:
return
self.listener_logger.debug(
LOG.debug(
"\nMessage No: %s\n", method_frame.delivery_tag)
self.listener_logger.debug(
LOG.debug(
json.dumps(message, sort_keys=True, indent=2))
channel.basic_ack(delivery_tag=method_frame.delivery_tag)
except Exception:
self.listener_logger.error(traceback.format_exc())
LOG.error("Could not specify action for message: %s",
traceback.format_exc())
return
def is_message_wanted(self, message):

37
valet/engine/optimizer/app_manager/app_handler.py
View File

@ -18,10 +18,14 @@
import operator
import time
from oslo_log import log
from valet.engine.optimizer.app_manager.app_topology import AppTopology
from valet.engine.optimizer.app_manager.app_topology_base import VM
from valet.engine.optimizer.app_manager.application import App
LOG = log.getLogger(__name__)
class AppHistory(object):
@ -40,12 +44,11 @@ class AppHandler(object):
placement and updating topology info.
"""
def __init__(self, _resource, _db, _config, _logger):
def __init__(self, _resource, _db, _config):
"""Init App Handler Class."""
self.resource = _resource
self.db = _db
self.config = _config
self.logger = _logger
""" current app requested, a temporary copy """
self.apps = {}
@ -109,7 +112,7 @@ class AppHandler(object):
"""Add app and set or regenerate topology, return updated topology."""
self.apps.clear()
app_topology = AppTopology(self.resource, self.logger)
app_topology = AppTopology(self.resource)
stack_id = None
if "stack_id" in _app.keys():
@ -124,9 +127,7 @@ class AppHandler(object):
application_name = "none"
action = _app["action"]
if action == "ping":
self.logger.info("got ping")
elif action == "replan" or action == "migrate":
if action == "replan" or action == "migrate":
re_app = self._regenerate_app_topology(stack_id, _app,
app_topology, action)
if re_app is None:
@ -136,14 +137,15 @@ class AppHandler(object):
return None
if action == "replan":
self.logger.info("got replan: " + stack_id)
LOG.info("got replan: " + stack_id)
elif action == "migrate":
self.logger.info("got migration: " + stack_id)
LOG.info("got migration: " + stack_id)
app_id = app_topology.set_app_topology(re_app)
if app_id is None:
self.logger.error(app_topology.status)
LOG.error("Could not set app topology for regererated graph." +
app_topology.status)
self.status = app_topology.status
self.apps[stack_id] = None
return None
@ -151,12 +153,13 @@ class AppHandler(object):
app_id = app_topology.set_app_topology(_app)
if len(app_topology.candidate_list_map) > 0:
self.logger.info("got ad-hoc placement: " + stack_id)
LOG.info("got ad-hoc placement: " + stack_id)
else:
self.logger.info("got placement: " + stack_id)
LOG.info("got placement: " + stack_id)
if app_id is None:
self.logger.error(app_topology.status)
LOG.error("Could not set app topology for app graph" +
app_topology.status)
self.status = app_topology.status
self.apps[stack_id] = None
return None
@ -216,8 +219,8 @@ class AppHandler(object):
if self.db is not None:
for appk, _ in self.apps.iteritems():
if self.db.add_app(appk, None) is False:
self.logger.error("AppHandler: error while adding app "
"info to MUSIC")
LOG.error("AppHandler: error while adding app "
"info to MUSIC")
def get_vm_info(self, _s_uuid, _h_uuid, _host):
"""Return vm_info from database."""
@ -241,12 +244,10 @@ class AppHandler(object):
old_app = self.db.get_app_info(_stack_id)
if old_app is None:
self.status = "error while getting old_app from MUSIC"
self.logger.error(self.status)
LOG.error("Error while getting old_app from MUSIC")
return None
elif len(old_app) == 0:
self.status = "cannot find the old app in MUSIC"
self.logger.error(self.status)
LOG.error("Cannot find the old app in MUSIC")
return None
re_app["action"] = "create"

5
valet/engine/optimizer/app_manager/app_topology.py
View File

@ -24,7 +24,7 @@ class AppTopology(object):
calculating and setting optimization.
"""
def __init__(self, _resource, _logger):
def __init__(self, _resource):
"""Init App Topology Class."""
self.vgroups = {}
self.vms = {}
@ -38,14 +38,13 @@ class AppTopology(object):
self.exclusion_list_map = {}
self.resource = _resource
self.logger = _logger
# restriction of host naming convention
high_level_allowed = True
if "none" in self.resource.datacenter.region_code_list:
high_level_allowed = False
self.parser = Parser(high_level_allowed, self.logger)
self.parser = Parser(high_level_allowed)
self.total_CPU = 0
self.total_mem = 0

16
valet/engine/optimizer/app_manager/app_topology_parser.py
View File

@ -24,13 +24,15 @@
OS::Heat::ResourceGroup
OS::Heat::ResourceGroup
"""
from oslo_log import log
import six
from valet.engine.optimizer.app_manager.app_topology_base import LEVELS
from valet.engine.optimizer.app_manager.app_topology_base import VGroup
from valet.engine.optimizer.app_manager.app_topology_base import VM
LOG = log.getLogger(__name__)
class Parser(object):
"""Parser Class.
@ -41,10 +43,8 @@ class Parser(object):
OS::Heat::Stack OS::Heat::ResourceGroup
"""
def __init__(self, _high_level_allowed, _logger):
def __init__(self, _high_level_allowed):
"""Init Parser Class."""
self.logger = _logger
self.high_level_allowed = _high_level_allowed
self.format_version = None
@ -109,10 +109,10 @@ class Parser(object):
if len(r["locations"]) > 0:
self.candidate_list_map[rk] = r["locations"]
vms[vm.uuid] = vm
self.logger.info("vm = " + vm.uuid)
LOG.info("vm = " + vm.uuid)
elif r["type"] == "OS::Cinder::Volume":
self.logger.warn("Parser: do nothing for volume at this "
"version")
LOG.warning("Parser: do nothing for volume at this "
"version")
elif r["type"] == "ATT::Valet::GroupAssignment":
vgroup = VGroup(self.stack_id, rk)
@ -154,7 +154,7 @@ class Parser(object):
return {}, {}
vgroups[vgroup.uuid] = vgroup
msg = "group = %s, type = %s"
self.logger.info(msg % (vgroup.name, vgroup.vgroup_type))
LOG.info(msg % (vgroup.name, vgroup.vgroup_type))
if self._merge_diversity_groups(_elements, vgroups, vms) is False:
return {}, {}

148
valet/engine/optimizer/db_connect/music_handler.py
View File

@ -17,9 +17,14 @@
import json
import operator
from oslo_log import log
from valet.common.music import Music
from valet.engine.optimizer.db_connect.event import Event
LOG = log.getLogger(__name__)
def ensurekey(d, k):
return d.setdefault(k, {})
@ -33,21 +38,19 @@ class MusicHandler(object):
database for valet and returns/deletes/updates objects within it.
"""
def __init__(self, _config, _logger):
def __init__(self, _config):
"""Init Music Handler."""
self.config = _config
self.logger = _logger
self.music = Music(
hosts=self.config.hosts, port=self.config.port,
replication_factor=self.config.replication_factor,
music_server_retries=self.config.music_server_retries,
logger=self.logger)
music_server_retries=self.config.music_server_retries)
if self.config.hosts is not None:
self.logger.info("DB: music host = %s", self.config.hosts)
LOG.info("DB: music host = %s", self.config.hosts)
if self.config.replication_factor is not None:
self.logger.info("DB: music replication factor = %s ",
str(self.config.replication_factor))
LOG.info("DB: music replication factor = %s ",
str(self.config.replication_factor))
# FIXME(GJ): this may not need
def init_db(self):
@ -57,12 +60,12 @@ class MusicHandler(object):
necessary tables with the proper schemas in Music using API calls.
Return True if no exceptions are caught.
"""
self.logger.info("MusicHandler.init_db: create table")
LOG.info("MusicHandler.init_db: create table")
try:
self.music.create_keyspace(self.config.db_keyspace)
except Exception as e:
self.logger.error("DB: " + str(e))
LOG.error("DB could not create keyspace: " + str(e))
return False
schema = {
@ -74,7 +77,7 @@ class MusicHandler(object):
self.music.create_table(self.config.db_keyspace,
self.config.db_request_table, schema)
except Exception as e:
self.logger.error("DB: " + str(e))
LOG.error("DB could not create request table: " + str(e))
return False
schema = {
@ -86,7 +89,7 @@ class MusicHandler(object):
self.music.create_table(self.config.db_keyspace,
self.config.db_response_table, schema)
except Exception as e:
self.logger.error("DB: " + str(e))
LOG.error("DB could not create response table: " + str(e))
return False
schema = {
@ -100,7 +103,7 @@ class MusicHandler(object):
self.music.create_table(self.config.db_keyspace,
self.config.db_event_table, schema)
except Exception as e:
self.logger.error("DB: " + str(e))
LOG.error("DB could not create event table: " + str(e))
return False
schema = {
@ -112,7 +115,7 @@ class MusicHandler(object):
self.music.create_table(self.config.db_keyspace,
self.config.db_resource_table, schema)
except Exception as e:
self.logger.error("DB: " + str(e))
LOG.error("DB could not create resource table: " + str(e))
return False
schema = {
@ -124,7 +127,7 @@ class MusicHandler(object):
self.music.create_table(self.config.db_keyspace,
self.config.db_app_table, schema)
except Exception as e:
self.logger.error("DB: " + str(e))
LOG.error("DB could not create app table: " + str(e))
return False
schema = {
@ -137,7 +140,7 @@ class MusicHandler(object):
self.music.create_table(self.config.db_keyspace,
self.config.db_uuid_table, schema)
except Exception as e:
self.logger.error("DB: " + str(e))
LOG.error("DB could not create uuid table: " + str(e))
return False
return True
@ -157,7 +160,7 @@ class MusicHandler(object):
events = self.music.read_all_rows(self.config.db_keyspace,
self.config.db_event_table)
except Exception as e:
self.logger.error("DB:event: " + str(e))
LOG.error("DB:event: " + str(e))
# FIXME(GJ): return None?
return {}
@ -168,13 +171,13 @@ class MusicHandler(object):
method = row['method']
args_data = row['args']
self.logger.debug("MusicHandler.get_events: event (" +
event_id + ") is entered")
LOG.debug("MusicHandler.get_events: event (" +
event_id + ") is entered")
if exchange != "nova":
if self.delete_event(event_id) is False:
return None
self.logger.debug(
LOG.debug(
"MusicHandler.get_events: event exchange "
"(" + exchange + ") is not supported")
continue
@ -183,23 +186,23 @@ class MusicHandler(object):
'instance':
if self.delete_event(event_id) is False:
return None
self.logger.debug("MusicHandler.get_events: event method "
"(" + method + ") is not considered")
LOG.debug("MusicHandler.get_events: event method "
"(" + method + ") is not considered")
continue
if len(args_data) == 0:
if self.delete_event(event_id) is False:
return None
self.logger.debug("MusicHandler.get_events: "
"event does not have args")
LOG.debug("MusicHandler.get_events: "
"event does not have args")
continue
try:
args = json.loads(args_data)
except (ValueError, KeyError, TypeError):
self.logger.warn("MusicHandler.get_events: error while "
"decoding to JSON event = " + method +
":" + event_id)
LOG.warning("MusicHandler.get_events: error while "
"decoding to JSON event = " + method +
":" + event_id)
continue
# TODO(lamt) this block of code can use refactoring
@ -227,11 +230,11 @@ class MusicHandler(object):
event_list.append(e)
else:
msg = "unknown vm_state = %s"
self.logger.warn(
LOG.warning(
msg % change_data["vm_state"])
if 'uuid' in change_data.keys():
msg = " uuid = %s"
self.logger.warn(
LOG.warning(
msg % change_data['uuid'])
if not self.delete_event(event_id):
return None
@ -296,20 +299,20 @@ class MusicHandler(object):
e.host is None or e.host == "none" or \
e.vcpus == -1 or e.mem == -1:
error_event_list.append(e)
self.logger.warn("MusicHandler.get_events: data "
"missing in instance object event")
LOG.warning("MusicHandler.get_events: data "
"missing in instance object event")
elif e.object_name == 'ComputeNode':
if e.host is None or e.host == "none":
error_event_list.append(e)
self.logger.warn("MusicHandler.get_events: data "
"missing in compute object event")
LOG.warning("MusicHandler.get_events: data "
"missing in compute object event")
elif e.method == "build_and_run_instance":
if e.uuid is None or e.uuid == "none":
error_event_list.append(e)
self.logger.warn("MusicHandler.get_events: data missing "
"in build event")
LOG.warning("MusicHandler.get_events: data missing "
"in build event")
if len(error_event_list) > 0:
event_list[:] = [
@ -327,7 +330,7 @@ class MusicHandler(object):
self.config.db_event_table,
'timestamp', _event_id)
except Exception as e:
self.logger.error("DB: while deleting event: " + str(e))
LOG.error("DB: while deleting event: " + str(e))
return False
return True
@ -342,7 +345,7 @@ class MusicHandler(object):
row = self.music.read_row(self.config.db_keyspace,
self.config.db_uuid_table, 'uuid', _uuid)
except Exception as e:
self.logger.error("DB: while reading uuid: " + str(e))
LOG.error("DB: while reading uuid: " + str(e))
return None
if len(row) > 0:
@ -376,7 +379,7 @@ class MusicHandler(object):
self.music.create_row(self.config.db_keyspace,
self.config.db_uuid_table, data)
except Exception as e:
self.logger.error("DB: while inserting uuid: " + str(e))
LOG.error("DB: while inserting uuid: " + str(e))
return False
return True
@ -388,7 +391,7 @@ class MusicHandler(object):
self.config.db_uuid_table, 'uuid',
_k)
except Exception as e:
self.logger.error("DB: while deleting uuid: " + str(e))
LOG.error("DB: while deleting uuid: " + str(e))
return False
return True
@ -402,16 +405,15 @@ class MusicHandler(object):
requests = self.music.read_all_rows(self.config.db_keyspace,
self.config.db_request_table)
except Exception as e:
self.logger.error("DB: while reading requests: " + str(e))
LOG.error("DB: while reading requests: " + str(e))
# FIXME(GJ): return None?
return {}
if len(requests) > 0:
self.logger.info("MusicHandler.get_requests: placement request "
"arrived")
LOG.info("MusicHandler.get_requests: placement request arrived")
for _, row in requests.iteritems():
self.logger.info(" request_id = " + row['stack_id'])
LOG.info(" request_id = " + row['stack_id'])
r_list = json.loads(row['request'])
for r in r_list:
@ -431,8 +433,8 @@ class MusicHandler(object):
self.music.create_row(self.config.db_keyspace,
self.config.db_response_table, data)
except Exception as e:
self.logger.error("MUSIC error while putting placement "
"result: " + str(e))
LOG.error("MUSIC error while putting placement "
"result: " + str(e))
return False
for appk in _result.keys():
@ -441,8 +443,8 @@ class MusicHandler(object):
self.config.db_request_table,
'stack_id', appk)
except Exception as e:
self.logger.error("MUSIC error while deleting handled "
"request: " + str(e))
LOG.error("MUSIC error while deleting handled "
"request: " + str(e))
return False
return True
@ -455,10 +457,10 @@ class MusicHandler(object):
try:
row = self.music.read_row(self.config.db_keyspace,
self.config.db_resource_table,
'site_name', _k, self.logger)
'site_name', _k)
except Exception as e:
self.logger.error("MUSIC error while reading resource status: " +
str(e))
LOG.error("MUSIC error while reading resource status: " +
str(e))
return None
if len(row) > 0:
@ -475,8 +477,8 @@ class MusicHandler(object):
self.config.db_resource_table,
'site_name', _k)
except Exception as e:
self.logger.error("MUSIC error while reading resource status: " +
str(e))
LOG.error("MUSIC error while reading resource status: " +
str(e))
return False
json_resource = {}
@ -525,8 +527,8 @@ class MusicHandler(object):
self.config.db_resource_table,
'site_name', _k)
except Exception as e:
self.logger.error("MUSIC error while deleting resource "
"status: " + str(e))
LOG.error("MUSIC error while deleting resource "
"status: " + str(e))
return False
else:
@ -541,10 +543,10 @@ class MusicHandler(object):
self.music.create_row(self.config.db_keyspace,
self.config.db_resource_table, data)
except Exception as e:
self.logger.error("DB: " + str(e))
LOG.error("DB could not create row in resource table: " + str(e))
return False
self.logger.info("DB: resource status updated")
LOG.info("DB: resource status updated")
return True
@ -555,7 +557,7 @@ class MusicHandler(object):
self.config.db_keyspace, self.config.db_app_table,
'stack_id', _k)
except Exception as e:
self.logger.error("DB: while deleting app: " + str(e))
LOG.error("DB: while deleting app: " + str(e))
return False
if _app_data is not None:
@ -568,7 +570,7 @@ class MusicHandler(object):
self.music.create_row(self.config.db_keyspace,
self.config.db_app_table, data)
except Exception as e:
self.logger.error("DB: while inserting app: " + str(e))
LOG.error("DB: while inserting app: " + str(e))
return False
return True
@ -583,7 +585,7 @@ class MusicHandler(object):
self.config.db_app_table, 'stack_id',
_s_uuid)
except Exception as e:
self.logger.error("DB: while reading app info: " + str(e))
LOG.error("DB: while reading app info: " + str(e))
return None
if len(row) > 0:
@ -606,7 +608,7 @@ class MusicHandler(object):
self.config.db_app_table, 'stack_id',
_s_uuid)
except Exception as e:
self.logger.error("DB: " + str(e))
LOG.error("DB could not read row in app table: " + str(e))
return None
if len(row) > 0:
@ -620,25 +622,25 @@ class MusicHandler(object):
if vm["host"] != _host:
vm["planned_host"] = vm["host"]
vm["host"] = _host
self.logger.warn("db: conflicted placement "
"decision from Ostro")
LOG.warning("DB: conflicted placement "
"decision from Ostro")
# TODO(GY): affinity, diversity, exclusivity
# validation check
updated = True
else:
vm["status"] = "scheduled"
self.logger.warn("DB: vm was deleted")
LOG.warning("DB: vm was deleted")
updated = True
vm_info = vm
break
else:
self.logger.error("MusicHandler.get_vm_info: vm is missing "
"from stack")
LOG.error("MusicHandler.get_vm_info: vm is missing "
"from stack")
else:
self.logger.warn("MusicHandler.get_vm_info: not found stack for "
"update = " + _s_uuid)
LOG.warning("MusicHandler.get_vm_info: not found stack for "
"update = " + _s_uuid)
if updated is True:
if self.add_app(_s_uuid, json_app) is False:
@ -657,7 +659,7 @@ class MusicHandler(object):
self.config.db_app_table, 'stack_id',
_s_uuid)
except Exception as e:
self.logger.error("DB: " + str(e))
LOG.error("DB could not read row in app table: " + str(e))
return False
if len(row) > 0:
@ -669,18 +671,18 @@ class MusicHandler(object):
if vmk == _h_uuid:
if vm["status"] != "deleted":
vm["status"] = "deleted"
self.logger.warn("DB: deleted marked")
LOG.warning("DB: deleted marked")
updated = True
else:
self.logger.warn("DB: vm was already deleted")
LOG.warning("DB: vm was already deleted")
break
else:
self.logger.error("MusicHandler.update_vm_info: vm is missing "
"from stack")
LOG.error("MusicHandler.update_vm_info: vm is missing "
"from stack")
else:
self.logger.warn("MusicHandler.update_vm_info: not found "
"stack for update = " + _s_uuid)
LOG.warning("MusicHandler.update_vm_info: not found "
"stack for update = " + _s_uuid)
if updated is True:
if self.add_app(_s_uuid, json_app) is False:

44
valet/engine/optimizer/ostro/constraint_solver.py
View File

@ -12,6 +12,7 @@
# 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.
from oslo_log import log
from valet.engine.optimizer.app_manager.app_topology_base import LEVELS
from valet.engine.optimizer.app_manager.app_topology_base import VGroup
@ -24,20 +25,21 @@ from valet.engine.optimizer.ostro.openstack_filters import CoreFilter
from valet.engine.optimizer.ostro.openstack_filters import DiskFilter
from valet.engine.optimizer.ostro.openstack_filters import RamFilter
LOG = log.getLogger(__name__)
class ConstraintSolver(object):
"""ConstraintSolver."""
def __init__(self, _logger):
def __init__(self):
"""Initialization."""
"""Instantiate filters to help enforce constraints."""
self.logger = _logger
self.openstack_AZ = AvailabilityZoneFilter(self.logger)
self.openstack_AIES = AggregateInstanceExtraSpecsFilter(self.logger)
self.openstack_R = RamFilter(self.logger)
self.openstack_C = CoreFilter(self.logger)
self.openstack_D = DiskFilter(self.logger)
self.openstack_AZ = AvailabilityZoneFilter()
self.openstack_AIES = AggregateInstanceExtraSpecsFilter()
self.openstack_R = RamFilter()
self.openstack_C = CoreFilter()
self.openstack_D = DiskFilter()
self.status = "success"
@ -57,11 +59,11 @@ class ConstraintSolver(object):
candidate_list.append(r)
if len(candidate_list) == 0:
self.status = "no candidate for node = " + _n.node.name
self.logger.warn(self.status)
LOG.warning(self.status)
return candidate_list
else:
self.logger.debug("ConstraintSolver: num of candidates = " +
str(len(candidate_list)))
LOG.debug("ConstraintSolver: num of candidates = " +
str(len(candidate_list)))
"""Availability zone constraint."""
if isinstance(_n.node, VGroup) or isinstance(_n.node, VM):
@ -72,7 +74,7 @@ class ConstraintSolver(object):
if len(candidate_list) == 0:
self.status = "violate availability zone constraint for " \
"node = " + _n.node.name
self.logger.error("ConstraintSolver: " + self.status)
LOG.error("ConstraintSolver: " + self.status)
return candidate_list
"""Host aggregate constraint."""
@ -82,7 +84,7 @@ class ConstraintSolver(object):
if len(candidate_list) == 0:
self.status = "violate host aggregate constraint for " \
"node = " + _n.node.name
self.logger.error("ConstraintSolver: " + self.status)
LOG.error("ConstraintSolver: " + self.status)
return candidate_list
"""CPU capacity constraint."""
@ -91,7 +93,7 @@ class ConstraintSolver(object):
if len(candidate_list) == 0:
self.status = "violate cpu capacity constraint for " \
"node = " + _n.node.name
self.logger.error("ConstraintSolver: " + self.status)
LOG.error("ConstraintSolver: " + self.status)
return candidate_list
"""Memory capacity constraint."""
@ -100,7 +102,7 @@ class ConstraintSolver(object):
if len(candidate_list) == 0:
self.status = "violate memory capacity constraint for " \
"node = " + _n.node.name
self.logger.error("ConstraintSolver: " + self.status)
LOG.error("ConstraintSolver: " + self.status)
return candidate_list
"""Local disk capacity constraint."""
@ -109,7 +111,7 @@ class ConstraintSolver(object):
if len(candidate_list) == 0:
self.status = "violate local disk capacity constraint for " \
"node = " + _n.node.name
self.logger.error("ConstraintSolver: " + self.status)
LOG.error("ConstraintSolver: " + self.status)
return candidate_list
""" diversity constraint """
@ -125,7 +127,7 @@ class ConstraintSolver(object):
if len(candidate_list) == 0:
self.status = "violate diversity constraint for " \
"node = " + _n.node.name
self.logger.error("ConstraintSolver: " + self.status)
LOG.error("ConstraintSolver: " + self.status)
return candidate_list
else:
self._constrain_diversity(_level, _n, _node_placements,
@ -133,7 +135,7 @@ class ConstraintSolver(object):
if len(candidate_list) == 0:
self.status = "violate diversity constraint for " \
"node = " + _n.node.name
self.logger.error("ConstraintSolver: " + self.status)
LOG.error("ConstraintSolver: " + self.status)
return candidate_list
"""Exclusivity constraint."""
@ -142,7 +144,7 @@ class ConstraintSolver(object):
if len(exclusivities) > 1:
self.status = "violate exclusivity constraint (more than one " \
"exclusivity) for node = " + _n.node.name
self.logger.error("ConstraintSolver: " + self.status)
LOG.error("ConstraintSolver: " + self.status)
return []
else:
if len(exclusivities) == 1:
@ -153,14 +155,14 @@ class ConstraintSolver(object):
if len(candidate_list) == 0:
self.status = "violate exclusivity constraint for " \
"node = " + _n.node.name
self.logger.error("ConstraintSolver: " + self.status)
LOG.error("ConstraintSolver: " + self.status)
return candidate_list
else:
self._constrain_non_exclusivity(_level, candidate_list)
if len(candidate_list) == 0:
self.status = "violate non-exclusivity constraint for " \
"node = " + _n.node.name
self.logger.error("ConstraintSolver: " + self.status)
LOG.error("ConstraintSolver: " + self.status)
return candidate_list
"""Affinity constraint."""
@ -173,7 +175,7 @@ class ConstraintSolver(object):
if len(candidate_list) == 0:
self.status = "violate affinity constraint for " \
"node = " + _n.node.name
self.logger.error("ConstraintSolver: " + self.status)
LOG.error("ConstraintSolver: " + self.status)
return candidate_list
return candidate_list

15
valet/engine/optimizer/ostro/openstack_filters.py
View File

@ -27,9 +27,8 @@ class AggregateInstanceExtraSpecsFilter(object):
# Aggregate data and instance type does not change within a request
run_filter_once_per_request = True
def __init__(self, _logger):
def __init__(self):
"""Initialization."""
self.logger = _logger
def host_passes(self, _level, _host, _v):
"""Return a list of hosts that can create instance_type."""
@ -109,9 +108,8 @@ class AvailabilityZoneFilter(object):
# Availability zones do not change within a request
run_filter_once_per_request = True
def __init__(self, _logger):
def __init__(self):
"""Initialization."""
self.logger = _logger
def host_passes(self, _level, _host, _v):
"""Return True if all availalibility zones in _v exist in the host."""
@ -138,9 +136,8 @@ class AvailabilityZoneFilter(object):
class RamFilter(object):
"""RamFilter."""
def __init__(self, _logger):
def __init__(self):
"""Initialization."""
self.logger = _logger
def host_passes(self, _level, _host, _v):
"""Return True if host has sufficient available RAM."""
@ -161,9 +158,8 @@ class RamFilter(object):
class CoreFilter(object):
"""CoreFilter."""
def __init__(self, _logger):
def __init__(self):
"""Initialization."""
self.logger = _logger
def host_passes(self, _level, _host, _v):
"""Return True if host has sufficient CPU cores."""
@ -185,9 +181,8 @@ class CoreFilter(object):
class DiskFilter(object):
"""DiskFilter."""
def __init__(self, _logger):
def __init__(self):
"""Initialization."""
self.logger = _logger
def host_passes(self, _level, _host, _v):
"""Filter based on disk usage."""

12
valet/engine/optimizer/ostro/optimizer.py
View File

@ -12,22 +12,25 @@
# 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.
from oslo_log import log
from valet.engine.optimizer.app_manager.app_topology_base import VGroup
from valet.engine.optimizer.app_manager.app_topology_base import VM
from valet.engine.optimizer.ostro.search import Search
LOG = log.getLogger(__name__)
# FIXME(GJ): make search algorithm pluggable
# NOTE(GJ): do not deal with Volume placements at this version
class Optimizer(object):
"""Optimizer."""
def __init__(self, _resource, _logger):
def __init__(self, _resource):
"""Initialization."""
self.resource = _resource
self.logger = _logger
self.search = Search(self.logger)
self.search = Search()
self.status = "success"
@ -80,8 +83,7 @@ class Optimizer(object):
elif v.level == "cluster":
placement_map[v] = node_placement.cluster_name
self.logger.debug(" " + v.name + " placed in " +
placement_map[v])
LOG.debug(v.name + " placed in " + placement_map[v])
self._update_resource_status(uuid_map)

158
valet/engine/optimizer/ostro/ostro.py
View File

@ -12,13 +12,13 @@
# 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.
"""Valet Engine."""
from oslo_config import cfg
import threading
import time
import traceback
from oslo_config import cfg
from oslo_log import log
from valet.engine.listener.listener_manager import ListenerManager
from valet.engine.optimizer.app_manager.app_handler import AppHandler
from valet.engine.optimizer.app_manager.app_topology_base import VM
@ -29,37 +29,34 @@ from valet.engine.resource_manager.resource import Resource
from valet.engine.resource_manager.topology_manager import TopologyManager
CONF = cfg.CONF
LOG = log.getLogger(__name__)
class Ostro(object):
"""Valet Engine."""
def __init__(self, _config, _logger):
def __init__(self, _config):
"""Initialization."""
self.config = _config
self.logger = _logger
self.db = MusicHandler(self.config, self.logger)
self.db = MusicHandler(self.config)
if self.db.init_db() is False:
self.logger.error("error while initializing MUSIC database")
LOG.error("error while initializing MUSIC database")
self.resource = Resource(self.db, self.config, self.logger)
self.app_handler = AppHandler(self.resource, self.db, self.config,
self.logger)
self.optimizer = Optimizer(self.resource, self.logger)
self.resource = Resource(self.db, self.config)
self.app_handler = AppHandler(self.resource, self.db, self.config)
self.optimizer = Optimizer(self.resource)
self.data_lock = threading.Lock()
self.thread_list = []
self.topology = TopologyManager(
1, "Topology", self.resource,
self.data_lock, self.config, self.logger)
self.data_lock, self.config)
self.compute = ComputeManager(
2, "Compute", self.resource,
self.data_lock, self.config, self.logger)
self.data_lock, self.config)
self.listener = ListenerManager(3, "Listener", CONF)
@ -69,7 +66,7 @@ class Ostro(object):
self.batch_store_trigger = 10 # sec
def run_ostro(self):
self.logger.info("start Ostro ......")
LOG.info("start Ostro ......")
self.topology.start()
self.compute.start()
@ -114,7 +111,7 @@ class Ostro(object):
for t in self.thread_list:
t.join()
self.logger.info("exit Ostro")
LOG.info("exit Ostro")
def stop_ostro(self):
"""Stop main engine process."""
@ -131,22 +128,22 @@ class Ostro(object):
def bootstrap(self):
"""Start bootstrap and update the engine's resource topology."""
self.logger.info("Ostro.bootstrap: start bootstrap")
LOG.info("Ostro.bootstrap: start bootstrap")
try:
resource_status = self.db.get_resource_status(
self.resource.datacenter.name)
if resource_status is None:
self.logger.error("failed to read from table: %s" %
self.config.db_resource_table)
LOG.error("failed to read from table: %s" %
self.config.db_resource_table)
return False
if len(resource_status) > 0:
self.logger.info("bootstrap from DB")
LOG.info("bootstrap from DB")
if not self.resource.bootstrap_from_db(resource_status):
self.logger.error("failed to parse bootstrap data!")
LOG.error("failed to parse bootstrap data!")
self.logger.info("bootstrap from OpenStack")
LOG.info("bootstrap from OpenStack")
if not self._set_hosts():
return False
@ -159,42 +156,42 @@ class Ostro(object):
self.resource.update_topology()
except Exception:
self.logger.critical("Ostro.bootstrap failed: %s" %
traceback.format_exc())
LOG.critical("Ostro.bootstrap failed: ",
traceback.format_exc())
self.logger.info("done bootstrap")
LOG.info("done bootstrap")
return True
def _set_topology(self):
if not self.topology.set_topology():
self.logger.error("failed to read datacenter topology")
LOG.error("failed to read datacenter topology")
return False
self.logger.info("done topology bootstrap")
LOG.info("done topology bootstrap")
return True
def _set_hosts(self):
if not self.compute.set_hosts():
self.logger.error("failed to read hosts from OpenStack (Nova)")
LOG.error("failed to read hosts from OpenStack (Nova)")
return False
self.logger.info("done hosts & groups bootstrap")
LOG.info("done hosts & groups bootstrap")
return True
def _set_flavors(self):
if not self.compute.set_flavors():
self.logger.error("failed to read flavors from OpenStack (Nova)")
LOG.error("failed to read flavors from OpenStack (Nova)")
return False
self.logger.info("done flavors bootstrap")
LOG.info("done flavors bootstrap")
return True
# TODO(GJ): evaluate delay
def place_app(self, _app_data):
for req in _app_data:
if req["action"] == "query":
self.logger.info("start query")
LOG.info("start query")
query_result = self._query(req)
result = self._get_json_results("query", "ok",
@ -203,9 +200,9 @@ class Ostro(object):
if not self.db.put_result(result):
return False
self.logger.info("done query")
LOG.info("done query")
else:
self.logger.info("start app placement")
LOG.info("start app placement")
result = None
(decision_key, old_decision) = self.app_handler.check_history(
@ -221,14 +218,13 @@ class Ostro(object):
if decision_key is not None:
self.app_handler.put_history(decision_key, result)
else:
self.logger.warn("decision(%s) already made" %
decision_key)
LOG.info("decision(%s) already made" % decision_key)
result = old_decision
if not self.db.put_result(result):
return False
self.logger.info("done app placement")
LOG.info("done app placement")
return True
@ -247,11 +243,11 @@ class Ostro(object):
query_result[_q["stack_id"]] = vm_list
else:
self.status = "unknown paramenter in query"
self.logger.warn("unknown paramenter in query")
LOG.warning("unknown paramenter in query")
query_result[_q["stack_id"]] = None
else:
self.status = "no paramenter in query"
self.logger.warn("no parameters in query")
LOG.warning("no parameters in query")
query_result[_q["stack_id"]] = None
elif _q["type"] == "all_groups":
self.data_lock.acquire()
@ -259,11 +255,11 @@ class Ostro(object):
self.data_lock.release()
else:
self.status = "unknown query type"
self.logger.warn("unknown query type")
LOG.warning("unknown query type")
query_result[_q["stack_id"]] = None
else:
self.status = "unknown type in query"
self.logger.warn("no type in query")
LOG.warning("no type in query")
query_result[_q["stack_id"]] = None
return query_result
@ -284,7 +280,7 @@ class Ostro(object):
if vm_id[2] != "none": # if physical_uuid != 'none'
vm_list.append(vm_id[2])
else:
self.logger.warn("found pending vms in this group while query")
LOG.warning("found pending vms in this group while query")
return vm_list
@ -301,20 +297,20 @@ class Ostro(object):
app_topology = self.app_handler.add_app(_app)
if app_topology is None:
self.status = self.app_handler.status
self.logger.error("Ostro._place_app: error while register"
"requested apps: " + self.app_handler.status)
LOG.error("Ostro._place_app: error while register"
"requested apps: " + self.app_handler.status)
return None
"""Check and set vm flavor information."""
for _, vm in app_topology.vms.iteritems():
if self._set_vm_flavor_information(vm) is False:
self.status = "fail to set flavor information"
self.logger.error(self.status)
LOG.error(self.status)
return None
for _, vg in app_topology.vgroups.iteritems():
if self._set_vm_flavor_information(vg) is False:
self.status = "fail to set flavor information in a group"
self.logger.error(self.status)
LOG.error(self.status)
return None
self.data_lock.acquire()
@ -359,8 +355,8 @@ class Ostro(object):
flavor = self.resource.get_flavor(_vm.flavor)
if flavor is None:
self.logger.warn("Ostro._set_vm_flavor_properties: does not exist "
"flavor (" + _vm.flavor + ") and try to refetch")
LOG.warning("Ostro._set_vm_flavor_properties: does not exist "
"flavor (" + _vm.flavor + ") and try to refetch")
# Reset flavor resource and try again
if self._set_flavors() is False:
@ -395,14 +391,14 @@ class Ostro(object):
for e in _event_list:
if e.host is not None and e.host != "none":
if self._check_host(e.host) is False:
self.logger.warn("Ostro.handle_events: host (" + e.host +
") related to this event not exists")
LOG.warning("Ostro.handle_events: host (" + e.host +
") related to this event not exists")
continue
if e.method == "build_and_run_instance":
# VM is created (from stack)
self.logger.info("Ostro.handle_events: got build_and_run "
"event for %s" % e.uuid)
LOG.info("Ostro.handle_events: got build_and_run "
"event for %s" % e.uuid)
if self.db.put_uuid(e) is False:
self.data_lock.release()
return False
@ -417,22 +413,22 @@ class Ostro(object):
return False
if e.vm_state == "active":
self.logger.info("Ostro.handle_events: got instance_"
"active event for " + e.uuid)
LOG.info("Ostro.handle_events: got instance_"
"active event for " + e.uuid)
vm_info = self.app_handler.get_vm_info(
orch_id[1], orch_id[0], e.host)
if vm_info is None:
self.logger.error("Ostro.handle_events: error "
"while getting app info "
"from MUSIC")
LOG.error("Ostro.handle_events: error "
"while getting app info "
"from MUSIC")
self.data_lock.release()
return False
if len(vm_info) == 0:
# Stack not found because vm is created by the
# other stack
self.logger.warn("EVENT: no vm_info found in app "
"placement record")
LOG.warning("EVENT: no vm_info found in app "
"placement record")
self._add_vm_to_host(
e.uuid, orch_id[0], e.host, e.vcpus,
e.mem, e.local_disk)
@ -440,8 +436,8 @@ class Ostro(object):
if ("planned_host" in vm_info.keys() and
vm_info["planned_host"] != e.host):
# VM is activated in the different host
self.logger.warn("EVENT: vm activated in the "
"different host")
LOG.warning("EVENT: vm activated in the "
"different host")
self._add_vm_to_host(
e.uuid, orch_id[0], e.host, e.vcpus,
e.mem, e.local_disk)
@ -461,8 +457,8 @@ class Ostro(object):
# Possibly the vm deleted in the host while
# batch cleanup
if not self._check_h_uuid(orch_id[0], e.host):
self.logger.warn("EVENT: planned vm was "
"deleted")
LOG.warning("EVENT: planned vm was "
"deleted")
if self._check_uuid(e.uuid, e.host):
self._update_h_uuid_in_host(orch_id[0],
e.uuid,
@ -470,7 +466,7 @@ class Ostro(object):
self._update_h_uuid_in_logical_groups(
orch_id[0], e.uuid, e.host)
else:
self.logger.info(
LOG.info(
"EVENT: vm activated as planned")
self._update_uuid_in_host(
orch_id[0], e.uuid, e.host)
@ -480,8 +476,8 @@ class Ostro(object):
resource_updated = True
elif e.vm_state == "deleted":
self.logger.info("EVENT: got instance_delete for %s" %
e.uuid)
LOG.info("EVENT: got instance_delete for %s" %
e.uuid)
self._remove_vm_from_host(
e.uuid, orch_id[0], e.host, e.vcpus,
@ -491,24 +487,24 @@ class Ostro(object):
if not self.app_handler.update_vm_info(
orch_id[1], orch_id[0]):
self.logger.error("EVENT: error while updating "
"app in MUSIC")
LOG.error("EVENT: error while updating "
"app in MUSIC")
self.data_lock.release()
return False
resource_updated = True
else:
self.logger.warn("Ostro.handle_events: unknown vm_"
"state = " + e.vm_state)
LOG.warning("Ostro.handle_events: unknown vm_"
"state = " + e.vm_state)
elif e.object_name == 'ComputeNode':
# Host resource is updated
self.logger.debug("Ostro.handle_events: got compute event")
LOG.debug("Ostro.handle_events: got compute event")
elif e.object_name == 'ComputeNode':
# Host resource is updated
self.logger.info("EVENT: got compute for " + e.host)
LOG.info("EVENT: got compute for " + e.host)
# NOTE: what if host is disabled?
if self.resource.update_host_resources(
e.host, e.status, e.vcpus, e.vcpus_used, e.mem,
@ -519,11 +515,11 @@ class Ostro(object):
resource_updated = True
else:
self.logger.warn("Ostro.handle_events: unknown object_"
"name = " + e.object_name)
LOG.warning("Ostro.handle_events: unknown object_"
"name = " + e.object_name)
else:
self.logger.warn("Ostro.handle_events: unknown event "
"method = " + e.method)
LOG.warning("Ostro.handle_events: unknown event "
"method = " + e.method)
if resource_updated is True:
self.resource.update_topology(store=False)
@ -577,7 +573,7 @@ class Ostro(object):
_local_disk)
self.resource.update_host_time(_host_name)
else:
self.logger.warn("vm (%s) is missing while removing" % _uuid)
LOG.warning("vm (%s) is missing while removing" % _uuid)
def _remove_vm_from_logical_groups(self, _uuid, _h_uuid, _host_name):
host = self.resource.hosts[_host_name]
@ -618,8 +614,8 @@ class Ostro(object):
if host.update_uuid(_h_uuid, _uuid) is True:
self.resource.update_host_time(_host_name)
else:
self.logger.warn("Ostro._update_uuid_in_host: fail to update uuid "
"in host = %s" % host.name)
LOG.warning("Ostro._update_uuid_in_host: fail to update uuid "
"in host = %s" % host.name)
def _update_h_uuid_in_host(self, _h_uuid, _uuid, _host_name):
host = self.resource.hosts[_host_name]

57
valet/engine/optimizer/ostro/search.py
View File

@ -15,6 +15,8 @@
import copy
import operator
from oslo_log import log
from valet.engine.optimizer.app_manager.app_topology_base import LEVELS
from valet.engine.optimizer.app_manager.app_topology_base import VGroup
from valet.engine.optimizer.app_manager.app_topology_base import VM
@ -24,13 +26,14 @@ from valet.engine.optimizer.ostro.search_base import Node
from valet.engine.optimizer.ostro.search_base import Resource
from valet.engine.resource_manager.resource_base import Datacenter
LOG = log.getLogger(__name__)
class Search(object):
'''A bin-packing with maximal consolidation approach '''
def __init__(self, _logger):
def __init__(self):
"""Initialization."""
self.logger = _logger
# search inputs
self.resource = None
@ -92,9 +95,9 @@ class Search(object):
self.resource = _resource
self.constraint_solver = ConstraintSolver(self.logger)
self.constraint_solver = ConstraintSolver()
self.logger.info("start search")
LOG.info("start search")
self._create_avail_logical_groups()
self._create_avail_hosts()
@ -118,9 +121,9 @@ class Search(object):
self.app_topology = _app_topology
self.resource = _resource
self.constraint_solver = ConstraintSolver(self.logger)
self.constraint_solver = ConstraintSolver()
self.logger.info("start search for replan")
LOG.info("start search for replan")
self._create_avail_logical_groups()
self._create_avail_hosts()
@ -130,7 +133,7 @@ class Search(object):
self._compute_resource_weights()
self.logger.info("first, place already-planned nodes")
LOG.info("first, place already-planned nodes")
# reconsider all vms to be migrated together
if len(_app_topology.exclusion_list_map) > 0:
@ -138,17 +141,17 @@ class Search(object):
if self._place_planned_nodes() is False:
self.status = "cannot replan VMs that was planned"
self.logger.error(self.status)
LOG.error(self.status)
return False
self.logger.info("second, re-place not-planned nodes")
LOG.info("second, re-place not-planned nodes")
init_level = LEVELS[len(LEVELS) - 1]
(open_node_list, level) = self._open_list(self.app_topology.vms,
self.app_topology.vgroups,
init_level)
if open_node_list is None:
self.logger.error("fail to replan")
LOG.error("fail to replan")
return False
for v, ah in self.planned_placements.iteritems():
@ -169,8 +172,8 @@ class Search(object):
if vk in self.app_topology.planned_vm_map.keys():
del self.app_topology.planned_vm_map[vk]
else:
self.logger.error("Search: migrated " + migrated_vm_id +
" is missing while replan")
LOG.error("Search: migrated " + migrated_vm_id +
" is missing while replan")
def _get_child_vms(self, _g, _vm_list, _e_vmk):
for sgk, sg in _g.subvgroups.iteritems():
@ -212,8 +215,8 @@ class Search(object):
vgroup.host = []
host_name = self._get_host_of_vgroup(hk, vgroup.level)
if host_name is None:
self.logger.error("Search: host does not exist while "
"replan with vgroup")
LOG.warning("Search: host does not exist while "
"replan with vgroup")
else:
if host_name not in vgroup.host:
vgroup.host.append(host_name)
@ -301,7 +304,7 @@ class Search(object):
self._deduct_reservation(_level, best_resource, n)
self._close_planned_placement(_level, best_resource, n.node)
else:
self.logger.error("fail to place already-planned VMs")
LOG.error("fail to place already-planned VMs")
return False
return True
@ -332,7 +335,7 @@ class Search(object):
host_name = self._get_host_of_level(_n, _level)
if host_name is None:
self.logger.warn("cannot find host while replanning")
LOG.warning("cannot find host while replanning")
return None
avail_hosts = {}
@ -385,8 +388,8 @@ class Search(object):
for hk, host in self.resource.hosts.iteritems():
if host.check_availability() is False:
self.logger.debug("Search: host (" + host.name +
") not available at this time")
LOG.debug("Search: host (" + host.name +
") not available at this time")
continue
r = Resource()
@ -460,7 +463,7 @@ class Search(object):
for lgk, lg in self.resource.logical_groups.iteritems():
if lg.status != "enabled":
self.logger.warn("group (" + lg.name + ") disabled")
LOG.warning("group (" + lg.name + ") disabled")
continue
lgr = LogicalGroupResource()
@ -715,8 +718,8 @@ class Search(object):
if host_name not in host_list:
host_list.append(host_name)
else:
self.logger.warn("Search: cannot find candidate "
"host while replanning")
LOG.warning("Search: cannot find candidate "
"host while replanning")
_n.node.host = host_list
candidate_list = []
@ -779,7 +782,7 @@ class Search(object):
else:
debug_candidate_name = cr.get_resource_name(_level)
msg = "rollback of candidate resource = {0}"
self.logger.warn(msg.format(debug_candidate_name))
LOG.warning(msg.format(debug_candidate_name))
if planned_host is None:
# recursively rollback deductions of all
@ -792,7 +795,7 @@ class Search(object):
if best_resource is None and len(candidate_list) == 0:
self.status = "no available hosts"
self.logger.warn(self.status)
LOG.warning(self.status)
return best_resource
@ -858,7 +861,7 @@ class Search(object):
lgr.group_type = "EX"
self.avail_logical_groups[lgr.name] = lgr
self.logger.info(
LOG.info(
"Search: add new exclusivity (%s)" % _exclusivity_id)
else:
@ -913,7 +916,7 @@ class Search(object):
lgr.group_type = "AFF"
self.avail_logical_groups[lgr.name] = lgr
self.logger.info("add new affinity (" + _affinity_id + ")")
LOG.info("add new affinity (" + _affinity_id + ")")
else:
lgr = self.avail_logical_groups[_affinity_id]
@ -963,8 +966,8 @@ class Search(object):
lgr.group_type = "DIV"
self.avail_logical_groups[lgr.name] = lgr
self.logger.info(
"Search: add new diversity (%s)" % _diversity_id)
LOG.info(
"Search: add new diversity (%s)", _diversity_id)
else:
lgr = self.avail_logical_groups[_diversity_id]

24
valet/engine/optimizer/ostro_server/daemon.py
View File

@ -17,12 +17,15 @@
import atexit
import os
from oslo_config import cfg
from signal import SIGTERM
import signal
import sys
import time
from oslo_config import cfg
from oslo_log import log
CONF = cfg.CONF
LOG = log.getLogger(__name__)
class Daemon(object):
@ -31,7 +34,7 @@ class Daemon(object):
"""Usage: subclass the Daemon class and override the run() method
"""
def __init__(self, priority, pidfile, logger, stdin='/dev/null',
def __init__(self, priority, pidfile, stdin='/dev/null',
stdout='/dev/null', stderr='/dev/null'):
"""Initialization."""
self.stdin = stdin
@ -39,7 +42,6 @@ class Daemon(object):
self.stderr = stderr
self.pidfile = pidfile
self.priority = priority
self.logger = logger
def daemonize(self):
"""Do the UNIX double-fork magic."""
@ -53,9 +55,9 @@ class Daemon(object):
# exit first parent
sys.exit(0)
except OSError as e:
self.logger.error("Daemon error at step1: " + e.strerror)
sys.stderr.write("fork #1 failed: %d (%s)\n" %
(e.errno, e.strerror))
LOG.error("Daemon error at step1: ", e.strerror)
LOG.error("fork #1 failed: %d (%s)\n",
e.errno, e.strerror)
sys.exit(1)
# decouple from parent environment
@ -70,9 +72,9 @@ class Daemon(object):
# exit from second parent
sys.exit(0)
except OSError as e:
self.logger.error("Daemon error at step2: " + e.strerror)
sys.stderr.write("fork #2 failed: %d (%s)\n" %
(e.errno, e.strerror))
LOG.error("Daemon error at step2: ", e.strerror)
LOG.error("fork #2 failed: %d (%s)\n",
e.errno, e.strerror)
sys.exit(1)
# redirect standard file descriptors
@ -140,7 +142,7 @@ class Daemon(object):
# Try killing the daemon process
try:
while 1:
os.kill(pid, SIGTERM)
os.kill(pid, signal.SIGTERM)
time.sleep(0.1)
except OSError as err:
err = str(err)

16
valet/engine/optimizer/ostro_server/health_checker.py
View File

@ -26,6 +26,7 @@ from valet.common.music import REST
from valet.engine.conf import init_engine
CONF = cfg.CONF
LOG = get_logger(__name__)
class HealthCheck(object):
@ -107,7 +108,7 @@ class HealthCheck(object):
engine_id = placement['resources']['id']
break
except Exception as e:
logger.warn("HealthCheck exception in read response " + str(e))
LOG.warning("HealthCheck exception in read response, ", str(e))
return engine_id
@ -126,7 +127,7 @@ class HealthCheck(object):
}
self.rest.request(method='delete', path=path, data=data)
except Exception as e:
logger.warn("HealthCheck exception in delete request - " + str(e))
LOG.warning("HealthCheck exception in delete request, ", str(e))
try:
path = base % {
@ -136,7 +137,7 @@ class HealthCheck(object):
}
self.rest.request(method='delete', path=path, data=data)
except Exception as e:
logger.warn("HealthCheck exception in delete response - " + str(e))
LOG.warning("HealthCheck exception in delete response, ", str(e))
if __name__ == "__main__":
@ -144,20 +145,19 @@ if __name__ == "__main__":
respondent_id = None
code = 0
init_engine(default_config_files=['/etc/valet/valet.conf'])
logger = get_logger("ostro_daemon")
if os.path.exists(CONF.engine.pid):
respondent_id = HealthCheck().ping()
if respondent_id == CONF.engine.priority:
code = CONF.engine.priority
logger.info("HealthCheck - Alive, "
"respondent instance id: {}".format(respondent_id))
LOG.info("HealthCheck - Alive, "
"respondent instance id: {}".format(respondent_id))
else:
logger.warn("HealthCheck - pid file exists, "
LOG.warning("HealthCheck - pid file exists, "
"engine {} did not respond in a timely manner "
"(respondent id {})".format(CONF.engine.priority,
respondent_id))
else:
logger.info("HealthCheck - no pid file, engine is not running!")
LOG.info("HealthCheck - no pid file, engine is not running!")
sys.exit(code)

29
valet/engine/resource_manager/compute.py
View File

@ -16,6 +16,7 @@ import traceback
from novaclient import client as nova_client
from oslo_config import cfg
from oslo_log import log
from resource_base import Flavor
from resource_base import Host
@ -26,6 +27,7 @@ from resource_base import LogicalGroup
VERSION = 2
CONF = cfg.CONF
LOG = log.getLogger(__name__)
class Compute(object):
@ -37,9 +39,8 @@ class Compute(object):
Interacts with nova client to perform these actions.
"""
def __init__(self, _logger):
def __init__(self):
"""Compute init."""
self.logger = _logger
self.nova = None
def set_hosts(self, _hosts, _logical_groups):
@ -48,22 +49,22 @@ class Compute(object):
status = self._set_availability_zones(_hosts, _logical_groups)
if status != "success":
self.logger.error('_set_availability_zones failed')
LOG.error('_set_availability_zones failed')
return status
status = self._set_aggregates(_hosts, _logical_groups)
if status != "success":
self.logger.error('_set_aggregates failed')
LOG.error('_set_aggregates failed')
return status
status = self._set_placed_vms(_hosts, _logical_groups)
if status != "success":
self.logger.error('_set_placed_vms failed')
LOG.error('_set_placed_vms failed')
return status
status = self._set_resources(_hosts)
if status != "success":
self.logger.error('_set_resources failed')
LOG.error('_set_resources failed')
return status
return "success"
@ -102,11 +103,11 @@ class Compute(object):
_hosts[host.name] = host
except (ValueError, KeyError, TypeError):
self.logger.error(traceback.format_exc())
LOG.error(traceback.format_exc())
return "Error while setting host zones from Nova"
except Exception:
self.logger.critical(traceback.format_exc())
LOG.critical(traceback.format_exc())
return "success"
@ -134,7 +135,7 @@ class Compute(object):
aggregate.vms_per_host[host.name] = []
except (ValueError, KeyError, TypeError):
self.logger.error(traceback.format_exc())
LOG.error(traceback.format_exc())
return "Error while setting host aggregates from Nova"
return "success"
@ -182,7 +183,7 @@ class Compute(object):
_vm_list.append(s['uuid'])
except (ValueError, KeyError, TypeError):
self.logger.error(traceback.format_exc())
LOG.error(traceback.format_exc())
return "Error while getting existing vms"
return "success"
@ -201,7 +202,7 @@ class Compute(object):
_vm_detail.append(status)
except (ValueError, KeyError, TypeError):
self.logger.error(traceback.format_exc())
LOG.error(traceback.format_exc())
return "Error while getting vm detail"
return "success"
@ -226,7 +227,7 @@ class Compute(object):
host.disk_available_least = float(hv.disk_available_least)
except (ValueError, KeyError, TypeError):
self.logger.error(traceback.format_exc())
LOG.error(traceback.format_exc())
return "Error while setting host resources from Nova"
return "success"
@ -287,7 +288,7 @@ class Compute(object):
_flavors[flavor.name] = flavor
except (ValueError, KeyError, TypeError):
self.logger.error(traceback.format_exc())
LOG.error(traceback.format_exc())
return "Error while getting flavors"
return "success"
@ -308,7 +309,7 @@ class Compute(object):
break
except (ValueError, KeyError, TypeError):
self.logger.error(traceback.format_exc())
LOG.error(traceback.format_exc())
return "Error while getting flavor extra spec"
return "success"

118
valet/engine/resource_manager/compute_manager.py
View File

@ -15,13 +15,17 @@
"""Compute Manager."""
from copy import deepcopy
import threading
import time
from copy import deepcopy
from oslo_log import log
from valet.engine.resource_manager.compute import Compute
from valet.engine.resource_manager.resource_base import Host
LOG = log.getLogger(__name__)
class ComputeManager(threading.Thread):
"""Compute Manager Class.
@ -30,7 +34,7 @@ class ComputeManager(threading.Thread):
flavors, etc. Calls many functions from Resource.
"""
def __init__(self, _t_id, _t_name, _rsc, _data_lock, _config, _logger):
def __init__(self, _t_id, _t_name, _rsc, _data_lock, _config):
"""Init Compute Manager."""
threading.Thread.__init__(self)
@ -43,8 +47,6 @@ class ComputeManager(threading.Thread):
self.config = _config
self.logger = _logger
self.admin_token = None
self.project_token = None
@ -52,8 +54,8 @@ class ComputeManager(threading.Thread):
def run(self):
"""Start Compute Manager thread to run setup."""
self.logger.info("ComputeManager: start " + self.thread_name +
" ......")
LOG.info("ComputeManager: start " + self.thread_name +
" ......")
if self.config.compute_trigger_freq > 0:
period_end = time.time() + self.config.compute_trigger_freq
@ -71,21 +73,21 @@ class ComputeManager(threading.Thread):
self.config.compute_trigger_freq)
# NOTE(GJ): do not timer based batch
self.logger.info("exit compute_manager " + self.thread_name)
LOG.info("exit compute_manager " + self.thread_name)
def _run(self):
self.logger.info("ComputeManager: --- start compute_nodes "
"status update ---")
LOG.info("ComputeManager: --- start compute_nodes "
"status update ---")
triggered_host_updates = self.set_hosts()
if triggered_host_updates is not True:
self.logger.warn("fail to set hosts from nova")
LOG.warning("fail to set hosts from nova")
triggered_flavor_updates = self.set_flavors()
if triggered_flavor_updates is not True:
self.logger.warn("fail to set flavor from nova")
LOG.warning("fail to set flavor from nova")
self.logger.info("ComputeManager: --- done compute_nodes "
"status update ---")
LOG.info("ComputeManager: --- done compute_nodes "
"status update ---")
return True
@ -94,7 +96,7 @@ class ComputeManager(threading.Thread):
hosts = {}
logical_groups = {}
compute = Compute(self.logger)
compute = Compute()
status = compute.set_hosts(hosts, logical_groups)
if status != "success":
@ -125,8 +127,8 @@ class ComputeManager(threading.Thread):
_logical_groups[lk])
self.resource.logical_groups[lk].last_update = time.time()
self.logger.warn("ComputeManager: new logical group (" +
lk + ") added")
LOG.warning("ComputeManager: new logical group (" +
lk + ") added")
updated = True
for rlk in self.resource.logical_groups.keys():
@ -137,8 +139,8 @@ class ComputeManager(threading.Thread):
self.resource.logical_groups[rlk].status = "disabled"
self.resource.logical_groups[rlk].last_update = time.time()
self.logger.warn("ComputeManager: logical group (" +
rlk + ") removed")
LOG.warning("ComputeManager: logical group (" +
rlk + ") removed")
updated = True
for lk in _logical_groups.keys():
@ -149,8 +151,8 @@ class ComputeManager(threading.Thread):
if self._check_logical_group_metadata_update(lg, rlg) is True:
rlg.last_update = time.time()
self.logger.warn("ComputeManager: logical group (" +
lk + ") updated")
LOG.warning("ComputeManager: logical group (" +
lk + ") updated")
updated = True
return updated
@ -193,8 +195,8 @@ class ComputeManager(threading.Thread):
self.resource.hosts[new_host.name] = new_host
new_host.last_update = time.time()
self.logger.warn("ComputeManager: new host (" +
new_host.name + ") added")
LOG.warning("ComputeManager: new host (" +
new_host.name + ") added")
updated = True
for rhk, rhost in self.resource.hosts.iteritems():
@ -203,8 +205,8 @@ class ComputeManager(threading.Thread):
rhost.tag.remove("nova")
rhost.last_update = time.time()
self.logger.warn("ComputeManager: host (" +
rhost.name + ") disabled")
LOG.warning("ComputeManager: host (" +
rhost.name + ") disabled")
updated = True
for hk in _hosts.keys():
@ -217,8 +219,8 @@ class ComputeManager(threading.Thread):
for hk, h in self.resource.hosts.iteritems():
if h.clean_memberships() is True:
h.last_update = time.time()
self.logger.warn("ComputeManager: host (" + h.name +
") updated (delete EX/AFF/DIV membership)")
LOG.warning("ComputeManager: host (" + h.name +
") updated (delete EX/AFF/DIV membership)")
updated = True
for hk, host in self.resource.hosts.iteritems():
@ -247,20 +249,20 @@ class ComputeManager(threading.Thread):
if "nova" not in _rhost.tag:
_rhost.tag.append("nova")
topology_updated = True
self.logger.warn("ComputeManager: host (" + _rhost.name +
") updated (tag added)")
LOG.warning("ComputeManager: host (" + _rhost.name +
") updated (tag added)")
if _host.status != _rhost.status:
_rhost.status = _host.status
topology_updated = True
self.logger.warn("ComputeManager: host (" + _rhost.name +
") updated (status changed)")
LOG.warning("ComputeManager: host (" + _rhost.name +
") updated (status changed)")
if _host.state != _rhost.state:
_rhost.state = _host.state
topology_updated = True
self.logger.warn("ComputeManager: host (" + _rhost.name +
") updated (state changed)")
LOG.warning("ComputeManager: host (" + _rhost.name +
") updated (state changed)")
return topology_updated
@ -274,8 +276,8 @@ class ComputeManager(threading.Thread):
_rhost.original_vCPUs = _host.original_vCPUs
_rhost.avail_vCPUs = _host.avail_vCPUs
topology_updated = True
self.logger.warn("ComputeManager: host (" + _rhost.name +
") updated (CPU updated)")
LOG.warning("ComputeManager: host (" + _rhost.name +
") updated (CPU updated)")
if _host.mem_cap != _rhost.mem_cap or \
_host.original_mem_cap != _rhost.original_mem_cap or \
@ -284,8 +286,8 @@ class ComputeManager(threading.Thread):
_rhost.original_mem_cap = _host.original_mem_cap
_rhost.avail_mem_cap = _host.avail_mem_cap
topology_updated = True
self.logger.warn("ComputeManager: host (" + _rhost.name +
") updated (mem updated)")
LOG.warning("ComputeManager: host (" + _rhost.name +
") updated (mem updated)")
if _host.local_disk_cap != _rhost.local_disk_cap or \
_host.original_local_disk_cap != _rhost.original_local_disk_cap or \
@ -294,8 +296,8 @@ class ComputeManager(threading.Thread):
_rhost.original_local_disk_cap = _host.original_local_disk_cap
_rhost.avail_local_disk_cap = _host.avail_local_disk_cap
topology_updated = True
self.logger.warn("ComputeManager: host (" + _rhost.name +
") updated (local disk space updated)")
LOG.warning("ComputeManager: host (" + _rhost.name +
") updated (local disk space updated)")
if _host.vCPUs_used != _rhost.vCPUs_used or \
_host.free_mem_mb != _rhost.free_mem_mb or \
@ -306,8 +308,8 @@ class ComputeManager(threading.Thread):
_rhost.free_disk_gb = _host.free_disk_gb
_rhost.disk_available_least = _host.disk_available_least
topology_updated = True
self.logger.warn("ComputeManager: host (" + _rhost.name +
") updated (other resource numbers)")
LOG.warning("ComputeManager: host (" + _rhost.name +
") updated (other resource numbers)")
return topology_updated
@ -318,8 +320,8 @@ class ComputeManager(threading.Thread):
if mk not in _rhost.memberships.keys():
_rhost.memberships[mk] = self.resource.logical_groups[mk]
topology_updated = True
self.logger.warn("ComputeManager: host (" + _rhost.name +
") updated (new membership)")
LOG.warning("ComputeManager: host (" + _rhost.name +
") updated (new membership)")
for mk in _rhost.memberships.keys():
m = _rhost.memberships[mk]
@ -328,8 +330,8 @@ class ComputeManager(threading.Thread):
if mk not in _host.memberships.keys():
del _rhost.memberships[mk]
topology_updated = True
self.logger.warn("ComputeManager: host (" + _rhost.name +
") updated (delete membership)")
LOG.warning("ComputeManager: host (" + _rhost.name +
") updated (delete membership)")
return topology_updated
@ -343,7 +345,7 @@ class ComputeManager(threading.Thread):
if alen != blen:
topology_updated = True
msg = "host ({0}) {1} none vms removed"
self.logger.warn(msg.format(_rhost.name, str(blen - alen)))
LOG.warning(msg.format(_rhost.name, str(blen - alen)))
self.resource.clean_none_vms_from_logical_groups(_rhost)
@ -351,16 +353,16 @@ class ComputeManager(threading.Thread):
if _rhost.exist_vm_by_uuid(vm_id[2]) is False:
_rhost.vm_list.append(vm_id)
topology_updated = True
self.logger.warn("ComputeManager: host (" + _rhost.name +
") updated (new vm placed)")
LOG.warning("ComputeManager: host (" + _rhost.name +
") updated (new vm placed)")
for rvm_id in _rhost.vm_list:
if _host.exist_vm_by_uuid(rvm_id[2]) is False:
self.resource.remove_vm_by_uuid_from_logical_groups(
_rhost, rvm_id[2])
topology_updated = True
self.logger.warn("ComputeManager: host (" + _rhost.name +
") updated (vm removed)")
LOG.warning("ComputeManager: host (" + _rhost.name +
") updated (vm removed)")
blen = len(_rhost.vm_list)
_rhost.vm_list = [
@ -369,7 +371,7 @@ class ComputeManager(threading.Thread):
if alen != blen:
topology_updated = True
msg = "host ({0}) {1} vms removed"
self.logger.warn(msg.format(_rhost.name, str(blen - alen)))
LOG.warning(msg.format(_rhost.name, str(blen - alen)))
return topology_updated
@ -377,11 +379,11 @@ class ComputeManager(threading.Thread):
"""Return True if compute set flavors returns success."""
flavors = {}
compute = Compute(self.logger)
compute = Compute()
status = compute.set_flavors(flavors)
if status != "success":
self.logger.error(status)
LOG.error(status)
return False
self.data_lock.acquire()
@ -399,8 +401,8 @@ class ComputeManager(threading.Thread):
self.resource.flavors[fk] = deepcopy(_flavors[fk])
self.resource.flavors[fk].last_update = time.time()
self.logger.warn("ComputeManager: new flavor (" +
fk + ":" + _flavors[fk].flavor_id + ") added")
LOG.warning("ComputeManager: new flavor (" +
fk + ":" + _flavors[fk].flavor_id + ") added")
updated = True
for rfk in self.resource.flavors.keys():
@ -409,8 +411,8 @@ class ComputeManager(threading.Thread):
rf.status = "disabled"
rf.last_update = time.time()
self.logger.warn("ComputeManager: flavor (" + rfk + ":" +
rf.flavor_id + ") removed")
LOG.warning("ComputeManager: flavor (" + rfk + ":" +
rf.flavor_id + ") removed")
updated = True
for fk in _flavors.keys():
@ -419,8 +421,8 @@ class ComputeManager(threading.Thread):
if self._check_flavor_spec_update(f, rf) is True:
rf.last_update = time.time()
self.logger.warn("ComputeManager: flavor (" + fk + ":" +
rf.flavor_id + ") spec updated")
LOG.warning("ComputeManager: flavor (" + fk + ":" +
rf.flavor_id + ") spec updated")
updated = True
return updated

86
valet/engine/resource_manager/resource.py
View File

@ -15,6 +15,8 @@
import time
import traceback
from oslo_log import log
from valet.engine.optimizer.app_manager.app_topology_base import LEVELS
from valet.engine.resource_manager.resource_base import Datacenter
from valet.engine.resource_manager.resource_base import Flavor
@ -22,6 +24,8 @@ from valet.engine.resource_manager.resource_base import Host
from valet.engine.resource_manager.resource_base import HostGroup
from valet.engine.resource_manager.resource_base import LogicalGroup
LOG = log.getLogger(__name__)
class Resource(object):
"""Resource Class.
@ -32,12 +36,11 @@ class Resource(object):
updates to base resource types.
"""
def __init__(self, _db, _config, _logger):
def __init__(self, _db, _config):
"""Init Resource Class."""
self.db = _db
self.config = _config
self.logger = _logger
""" resource data """
self.datacenter = Datacenter(self.config.datacenter_name)
@ -64,7 +67,7 @@ class Resource(object):
def bootstrap_from_db(self, _resource_status):
"""Return True if bootsrap resource from database successful."""
try:
self.logger.info("Resource status from DB = %s", _resource_status)
LOG.info("Resource status from DB = %s", _resource_status)
logical_groups = _resource_status.get("logical_groups")
if logical_groups:
for lgk, lg in logical_groups.iteritems():
@ -78,7 +81,7 @@ class Resource(object):
self.logical_groups[lgk] = logical_group
if len(self.logical_groups) == 0:
self.logger.warn("no logical_groups")
LOG.warning("no logical_groups")
flavors = _resource_status.get("flavors")
if flavors:
@ -94,7 +97,7 @@ class Resource(object):
self.flavors[fk] = flavor
if len(self.flavors) == 0:
self.logger.error("fail loading flavors")
LOG.error("fail loading flavors")
hosts = _resource_status.get("hosts")
if hosts:
@ -124,7 +127,7 @@ class Resource(object):
self.hosts[hk] = host
if len(self.hosts) == 0:
self.logger.error("fail loading hosts")
LOG.error("fail loading hosts")
host_groups = _resource_status.get("host_groups")
if host_groups:
@ -151,7 +154,7 @@ class Resource(object):
self.host_groups[hgk] = host_group
if len(self.host_groups) == 0:
self.logger.warn("fail loading host_groups")
LOG.warning("fail loading host_groups")
dc = _resource_status.get("datacenter")
if dc:
@ -181,7 +184,7 @@ class Resource(object):
self.datacenter.resources[ck] = self.hosts[ck]
if len(self.datacenter.resources) == 0:
self.logger.error("fail loading datacenter")
LOG.error("fail loading datacenter")
hgs = _resource_status.get("host_groups")
if hgs:
@ -215,8 +218,8 @@ class Resource(object):
self._update_compute_avail()
except Exception:
self.logger.error("while bootstrap_from_db: ",
traceback.format_exc())
LOG.error("while bootstrap_from_db: ",
traceback.format_exc())
return True
@ -314,7 +317,7 @@ class Resource(object):
host_group_updates = {}
datacenter_update = None
self.logger.info("check and store resource status")
LOG.info("check and store resource status")
for fk, flavor in self.flavors.iteritems():
if flavor.last_update >= self.curr_db_timestamp:
@ -366,66 +369,66 @@ class Resource(object):
return True
def show_current_logical_groups(self):
self.logger.debug("--- track logical groups info ---")
LOG.debug("--- track logical groups info ---")
for lgk, lg in self.logical_groups.iteritems():
if lg.status == "enabled":
self.logger.debug("lg name = " + lgk)
self.logger.debug(" type = " + lg.group_type)
LOG.debug("lg name = " + lgk)
LOG.debug(" type = " + lg.group_type)
if lg.group_type == "AGGR":
for k in lg.metadata.keys():
self.logger.debug(" metadata key = " + k)
self.logger.debug(" vms")
LOG.debug(" metadata key = " + k)
LOG.debug(" vms")
debug_msg = " orch_id = %s uuid = %s"
for v in lg.vm_list:
self.logger.debug(debug_msg, v[0], v[2])
self.logger.debug(" hosts")
LOG.debug(debug_msg % (v[0], v[2]))
LOG.debug(" hosts")
for h, v in lg.vms_per_host.iteritems():
self.logger.debug(" host = %s", h)
self.logger.debug(" vms = %s",
str(len(lg.vms_per_host[h])))
LOG.debug(" host = %s" % h)
LOG.debug(" vms = %s" %
str(len(lg.vms_per_host[h])))
host = None
if h in self.hosts.keys():
host = self.hosts[h]
elif h in self.host_groups.keys():
host = self.host_groups[h]
else:
self.logger.error("TEST: lg member not exist")
LOG.error("TEST: lg member not exist")
if host is not None:
self.logger.debug(" status = " + host.status)
LOG.debug(" status = " + host.status)
if lgk not in host.memberships.keys():
self.logger.error("TEST: membership missing")
LOG.error("TEST: membership missing")
def show_current_host_status(self):
self.logger.debug("--- track host info ---")
LOG.debug("--- track host info ---")
for hk, h in self.hosts.iteritems():
self.logger.debug("host name = " + hk)
self.logger.debug(" status = " + h.status + ", " + h.state)
self.logger.debug(" vms = " + str(len(h.vm_list)))
self.logger.debug(" resources (org, total, avail, used)")
LOG.debug("host name = " + hk)
LOG.debug(" status = " + h.status + ", " + h.state)
LOG.debug(" vms = " + str(len(h.vm_list)))
LOG.debug(" resources (org, total, avail, used)")
cpu_org = str(h.original_vCPUs)
cpu_tot = str(h.vCPUs)
cpu_avail = str(h.avail_vCPUs)
cpu_used = str(h.vCPUs_used)
msg = " {0} = {1}, {2}, {3}, {4}"
self.logger.debug(
LOG.debug(
msg.format('cpu', cpu_org, cpu_tot, cpu_avail, cpu_used))
mem_org = str(h.original_mem_cap)
mem_tot = str(h.mem_cap)
mem_avail = str(h.avail_mem_cap)
mem_used = str(h.free_mem_mb)
self.logger.debug(
LOG.debug(
msg.format('mem', mem_org, mem_tot, mem_avail, mem_used))
dsk_org = str(h.original_local_disk_cap)
dsk_tot = str(h.local_disk_cap)
dsk_avail = str(h.avail_local_disk_cap)
dsk_used = str(h.free_disk_gb)
self.logger.debug(
LOG.debug(
msg.format('disk', dsk_org, dsk_tot, dsk_avail, dsk_used))
self.logger.debug(" memberships")
LOG.debug(" memberships")
for mk in h.memberships.keys():
self.logger.debug(" " + mk)
LOG.debug(" " + mk)
if mk not in self.logical_groups.keys():
self.logger.error("TEST: lg missing")
LOG.error("TEST: lg missing")
def update_rack_resource(self, _host):
"""Update resources for rack (host), then update cluster."""
@ -509,8 +512,9 @@ class Resource(object):
if host.status != _st:
host.status = _st
self.logger.info(
"Resource.update_host_resources: host(%s) status changed", _hn)
LOG.warning(
"Resource.update_host_resources: host(%s) status changed" %
_hn)
updated = True
# FIXME(GJ): should check cpu, memm and disk here?
@ -577,8 +581,8 @@ class Resource(object):
"""Remove vm by orchestration id from lgs. Update host and lgs."""
for lgk in _host.memberships.keys():
if lgk not in self.logical_groups.keys():
self.logger.warn("logical group (%s) missing while "
"removing %s", lgk, _h_uuid)
LOG.warning("logical group (%s) missing while "
"removing %s" % (lgk, _h_uuid))
continue
lg = self.logical_groups[lgk]
@ -617,8 +621,8 @@ class Resource(object):
"""Remove vm by uuid from lgs and update proper host and lgs."""
for lgk in _host.memberships.keys():
if lgk not in self.logical_groups.keys():
self.logger.warn("logical group (%s) missing while "
"removing %s", lgk, _uuid)
LOG.warning("logical group (%s) missing while "
"removing %s" % (lgk, _uuid))
continue
lg = self.logical_groups[lgk]

13
valet/engine/resource_manager/topology.py
View File

@ -16,10 +16,14 @@
"""Topology class - performs actual setting up of Topology object."""
import copy
from oslo_log import log
from sre_parse import isdigit
from valet.engine.resource_manager.resource_base import HostGroup
LOG = log.getLogger(__name__)
class Topology(object):
"""Topology class.
@ -27,10 +31,9 @@ class Topology(object):
currently, using cannonical naming convention to find the topology
"""
def __init__(self, _config, _logger):
def __init__(self, _config):
"""Init config and logger."""
self.config = _config
self.logger = _logger
# Triggered by rhosts change
def set_topology(self, _datacenter, _host_groups, _hosts, _rhosts):
@ -52,7 +55,7 @@ class Topology(object):
(region_name, rack_name, _, status) = self._set_layout_by_name(rhk)
if status != "success":
self.logger.warn(status + " in host_name (" + rhk + ")")
LOG.warning(status + " in host_name (" + rhk + ")")
if region_name not in _datacenter.region_code_list:
_datacenter.region_code_list.append(region_name)
@ -76,10 +79,10 @@ class Topology(object):
_datacenter.resources[hgk] = hg
if len(_datacenter.region_code_list) > 1:
self.logger.warn("more than one region code")
LOG.warning("more than one region code")
if "none" in _host_groups.keys():
self.logger.warn("some hosts are into unknown rack")
LOG.warning("some hosts are into unknown rack")
return "success"

81
valet/engine/resource_manager/topology_manager.py
View File

@ -15,17 +15,21 @@
import threading
import time
from oslo_log import log
from valet.engine.resource_manager.resource_base import Datacenter
from valet.engine.resource_manager.resource_base import Host
from valet.engine.resource_manager.resource_base import HostGroup
from valet.engine.resource_manager.topology import Topology
LOG = log.getLogger(__name__)
class TopologyManager(threading.Thread):
"""Topology Manager Class."""
def __init__(self, _t_id, _t_name, _resource,
_data_lock, _config, _logger):
_data_lock, _config):
"""Init Topology Manager."""
threading.Thread.__init__(self)
@ -37,14 +41,13 @@ class TopologyManager(threading.Thread):
self.resource = _resource
self.config = _config
self.logger = _logger
self.update_batch_wait = self.config.update_batch_wait
def run(self):
"""Function starts and tracks Topology Manager Thread."""
self.logger.info("TopologyManager: start " +
self.thread_name + " ......")
LOG.info("TopologyManager: start " +
self.thread_name + " ......")
if self.config.topology_trigger_freq > 0:
period_end = time.time() + self.config.topology_trigger_freq
@ -61,17 +64,17 @@ class TopologyManager(threading.Thread):
period_end = (curr_ts +
self.config.topology_trigger_freq)
# NOTE(GJ): do not timer based batch
self.logger.info("exit topology_manager " + self.thread_name)
LOG.info("exit topology_manager " + self.thread_name)
def _run(self):
self.logger.info("TopologyManager: --- start topology "
"status update ---")
LOG.info("TopologyManager: --- start topology "
"status update ---")
if self.set_topology() is not True:
self.logger.warn("fail to set topology")
LOG.warning("fail to set topology")
self.logger.info("--- done topology status update ---")
LOG.info("--- done topology status update ---")
def set_topology(self):
host_groups = {}
@ -80,7 +83,7 @@ class TopologyManager(threading.Thread):
# NOTE(GJ): do not consider switch topology at this version
datacenter = Datacenter(self.config.datacenter_name)
topology = Topology(self.config, self.logger)
topology = Topology(self.config)
status = topology.set_topology(datacenter, host_groups, hosts,
self.resource.hosts)
@ -104,8 +107,8 @@ class TopologyManager(threading.Thread):
new_host.last_update = time.time()
self.logger.info("TopologyManager: new host (" +
new_host.name + ") added from configuration")
LOG.warning("TopologyManager: new host (" +
new_host.name + ") added from configuration")
updated = True
for rhk in self.resource.hosts.keys():
@ -116,8 +119,8 @@ class TopologyManager(threading.Thread):
host.last_update = time.time()
self.logger.info("TopologyManager: host (" +
host.name + ") removed from configuration")
LOG.warning("TopologyManager: host (" +
host.name + ") removed from configuration")
updated = True
for hgk in _host_groups.keys():
@ -127,8 +130,8 @@ class TopologyManager(threading.Thread):
new_host_group.last_update = time.time()
self.logger.info("TopologyManager: new host_group (" +
new_host_group.name + ") added")
LOG.warning("TopologyManager: new host_group (" +
new_host_group.name + ") added")
updated = True
for rhgk in self.resource.host_groups.keys():
@ -138,8 +141,8 @@ class TopologyManager(threading.Thread):
host_group.last_update = time.time()
self.logger.info("TopologyManager: host_group (" +
host_group.name + ") disabled")
LOG.warning("TopologyManager: host_group (" +
host_group.name + ") disabled")
updated = True
for hk in _hosts.keys():
@ -191,8 +194,8 @@ class TopologyManager(threading.Thread):
if "infra" not in _rhost.tag:
_rhost.tag.append("infra")
updated = True
self.logger.info("TopologyManager: host (" + _rhost.name +
") updated (tag)")
LOG.warning("TopologyManager: host (" + _rhost.name +
") updated (tag)")
if (_rhost.host_group is None or
_host.host_group.name != _rhost.host_group.name):
@ -203,8 +206,8 @@ class TopologyManager(threading.Thread):
else:
_rhost.host_group = self.resource.datacenter
updated = True
self.logger.info("TopologyManager: host (" + _rhost.name +
") updated (host_group)")
LOG.warning("TopologyManager: host (" + _rhost.name +
") updated (host_group)")
return updated
@ -214,14 +217,14 @@ class TopologyManager(threading.Thread):
if _hg.host_type != _rhg.host_type:
_rhg.host_type = _hg.host_type
updated = True
self.logger.info("TopologyManager: host_group (" + _rhg.name +
") updated (hosting type)")
LOG.warning("TopologyManager: host_group (" + _rhg.name +
") updated (hosting type)")
if _rhg.status == "disabled":
_rhg.status = "enabled"
updated = True
self.logger.info("TopologyManager: host_group (" + _rhg.name +
") updated (enabled)")
LOG.warning("TopologyManager: host_group (" + _rhg.name +
") updated (enabled)")
if _hg.parent_resource != _rhg.parent_resource:
if _hg.parent_resource.name in self.resource.host_groups.keys():
@ -230,8 +233,8 @@ class TopologyManager(threading.Thread):
else:
_rhg.parent_resource = self.resource.datacenter
updated = True
self.logger.info("TopologyManager: host_group (" + _rhg.name +
") updated (parent host_group)")
LOG.warning("TopologyManager: host_group (" + _rhg.name +
") updated (parent host_group)")
for rk in _hg.child_resources.keys():
exist = False
@ -245,8 +248,8 @@ class TopologyManager(threading.Thread):
elif _rhg.host_type == "cluster":
_rhg.child_resources[rk] = self.resource.host_groups[rk]
updated = True
self.logger.info("TopologyManager: host_group (" + _rhg.name +
") updated (new child host)")
LOG.warning("TopologyManager: host_group (" + _rhg.name +
") updated (new child host)")
for rrk in _rhg.child_resources.keys():
exist = False
@ -257,8 +260,8 @@ class TopologyManager(threading.Thread):
if exist is False:
del _rhg.child_resources[rrk]
updated = True
self.logger.info("TopologyManager: host_group (" + _rhg.name +
") updated (child host removed)")
LOG.warning("TopologyManager: host_group (" + _rhg.name +
") updated (child host removed)")
return updated
@ -269,8 +272,8 @@ class TopologyManager(threading.Thread):
if rc not in self.resource.datacenter.region_code_list:
self.resource.datacenter.region_code_list.append(rc)
updated = True
self.logger.info("TopologyManager: datacenter updated "
"(new region code, " + rc + ")")
LOG.warning("TopologyManager: datacenter updated "
"(new region code, " + rc + ")")
code_list = self.resource.datacenter.region_code_list
blen = len(code_list)
@ -279,7 +282,7 @@ class TopologyManager(threading.Thread):
if alen != blen:
updated = True
self.resource.datacenter.region_code_list = code_list
self.logger.info("datacenter updated (region code removed)")
LOG.warning("datacenter updated (region code removed)")
for rk in _datacenter.resources.keys():
exist = False
@ -296,8 +299,8 @@ class TopologyManager(threading.Thread):
self.resource.datacenter.resources[rk] = \
self.resource.hosts[rk]
updated = True
self.logger.info("TopologyManager: datacenter updated "
"(new resource)")
LOG.warning("TopologyManager: datacenter updated "
"(new resource)")
for rrk in self.resource.datacenter.resources.keys():
exist = False
@ -308,7 +311,7 @@ class TopologyManager(threading.Thread):
if exist is False:
del self.resource.datacenter.resources[rrk]
updated = True
self.logger.info("TopologyManager: datacenter updated "
"(resource removed)")
LOG.warning("TopologyManager: datacenter updated "
"(resource removed)")
return updated

9
valet/tests/unit/engine/test_search.py
View File

@ -12,16 +12,11 @@
# 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.
"""Test Search."""
import logging
import mock
from valet.engine.optimizer.ostro.search import Search
from valet.tests.base import Base
LOG = logging.getLogger(__name__)
class TestSearch(Base):
"""Unit tests for valet.engine.optimizer.ostro.search."""
@ -30,7 +25,7 @@ class TestSearch(Base):
"""Setup Test Search Class."""
super(TestSearch, self).setUp()
self.search = Search(LOG)
self.search = Search()
def test_copy_resource_status(self):
"""Test Copy Resource Status."""

2
valet/tests/unit/engine/test_topology.py
View File

@ -25,7 +25,7 @@ class TestTopology(Base):
def setUp(self):
"""Setup TestTopology Test Class."""
super(TestTopology, self).setUp()
self.topo = Topology(Config(), None)
self.topo = Topology(Config())
def test_simple_topology(self):
"""Validate simple topology (region, rack, node_type and status)."""