
from __future__ import absolute_import
from __future__ import division

from collections import namedtuple
from contextlib import closing, contextmanager
import errno
import io
import logging
import os
import re
import tarfile
import time
import threading
import xml.etree.ElementTree as ET
import zipfile

import libvirt

from vdsm.common import cmdutils
from vdsm.common import concurrent
from vdsm.common import libvirtconnection
from vdsm.common import password
from vdsm.common import response
from vdsm.common import zombiereaper
from vdsm.common.cmdutils import wrap_command
from vdsm.common.commands import execCmd, BUFFSIZE, terminating
from vdsm.common.compat import subprocess
from vdsm.common.config import config
from vdsm.common.define import errCode, doneCode
from vdsm.common.logutils import traceback
from vdsm.common.time import monotonic_time
from vdsm.constants import P_VDSM_LOG, P_VDSM_RUN, EXT_KVM_2_OVIRT
from vdsm.utils import NICENESS, IOCLASS

try:
    import ovirt_imageio_common
except ImportError:
    ovirt_imageio_common = None


_lock = threading.Lock()
_jobs = {}

_V2V_DIR = os.path.join(P_VDSM_RUN, 'v2v')
_LOG_DIR = os.path.join(P_VDSM_LOG, 'import')
_VIRT_V2V = cmdutils.CommandPath('virt-v2v', '/usr/bin/virt-v2v')
_SSH_AGENT = cmdutils.CommandPath('ssh-agent', '/usr/bin/ssh-agent')
_SSH_ADD = cmdutils.CommandPath('ssh-add', '/usr/bin/ssh-add')
_XEN_SSH_PROTOCOL = 'xen+ssh'
_VMWARE_PROTOCOL = 'vpx'
_KVM_PROTOCOL = 'qemu'
_SSH_AUTH_RE = b'(SSH_AUTH_SOCK)=([^;]+).*;\nSSH_AGENT_PID=(\d+)'
_OVF_RESOURCE_CPU = 3
_OVF_RESOURCE_MEMORY = 4
_OVF_RESOURCE_NETWORK = 10
_QCOW2_COMPAT_SUPPORTED = ('0.10', '1.1')

_OVF_ORIGIN_OVIRT = 3

_OVF_NS = 'http://schemas.dmtf.org/ovf/envelope/1'
_RASD_NS = 'http://schemas.dmtf.org/wbem/wscim/1/cim-schema/2/' \
           'CIM_ResourceAllocationSettingData'

ImportProgress = namedtuple('ImportProgress',
                            ['current_disk', 'disk_count', 'description'])
DiskProgress = namedtuple('DiskProgress', ['progress'])


class STATUS:
    STARTING = 'starting'
    COPYING_DISK = 'copying_disk'
    ABORTED = 'aborted'
    FAILED = 'error'
    DONE = 'done'


class V2VError(Exception):
    ''' Base class for v2v errors '''


class ClientError(Exception):
    err_name = 'unexpected'


class InvalidVMConfiguration(ValueError):
    ''' Unexpected error while parsing libvirt domain xml '''


class OutputParserError(V2VError):
    ''' Error while parsing virt-v2v output '''


class JobExistsError(ClientError):
    ''' Job already exists in _jobs collection '''
    err_name = 'JobExistsError'


class VolumeError(ClientError):
    ''' Error preparing volume '''


class NoSuchJob(ClientError):
    ''' Job not exists in _jobs collection '''
    err_name = 'NoSuchJob'


class JobNotDone(ClientError):
    ''' Import process still in progress '''
    err_name = 'JobNotDone'


class NoSuchOvf(V2VError):
    ''' Ovf path is not exists in /var/run/vdsm/v2v/ '''
    err_name = 'V2VNoSuchOvf'


class V2VProcessError(V2VError):
    ''' virt-v2v process had error in execution '''


class InvalidInputError(ClientError):
    ''' Invalid input received '''


def get_external_vms(uri, username, password, vm_names=None):
    if vm_names is not None:
        if not vm_names:
            vm_names = None
        else:
            vm_names = frozenset(vm_names)

    try:
        conn = libvirtconnection.open_connection(uri=uri,
                                                 username=username,
                                                 passwd=password)
    except libvirt.libvirtError as e:
        logging.exception('error connecting to hypervisor')
        return {'status': {'code': errCode['V2VConnection']['status']['code'],
                           'message': str(e)}}

    with closing(conn):
        vms = []
        for vm in _list_domains(conn):
            if vm_names is not None and vm.name() not in vm_names:
                continue
            _add_vm(conn, vms, vm)
        return {'status': doneCode, 'vmList': vms}


def get_external_vm_names(uri, username, password):
    try:
        conn = libvirtconnection.open_connection(uri=uri,
                                                 username=username,
                                                 passwd=password)
    except libvirt.libvirtError as e:
        logging.exception('error connecting to hypervisor')
        return response.error('V2VConnection', str(e))

    with closing(conn):
        vms = [vm.name() for vm in _list_domains(conn)]
        return response.success(vmNames=vms)


