# -*- coding: utf-8 -*-
from __init__ import __version__
from sjrpc.utils import RpcHandler
from sjrpc.utils import pure
from logging import debug, info
from fnmatch import fnmatchcase
from exceptions import FeatureNotImplemented
from common import LocalHost
from threading import Timer
from sjrpc.core import RpcError
_MOD_KVM = True
try:
import kvm
except ImportError:
_MOD_KVM = False
_MOD_XEN = True
try:
import xen
except ImportError:
_MOD_XEN = False
class NodeHandler(RpcHandler):
'''
Main node handler that exports the host capabilities to the server.
'''
def __init__(self, connection, detect_hv, allow_exec):
'''
'''
super(RpcHandler, self).__init__()
self._connection = connection
self._allow_cmd_exec = allow_exec
self._host_handle = None
if detect_hv:
debug('Hypervisor detection in progress')
if _MOD_KVM:
debug('Initializing connection to the local KVM hypervisor')
self._host_handle = kvm.KvmHypervisor()
elif _MOD_XEN:
debug('Initializing connection to the local Xen hypervisor')
self._host_handle = xen.XenHypervisor()
if self._host_handle is None:
debug('Hypervisor detection failed')
if not detect_hv or self._host_handle is None:
debug('Hypervisor detection disabled, running as regular node')
self._host_handle = LocalHost()
self.EXEC_METHODS = ['execute_command', 'shutdown']
self.HV_TAG_MANDATORY = ['h']
self.HV_TAG_MAP = {
'version' : ( lambda o: True,
lambda o,t: str(__version__)),
'h' : self._tag_map_direct('get_name'),
'htype' : self._tag_map_direct('get_hv_type'),
'status' : self._tag_map_direct('get_status'),
'hserial' : self._tag_map_direct('get_hw_serial'),
'hvendor' : self._tag_map_direct('get_hw_vendor'),
'hmachine' : self._tag_map_direct('get_hw_product'),
'hbios' : self._tag_map_direct('get_hw_bios'),
'arch' : self._tag_map_direct('get_arch'),
'platform' : self._tag_map_direct('get_platform'),
'uname' : self._tag_map_direct('get_uname'),
'uptime' : self._tag_map_direct('get_uptime'),
'hvm' : self._tag_map_direct('get_hvm_available'),
'libvirtver': self._tag_map_direct('get_libvirt_version'),
'hvver' : self._tag_map_direct('get_hv_version'),
'load' : self._tag_map_direct('get_loadavg'),
'cpu' : self._tag_map_direct('get_cpu'),
'cpucore' : self._tag_map_direct('get_cpu_core'),
'cputhread' : self._tag_map_direct('get_cpu_threads'),
'cpufreq' : self._tag_map_direct('get_cpu_frequency'),
'cpuuse' : self._tag_map_direct('get_cpu_usage'),
'mem' : self._tag_map_direct('get_mem'),
'memfree' : self._tag_map_direct('get_mem_free'),
'memused' : self._tag_map_direct('get_mem_used'),
'disk' : self._tag_map_keys('get_disks'),
'sto' : ( lambda o: hasattr(o, 'storage'),
lambda o,t: ' '.join(
getattr(o, 'storage')().pool_list())),
'nvm' : self._tag_map_counter('vm_list'),
'vmstarted' : self._tag_map_counter('vm_list_running'),
'vmstopped' : self._tag_map_counter('vm_list_stopped'),
'vmpaused' : self._tag_map_counter('vm_list_paused'),
}
self.HV_TAG_GLOB = {
'disk*' : self._tag_map_helper(self._helper_hv_disk),
'sto*' : self._tag_map_helper(self._helper_hv_sto),
}
self.VM_TAG_MANDATORY = ['hv', 'h']
self.VM_TAG_MAP = {
'version' : ( lambda o: True,
lambda o,t: str(__version__)),
'hv' : ( lambda o: hasattr(o, 'hypervisor'),
lambda o,t: o.hypervisor().get_name()),
# FIXME crappy tag implementation
'status' : ( lambda o: True,
lambda o,t: 'running' if o.is_active()
else 'paused' if o.is_paused()
else 'stopped'),
'h' : self._tag_map_direct('get_name'),
'arch' : self._tag_map_direct('get_arch'),
'cpu' : self._tag_map_direct('get_cpu'),
'mem' : self._tag_map_direct('get_mem'),
'memused' : self._tag_map_direct('get_mem_used'),
'memfree' : self._tag_map_direct('get_mem_free'),
}
self.VM_TAG_GLOB = {
'disk*' : self._tag_map_helper(self._helper_vm_disk),
}
# populate the server with running VMs
self._register_vm = []
self._register_timer = Timer(5.0, self._register)
def __getitem__(self, name):
'''
'''
# filter the private members access
if name.startswith('_'):
raise KeyError('Remote name `%s` is private' % repr(name))
# filter command execution methods
elif not self._allow_cmd_exec and name in self.EXEC_METHODS:
raise KeyError('Remote name `%s` is disabled by configuration'
% repr(name))
else:
debug('Called %s.%s' % (self.__class__.__name__, name))
return super(NodeHandler, self).__getitem__(name)
def _tag_map_direct(self, method):
'''
'''
return ( lambda o: hasattr(o, method),
lambda o,t: getattr(o, method)())
def _tag_map_counter(self, method):
'''
'''
return ( lambda o: hasattr(o, method),
lambda o,t: len(getattr(o, method)()))
def _tag_map_keys(self, method):
'''
'''
return ( lambda o: hasattr(o, method),
lambda o,t: ' '.join(getattr(o, method)().keys()))
def _tag_map_helper(self, helper):
'''
'''
return ( lambda o, resolve=False: helper(o, resolve=resolve),
lambda o, tag_name=None, resolve=False:
helper(o, tag_name=tag_name, resolve=resolve))
def _helper_hv_disk(self, hv, tag_name=None, resolve=True):
'''
'''
result = {}
disks = hv.get_disks()
if len(disks):
result['disk'] = ' '.join(disks.keys())
for name, size in disks.iteritems():
if size:
result['disk%s_size' % name] = str(size)
if not result:
result = None
return result
def _helper_hv_sto(self, hv, tag_name=None, resolve=True):
'''
'''
result = {}
if hasattr(hv, 'storage'):
pools = hv.storage().pool_list()
if len(pools):
result['sto'] = ' '.join(pools)
for pool_name in pools:
pool = hv.storage().pool_get(pool_name)
capa = pool.get_space_capacity()
if capa:
result['sto%s_size' % pool_name] = str(capa)
free = pool.get_space_free()
if free:
result['sto%s_free' % pool_name] = str(free)
used = pool.get_space_used()
if used:
result['sto%s_used' % pool_name] = str(used)
vol = pool.volume_list()
if vol:
result['sto%s_vol' % pool_name] = ' '.join(vol)
if not result:
result = None
return result
def _helper_vm_disk(self, vm, tag_name=None, resolve=True):
'''
'''
result = {}
volumes = vm.get_volumes()
if len(volumes):
result['disk'] = ' '.join([str(i) for i in range(0, len(volumes))])
for vol_id, vol in enumerate(volumes):
path = vol.get_path()
if path:
result['disk%i_path' % vol_id] = str(path)
capa = vol.get_space_capacity()
if capa:
result['disk%i_size' % vol_id] = str(capa)
if not result:
result = None
return result
def _register(self):
try:
vm_current = self._host_handle.vm_list()
for vm in vm_current:
if vm not in self._register_vm:
try:
self._connection.get_server().call('register', vm, 'vm')
except RpcError as e:
if e.exception == '#FIXME':
self._register_vm.append(vm)
else:
raise e
else:
self._register_vm.append(vm)
for vm in self._register_vm:
if vm not in vm_current:
try:
self._connection.get_server().call('unregister', vm)
except RpcError as e:
if e.exception == '#FIXME':
vm_current.pop(vm)
else:
raise e
else:
vm_current.pop(vm)
except Exception as e:
raise e
finally:
self._register_timer = Timer(1.0, self._register)
@pure
def get_tags(self, tags=None, resolve=True):
'''
'''
result = {}
debug('get_tags: server requested tags `%s`' % tags)
# return all tags if server does not request a subset
if tags is None:
# add simple tags
tags = self.HV_TAG_MAP.keys()
# add globbing tags
for pattern, handler in self.HV_TAG_GLOB.iteritems():
# helper is available on the current host
if handler[0](self._host_handle):
debug('get_tags: host implements `%s`' % pattern)
# get tags from helper
htags = handler[0](self._host_handle, resolve=False)
# append all tags
tags.extend(htags)
debug('get_tags: no tag specified, expanded list to `%s`' % tags)
# add mandatory tags if missing in the list
else:
for mtag in self.HV_TAG_MANDATORY:
if mtag not in tags:
debug('get_tags: add missing mandatory tag `%s`' % mtag)
tags.append(mtag)
# query host
debug('get_tags: query host with tag list `%s`' % tags)
for tag in tags:
# first, test tag name againts list of plain name
if tag in self.HV_TAG_MAP:
debug('get_tags: plain mapping found for tag `%s`' % tag)
if self.HV_TAG_MAP[tag][0](self._host_handle):
# fetch tag data
q = self.HV_TAG_MAP[tag][1](self._host_handle, tag)
debug('get_tags: host returned `%s`' % q)
# append tag information
if q is not None:
# when a dict is returned, it may contain >1 tags
# in this case the real tag name is given by the
# wrapper and it may differ from the mapping name
if isinstance(q, dict):
for key, val in q.iteritems():
result[key] = {}
result[key]['value'] = str(val)
# FIXME really fast
result[key]['ttl'] = -1
# or there's only one value
else:
result[tag] = {}
result[tag]['value'] = str(q)
# FIXME really fast
result[tag]['ttl'] = -1
else:
debug('get_tags: I wont return `%s`=`None`' % tag)
else:
debug('get_tags: tag `%s` is NOT implemented' % tag)
# if no direct tag mapping exists, test name against globbing a list
else:
debug('get_tags: searching for `%s` in globbing tags' % tag)
# iter on globbing patterns, and get helper references
# process the first globbing tag that match then exit because
# there should not exist two globbing pattern matching
# one tag, ideally
for pattern, handler in self.HV_TAG_GLOB.iteritems():
# helper is available on the current host
if handler[0](self._host_handle, tag):
if fnmatchcase(tag, pattern):
debug('get_tags: processing tag `%s` with '
'pattern `%s`' % (tag, pattern))
# get tags from helper
htags = handler[0](self._host_handle, tag)
# FIXME intead of extracting one tag, try not
# to build the whole list. Maybe it's too
# difficult and not worth to implement
if tag in htags:
debug('get_tags: found tag in helper result'
'with value `%s`' % htags[tag])
result[tag] = {}
result[tag]['value'] = str(htags[tag])
# FIXME
result[tag]['ttl'] = -1
break
return result
@pure
def node_shutdown(self, reboot=True, gracefull=True):
'''
'''
info('shutdown: server requested shutdown of local host')
debug('shutdown: reboot=%s gracefull=%s' % (reboot, gracefull))
if reboot:
method = 'power_reboot' if gracefull else 'power_force_reboot'
else:
method = 'power_shutdown' if gracefull else 'power_off'
if hasattr(self._host_handle, method):
result = getattr(self._host_handle, method)()
debug('shutdown: in progress ... action returned `%s`' % result)
return result
else:
debug('shutdown: unable to proceed, this feature is not available')
raise FeatureNotImplemented('host handler has no method `%s`'
% method)
@pure
def vm_tags(self, vm_names=None, tags=None, resolve=True):
'''
'''
# list all VMs if the server did not provide names
debug('vm_list: server requested list of vm `%s`' % vm_names)
if vm_names is None:
vm_names = self._host_handle.vm_list()
debug('vm_list: no vm specified, expanded list to `%s`' % vm_names)
# return all tags if server does not request a subset
get_all = tags is None
debug('vm_list: server requested tags `%s`' % tags)
if get_all:
# add simple tags
tags = self.VM_TAG_MAP.keys()
debug('vm_list: no tag specified, expanded list to `%s`' % tags)
# add mandatory tags if missing in the list
else:
for mtag in self.VM_TAG_MANDATORY:
if mtag not in tags:
debug('vm_list: add missing mandatory tag `%s`' % mtag)
tags.append(mtag)
# query each vm
result = {}
for vm_name in vm_names:
vm_tag = {}
try:
# copy tag list for local modifications (globbing)
mytags = tags
# open a wrapper to the VM
debug('vm_list: fetching vm data for `%s`' % vm_name)
vm = self._host_handle.vm_get(vm_name)
# expand tag list with globbing tags
if get_all:
for pattern, handler in self.VM_TAG_GLOB.iteritems():
# helper is available on this VM
if handler[0](vm):
debug('vm_list: vm implements `%s`' % pattern)
# get tags from helper
htags = handler[0](vm, resolve=False)
# append all tags
mytags.extend(htags)
# query the VM with each tag
for tag in mytags:
# first, search tag in plain mappings
if tag in self.VM_TAG_MAP:
debug('vm_list: plain mapping found for tag `%s`' % tag)
# proceed if tag can be resolved on this VM
if self.VM_TAG_MAP[tag][0](vm):
vm_tag[tag] = {}
# fetch data only if we only built the tag list
debug('vm_list: resolving tag %s`' % tag)
# call the wrapper mapping lambda
q = self.VM_TAG_MAP[tag][1](vm, tag)
debug('vm_list: query returned `%s`' % q)
# when a dict is returned, it may contain >1 tag
# in this case the real tag name is given by the
# wrapper and it may differ from the mapping nam
if isinstance(q, dict):
for key, val in q.iteritems():
if val is not None:
if resolve:
vm_tag[key]['value'] = str(q)
# FIXME really fast
vm_tag[key]['ttl'] = -1
else:
debug('vm_list: I wont return '
'`%s`=`None`' % key)
# or there's only one value
elif q is not None:
if resolve:
vm_tag[tag]['value'] = str(q)
# FIXME really fast
vm_tag[tag]['ttl'] = -1
else:
debug('vm_list: I wont return `%s`=`None`'
% tag)
# no tag mapping exist, test name against the globbing list
else:
debug('vm_list: searching for `%s` in globbing tags'
% tag)
# iter on globbing patterns, and get helper references
# process the first globbing tag that match then exit
# because there should not exist two globbing pattern
# matching one tag, ideally
for pattern, handler in self.VM_TAG_GLOB.iteritems():
# helper is available on the current VM
if handler[0](vm, tag):
if fnmatchcase(tag, pattern):
debug('get_tags: processing tag `%s` with '
'pattern `%s`' % (tag, pattern))
# get tags from helper
htags = handler[0](vm, tag)
# FIXME intead of extracting one tag, try
# not to build the whole list. Maybe it's
# too difficult and not worth implementing
if tag in htags:
debug('get_tags: found tag in helper '
'result with value `%s`' % htags[tag])
vm_tag[tag] = {}
if resolve:
vm_tag[tag]['value'] = str(
htags[tag])
# FIXME
vm_tag[tag]['ttl'] = -1
break
# save the tag list
# FIXME: in case of exception, we won't return a single VM tag
result[vm_name] = vm_tag
except Exception as e:
debug('(%s) : %s' % (repr(e), e))
return result
@pure
def vm_stop(self, vm_names=None, force=False):
'''
'''
info('vm_stop: server requested stop of `%s`' % vm_names)
debug('vm_stop: force stop is `%s`' % force)
if vm_names is None:
vm_names = self._host_handle.vm_list_running()
debug('vm_stop: no vm specified, expanded list to `%s`' % vm_names)
for vm_name in vm_names:
try:
debug('vm_stop: fetching vm data for `%s`' % vm_name)
vm = self._host_handle.vm_get(vm_name)
if force:
debug('vm_stop: powering off `%s`' % vm_name)
vm.power_off()
else:
info('vm_stop: shutdown requested for `%s`' % vm_name)
vm.power_shutdown()
except:
pass
@pure
def vm_start(self, vm_names=None):
'''
'''
info('vm_start: server requested start of `%s`' % vm_names)
if vm_names is None:
vm_names = self._host_handle.vm_list_stopped()
debug('vm_start: no vm specified, expanded list to `%s`' % vm_names)
for vm_name in vm_names:
try:
debug('vm_start: fetching vm data for `%s`' % vm_name)
vm = self._host_handle.vm_get(vm_name)
info('vm_start: powering on `%s`' % vm_name)
vm.power_on()
except:
pass
@pure
def vm_suspend(self, vm_names=None):
'''
'''
info('vm_suspend: server requested suspend of `%s`' % vm_names)
if vm_names is None:
vm_names = self._host_handle.vm_list_running()
debug('vm_suspend: no vm specified, expanded list to `%s`'
% vm_names)
for vm_name in vm_names:
try:
debug('vm_suspend: fetching vm data for `%s`' % vm_name)
vm = self._host_handle.vm_get(vm_name)
info('vm_suspend: pause execution of `%s`' % vm_name)
vm.power_suspend()
except:
pass
@pure
def vm_resume(self, vm_names=None):
'''
'''
info('vm_resume: server requested resume of `%s`' % vm_names)
if vm_names is None:
vm_names = self._host_handle.vm_list_running()
debug('vm_resume: no vm specified, expanded list to `%s`'
% vm_names)
for vm_name in vm_names:
try:
debug('vm_resume: fetching vm data for `%s`' % vm_name)
vm = self._host_handle.vm_get(vm_name)
info('vm_resume: resume execution of `%s`' % vm_name)
vm.power_resume()
except:
pass
@pure
def execute_command(self, command):
'''
'''
info('execute_command: starting execution of `%s`' % command)
output = self._host_handle.execute(command)
info('execute_command: finished execution of `%s`' % command)
return output