Newer
Older
import libvirt
from ccnode.host import Handler as HostHandler
from ccnode.tags import Tag, tag_inspector, get_tags
from ccnode.hypervisor import tags
from ccnode.hypervisor.lib import (
DOMAIN_STATES, EVENTS, STORAGE_STATES,
EventLoop as VirEventLoop,
)
from ccnode.hypervisor.domains import VirtualMachine
logger = logging.getLogger(__name__)
# FIXME find a way to refactor Handler and Hypervisor class
class Handler(HostHandler):
def __init__(self, *args, **kwargs):
"""
self.hypervisor_name = kwargs.pop('hypervisor_name')
HostHandler.__init__(self, *args, **kwargs)
#: keep index of asynchronous calls
self.async_calls = dict()
self.timer = self.main.evloop.timer(.0, 5., self.virt_connect_cb)
self.hypervisor = None
self._virt_connected = False
# register tags
self.tag_db.add_tags(tag_inspector(tags, self))
@property
def virt_connected(self):
return self._virt_connected
@virt_connected.setter
def virt_connected(self, value):
self._virt_connected = value
# update tags
for tag in ('vir_status', 'sto', 'nvm', 'vmpaused', 'vmstarted',
'vmstopped', 'hvver', 'libvirtver', 'hv'):
# TODO tag annoucement
self.tag_db['__main__'][tag].calculate_value()
def start(self):
self.timer.start()
HostHandler.start(self)
def stop(self):
self.timer.stop()
if self.hypervisor is not None:
self.hypervisor.stop()
HostHandler.stop(self)
def virt_connect_cb(self, *args):
# initialize hypervisor instance
try:
self.hypervisor = Hypervisor(
name=self.hypervisor_name,
handler=self,
)
except libvirt.libvirtError:
logger.exception('Error while connecting to libvirt')
return
self.virt_connected = True
# register hypervisor storage tags
for name, storage in self.hypervisor.storage.storages.iteritems():
self.tag_db.add_tags((
Tag('sto%s_state' % name, lambda: storage.state, 5, 5),
Tag('sto%s_size' % name, lambda: storage.capacity, 5, 5),
Tag('sto%s_free' % name, lambda: storage.available, 5, 5),
lambda: storage.capacity - storage.available, 5, 5),
))
self.tag_db.add_sub_object(dom.name, dom.tags.itervalues())
self.rpc_handler.update(dict(
vm_define=self.vm_define,
vm_undefine=self.vm_undefine,
vm_export=self.vm_export,
vm_stop=self.vm_stop,
vm_start=self.vm_start,
vm_suspend=self.vm_suspend,
vm_resume=self.vm_resume,
))
self.main.reset_handler('vm_define', self.vm_define)
self.main.reset_handler('vm_undefine', self.vm_undefine)
self.main.reset_handler('vm_export', self.vm_export)
self.main.reset_handler('vm_stop', self.vm_stop)
self.main.reset_handler('vm_start', self.vm_start)
self.main.reset_handler('vm_suspend', self.vm_suspend)
self.main.reset_handler('vm_resume', self.vm_resume)
# if everything went fine, unregister the timer
self.timer.stop()
def virt_connect_restart(self):
"""Restart libvirt connection.
This method might be called when libvirt connection is lost.
"""
if not self.virt_connected:
return
logger.error('Connection to libvirt lost, trying to restart')
# update connection state
self.virt_connected = False
# unregister tags that will be re registered later
for storage in self.hypervisor.storage.storages:
self.tag_db.remove_tags((
'sto%s_state' % storage,
'sto%s_size' % storage,
'sto%s_free' % storage,
'sto%s_used' % storage,
))
# unregister sub objects (for the same reason)
for sub_id in self.tag_db.keys():
if sub_id == '__main__':
continue
self.tag_db.remove_sub_object(sub_id)
# stop and delete hypervisor instance
self.hypervisor.stop()
self.hypervisor = None
# remove handlers related to libvirt
self.main.remove_handler('vm_define')
self.main.remove_handler('vm_undefine')
self.main.remove_handler('vm_export')
self.main.remove_handler('vm_stop')
self.main.remove_handler('vm_start')
self.main.remove_handler('vm_suspend')
self.main.remove_handler('vm_resume')
# launch connection timer
self.timer.start()
def iter_vms(self, vm_names):
"""Utility function to iterate over VM objects using their names."""
if vm_names is None:
return
get_domain = self.hypervisor.domains.get
for name in vm_names:
dom = get_domain(name)
if dom is not None:
yield dom
def vm_define(self, data, format='xml'):
logger.debug('VM define')
if format != 'xml':
raise NotImplementedError('Format not supported')
return self.hypervisor.vm_define(data)
def vm_undefine(self, name):
vm = self.hypervisor.domains.get(name)
if vm is not None:
vm.undefine()
def vm_export(self, name, format='xml'):
if format != 'xml':
raise NotImplementedError('Format not supported')
vm = self.hypervisor.domains.get(name)
if vm is None:
return
return vm.lv_dom.XMLDesc(0)
def vm_stop(self, vm_names=None, force=False):
try:
if force:
vm.destroy()
else:
vm.stop()
except libvirt.libvirtError:
# should we return errors ?
pass
def vm_start(self, vm_names=None):
for vm in self.iter_vms(vm_names):
vm.start()
def vm_suspend(self, vm_names=None):
for vm in self.iter_vms(vm_names):
vm.suspend()
def vm_resume(self, vm_names=None):
for vm in self.iter_vms(vm_names):
vm.resume()
class Hypervisor(object):
"""Container for all hypervisor related state."""
def __init__(self, name, handler):
"""
:param str name: name of hypervisor instance
:param Handler handler: hypervisor handler
self.handler = weakref.proxy(handler)
#: hv attributes
self.name = name
self.type = u'kvm'
# register libvirt error handler
libvirt.registerErrorHandler(self.vir_error_cb, None)
self.vir_event_loop = VirEventLoop(self.handler.main.evloop)
# This tells libvirt what event loop implementation it
# should use
libvirt.virEventRegisterImpl(
self.vir_event_loop.add_handle,
self.vir_event_loop.update_handle,
self.vir_event_loop.remove_handle,
self.vir_event_loop.add_timer,
self.vir_event_loop.update_timer,
self.vir_event_loop.remove_timer,
self.vir_con = libvirt.open('qemu:///system') # currently only support KVM
self.storage = StorageIndex(self.vir_con)
logger.debug('Storages: %s', self.storage.paths)
#: domains: vms, containers...
self.domains = dict()
# find defined domains
for dom_name in self.vir_con.listDefinedDomains():
dom = self.vir_con.lookupByName(dom_name)
self.domains[dom.name()] = VirtualMachine(dom, self)
for dom_id in self.vir_con.listDomainsID():
dom = self.vir_con.lookupByID(dom_id)
self.domains[dom.name()] = VirtualMachine(dom, self)
logger.debug('Domains: %s', self.domains)
self.vir_con.domainEventRegister(self.vir_cb, None) # TODO find out args
def stop(self):
self.vir_event_loop.stop()
# unregister callback
try:
self.vir_con.domainEventDeregister(self.vir_cb)
except libvirt.libvirtError:
# in case the libvirt connection is broken, it will raise the error
pass
ret = self.vir_con.close()
logger.debug('Libvirt still handling %s ref connections', ret)
def vir_error_cb(self, ctxt, err):
"""Libvirt error callback.
See http://libvirt.org/errors.html for more informations.
:param ctxt: arbitrary context data (not needed because context is
givent by self
:param err: libvirt error code
"""
logger.error('Libvirt error %s', err)
def vir_cb(self, conn, dom, event, detail, opaque):
"""Callback for libvirt event loop."""
logger.debug('Received event %s on domain %s, detail %s', event,
dom.name(), detail)
event = EVENTS[event]
if event == 'Added':
vm = VirtualMachine(dom, self)
logger.info('Created domain %s', vm.name)
self.handler.tag_db.add_sub_object(vm.name, vm.tags.itervalues())
elif event == 'Removed':
logger.info('Removed domain %s', vm.name)
self.handler.tag_db.remove_sub_object(vm.name)
elif event in ('Started', 'Suspended', 'Resumed', 'Stopped', 'Saved',
'Restored'):
# sometimes libvirt sent a start event before a created event so be
# careful
if vm is not None:
state = DOMAIN_STATES[dom.info()[0]]
logger.info('Domain change state from %s to %s', vm.state,
state)
vm.state = state
# update domain state counts
for tag in ('nvm', 'vmpaused', 'vmstarted', 'vmstopped'):
# TODO tag announcement
self.handler.tag_db['__main__'][tag].calculate_value()
def vm_define(self, xml_desc):
"""Create a VM on the Hypervisor
:param str xml_desc: XML description in libvirt format
:return: VM name created
"""
try:
return self.vir_con.defineXML(xml_desc).name()
except libvirt.libvirtError:
logger.exception('Error while creating domain')
# reraise exception for the cc-server
raise
def _count_domain(self, filter=lambda d: True):
count = 0
for dom in self.domains.itervalues():
if filter(dom):
count += 1
return count
@property
def vm_started(self):
"""Number of VMs started."""
return self._count_domain(lambda d: d.state == 'running')
@property
def vm_stopped(self):
"""Number of VMs stopped."""
return self._count_domain(lambda d: d.state == 'stopped')
@property
def vm_paused(self):
"""Number of VMs paused."""
return self._count_domain(lambda d: d.state == 'paused')
@property
def vm_total(self):
"""Total number of VMs on the hypervisor."""
return self._count_domain()
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
class StorageIndex(object):
"""Keep an index of all storage volume paths."""
def __init__(self, lv_con):
"""
:param lv_con: Libvirt connection
"""
self.storages = dict(
(s.name, s) for s in imap(
Storage,
imap(
lv_con.storagePoolLookupByName,
chain(
lv_con.listDefinedStoragePools(),
lv_con.listStoragePools(),
),
),
),
)
self.paths = dict(
(v.path, v) for v in chain.from_iterable(map(
lambda s: s.volumes,
self.storages.itervalues(),
)),
)
def get_volume(self, path):
return self.paths.get(path)
def get_storage(self, name):
return self.Storage.get(name)
class Storage(object):
"""Storage abstraction."""
def __init__(self, lv_storage):
"""
:param lv_storage: Libvirt pool storage instance
"""
self.uuid = lv_storage.UUID()
self.name = lv_storage.name()
self.state, self.capacity, self.allocation, self.available = lv_storage.info()
self.state = STORAGE_STATES[self.state]
(lv_storage.storageVolLookupByName(n) for n in
lv_storage.listVolumes()),
)
class Volume(object):
"""Volume abstraction."""
"""
:param lv_volume: Libvirt volume instance
"""
self.storage = lv_volume.storagePoolLookupByVolume().name()
self.path = lv_volume.path()
self.name = lv_volume.name()
self.capacity, self.allocation = lv_volume.info()[1:]