def convert_external_vm(uri, username, password, vminfo, job_id, irs):
    if uri.startswith(_XEN_SSH_PROTOCOL):
        command = XenCommand(uri, vminfo, job_id, irs)
    elif uri.startswith(_VMWARE_PROTOCOL):
        command = LibvirtCommand(uri, username, password, vminfo, job_id,
                                 irs)
    elif uri.startswith(_KVM_PROTOCOL):
        if ovirt_imageio_common is None:
            raise V2VError('Unsupported protocol KVM, ovirt_imageio_common'
                           'package is needed for importing KVM images')
        command = KVMCommand(uri, username, password, vminfo, job_id, irs)
    else:
        raise ClientError('Unknown protocol for Libvirt uri: %s', uri)
    job = ImportVm(job_id, command)
    job.start()
    _add_job(job_id, job)
    return {'status': doneCode}


def convert_ova(ova_path, vminfo, job_id, irs):
    command = OvaCommand(ova_path, vminfo, job_id, irs)
    job = ImportVm(job_id, command)
    job.start()
    _add_job(job_id, job)
    return response.success()


def get_ova_info(ova_path):
    ns = {'ovf': _OVF_NS, 'rasd': _RASD_NS}

    ovf_str = _read_ovf_from_ova(ova_path)
    try:
        root = ET.fromstring(ovf_str)
    except ET.ParseError as e:
        raise V2VError('Error reading ovf from ova, position: %r' % e.position)

    vm = {}
    _add_origin_ovf_info(vm, root, ovf_str)
    _add_general_ovf_info(vm, root, ns, ova_path)
    _add_disks_ovf_info(vm, root, ns)
    _add_networks_ovf_info(vm, root, ns)

    return response.success(vmList=vm)


def get_converted_vm(job_id):
    try:
        job = _get_job(job_id)
        _validate_job_done(job)
        ovf = _read_ovf(job_id)
    except ClientError as e:
        logging.info('Converted VM error %s', e)
        return errCode[e.err_name]
    except V2VError as e:
        logging.error('Converted VM error %s', e)
        return errCode[e.err_name]
    return {'status': doneCode, 'ovf': ovf}


def delete_job(job_id):
    try:
        job = _get_job(job_id)
        _validate_job_finished(job)
        _remove_job(job_id)
    except ClientError as e:
        logging.info('Cannot delete job, error: %s', e)
        return errCode[e.err_name]
    return {'status': doneCode}


def abort_job(job_id):
    try:
        job = _get_job(job_id)
        job.abort()
    except ClientError as e:
        logging.info('Cannot abort job, error: %s', e)
        return errCode[e.err_name]
    return {'status': doneCode}


def get_jobs_status():
    ret = {}
        items = tuple(_jobs.items())
    for job_id, job in items:
        ret[job_id] = {
            'status': job.status,
            'description': job.description,
            'progress': job.progress
        }
    return ret


def _add_job(job_id, job):
        if job_id in _jobs:
            raise JobExistsError("Job %r exists" % job_id)
        _jobs[job_id] = job


def _get_job(job_id):
        if job_id not in _jobs:
            raise NoSuchJob("No such job %r" % job_id)
        return _jobs[job_id]


def _remove_job(job_id):
        if job_id not in _jobs:
            raise NoSuchJob("No such job %r" % job_id)
        del _jobs[job_id]


def _validate_job_done(job):
    if job.status != STATUS.DONE:
        raise JobNotDone("Job %r is %s" % (job.id, job.status))


def _validate_job_finished(job):
    if job.status not in (STATUS.DONE, STATUS.FAILED, STATUS.ABORTED):
        raise JobNotDone("Job %r is %s" % (job.id, job.status))


def _read_ovf(job_id):
    file_name = os.path.join(_V2V_DIR, "%s.ovf" % job_id)
    try:
        with open(file_name, 'r') as f:
            return f.read()
    except IOError as e:
        if e.errno != errno.ENOENT:
            raise
        raise NoSuchOvf("No such ovf %r" % file_name)


class SSHAgent(object):
    def __init__(self):
        self._auth = None
        self._agent_pid = None
        self._ssh_auth_re = re.compile(_SSH_AUTH_RE)

    def __enter__(self):
        rc, out, err = execCmd([_SSH_AGENT.cmd], raw=True)
        if rc != 0:
            raise V2VError('Error init ssh-agent, exit code: %r'
                           ', out: %r, err: %r' %
                           (rc, out, err))

        m = self._ssh_auth_re.match(out)
        self._auth = {m.group(1): m.group(2)}
        self._agent_pid = m.group(3)

        try:
            rc, out, err = execCmd([_SSH_ADD.cmd], env=self._auth)
        except:
            self._kill_agent()
            raise

        if rc != 0:
            if rc != 2:
                self._kill_agent()
            raise V2VError('Error init ssh-add, exit code: %r'
                           ', out: %r, err: %r' %
                           (rc, out, err))

    def __exit__(self, *args):
        rc, out, err = execCmd([_SSH_ADD.cmd, '-d'], env=self._auth)
        if rc != 0:
            logging.error('Error deleting ssh-add, exit code: %r'
                          ', out: %r, err: %r' %
                          (rc, out, err))

        self._kill_agent()

    def _kill_agent(self):
        rc, out, err = execCmd([_SSH_AGENT.cmd, '-k'],
                               env={'SSH_AGENT_PID': self._agent_pid})
        if rc != 0:
            logging.error('Error killing ssh-agent (PID=%r), exit code: %r'
                          ', out: %r, err: %r' %
                          (self._agent_pid, rc, out, err))

    @property
    def auth(self):
        return self._auth


