# -*- coding: utf-8 -*-
from logging import debug, error, warning, info
from fnmatch import fnmatchcase
from lvm import LVM
from time import sleep
from sjrpc.core import RpcError
from sjrpc.utils import RpcHandler
from sjrpc.utils import pure
from errors import HostError, HypervisorError
from common import LocalHost
from jobs import ReceiveFileJob, SendFileJob, DrbdCopyJob, TCPTunnelJob
from __init__ import __version__
#FIXME should not be needed
import traceback, sys
from libvirt import libvirtError
#FIXME clean this up
_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=True, safe_mode=True,
force_xen=False):
'''
'''
super(NodeHandler, self).__init__()
self._connection = connection
self._safe_mode = safe_mode
# create interface with host
self._host_handle = None
if detect_hv:
debug('Hypervisor detection in progress')
if _MOD_KVM and not force_xen:
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()
# methods that execute administrative commands, to be banned when
# running in safe mode
self.UNSAFE_METHODS = ['execute_command', 'shutdown']
# hypervisor tags
self.HV_TAG_MANDATORY = ['h']
self.HV_TAG_MAP = {
# infinite TTL
'version' : ( lambda o: True,
lambda o,t: str(__version__),
-1),
'libvirtver': self._tag_map_direct('get_libvirt_version', -1),
'htype' : self._tag_map_direct('get_hv_type', -1),
'hserial' : self._tag_map_direct('get_hw_serial', -1),
'hvendor' : self._tag_map_direct('get_hw_vendor', -1),
'hmodel' : self._tag_map_direct('get_hw_product', -1),
'arch' : self._tag_map_direct('get_arch', -1),
'hvm' : self._tag_map_direct('get_hvm_available', -1),
'cpu' : self._tag_map_direct('get_cpu', -1),
'cpulogical': self._tag_map_direct('get_cpu_thread', -1),
'chaserial' : self._tag_map_direct('get_chassis_serial', -1),
'chaasset' : self._tag_map_direct('get_chassis_asset', -1),
# one day
'hbios' : self._tag_map_direct('get_hw_bios', 24*3600),
'hvver' : self._tag_map_direct('get_hv_version', 24*3600),
'platform' : self._tag_map_direct('get_platform', 24*3600),
'os' : self._tag_map_direct('get_system', 24*3600),
'uname' : self._tag_map_direct('get_uname', 24*3600),
'cpufreq' : self._tag_map_direct('get_cpu_frequency', 24*3600),
'mem' : self._tag_map_direct('get_mem', 24*3600),
'disk' : self._tag_map_keys('get_disks', 24*3600),
'h' : self._tag_map_direct('get_name', 24*3600),
# one hour
# one minute
'memfree' : self._tag_map_direct('get_mem_free', 60),
'memused' : self._tag_map_direct('get_mem_used', 60),
'sto' : ( lambda o: hasattr(o, 'storage'),
lambda o,t: ' '.join(
getattr(o, 'storage')().pool_list()),
60),
# 5 seconds
'uptime' : self._tag_map_direct('get_uptime', 5),
'cpuuse' : self._tag_map_direct('get_cpu_usage', 5),
'load' : self._tag_map_direct('get_loadavg', 5),
'nvm' : self._tag_map_counter('vm_list', 5),
'vmstarted' : self._tag_map_counter('vm_list_running', 5),
'vmstopped' : self._tag_map_counter('vm_list_stopped', 5),
'vmpaused' : self._tag_map_counter('vm_list_paused', 5),
'rjobs' : ( lambda o: True,
lambda o,t: str(len(
self._host_handle.jobmgr.list()['running'])),
5),
# 5 seconds
}
self.HV_TAG_GLOB = {
'disk*' : self._tag_map_helper(self._helper_hv_disk, 24*3600),
'sto*' : self._tag_map_helper(self._helper_hv_sto, 60),
}
# guest VM tags
self.VM_TAG_MANDATORY = ['hv', 'h']
self.VM_TAG_MAP = {
# infinite TTL
'hv' : ( lambda o: hasattr(o, 'hypervisor'),
lambda o,t: o.hypervisor().get_name(),
-1),
'htype' : ( lambda o: hasattr(o, 'hypervisor'),
lambda o,t: o.hypervisor().get_hv_type(),
-1),
'arch' : self._tag_map_direct('get_arch', -1),
'h' : self._tag_map_direct('get_name', -1),
# one day
# one hour
'cpu' : self._tag_map_direct('get_cpu_core', 3600),
'mem' : self._tag_map_direct('get_mem', 3600),
'memmax' : self._tag_map_direct('get_mem_max', 3600),
'vncport' : self._tag_map_direct('get_vnc_port', 3600),
# one minute
# 5 seconds
'status' : ( lambda o: True,
lambda o,t: 'running' if o.is_active()
else 'paused' if o.is_paused()
else 'stopped',
5), # FIXME crappy tag implementation
#'cpuuse' : self._tag_map_direct('get_cpu_usage'),
}
self.VM_TAG_GLOB = {
'disk*' : self._tag_map_helper(self._helper_vm_disk, 3600),
'nic*' : self._tag_map_helper(self._helper_vm_nic, 3600),
}
# FIXME
self._register_vm = []
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._safe_mode and name in self.UNSAFE_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, ttl):
'''
'''
return ( lambda o: hasattr(o, method),
lambda o,t: getattr(o, method)(),
ttl)
def _tag_map_counter(self, method, ttl):
'''
'''
return ( lambda o: hasattr(o, method),
lambda o,t: len(getattr(o, method)()),
ttl)
def _tag_map_keys(self, method, ttl):
'''
'''
return ( lambda o: hasattr(o, method),
lambda o,t: ' '.join(getattr(o, method)().keys()),
ttl)
def _tag_map_helper(self, helper, ttl):
'''
'''
return ( lambda o, resolve=False: helper(o, resolve=resolve),
lambda o, tag_name=None, resolve=False:
helper(o, tag_name=tag_name, resolve=resolve),
ttl)
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 is not None:
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 is not None:
result['sto%s_size' % pool_name] = str(capa)
free = pool.get_space_free()
if free is not None:
result['sto%s_free' % pool_name] = str(free)
used = pool.get_space_used()
if used is not None:
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):
name = vol.get_name()
if name:
result['disk%i_vol' % vol_id] = str(name)
pool = vol.get_pool()
if pool:
result['disk%i_pool' % vol_id] = str(pool.name())
path = vol.get_path()
if path:
result['disk%i_path' % vol_id] = str(path)
capa = vol.get_space_capacity()
if capa is not None:
result['disk%i_size' % vol_id] = str(capa)
if not result:
result = None
return result
def _helper_vm_nic(self, vm, tag_name=None, resolve=True):
'''
'''
result = {}
nics = vm.get_nics()
if len(nics):
result['nic'] = ' '.join([str(i) for i in range(0, len(nics))])
for nic_id, nic in enumerate(nics):
mac = nic.get('mac')
if mac:
result['nic%i_mac' % nic_id] = str(mac)
model = nic.get('model')
if model:
result['nic%i_model' % nic_id] = str(model)
source = nic.get('source')
if source:
result['nic%i_source' % nic_id] = str(source)
if not result:
result = None
return result
def scheduler_run(self):
'''
'''
# call host scheduler
if hasattr(self._host_handle, 'scheduler_run'):
self._host_handle.scheduler_run()
# (un)register sub nodes if this host has the capability
if hasattr(self._host_handle, 'vm_list'):
try:
vm_current = self._host_handle.vm_list()
for vm in vm_current:
if vm not in self._register_vm:
try:
info('registering vm `%s`' % vm)
self._connection.get_server().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:
info('unregistering vm `%s`' % vm)
self._connection.get_server().unregister(vm)
except RpcError as e:
if e.exception == '#FIXME':
self._register_vm.remove(vm)
else:
raise e
else:
self._register_vm.remove(vm)
except Exception as e:
debug("REGISTER except `%s`:`%s`" % (repr(e), e))
pass
##################################
# Tag query
##################################
@pure
def get_tags(self, tags=None, noresolve_tags=None):
'''
'''
result = {}
info('server requested tags=`%s` noresolve_tags=`%s`', tags,
noresolve_tags)
# build a single dict of tags, boolean means "resolve"
mytags = {}
if tags:
for t in tags:
mytags[t] = True
if noresolve_tags:
for t in noresolve_tags:
if t not in mytags:
mytags[t] = False
# return all tags if server does not request a subset
if not mytags:
# add simple tags
for t in self.HV_TAG_MAP.keys():
mytags[t] = True
# add globbing tags
for pattern, handler in self.HV_TAG_GLOB.iteritems():
try:
# helper is available on the current host
if handler[0](self._host_handle):
debug('host implements `%s`', pattern)
# get tags from helper
htags = handler[0](self._host_handle, resolve=False)
# append all tags
for t in htags:
mytags[t] = True
except Exception as err:
debug('error adding globbing tags `%r`:`%s`', err, err)
debug('no tag specified, expanded list to `%s`', mytags.keys())
# add mandatory tags if missing in the list, or set noresolve
else:
for t in self.HV_TAG_MANDATORY:
if t not in mytags or not mytags[t]:
debug('add/correct mandatory tag `%s`', t)
mytags[t] = True
# query host
info('query host with tag list `%s`', mytags.keys())
for tag, resolve in mytags.iteritems():
# first, test tag name againts list of plain name
if tag in self.HV_TAG_MAP:
debug('A1: plain mapping found for tag `%s`', tag)
try:
if self.HV_TAG_MAP[tag][0](self._host_handle):
debug('B1: tag `%s` is available on host', tag)
result[tag] = {}
result[tag]['ttl'] = self.HV_TAG_MAP[tag][2]
if resolve:
debug('C1: now resolving tag `%s`', tag)
# fetch tag data
q = self.HV_TAG_MAP[tag][1](self._host_handle, tag)
debug('D1: host returned `%s`=`%s`', tag, q)
if q is not None:
# append tag data
result[tag]['value'] = str(q)
else:
debug('E1: I wont return `%s`=`None`', tag)
else:
debug('tag `%s` is not implemented !!!', tag)
except Exception as err:
debug('error with tag mapping `%r`:`%s`', err, err)
# if no direct tag mapping exists, test name against globbing
# a list
else:
debug('A2: searching for `%s` in globbing tags', tag)
# iter on globbing patterns, and get helper references
# process the first globbing tag that match then exit bcause
# there should not exist two globbing pattern matching
# one tag, ideally
for pattern, handler in self.HV_TAG_GLOB.iteritems():
try:
# helper is available on the current host
if handler[0](self._host_handle, tag):
debug('B2: testing pattern `%s`', pattern)
if fnmatchcase(tag, pattern):
debug('C2: processing tag `%s` with pattern `%s`',
tag, pattern)
# get tags from helper
htags = handler[1](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('D2: found tag in helper result with'
' value `%s`', htags[tag])
result[tag] = {}
result[tag]['ttl'] = handler[2]
if resolve and htags[tag] is not None:
result[tag]['value'] = str(htags[tag])
break
except Exception as err:
debug('error with globbing tag `%r`:`%s`', err, err)
return result
def _sub_tag_list(self, sub_obj):
'''
'''
result = []
# add simple tags
result.extend(self.VM_TAG_MAP.keys())
# add globbing tags
for pattern, handler in self.VM_TAG_GLOB.iteritems():
try:
# helper is available on the current host
if handler[0](sub_obj):
debug('sub node implements `%s`' % pattern)
# get tags from helper
htags = handler[0](sub_obj, resolve=False)
debug('handler provides `%s`' % htags)
# append all tags
for t in htags.keys():
result.append(t)
except Exception as err:
debug('error while listing sub node tags `%r`:`%s`', err, err)
return result
@pure
def sub_tags(self, sub_id, tags=None, noresolve_tags=None):
'''
'''
info('server requested tags for `%s` with tags=`%s` noresolve_tags=`%s`'
, sub_id, tags, noresolve_tags)
if sub_id not in self._host_handle.vm_list():
debug('sub node `%s` is unknown !', sub_id)
raise HypervisorError('sub node `%s` is unknown' % sub_id)
else:
# open a wrapper to the VM
debug('fetching vm data for `%s`', sub_id)
sub = self._host_handle.vm_get(sub_id)
# build a single list of tags
debug('server requested tags `%s` + `%s`', tags, noresolve_tags)
available_tags = self._sub_tag_list(sub)
mytags = {}
# return all resolved tags case
if tags is None and noresolve_tags is None:
for t in available_tags:
mytags[t] = True
elif tags is None or noresolve_tags is None:
if tags is None:
for t in available_tags:
mytags[t] = True
for t in noresolve_tags:
mytags[t] = False
else:
for t in available_tags:
mytags[t] = False
for t in tags:
mytags[t] = True
else:
for t in noresolve_tags:
mytags[t] = False
for t in tags:
mytags[t] = True
debug('expanded list to `%s`', mytags.keys())
# add mandatory tags if missing in the list, or set noresolve
for t in self.VM_TAG_MANDATORY:
if t not in mytags or not mytags[t]:
debug('add/correct mandatory tag `%s`' % t)
mytags[t] = True
# query the subnode
result = {}
try:
# query the VM with each tag
for tag, resolve in mytags.iteritems():
# first, search tag in plain mappings
if tag in self.VM_TAG_MAP:
debug('A1: plain mapping found for tag `%s`', tag)
# proceed if tag can be resolved on this VM
try:
if self.VM_TAG_MAP[tag][0](sub):
result[tag] = {}
result[tag]['ttl'] = self.VM_TAG_MAP[tag][2]
if resolve:
debug('B1: resolving tag `%s`', tag)
# call the wrapper mapping lambda
q = self.VM_TAG_MAP[tag][1](sub, tag)
debug('C1: tag query returned `%s`=`%s`', tag, q)
if q is not None:
if resolve:
result[tag]['value'] = str(q)
else:
warning('D1: I wont return `%s`=`None`', tag)
except Exception as err:
debug('error with plain mapping `%r`:`%s`', err, err)
# no tag mapping exist, test name against the globbing list
else:
debug('A2: 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():
try:
# helper is available on the current VM
if handler[0](sub, tag):
if fnmatchcase(tag, pattern):
debug('B2: processing tag `%s` with pattern'
' `%s`' % (tag, pattern))
# get tags from helper
htags = handler[1](sub, 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('C2: found tag in helper result with'
' value `%s`' % htags[tag])
result[tag] = {}
result[tag]['ttl'] = handler[2]
if resolve:
result[tag]['value'] = str(
htags[tag])
break
except Exception as err:
debug('error with globbing tag `%r`:`%s`', err, err)
except Exception as err:
debug('global error `%r`:`%s`', err, err)
return result
##################################
# Host control
##################################
@pure
def node_shutdown(self, reboot=True, gracefull=True):
'''
'''
info('server requested shutdown of local host with options '
'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)()
info('in progress... action returned `%s`', result)
return result
else:
info('unable to proceed, this feature is not available')
raise NotImplementedError('host handler has no method `%s`' %method)
@pure
def execute_command(self, command):
'''
'''
info('starting execution of `%s`' % command)
output = self._host_handle.execute(command)
info('finished execution of `%s`' % command)
return output
##################################
# VM management
##################################
###
# Definition and migration of domains
@pure
def vm_define(self, data, format='xml'):
'''
'''
debug('request for creation of a new VM with data=`%s`', data)
name = self._host_handle.vm_define(data)
debug('new VM has name `%s`', name)
return name
@pure
def vm_undefine(self, name):
'''
'''
debug('request for deletion of VM definition named `%s`', name)
vm = self._host_handle.vm_get(name)
vm.undefine()
debug('succesfully undefined VM `%s`', name)
@pure
def vm_export(self, name, format='xml'):
'''
'''
debug('request for definition data of VM `%s`', name)
vm = self._host_handle.vm_get(name)
return vm.get_config()
###
# Life cycle control
@pure
def vm_stop(self, vm_names=None, force=False):
'''
'''
info('server requested stop of `%s`, force is `%s`', vm_names, force)
if vm_names is None:
vm_names = self._host_handle.vm_list_running()
debug('no vm specified, expanded list to `%s`', vm_names)
for vm_name in vm_names:
try:
debug('fetching vm data for `%s`', vm_name)
vm = self._host_handle.vm_get(vm_name)
if force:
debug('powering off `%s`', vm_name)
vm.power_off()
else:
debug('shutdown requested for `%s`', vm_name)
vm.power_shutdown()
except:
pass
@pure
def vm_start(self, vm_names=None):
'''
'''
info('server requested start of `%s`', vm_names)
if vm_names is None:
vm_names = self._host_handle.vm_list_stopped()
debug('no vm specified, expanded list to `%s`', vm_names)
for vm_name in vm_names:
try:
debug('fetching vm data for `%s`', vm_name)
vm = self._host_handle.vm_get(vm_name)
debug('powering on `%s`', vm_name)
vm.power_on()
except:
pass
@pure
def vm_suspend(self, vm_names=None):
'''
'''
info('server requested suspend of `%s`', vm_names)
if vm_names is None:
vm_names = self._host_handle.vm_list_running()
debug('no vm specified, expanded list to `%s`',vm_names)
for vm_name in vm_names:
try:
debug('fetching vm data for `%s`', vm_name)
vm = self._host_handle.vm_get(vm_name)
debug('pause execution of `%s`', vm_name)
vm.power_suspend()
except:
pass
@pure
def vm_resume(self, vm_names=None):
'''
'''
info('server requested resume of `%s`', vm_names)
if vm_names is None:
vm_names = self._host_handle.vm_list_paused()
debug('no vm specified, expanded list to `%s`', vm_names)
for vm_name in vm_names:
try:
debug('fetching vm data for `%s`', vm_name)
vm = self._host_handle.vm_get(vm_name)
debug('resume execution of `%s`', vm_name)
vm.power_resume()
except:
pass
###
# Migration and helpers
@pure
def tun_setup(self, local=True):
'''
'''
# facilities
jobmgr = self._host_handle.jobmgr
# create the job
job = TCPTunnelJob(jobmgr)
job.prepare()
job.start_now()
# start listening
job.tunnel_listen(local)
# build a ressource
return {
'jid' : job.get_id(),
'key' : 'FIXME',
'port' : job.tunnel_get_server_port(),
}
@pure
def tun_connect(self, res, remote_res, remote_ip):
'''
'''
# facilities
jobmgr = self._host_handle.jobmgr
# get the job
job = jobmgr.get_job(res['jid'])
# connect to the remote endpoint
job.tunnel_connect((remote_ip, remote_res['port']))
@pure
def tun_connect_hv(self, res, migration=False):
'''
'''
# FIXME magic values
ip = '127.0.0.1'
if migration and self._host_handle.get_hv_type() == 'xen':
port = 8002
else:
port = 16509
# facilities
jobmgr = self._host_handle.jobmgr
# get the job
job = jobmgr.get_job(res['jid'])
# connect to libvirt
job.tunnel_connect((ip, port))
@pure
def tun_destroy(self, res):
'''
'''
# facilities
jobmgr = self._host_handle.jobmgr
# stop the tunnel
jobmgr.cancel(res['jid'])
@pure
def vm_migrate_tunneled(self, name, tun_res):
'''
'''
debug('request for tunneled migration of VM `%s`', name)
# local listenning tunnel port
tun_port = tun_res.get('port')
debug('tunnel port = `%d`', tun_port)
# acquire vm object
vm = self._host_handle.vm_get(name)
# migrate
vm.migrate('127.0.0.1', tun_port)
##################################
# Storage control
##################################
###
# Volume management
@pure
def vol_create(self, pool, name, size):
'''
'''
size = int(size)
if hasattr(self._host_handle, 'storage'):
pool = self._host_handle.storage().pool_get(pool)
pool.volume_create(name, size)
else:
raise NotImplementedError('host handler has no storage support')
@pure
def vol_delete(self, pool, name, wipe=False):
'''
'''
if hasattr(self._host_handle, 'storage'):
pool = self._host_handle.storage().pool_get(pool)
vol = pool.volume_get(name)
if wipe:
vol.wipe()
vol.delete()
else:
raise NotImplementedError('host handler has no storage support')
@pure
def vol_copy(self, pool, name, dest_pool, dest_name):
raise NotImplementedError()
###
# Basic network copy
@pure
def vol_export(self, pool, name, raddr, rport):
'''
'''
rport = int(rport)
if hasattr(self._host_handle, 'storage'):
# facilities
sto = self._host_handle.storage()
jmgr = self._host_handle.jobmgr
# get device path
vol_path = sto.pool_get(pool).volume_get(name).get_path()
# create job
job = SendFileJob(jmgr, vol_path, raddr, rport)
job.start_now()
jid = job.get_id()
# wait for job completion
while self._host_handle.jobmgr.is_job_running(jid):
sleep(2)
# return job report
res = {}
res['id'] = jid
res['log'] = job.get_log()
if jmgr.is_job_finished(jid):
res['checksum'] = job.get_checksum()
return res
else:
raise NotImplementedError('host handler has no storage support')
@pure
def vol_import(self, pool, name):
'''
'''
if hasattr(self._host_handle, 'storage'):
# facilities
sto = self._host_handle.storage()
# get device path
vol_path = sto.pool_get(pool).volume_get(name).get_path()
# create job
job = ReceiveFileJob(self._host_handle.jobmgr, vol_path)
# do the preparation immediately, so we can fetch the port number
# before the job starts
job.prepare()
job.start_now()
# return job info
res = {}
res['id'] = job.get_id()
res['port'] = job.get_port()
return res
else:
raise NotImplementedError('host handler has no storage support')
@pure
def vol_import_wait(self, jid):
'''
'''
jid = int(jid)
jmgr = self._host_handle.jobmgr
# wait for job completion
while jmgr.is_job_running(jid):
sleep(2)
# get the job
job = jmgr.get_job(jid)
# return job report
res = {}
res['id'] = job.get_id()
res['log'] = job.get_log()
if jmgr.is_job_finished(job.get_id()):
res['checksum'] = job.get_checksum()
return res
@pure
def vol_import_cancel(self, jid):
'''
'''
jid = int(jid)
# facilities
jmgr = self._host_handle.jobmgr
# get the job
job = jmgr.get_job(jid)
# check job type
if job.get_type() != 'ReceiveFileJob':
raise HostError('this job ID does not match a volume import')
# cancel it
jmgr.cancel(jid)
###
# Live network copy
@pure
def drbd_setup(self, pool, name):
'''
'''
if hasattr(self._host_handle, 'storage'):
# facilities
sto = self._host_handle.storage()
lvm = self._host_handle.lvm
jobmgr = self._host_handle.jobmgr
drbdpool = self._host_handle.drbdpool
# get pool and volume handles
vol_pool = sto.pool_get(pool)
vol = vol_pool.volume_get(name)
# stop if it's not an LVM volume
if vol_pool.get_source_format() != 'lvm2':
raise HostError('not a LVM2 volume, cannot use DRBD')
# fetch LV/VG info
vg_name = vol_pool.get_source_name()
lv_name = vol.get_name()
# create job
job = DrbdCopyJob(jobmgr, lvm, drbdpool, vg_name, lv_name)
job.start_now()
return {
'jid' : job.get_id(),
'port' : job.drbd_get_port(),
}
else:
raise NotImplementedError('host handler has no storage support')
@pure
def drbd_connect(self, res, remote_res, remote_ip):
'''
Connect the local live copy engine to the designated remote host that
will push or pull data.
'''
# facilities
jobmgr = self._host_handle.jobmgr
# get the job
job = jobmgr.get_job(res['jid'])
# connect node
job.drbd_connect(remote_ip, remote_res['port'])
@pure
def drbd_role(self, res, primary):
'''
'''
# facilities
jobmgr = self._host_handle.jobmgr
# get the job
job = jobmgr.get_job(res['jid'])
# change role
job.drbd_role(primary=primary)
@pure
def drbd_takeover(self, res, state):
'''
'''
# facilities
jobmgr = self._host_handle.jobmgr
# get the job
job = jobmgr.get_job(res['jid'])
# hijack the VM disk DM
job.drbd_takeover(state)
@pure
def drbd_sync_status(self, res):
'''
Get status information about a running sync
'''
# facilities
jobmgr = self._host_handle.jobmgr
# get the job
job = jobmgr.get_job(res['jid'])
# get drbd status
status = job.drbd_status()
# return status
return {
'done' : status['disk'] == job._drbd.DSTATES.UP_TO_DATE,
'completion' : status['percent'],
'speed' : None,
}
@pure
def drbd_shutdown(self, res):
'''
Close connection and destroy all ressources allocated for a live copy.
'''
# facilities
jobmgr = self._host_handle.jobmgr
# cancel the job so it can clean itself
# FIXME: check type, we don't want to cancel anything
jobmgr.cancel(res['jid'])
##################################
# Job management
##################################
@pure
def job_list(self):
'''
List all existing jobs sorted by state, with ID and type.
'''
return self._host_handle.jobmgr.list()
@pure
def job_log(self, jid):
'''
Get log messages of a given job (human friendly text string with
carriage return).
'''
# get the job
job = self._host_handle.jobmgr.get_job(jid)
# return log string
return job.get_log()