handlers.py

# -*- coding: utf-8 -*-

from logging import debug, warning, info
from fnmatch import fnmatchcase
from time import sleep
from sjrpc.core import RpcError
from sjrpc.utils import RpcHandler
from sjrpc.utils import pure
from errors import HypervisorError
from common import LocalHost
from jobs import ReceiveFileJob, SendFileJob
from __init__ import __version__


_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),
            '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),
        }
        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_maxbaloon', 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:
                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):
            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:
                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 handler 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('get_tags: 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('get_tags: 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:
                    warning('get_tags: `%s` -> `%s`', repr(err), err)
            debug('get_tags: 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('get_tags: add/correct mandatory tag `%s`', t)
                    mytags[t] = True
        # query host
        info('get_tags: 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('get_tags: plain mapping found for tag `%s`', tag)
                try:
                    if self.HV_TAG_MAP[tag][0](self._host_handle):
                        debug('get_tags: tag `%s` is available on host', tag)
                        result[tag] = {}
                        result[tag]['ttl'] = self.HV_TAG_MAP[tag][2]
                        if resolve:
                            debug('get_tags: resolving now tag `%s`', tag)
                            # fetch tag data
                            q = self.HV_TAG_MAP[tag][1](self._host_handle, tag)
                            debug('get_tags: host returned `%s`', q)
                            if q is not None:
                                # append tag data
                                result[tag]['value'] = str(q)
                            else:
                                debug('get_tags: I wont return `%s`=`None`',tag)
                    else:
                        debug('get_tags: tag `%s` is NOT implemented', tag)
                except Exception as err:
                    warning('get_tags: `%s` -> `%s`', repr(err), err)
            # 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 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('get_tags: testing pattern `%s`', pattern)
                            if fnmatchcase(tag, pattern):
                                debug('get_tags: 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('get_tags: 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:
                        warning('get_tags: `%s` -> `%s`', repr(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_tags: sub node implements `%s`' % pattern)
                    # get tags from helper
                    htags = handler[0](sub_obj, resolve=False)
                    debug('sub_tags: handler provides `%s`' % htags)
                    # append all tags
                    for t in htags.keys():
                        result.append(t)
            except Exception as err:
                warning('_sub_tag_list: `%s` -> `%s`', repr(err), err)
        return result
    
    @pure
    def sub_tags(self, sub_id, tags=None, noresolve_tags=None):
        '''
        '''
        info('sub_tags: server requested tags for `%s`', sub_id)
        if sub_id not in self._host_handle.vm_list():
            warning('sub_tags: sub node `%s` is unknown !', sub_id)
            raise HypervisorError('sub node `%s` is unknown' % sub_id)
        else:
            # open a wrapper to the VM
            debug('sub_tags: fetching vm data for `%s`', sub_id)
            sub = self._host_handle.vm_get(sub_id)
        # build a single list of tags
        info('sub_tags: 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('sub_tags: 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('sub_tags: 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:
                    info('sub_tags: 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('sub_tags: resolving tag `%s`', tag)
                                # call the wrapper mapping lambda
                                q = self.VM_TAG_MAP[tag][1](sub, tag)
                                info('sub_tags: tag query returned `%s`', q)
                                if q is not None:
                                    if resolve:
                                        result[tag]['value'] = str(q)
                                else:
                                    warning('sub_tags: I wont return `%s`=`None`'
                                                                        , tag)
                    except Exception as err:
                        warning('sub_tags: plain: `%s` -> `%s`', repr(err), err)
                # no tag mapping exist, test name against the globbing list
                else:
                    info('sub_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.VM_TAG_GLOB.iteritems():
                        try:
                            # helper is available on the current VM
                            if handler[0](sub, tag):
                                if fnmatchcase(tag, pattern):
                                    debug('sub_tags: 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:
                                        info('sub_tags: 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:
                            warning('sub_tags: glob: `%s` -> `%s`', repr(err),
                                                                            err)
        except Exception as err:
            warning('sub_tags: global: `%s` -> `%s`' % (repr(err), err))
        return result
    
    ##################################
    #   Host control
    ##################################
    
    @pure
    def node_shutdown(self, reboot=True, gracefull=True):
        '''
        '''
        warning('node_shutdown: server requested shutdown of local host')
        info('node_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)()
            info('node_shutdown: in progress ... action returned `%s`', result)
            return result
        else:
            error('node_shutdown: unable to proceed, this feature is not available')
            raise NotImplementedError('host handler has no method `%s`' %method)
    
    @pure
    def execute_command(self, command):
        '''
        '''
        error('execute_command: starting execution of `%s`' % command)
        output = self._host_handle.execute(command)
        info('execute_command: finished execution of `%s`' % command)
        return output
    
    ##################################
    #   VM management
    ##################################
    
    @pure
    def vm_define(self, data, format='xml'):
        '''
        '''
        warning('vm_define: server requested creation of a new VM')
        debug('vm_define: with description data `%s`' % data)
        if hasattr(self._host_handle, 'vm_define'):
            name = self._host_handle.vm_define(data)
            debug('vm_define: new VM has name `%s`' % name)
            return name
        else:
            raise NotImplementedError('host do not support VM creation')
    
    @pure
    def vm_undefine(self, name):
        '''
        '''
        warning('vm_undefine: server requested deletion of `%s`' % name)
        if hasattr(self._host_handle, 'vm_get'):
            vm = self._host_handle.vm_get(name)
            if hasattr(vm, 'undefine'):
                vm.undefine()
            else:
                raise NotImplementedError('VM object has not method `undefine`')
        else:
            raise NotImplementedError('host do not support ')
    
    @pure
    def vm_export(self, name, format='xml'):
        '''
        '''
        warning('vm_export: server requested configuration of `%s`' % name)
        if hasattr(self._host_handle, 'vm_get'):
            vm = self._host_handle.vm_get(name)
            if hasattr(vm, 'get_config'):
                return vm.get_config()
            else:
                raise NotImplementedError('VM object has not method `get_config`')
        else:
            raise NotImplementedError('host handler has not method `vm_get`')
    
    @pure
    def vm_stop(self, vm_names=None, force=False):
        '''
        '''
        warning('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):
        '''
        '''
        warning('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):
        '''
        '''
        warning('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):
        '''
        '''
        warning('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
    
    ##################################
    #   Storage control
    ##################################
    
    @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_export(self, pool, name, raddr, rport):
        rport = int(rport)
        if hasattr(self._host_handle, 'storage'):
            # get device path info
            sto = self._host_handle.storage()
            vol_path = sto.pool_get(pool).volume_get(name).get_path()
            # create job
            job = SendFileJob(self._host_handle.jobmgr, 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()
            res['checksum'] = job.get_checksum()
            return res
        else:
            raise NotImplementedError('host handler has no storage support')
    
    @pure
    def vol_import(self, pool, name, timeout=30):
        timeout = int(timeout)
        if hasattr(self._host_handle, 'storage'):
            # get device path info
            sto = self._host_handle.storage()
            vol_path = sto.pool_get(pool).volume_get(name).get_path()
            # create job
            # FIXME timeout
            job = ReceiveFileJob(self._host_handle.jobmgr, vol_path)
            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 the job to be finished
        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()
        res['checksum'] = job.get_checksum()
        return res
    
    @pure
    def vol_import_list(self):
        raise NotImplementedError()
    
    @pure
    def vol_import_cancel(self, jid):
        jid = int(jid)
        raise NotImplementedError()