class V2VCommand(object):
    def __init__(self, vminfo, vmid, irs):
        self._vminfo = vminfo
        self._vmid = vmid
        self._irs = irs
        self._prepared_volumes = []
        self._passwd_file = os.path.join(_V2V_DIR, "%s.tmp" % vmid)
        self._password = password.ProtectedPassword('')
        self._base_command = [_VIRT_V2V.cmd, '-v', '-x']
        self._query_v2v_caps()
        if 'qcow2_compat' in vminfo:
            qcow2_compat = vminfo['qcow2_compat']
            if qcow2_compat not in _QCOW2_COMPAT_SUPPORTED:
                logging.error('Invalid QCOW2 compat version %r' %
                              qcow2_compat)
                raise ValueError('Invalid QCOW2 compat version %r' %
                                 qcow2_compat)
            if 'vdsm-compat-option' in self._v2v_caps:
                self._base_command.extend(['--vdsm-compat', qcow2_compat])
            elif qcow2_compat != '0.10':
                logging.info('virt-v2v not supporting qcow2 compat version: '
                             '%r', qcow2_compat)

    def execute(self):
        raise NotImplementedError("Subclass must implement this")

    def _command(self):
        raise NotImplementedError("Subclass must implement this")

    def _start_helper(self):
        timestamp = time.strftime('%Y%m%dT%H%M%S')
        log = os.path.join(_LOG_DIR,
                           "import-%s-%s.log" % (self._vmid, timestamp))
        logging.info("Storing import log at: %r", log)
        v2v = _simple_exec_cmd(self._command(),
                               nice=NICENESS.HIGH,
                               ioclass=IOCLASS.IDLE,
                               env=self._environment(),
                               stdout=subprocess.PIPE,
                               stderr=subprocess.STDOUT)
        tee = _simple_exec_cmd(['tee', log],
                               nice=NICENESS.HIGH,
                               ioclass=IOCLASS.IDLE,
                               stdin=v2v.stdout,
                               stdout=subprocess.PIPE)

        return PipelineProc(v2v, tee)

    def _get_disk_format(self):
        fmt = self._vminfo.get('format', 'raw').lower()
        return "qcow2" if fmt == "cow" else fmt

    def _disk_parameters(self):
        parameters = []
        for disk in self._vminfo['disks']:
            try:
                parameters.append('--vdsm-image-uuid')
                parameters.append(disk['imageID'])
                parameters.append('--vdsm-vol-uuid')
                parameters.append(disk['volumeID'])
            except KeyError as e:
                raise InvalidInputError('Job %r missing required property: %s'
                                        % (self._vmid, e))
        return parameters

    @contextmanager
    def _volumes(self):
        self._prepare_volumes()
        try:
            yield
        finally:
            self._teardown_volumes()

    def _prepare_volumes(self):
        if len(self._vminfo['disks']) < 1:
            raise InvalidInputError('Job %r cannot import vm with no disk',
                                    self._vmid)

        for disk in self._vminfo['disks']:
            drive = {'poolID': self._vminfo['poolID'],
                     'domainID': self._vminfo['domainID'],
                     'volumeID': disk['volumeID'],
                     'imageID': disk['imageID']}
            res = self._irs.prepareImage(drive['domainID'],
                                         drive['poolID'],
                                         drive['imageID'],
                                         drive['volumeID'])
            if res['status']['code']:
                raise VolumeError('Job %r bad volume specification: %s' %
                                  (self._vmid, drive))

            drive['path'] = res['path']
            self._prepared_volumes.append(drive)

    def _teardown_volumes(self):
        for drive in self._prepared_volumes:
            try:
                self._irs.teardownImage(drive['domainID'],
                                        drive['poolID'],
                                        drive['imageID'])
            except Exception as e:
                logging.error('Job %r error tearing down drive: %s',
                              self._vmid, e)

    def _get_storage_domain_path(self, path):
        return path.rsplit(os.sep, 3)[0]

    def _environment(self):
        env = os.environ.copy()

        env['LIBGUESTFS_BACKEND'] = 'direct'
        if 'virtio_iso_path' in self._vminfo:
            env['VIRTIO_WIN'] = self._vminfo['virtio_iso_path']
        return env

    @contextmanager
    def _password_file(self):
        fd = os.open(self._passwd_file, os.O_WRONLY | os.O_CREAT, 0o600)
        try:
            if self._password.value is None:
                os.write(fd, b"")
            else:
                os.write(fd, self._password.value.encode())
        finally:
            os.close(fd)
        try:
            yield
        finally:
            try:
                os.remove(self._passwd_file)
            except Exception:
                logging.exception("Job %r error removing passwd file: %s",
                                  self._vmid, self._passwd_file)

    def _query_v2v_caps(self):
        self._v2v_caps = frozenset()
        p = _simple_exec_cmd([_VIRT_V2V.cmd, '--machine-readable'],
                             env=os.environ.copy(),
                             stdout=subprocess.PIPE,
                             stderr=subprocess.PIPE)
        with terminating(p):
            try:
                out, err = p.communicate()
            except Exception:
                logging.exception('Terminating virt-v2v process after error')
                raise
        if p.returncode != 0:
            raise V2VProcessError(
                'virt-v2v exited with code: %d, stderr: %r' %
                (p.returncode, err))

        self._v2v_caps = frozenset(out.decode('utf8').splitlines())
        logging.debug("Detected virt-v2v capabilities: %r", self._v2v_caps)


class LibvirtCommand(V2VCommand):
    def __init__(self, uri, username, password, vminfo, vmid, irs):
        super(LibvirtCommand, self).__init__(vminfo, vmid, irs)
        self._uri = uri
        self._username = username
        self._password = password

    def _command(self):
        cmd = self._base_command
        cmd.extend(['-ic', self._uri,
                    '-o', 'vdsm',
                    '-of', self._get_disk_format(),
                    '-oa', self._vminfo.get('allocation', 'sparse').lower()])
        cmd.extend(self._disk_parameters())
        cmd.extend(['--password-file',
                    self._passwd_file,
                    '--vdsm-vm-uuid',
                    self._vmid,
                    '--vdsm-ovf-output',
                    _V2V_DIR,
                    '--machine-readable',
                    '-os',
                    self._get_storage_domain_path(
                        self._prepared_volumes[0]['path']),
                    self._vminfo['vmName']])
        return cmd

    @contextmanager
    def execute(self):
        with self._volumes(), self._password_file():
            yield self._start_helper()


class OvaCommand(V2VCommand):
    def __init__(self, ova_path, vminfo, vmid, irs):
        super(OvaCommand, self).__init__(vminfo, vmid, irs)
        self._ova_path = ova_path

    def _command(self):
        cmd = self._base_command
        cmd.extend(['-i', 'ova', self._ova_path,
                    '-o', 'vdsm',
                    '-of', self._get_disk_format(),
                    '-oa', self._vminfo.get('allocation', 'sparse').lower(),
                    '--vdsm-vm-uuid',
                    self._vmid,
                    '--vdsm-ovf-output',
                    _V2V_DIR,
                    '--machine-readable',
                    '-os',
                    self._get_storage_domain_path(
                        self._prepared_volumes[0]['path'])])
        cmd.extend(self._disk_parameters())
        return cmd

    @contextmanager
    def execute(self):
        with self._volumes():
            yield self._start_helper()


class XenCommand(V2VCommand):
    def __init__(self, uri, vminfo, job_id, irs):
        super(XenCommand, self).__init__(vminfo, job_id, irs)
        self._uri = uri
        self._ssh_agent = SSHAgent()

    def _command(self):
        cmd = self._base_command
        cmd.extend(['-ic', self._uri,
                    '-o', 'vdsm',
                    '-of', self._get_disk_format(),
                    '-oa', self._vminfo.get('allocation', 'sparse').lower()])
        cmd.extend(self._disk_parameters())
        cmd.extend(['--vdsm-vm-uuid',
                    self._vmid,
                    '--vdsm-ovf-output',
                    _V2V_DIR,
                    '--machine-readable',
                    '-os',
                    self._get_storage_domain_path(
                        self._prepared_volumes[0]['path']),
                    self._vminfo['vmName']])
        return cmd

    @contextmanager
    def execute(self):
        with self._volumes(), self._ssh_agent:
            yield self._start_helper()

    def _environment(self):
        env = super(XenCommand, self)._environment()
        env.update(self._ssh_agent.auth)
        return env


class KVMCommand(V2VCommand):
    def __init__(self, uri, username, password, vminfo, vmid, irs):
        super(KVMCommand, self).__init__(vminfo, vmid, irs)
        self._uri = uri
        self._username = username
        self._password = password

    def _command(self):
        cmd = [EXT_KVM_2_OVIRT,
               '--uri', self._uri,
               '--bufsize',
               str(config.getint('v2v', 'kvm2ovirt_buffer_size'))]
        if self._username is not None:
            cmd.extend([
                '--username', self._username,
                '--password-file', self._passwd_file])
        src, fmt = self._source_images()
        cmd.append('--source')
        cmd.extend(src)
        cmd.append('--dest')
        cmd.extend(self._dest_images())
        cmd.append('--storage-type')
        cmd.extend(fmt)
        cmd.append('--vm-name')
        cmd.append(self._vminfo['vmName'])
        cmd.append('--allocation')
        cmd.append(self._vminfo.get('allocation', 'sparse').lower())
        return cmd

    @contextmanager
    def execute(self):
        with self._volumes(), self._password_file():
            yield self._start_helper()

    def _source_images(self):
        con = libvirtconnection.open_connection(uri=self._uri,
                                                username=self._username,
                                                passwd=self._password)

        with closing(con):
            vm = con.lookupByName(self._vminfo['vmName'])
            if vm:
                params = {}
                root = ET.fromstring(vm.XMLDesc(0))
                _add_disks(root, params)
                src = []
                fmt = []
                for disk in params['disks']:
                    disk_info = _get_disk_info(con, disk, vm)
                    if disk_info is None:
                        break
                    disk.update(disk_info)
                    if 'vol-key' in disk:
                        src.append(disk['vol-key'])
                        fmt.append('volume')
                    elif 'alias' in disk:
                        src.append(disk['alias'])
                        fmt.append('path')
                return src, fmt

    def _dest_images(self):
        ret = []
        for vol in self._prepared_volumes:
            ret.append(vol['path'])
        return ret


class PipelineProc(object):

    def __init__(self, proc1, proc2):
        self._procs = (proc1, proc2)
        self._stdout = proc2.stdout

    def kill(self):
        for p in self._procs:
            logging.debug("Killing pid=%d", p.pid)
            try:
                p.kill()
            except OSError as e:
                if e.errno != errno.ESRCH:
                    raise e

    @property
    def pids(self):
        return [p.pid for p in self._procs]

    @property
    def returncode(self):
        ret = 0
        for p in self._procs:
            p.poll()
            if p.returncode is None:
                return None
            if p.returncode != 0 and ret == 0:
                ret = p.returncode

        return ret

    @property
    def stdout(self):
        return self._stdout

    def wait(self, timeout=None):
        if timeout is not None:
            deadline = monotonic_time() + timeout

            for p in self._procs:
                try:
                    p.wait(deadline - monotonic_time())
                except subprocess.TimeoutExpired:
                    return False
        else:
            for p in self._procs:
                p.wait()

        return True


class ImportVm(object):
    TERM_DELAY = 30
    PROC_WAIT_TIMEOUT = 30

    def __init__(self, job_id, command):
        self._id = job_id
        self._command = command
        self._thread = None

        self._status = STATUS.STARTING
        self._description = ''
        self._disk_progress = 0
        self._disk_count = 1
        self._current_disk = 1
        self._aborted = False
        self._proc = None

    def start(self):
        self._thread = concurrent.thread(self._run, name="v2v/" + self._id[:8])
        self._thread.start()

    def wait(self):
        if self._thread is not None and self._thread.is_alive():
            self._thread.join()

    @property
    def id(self):
        return self._id

    @property
    def status(self):
        return self._status

    @property
    def description(self):
        return self._description

    @property
    def progress(self):
        completed = (self._disk_count - 1) * 100
        return (completed + self._disk_progress) // self._disk_count

    @traceback(msg="Error importing vm")
    def _run(self):
        try:
            self._import()
        except Exception as ex:
            if self._aborted:
                logging.debug("Job %r was aborted", self._id)
            else:
                logging.exception("Job %r failed", self._id)
                self._status = STATUS.FAILED
                self._description = str(ex)
                try:
                    if self._proc is not None:
                        self._abort()
                except Exception as e:
                    logging.exception('Job %r, error trying to abort: %r',
                                      self._id, e)

    def _import(self):
        logging.info('Job %r starting import', self._id)

        with self._command.execute() as self._proc:
            self._watch_process_output()
            self._wait_for_process()

            if self._proc.returncode != 0:
                raise V2VProcessError('Job %r process failed exit-code: %r' %
                                      (self._id,
                                       self._proc.returncode))

            if self._status != STATUS.ABORTED:
                self._status = STATUS.DONE
                logging.info('Job %r finished import successfully',
                             self._id)

    def _wait_for_process(self):
        if self._proc.returncode is not None:
            return
        logging.debug("Job %r waiting for virt-v2v process", self._id)
        if not self._proc.wait(timeout=self.PROC_WAIT_TIMEOUT):
            raise V2VProcessError("Job %r timeout waiting for process pid=%s",
                                  self._id, self._proc.pids)

    def _watch_process_output(self):
        out = io.BufferedReader(io.FileIO(self._proc.stdout.fileno(),
                                mode='rb', closefd=False), BUFFSIZE)
        parser = OutputParser()
        for event in parser.parse(out):
            if isinstance(event, ImportProgress):
                self._status = STATUS.COPYING_DISK
                logging.info("Job %r copying disk %d/%d",
                             self._id, event.current_disk, event.disk_count)
                self._disk_progress = 0
                self._current_disk = event.current_disk
                self._disk_count = event.disk_count
                self._description = event.description
            elif isinstance(event, DiskProgress):
                self._disk_progress = event.progress
                if event.progress % 10 == 0:
                    logging.info("Job %r copy disk %d progress %d/100",
                                 self._id, self._current_disk, event.progress)
            else:
                raise RuntimeError("Job %r got unexpected parser event: %s" %
                                   (self._id, event))

    def abort(self):
        self._status = STATUS.ABORTED
        logging.info('Job %r aborting...', self._id)
        self._abort()

    def _abort(self):
        self._aborted = True
        if self._proc is None:
            logging.warning(
                'Ignoring request to abort job %r; the job failed to start',
                self._id)
            return
        if self._proc.returncode is None:
            logging.debug('Job %r killing virt-v2v process', self._id)
            try:
                self._proc.kill()
            except OSError as e:
                if e.errno != errno.ESRCH:
                    raise
                logging.debug('Job %r virt-v2v process not running',
                              self._id)
            else:
                logging.debug('Job %r virt-v2v process was killed',
                              self._id)
            finally:
                for pid in self._proc.pids:
                    zombiereaper.autoReapPID(pid)


class OutputParser(object):
    COPY_DISK_RE = re.compile(br'.*(Copying disk (\d+)/(\d+)).*')
    DISK_PROGRESS_RE = re.compile(br'\s+\((\d+).*')

    def parse(self, stream):
        for line in stream:
            if b'Copying disk' in line:
                description, current_disk, disk_count = self._parse_line(line)
                yield ImportProgress(int(current_disk), int(disk_count),
                                     description)
                for chunk in self._iter_progress(stream):
                    progress = self._parse_progress(chunk)
                    if progress is not None:
                        yield DiskProgress(progress)
                    if progress == 100:
                        break

    def _parse_line(self, line):
        m = self.COPY_DISK_RE.match(line)
        if m is None:
            raise OutputParserError('unexpected format in "Copying disk"'
                                    ', line: %r' % line)
        return m.group(1), m.group(2), m.group(3)

    def _iter_progress(self, stream):
        chunk = b''
        while True:
            c = stream.read(1)
            if not c:
                raise OutputParserError('copy-disk stream closed unexpectedly')
            chunk += c
            if c == b'\r':
                yield chunk
                chunk = b''

    def _parse_progress(self, chunk):
        m = self.DISK_PROGRESS_RE.match(chunk)
        if m is None:
            return None
        try:
            return int(m.group(1))
        except ValueError:
            raise OutputParserError('error parsing progress regex: %r'
                                    % m.groups)


def _mem_to_mib(size, unit):
    lunit = unit.lower()
    if lunit in ('bytes', 'b'):
        return size // 1024 // 1024
    elif lunit in ('kib', 'k'):
        return size // 1024
    elif lunit in ('mib', 'm'):
        return size
    elif lunit in ('gib', 'g'):
        return size * 1024
    elif lunit in ('tib', 't'):
        return size * 1024 * 1024
    else:
        raise InvalidVMConfiguration("Invalid currentMemory unit attribute:"
                                     " %r" % unit)


def _list_domains(conn):
    try:
        for vm in conn.listAllDomains():
            yield vm
    except libvirt.libvirtError as e:
        if e.get_error_code() != libvirt.VIR_ERR_NO_SUPPORT:
            raise
        seen = set()
        for name in conn.listDefinedDomains():
            try:
                vm = conn.lookupByName(name)
            except libvirt.libvirtError as e:
                logging.error("Error looking up vm %r: %s", name, e)
            else:
                seen.add(name)
                yield vm
        for domainId in conn.listDomainsID():
            try:
                vm = conn.lookupByID(domainId)
            except libvirt.libvirtError as e:
                logging.error("Error looking up vm by id %r: %s", domainId, e)
            else:
                if vm.name() not in seen:
                    yield vm


def _add_vm(conn, vms, vm):
    params = {}
    try:
        _add_vm_info(vm, params)
    except libvirt.libvirtError as e:
        logging.error("error getting domain information: %s", e)
        return
    try:
        xml = vm.XMLDesc(0)
    except libvirt.libvirtError as e:
        logging.error("error getting domain xml for vm %r: %s",
                      vm.name(), e)
        return
    try:
        root = ET.fromstring(xml)
    except ET.ParseError as e:
        logging.error('error parsing domain xml: %s', e)
        return
    if not _block_disk_supported(conn, root):
        return
    try:
        _add_general_info(root, params)
    except InvalidVMConfiguration as e:
        logging.error("error adding general info: %s", e)
        return
    _add_snapshot_info(conn, vm, params)
    _add_networks(root, params)
    _add_disks(root, params)
    _add_graphics(root, params)
    _add_video(root, params)

    disk_info = None
    for disk in params['disks']:
        disk_info = _get_disk_info(conn, disk, vm)
        if disk_info is None:
            break
        disk.update(disk_info)
    if disk_info is not None:
        vms.append(params)
    else:
        logging.warning('Cannot add VM %s due to disk storage error',
                        vm.name())


def _block_disk_supported(conn, root):
    if conn.getType() == 'Xen':
        block_disks = root.findall('.//disk[@type="block"]')
        block_disks = [d for d in block_disks
                       if d.attrib.get('device', None) == "disk"]
        return len(block_disks) == 0

    return True


def _add_vm_info(vm, params):
    params['vmName'] = vm.name()
    if vm.isActive():
        params['status'] = "Up"
    else:
        params['status'] = "Down"


def _extract_ns_map(xml_str):
    try:
        xml_str = xml_str.decode('utf-8')
    except AttributeError:
    xml_src = io.StringIO(xml_str)
    events = ("start-ns",)
    return {
        name: uri
        for event, (name, uri) in ET.iterparse(xml_src, events)
        if event == "start-ns"
    }


def _add_origin_ovf_info(vm, root, ovf_str):
    ns_map = _extract_ns_map(ovf_str)
    if 'ovirt' in ns_map:
        vm['originType'] = _OVF_ORIGIN_OVIRT


def _add_general_info(root, params):
    e = root.find('./uuid')
    if e is not None:
        params['vmId'] = e.text

    e = root.find('./currentMemory')
    if e is not None:
        try:
            size = int(e.text)
        except ValueError:
            raise InvalidVMConfiguration("Invalid 'currentMemory' value: %r"
                                         % e.text)
        unit = e.get('unit', 'KiB')
        params['memSize'] = _mem_to_mib(size, unit)

    e = root.find('./vcpu')
    if e is not None:
        try:
            params['smp'] = int(e.text)
        except ValueError:
            raise InvalidVMConfiguration("Invalid 'vcpu' value: %r" % e.text)

    e = root.find('./os/type/[@arch]')
    if e is not None:
        params['arch'] = e.get('arch')


def _get_disk_info(conn, disk, vm):
    if 'alias' in disk.keys():
        if disk['disktype'] not in ('file', 'block', 'volume'):
            logging.error('Unsupported disk type: %r', disk['disktype'])
            return None

        out = {}
        try:
            vol = None
            if disk['disktype'] == 'volume':
                pool = conn.storagePoolLookupByName(disk['pool'])
                vol = pool.storageVolLookupByName(disk['volume'])
            elif disk['disktype'] == 'file':
                try:
                    vol = conn.storageVolLookupByPath(disk['alias'])
                except libvirt.libvirtError:
                    logging.exception('Failed to get volume by path')
                    if conn.getType() == 'Xen':
                        logging.error('Disk has to be in storage pool')
                        return None

            if vol:
                _, capacity, alloc = vol.info()
                out['vol-key'] = vol.key()
            else:
                vol = vm.blockInfo(disk['alias'])
                capacity, _, alloc = vol

        except libvirt.libvirtError:
            logging.exception("Error getting disk size")
            return None
        out.update({
            'capacity': str(capacity),
            'allocation': str(alloc)})
        return out
    return {}


def _convert_disk_format(format):
    if format == 'qcow2':
        return 'COW'
    elif format == 'raw':
        return 'RAW'
    raise KeyError


def _add_disks(root, params):
    params['disks'] = []
    disks = root.findall('.//disk[@type="file"]')
    disks = disks + root.findall('.//disk[@type="block"]')
    disks = disks + root.findall('.//disk[@type="volume"]')
    for disk in disks:
        d = {}
        disktype = disk.get('type')
        device = disk.get('device')
        if device is not None:
            if device == 'cdrom' or device == 'floppy':
                continue
            d['type'] = device
        target = disk.find('./target/[@dev]')
        if target is not None:
            d['dev'] = target.get('dev')
        if disktype == 'file':
            d['disktype'] = 'file'
            source = disk.find('./source/[@file]')
            if source is not None:
                d['alias'] = source.get('file')
        elif disktype == 'block':
            d['disktype'] = 'block'
            source = disk.find('./source/[@dev]')
            if source is not None:
                d['alias'] = source.get('dev')
        elif disktype == 'volume':
            d['disktype'] = 'volume'
            source = disk.find('./source/[@pool]')
            if source is not None:
                d['pool'] = source.get('pool')
                d['volume'] = source.get('volume')
                d['alias'] = '%s/%s' % (d['pool'], d['volume'])
        else:
            logging.error('Unsupported disk type: %r', type)

        driver = disk.find('./driver/[@type]')
        if driver is not None:
            try:
                d["format"] = _convert_disk_format(driver.get('type'))
            except KeyError:
                logging.warning("Disk %s has unsupported format: %r", d,
                                format)
        params['disks'].append(d)


def _add_graphics(root, params):
    e = root.find('./devices/graphics/[@type]')
    if e is not None:
        params['graphics'] = e.get('type')


def _add_video(root, params):
    e = root.find('./devices/video/model/[@type]')
    if e is not None:
        params['video'] = e.get('type')


def _add_networks(root, params):
    params['networks'] = []
    interfaces = root.findall('.//interface')
    for iface in interfaces:
        i = {}
        if 'type' in iface.attrib:
            i['type'] = iface.attrib['type']
        mac = iface.find('./mac/[@address]')
        if mac is not None:
            i['macAddr'] = mac.get('address')
        source = iface.find('./source/[@bridge]')
        if source is not None:
            i['bridge'] = source.get('bridge')
        target = iface.find('./target/[@dev]')
        if target is not None:
            i['dev'] = target.get('dev')
        model = iface.find('./model/[@type]')
        if model is not None:
            i['model'] = model.get('type')
        params['networks'].append(i)


def _add_snapshot_info(conn, vm, params):
    if conn.getType() == 'Xen':
        return

    try:
        ret = vm.hasCurrentSnapshot()
    except libvirt.libvirtError:
        logging.exception('Error checking for existing snapshots.')
    else:
        params['has_snapshots'] = ret > 0


def _vm_has_snapshot(vm):
    try:
        return vm.hasCurrentSnapshot() == 1
    except libvirt.libvirtError:
        logging.exception('Error checking if snapshot exist for vm: %s.',
                          vm.name())
        return False


def _read_ovf_from_ova(ova_path):
    if os.path.isdir(ova_path):
        return _read_ovf_from_ova_dir(ova_path)
    elif zipfile.is_zipfile(ova_path):
        return _read_ovf_from_zip_ova(ova_path)
    elif tarfile.is_tarfile(ova_path):
        return _read_ovf_from_tar_ova(ova_path)
    raise ClientError('Unknown ova format, supported formats:'
                      ' tar, zip or a directory')


def _find_ovf(entries):
    for entry in entries:
        if '.ovf' == os.path.splitext(entry)[1].lower():
            return entry
    return None


def _read_ovf_from_ova_dir(ova_path):
    files = os.listdir(ova_path)
    name = _find_ovf(files)
    if name is not None:
        with open(os.path.join(ova_path, name), 'r') as ovf_file:
            return ovf_file.read()
    raise ClientError('OVA directory %s does not contain ovf file' % ova_path)


def _read_ovf_from_zip_ova(ova_path):
    with open(ova_path, 'rb') as fh:
        zf = zipfile.ZipFile(fh)
        name = _find_ovf(zf.namelist())
        if name is not None:
            return zf.read(name)
    raise ClientError('OVA does not contains file with .ovf suffix')


def _read_ovf_from_tar_ova(ova_path):
    with tarfile.open(ova_path) as tar:
        for member in tar:
            if member.name.endswith('.ovf'):
                ovf = tar.extractfile(member)
                with closing(ovf):
                    return ovf.read()
        raise ClientError('OVA does not contains file with .ovf suffix')


def _add_general_ovf_info(vm, node, ns, ova_path):
    vm['status'] = 'Down'
    vmName = node.find('./ovf:VirtualSystem/ovf:Name', ns)
    if vmName is not None:
        vm['vmName'] = vmName.text
    else:
        vm['vmName'] = os.path.splitext(os.path.basename(ova_path))[0]

    memSize = node.find('.//ovf:Item[rasd:ResourceType="%d"]/'
                        'rasd:VirtualQuantity' % _OVF_RESOURCE_MEMORY, ns)
    if memSize is not None:
        vm['memSize'] = int(memSize.text)
    else:
        raise V2VError('Error parsing ovf information: no memory size')

    smp = node.find('.//ovf:Item[rasd:ResourceType="%d"]/'
                    'rasd:VirtualQuantity' % _OVF_RESOURCE_CPU, ns)
    if smp is not None:
        vm['smp'] = int(smp.text)
    else:
        raise V2VError('Error parsing ovf information: no cpu info')


def _get_max_disk_size(populated_size, size):
    if populated_size is None:
        return size
    if size is None:
        return populated_size
    return str(max(int(populated_size), int(size)))


def _parse_allocation_units(units):
    sp = '[ \t\n]?'
    base_unit = 'byte'
    modifier1 = '(?P<m1>{op}{sp}(?P<m1_num>{num}))'.format(
        op=operator,
        num=number,
        sp=sp)
    modifier2 = \
        '(?P<m2>{op}{sp}' \
        '(?P<m2_base>[0-9]+){sp}\^{sp}(?P<m2_exp>{exp}))'.format(
            op=operator,
            exp=exponent,
            sp=sp)
    r = '^{base_unit}({sp}{mod1})?({sp}{mod2})?$'.format(
        base_unit=base_unit,
        mod1=modifier1,
        mod2=modifier2,
        sp=sp)

    m = re.match(r, units, re.MULTILINE)
    if m is None:
        raise V2VError('Failed to parse allocation units: %r' % units)
    g = m.groupdict()

    ret = 1
    if g['m1'] is not None:
        try:
            ret *= int(g['m1_num'])
        except ValueError:
            raise V2VError("Failed to parse allocation units: %r" % units)
    if g['m2'] is not None:
        try:
            ret *= pow(int(g['m2_base']), int(g['m2_exp']))
        except ValueError:
            raise V2VError("Failed to parse allocation units: %r" % units)

    return ret


def _add_disks_ovf_info(vm, node, ns):
    vm['disks'] = []
    for d in node.findall(".//ovf:DiskSection/ovf:Disk", ns):
        disk = {'type': 'disk'}
        capacity = int(d.attrib.get('{%s}capacity' % _OVF_NS))
        if '{%s}capacityAllocationUnits' % _OVF_NS in d.attrib:
            units = d.attrib.get('{%s}capacityAllocationUnits' % _OVF_NS)
            capacity *= _parse_allocation_units(units)
        disk['capacity'] = str(capacity)
        fileref = d.attrib.get('{%s}fileRef' % _OVF_NS)
        alias = node.find('.//ovf:References/ovf:File[@ovf:id="%s"]' %
                          fileref, ns)
        if alias is not None:
            disk['alias'] = alias.attrib.get('{%s}href' % _OVF_NS)
            populated_size = d.attrib.get('{%s}populatedSize' % _OVF_NS, None)
            size = alias.attrib.get('{%s}size' % _OVF_NS)
            disk['allocation'] = _get_max_disk_size(populated_size, size)
        else:
            raise V2VError('Error parsing ovf information: disk href info')
        vm['disks'].append(disk)


def _add_networks_ovf_info(vm, node, ns):
    vm['networks'] = []
    for n in node.findall('.//ovf:Item[rasd:ResourceType="%d"]'
                          % _OVF_RESOURCE_NETWORK, ns):
        net = {}
        dev = n.find('./rasd:ElementName', ns)
        if dev is not None:
            net['dev'] = dev.text
        else:
            raise V2VError('Error parsing ovf information: '
                           'network element name')

        model = n.find('./rasd:ResourceSubType', ns)
        if model is not None:
            net['model'] = model.text
        else:
            raise V2VError('Error parsing ovf information: network model')

        bridge = n.find('./rasd:Connection', ns)
        if bridge is not None:
            net['bridge'] = bridge.text
            net['type'] = 'bridge'
        else:
            net['type'] = 'interface'
        vm['networks'].append(net)


def _simple_exec_cmd(command, env=None, nice=None, ioclass=None,
                     stdin=None, stdout=None, stderr=None):

    command = wrap_command(command, with_ioclass=ioclass,
                           ioclassdata=None, with_nice=nice,
                           with_setsid=False, with_sudo=False,
                           reset_cpu_affinity=True)

    logging.debug(cmdutils.command_log_line(command, cwd=None))

    p = subprocess.Popen(
        command, close_fds=True, cwd=None, env=env,
        stdin=stdin, stdout=stdout, stderr=stderr)
    return p
