
from __future__ import absolute_import
from __future__ import division

import io
import collections
import re
import threading
import time
import libvirt

from vdsm.common import concurrent
from vdsm.common import conv
from vdsm.common import logutils
from vdsm.common import response
from vdsm import sslutils
from vdsm import utils
from vdsm import jsonrpcvdscli
from vdsm.config import config
from vdsm.common.compat import pickle
from vdsm.common.define import NORMAL, Mbytes
from vdsm.common.network.address import normalize_literal_addr
from vdsm.virt.utils import DynamicBoundedSemaphore

from vdsm.virt import virdomain
from vdsm.virt import vmexitreason
from vdsm.virt import vmstatus
from six.moves import range


MODE_REMOTE = 'remote'
MODE_FILE = 'file'


METHOD_ONLINE = 'online'


incomingMigrations = DynamicBoundedSemaphore(
    max(1, config.getint('vars', 'max_incoming_migrations')))


CONVERGENCE_SCHEDULE_SET_DOWNTIME = "setDowntime"
CONVERGENCE_SCHEDULE_POST_COPY = "postcopy"
CONVERGENCE_SCHEDULE_SET_ABORT = "abort"


_MiB_IN_GiB = 1024


ADDRESS = '0'
PORT = 54321


class MigrationDestinationSetupError(RuntimeError):
    """
    Failed to create migration destination VM.
    """


class MigrationLimitExceeded(RuntimeError):
    """
    Cannot migrate right now: no resources on destination.
    """


class PostCopyPhase:
    NONE = 0
    REQUESTED = 1
    RUNNING = 2


class SourceThread(object):
    _RECOVERY_LOOP_PAUSE = 10

    ongoingMigrations = DynamicBoundedSemaphore(1)

    def __init__(self, vm, dst='', dstparams='',
                 mode=MODE_REMOTE, method=METHOD_ONLINE,
                 tunneled=False, dstqemu='', abortOnError=False,
                 consoleAddress=None, compressed=False,
                 autoConverge=False, recovery=False, **kwargs):
        self.log = vm.log
        self._vm = vm
        self._dst = dst
        self._mode = mode
        if method != METHOD_ONLINE:
            self.log.warning(
                'migration method %s is deprecated, forced to "online"',
                method)
        self._dstparams = dstparams
        self._enableGuestEvents = kwargs.get('enableGuestEvents', False)
        self._consoleAddress = consoleAddress
        self._dstqemu = dstqemu
        self._downtime = kwargs.get('downtime') or \
            config.get('vars', 'migration_downtime')
        self._maxBandwidth = int(
            kwargs.get('maxBandwidth') or
            config.getint('vars', 'migration_max_bandwidth')
        )
        self._incomingLimit = kwargs.get('incomingLimit')
        self._outgoingLimit = kwargs.get('outgoingLimit')
        self.status = {
            'status': {
                'code': 0,
                'message': 'Migration in progress'}}
        self._lock = threading.Lock()
        self._progress = 0
        self._thread = concurrent.thread(
            self.run, name='migsrc/' + self._vm.id[:8])
        self._preparingMigrationEvt = True
        self._migrationCanceledEvt = threading.Event()
        self._monitorThread = None
        self._destServer = None
        self._convergence_schedule = {
            'init': [],
            'stalling': []
        }
        self._use_convergence_schedule = False
        if 'convergenceSchedule' in kwargs:
            self._convergence_schedule = kwargs.get('convergenceSchedule')
            self._use_convergence_schedule = True
            self.log.debug('convergence schedule set to: %s',
                           str(self._convergence_schedule))
        self._started = False
        self._failed = False
        self._recovery = recovery
        tunneled = conv.tobool(tunneled)
        abortOnError = conv.tobool(abortOnError)
        compressed = conv.tobool(compressed)
        autoConverge = conv.tobool(autoConverge)
        self._migration_flags = self._calculate_migration_flags(tunneled,
                                                                abortOnError,
                                                                compressed,
                                                                autoConverge)

    def start(self):
        self._thread.start()

    def is_alive(self):
        return self._thread.is_alive()

    def migrating(self):
        return ((self.is_alive() and not self._failed) or
                (self._recovery and
                 self._vm.lastStatus == vmstatus.MIGRATION_SOURCE))

    @property
    def started(self):
        return self._started

    @property
    def hibernating(self):
        return self._mode == MODE_FILE

    def _update_progress(self):
        if self._monitorThread is None:
            return

        if self._monitorThread.progress is not None:
            progress = self._monitorThread.progress.percentage
        else:
            progress = 0

        with self._lock:
            old_progress = self._progress
            if progress >= old_progress:
                self._progress = progress

        if progress < old_progress:
            self.log.info(
                'new computed progress %d < than old value %d, discarded',
                progress, old_progress)

    def getStat(self):
        self._update_progress()
        self.status['progress'] = self._progress
        return self.status

    def _createClient(self, port):
        sslctx = sslutils.create_ssl_context()

        def is_ipv6_address(a):
            return (':' in a) and a.startswith('[') and a.endswith(']')

        if is_ipv6_address(self.remoteHost):
            host = self.remoteHost[1:-1]
        else:
            host = self.remoteHost

        client_socket = utils.create_connected_socket(host, int(port), sslctx)
        return self._vm.cif.createStompClient(client_socket)

    def _setupVdsConnection(self):
        if self.hibernating:
            return

        hostPort = _cannonize_host_port(
            self._dst,
            config.getint('addresses', 'management_port'))
        self.remoteHost, port = hostPort.rsplit(':', 1)

        client = self._createClient(port)
        requestQueues = config.get('addresses', 'request_queues')
        requestQueue = requestQueues.split(",")[0]
        self._destServer = jsonrpcvdscli.connect(requestQueue, client)
        self.log.debug('Initiating connection with destination')
        self._destServer.ping()

        self.log.debug('Destination server is: ' + hostPort)

    def _setupRemoteMachineParams(self):
        machineParams = self._vm.migration_parameters()
        machineParams['enableGuestEvents'] = self._enableGuestEvents
        if not self.hibernating:
            machineParams['migrationDest'] = 'libvirt'
        return machineParams

    def _prepareGuest(self):
        if self.hibernating:
            self.log.debug("Save State begins")
            if self._vm.guestAgent.isResponsive():
                lockTimeout = 30
            else:
                lockTimeout = 0
            self._vm.guestAgent.desktopLock()
            while lockTimeout:
                if self._vm.getStats()['session'] in ["Locked", "LoggedOff"]:
                    break
                time.sleep(1)
                lockTimeout -= 1
                if lockTimeout == 0:
                    self.log.warning('Agent ' + self._vm.id +
                                     ' unresponsive. Hiberanting without '
                                     'desktopLock.')
                    break
            self._vm.pause(vmstatus.SAVING_STATE)
        else:
            self.log.debug("Migration started")
            self._vm.lastStatus = vmstatus.MIGRATION_SOURCE

    def _recover(self, message):
        if not response.is_error(self.status):
            self.status = response.error('migrateErr')
        self.log.error(message)
        if not self.hibernating and self._destServer is not None:
            if self._vm.post_copy == PostCopyPhase.RUNNING:
                self._vm.handle_failed_post_copy(clean_vm=True)
                return
            try:
                self._destServer.destroy(self._vm.id)
            except Exception:
                self.log.exception("Failed to destroy remote VM")
        if self.hibernating:
            self._vm.cont(ignoreStatus=True)
            if self._enableGuestEvents:
                self._vm.guestAgent.events.after_hibernation_failure()
        elif self._enableGuestEvents:
            self._vm.guestAgent.events.after_migration_failure()
        self._failed = True
        if self._recovery:
            self._vm.set_last_status(vmstatus.UP, vmstatus.MIGRATION_SOURCE)
            self._recovery = False
        else:
            self._vm.lastStatus = vmstatus.UP
        self._started = False
        self._vm.send_status_event()

    def _finishSuccessfully(self, machineParams):
        with self._lock:
            self._progress = 100
        if not self.hibernating:
            self._vm.stopped_migrated_event_processed.wait()
            self._vm.setDownStatus(NORMAL, vmexitreason.MIGRATION_SUCCEEDED)
            self.status['status']['message'] = 'Migration done'
            if self._vm.post_copy == PostCopyPhase.RUNNING:
                self._vm.destroy()
        else:
            for ignoreParam in ('displayIp', 'display', 'pid'):
                if ignoreParam in machineParams:
                    del machineParams[ignoreParam]

            fname = self._vm.cif.prepareVolumePath(self._dstparams)
            try:
                with io.open(fname, "r+b") as f:
                    pickle.dump(machineParams, f)
            finally:
                self._vm.cif.teardownVolumePath(self._dstparams)

            self._vm.setDownStatus(NORMAL, vmexitreason.SAVE_STATE_SUCCEEDED)
            self.status['status']['message'] = 'SaveState done'

    @staticmethod
    def _raiseAbortError():
        e = libvirt.libvirtError(defmsg='')
        raise e

    def _update_outgoing_limit(self):
        if self._outgoingLimit:
            self.log.debug('Setting outgoing migration limit to %s',
                           self._outgoingLimit)
            SourceThread.ongoingMigrations.bound = self._outgoingLimit

    @property
    def recovery(self):
        return self._recovery

    def run(self):
        if self.recovery:
            self._recovery_run()
        else:
            self._regular_run()

    def _regular_run(self):
        self.log.debug("Starting migration source thread")
        self._recovery = False
        self._update_outgoing_limit()
        try:
            startTime = time.time()
            machineParams = self._setupRemoteMachineParams()
            self._setupVdsConnection()
            self._prepareGuest()

            while not self._started:
                try:
                    self.log.info("Migration semaphore: acquiring")
                    with SourceThread.ongoingMigrations:
                        self.log.info("Migration semaphore: acquired")
                        timeout = config.getint(
                            'vars', 'guest_lifecycle_event_reply_timeout')
                        if self.hibernating:
                            self._vm.guestAgent.events.before_hibernation(
                                wait_timeout=timeout)
                        elif self._enableGuestEvents:
                            self._vm.guestAgent.events.before_migration(
                                wait_timeout=timeout)
                        if self._migrationCanceledEvt.is_set():
                            self._raiseAbortError()
                        self.log.debug("migration semaphore acquired "
                                       "after %d seconds",
                                       time.time() - startTime)
                        migrationParams = {
                            'dst': self._dst,
                            'mode': self._mode,
                            'method': METHOD_ONLINE,
                            'dstparams': self._dstparams,
                            'dstqemu': self._dstqemu,
                        }
                        self._startUnderlyingMigration(
                            time.time(), migrationParams, machineParams
                        )
                        self._finishSuccessfully(machineParams)
                except libvirt.libvirtError as e:
                    if e.get_error_code() == libvirt.VIR_ERR_OPERATION_ABORTED:
                        self.status = response.error(
                            'migCancelErr', message='Migration canceled')
                    raise
                except MigrationLimitExceeded:
                    retry_timeout = config.getint('vars',
                                                  'migration_retry_timeout')
                    self.log.debug("Migration destination busy. Initiating "
                                   "retry in %d seconds.", retry_timeout)
                    self._migrationCanceledEvt.wait(retry_timeout)
        except MigrationDestinationSetupError as e:
            self._recover(str(e))
        except Exception as e:
            self._recover(str(e))
            self.log.exception("Failed to migrate")

    def _startUnderlyingMigration(self, startTime, migrationParams,
                                  machineParams):
        if self.hibernating:
            self._started = True
            self._vm.hibernate(self._dst)
        else:
            self._vm.prepare_migration()

            destCreateStartTime = time.time()
            result = self._destServer.migrationCreate(machineParams,
                                                      self._incomingLimit)
            destCreationTime = time.time() - destCreateStartTime
            startTime += destCreationTime
            self.log.info('Creation of destination VM took: %d seconds',
                          destCreationTime)

            if response.is_error(result):
                self.status = result
                if response.is_error(result, 'migrateLimit'):
                    raise MigrationLimitExceeded()
                else:
                    raise MigrationDestinationSetupError(
                        'migration destination error: ' +
                        result['status']['message'])

            self._started = True

            if config.getboolean('vars', 'ssl'):
                transport = 'tls'
            else:
                transport = 'tcp'
            duri = 'qemu+{}://{}/system'.format(
                transport, normalize_literal_addr(self.remoteHost))

            dstqemu = migrationParams['dstqemu']
            if dstqemu:
                muri = 'tcp://{}'.format(
                    normalize_literal_addr(dstqemu))
            else:
                muri = 'tcp://{}'.format(
                    normalize_literal_addr(self.remoteHost))

            self._vm.log.info('starting migration to %s '
                              'with miguri %s', duri, muri)

            self._monitorThread = MonitorThread(self._vm, startTime,
                                                self._convergence_schedule,
                                                self._use_convergence_schedule)

            if self._use_convergence_schedule:
                self._perform_with_conv_schedule(duri, muri)
            else:
                self._perform_with_downtime_thread(duri, muri)

            self.log.info("migration took %d seconds to complete",
                          (time.time() - startTime) + destCreationTime)

    def _perform_migration(self, duri, muri):
        if self._vm.hasSpice and self._vm.conf.get('clientIp'):
            SPICE_MIGRATION_HANDOVER_TIME = 120
            self._vm._reviveTicket(SPICE_MIGRATION_HANDOVER_TIME)

        self._preparingMigrationEvt = False
        if not self._migrationCanceledEvt.is_set():
            self._vm._dom.migrateToURI3(duri,
                                        self._migration_params(muri),
                                        self._migration_flags)
        else:
            self._raiseAbortError()

    def _migration_params(self, muri):
        params = {libvirt.VIR_MIGRATE_PARAM_BANDWIDTH: self._maxBandwidth}
        if not self.tunneled:
            params[libvirt.VIR_MIGRATE_PARAM_URI] = str(muri)
        if self._consoleAddress:
            graphics = 'spice' if self._vm.hasSpice else 'vnc'
            params[libvirt.VIR_MIGRATE_PARAM_GRAPHICS_URI] = str(
                '%s://%s' % (graphics, self._consoleAddress)
            )
        return params

    @property
    def tunneled(self):
        return self.migration_flags & libvirt.VIR_MIGRATE_TUNNELLED

    @property
    def migration_flags(self):
        return self._migration_flags

    def _calculate_migration_flags(self, tunneled, abort_on_error,
                                   compressed, auto_converge):
        flags = libvirt.VIR_MIGRATE_LIVE | libvirt.VIR_MIGRATE_PEER2PEER
        if tunneled:
            flags |= libvirt.VIR_MIGRATE_TUNNELLED
        if abort_on_error:
            flags |= libvirt.VIR_MIGRATE_ABORT_ON_ERROR
        if compressed:
            flags |= libvirt.VIR_MIGRATE_COMPRESSED
        if auto_converge:
            flags |= libvirt.VIR_MIGRATE_AUTO_CONVERGE
        if self._vm.min_cluster_version(4, 2):
            flags |= libvirt.VIR_MIGRATE_PERSIST_DEST
        for stalling in self._convergence_schedule.get('stalling', []):
            action = stalling.get('action', {}).get('name')
            if action == CONVERGENCE_SCHEDULE_POST_COPY:
                flags |= libvirt.VIR_MIGRATE_POSTCOPY
                break
        return flags

    def _perform_with_downtime_thread(self, duri, muri):
        self._vm.log.debug('performing migration with downtime thread')
        self._monitorThread.downtime_thread = DowntimeThread(
            self._vm,
            int(self._downtime),
            config.getint('vars', 'migration_downtime_steps')
        )

        with utils.running(self._monitorThread):
            self._perform_migration(duri, muri)

        self._monitorThread.join()

    def _perform_with_conv_schedule(self, duri, muri):
        self._vm.log.debug('performing migration with conv schedule')
        with utils.running(self._monitorThread):
            self._perform_migration(duri, muri)
        self._monitorThread.join()

    def set_max_bandwidth(self, bandwidth):
        self._vm.log.debug('setting migration max bandwidth to %d', bandwidth)
        self._maxBandwidth = bandwidth
        self._vm._dom.migrateSetMaxSpeed(bandwidth)

    def stop(self):
        try:
            self._migrationCanceledEvt.set()
            self._vm._dom.abortJob()
        except libvirt.libvirtError:
            if not self._preparingMigrationEvt:
                raise
        if self._recovery:
            self._recover("Migration stopped")

    def _recovery_run(self):
        self.log.debug("Starting migration recovery thread")
        while True:
            job_stats = self._vm._dom.jobStats()
            if not ongoing(job_stats):
                break
            time.sleep(self._RECOVERY_LOOP_PAUSE)
        self.log.debug("Recovered migration finished")
        if self._vm._dom.state(0)[0] == libvirt.VIR_DOMAIN_RUNNING:
            self.recovery_cleanup()

    def recovery_cleanup(self):
        if self._recovery and \
           self._vm.lastStatus == vmstatus.MIGRATION_SOURCE:
            self._recover("Migration failed")


def exponential_downtime(downtime, steps):
    if steps > 1:
        offset = downtime / float(steps)
        base = (downtime - offset) ** (1 / float(steps - 1))

        for i in range(steps):
            yield int(offset + base ** i)
    else:
        yield downtime


class DowntimeThread(object):


    def __init__(self, vm, downtime, steps):
        self._vm = vm
        self._downtime = downtime
        self._steps = steps
        self._stop = threading.Event()

        delay_per_gib = config.getint('vars', 'migration_downtime_delay')
        memSize = vm.mem_size_mb()
        self._wait = min(
            delay_per_gib * memSize / (_MiB_IN_GiB * self._steps),
            self._WAIT_STEP_LIMIT)
        self._downtimes = exponential_downtime(self._downtime, self._steps)
        self._initial_downtime = next(self._downtimes)

        self._thread = concurrent.thread(
            self.run, name='migdwn/' + self._vm.id[:8])

    def start(self):
        self._thread.start()

    def join(self):
        self._thread.join()

    def is_alive(self):
        return self._thread.is_alive()

    @logutils.traceback()
    def run(self):
        self._vm.log.debug('migration downtime thread started (%i steps)',
                           self._steps)

        for downtime in self._downtimes:
            if self._stop.is_set():
                break

            self._set_downtime(downtime)

            self._stop.wait(self._wait)

        self._vm.log.debug('migration downtime thread exiting')

    def set_initial_downtime(self):
        self._set_downtime(self._initial_downtime)

    def stop(self):
        self._vm.log.debug('stopping migration downtime thread')
        self._stop.set()

    def _set_downtime(self, downtime):
        self._vm.log.debug('setting migration downtime to %d', downtime)
        self._vm._dom.migrateSetMaxDowntime(downtime, 0)


class _FakeThreadInterface(object):

    def start(self):
        pass

    def stop(self):
        pass

    def join(self):
        pass

    def is_alive(self):
        return False

    def set_initial_downtime(self):
        pass


class MonitorThread(object):
    _MIGRATION_MONITOR_INTERVAL = config.getint(

    def __init__(self, vm, startTime, conv_schedule, use_conv_schedule):
        super(MonitorThread, self).__init__()
        self._stop = threading.Event()
        self._vm = vm
        self._startTime = startTime
        self.daemon = True
        self.progress = None
        self._conv_schedule = conv_schedule
        self._use_conv_schedule = use_conv_schedule
        self.downtime_thread = _FakeThreadInterface()
        self._thread = concurrent.thread(
            self.run, name='migmon/' + self._vm.id[:8])

    def start(self):
        self._thread.start()

    def join(self):
        self._thread.join()

    @property
    def enabled(self):
        return MonitorThread._MIGRATION_MONITOR_INTERVAL > 0

    @logutils.traceback()
    def run(self):
        if self.enabled:
            self._vm.log.debug('starting migration monitor thread')
            try:
                self.monitor_migration()
            except virdomain.NotConnectedError as e:
                self._vm.log.debug('domain disconnected in monitor thread: %s',
                                   e)
            finally:
                self.downtime_thread.stop()
            if self.downtime_thread.is_alive():
                self.downtime_thread.join()
            self._vm.log.debug('stopped migration monitor thread')
        else:
            self._vm.log.info('migration monitor thread disabled'
                              ' (monitoring interval set to 0)')

    def monitor_migration(self):
        memSize = self._vm.mem_size_mb()
        maxTimePerGiB = config.getint('vars',
                                      'migration_max_time_per_gib_mem')
        migrationMaxTime = (maxTimePerGiB * memSize + 1023) // 1024
        progress_timeout = config.getint('vars', 'migration_progress_timeout')
        lastProgressTime = time.time()
        lowmark = None
        initial_iteration = last_iteration = None

        self._execute_init(self._conv_schedule['init'])
        if not self._use_conv_schedule:
            self._vm.log.debug('setting initial migration downtime')
            self.downtime_thread.set_initial_downtime()

        while not self._stop.isSet():
            stopped = self._stop.wait(self._MIGRATION_MONITOR_INTERVAL)
            if stopped:
                break

            job_stats = self._vm._dom.jobStats()
            if not ongoing(job_stats):
                continue

            progress = Progress.from_job_stats(job_stats)
            if initial_iteration is None:
                initial_iteration = last_iteration = progress.mem_iteration

            self._vm.send_migration_status_event()

            now = time.time()
            if self._vm.post_copy != PostCopyPhase.NONE:
                if self._vm.post_copy == PostCopyPhase.RUNNING:
                    self._vm.log.debug(
                        'Post-copy migration still in progress: %d',
                        progress.data_remaining
                    )
            elif not self._use_conv_schedule and\
                    (0 < migrationMaxTime < now - self._startTime):
                self._vm.log.warn('The migration took %d seconds which is '
                                  'exceeding the configured maximum time '
                                  'for migrations of %d seconds. The '
                                  'migration will be aborted.',
                                  now - self._startTime,
                                  migrationMaxTime)
                self._vm._dom.abortJob()
                self.stop()
                break
            elif (lowmark is None) or (lowmark > progress.data_remaining):
                lowmark = progress.data_remaining
                lastProgressTime = now
            else:
                self._vm.log.warn(
                    'Migration stalling: remaining (%sMiB)'
                    ' > lowmark (%sMiB).',
                    progress.data_remaining // Mbytes, lowmark // Mbytes)

            if not self._vm.post_copy and\
               progress.mem_iteration > last_iteration:
                last_iteration = progress.mem_iteration
                current_iteration = last_iteration - initial_iteration
                self._vm.log.debug('new iteration: %i', current_iteration)
                if self._use_conv_schedule:
                    self._next_action(current_iteration)
                elif not self.downtime_thread.is_alive():
                    self.downtime_thread.start()

            if not self._use_conv_schedule and\
                    (now - lastProgressTime) > progress_timeout:
                self._vm.log.warn(
                    'Migration is stuck: Hasn\'t progressed in %s seconds. '
                    'Aborting.' % (now - lastProgressTime))
                self._vm._dom.abortJob()
                self.stop()

            if self._stop.isSet():
                break

            self.progress = progress
            self._vm.log.info('%s', progress)

    def stop(self):
        self._vm.log.debug('stopping migration monitor thread')
        self._stop.set()

    def _next_action(self, stalling):
        head = self._conv_schedule['stalling'][0]

        self._vm.log.debug('Stalling for %d iterations, '
                           'checking to make next action: '
                           '%s', stalling, head)
        if head['limit'] < stalling:
            self._execute_action_with_params(head['action'])
            self._conv_schedule['stalling'].pop(0)
            self._vm.log.debug('setting conv schedule to: %s',
                               self._conv_schedule)

    def _execute_init(self, init_actions):
        for action_with_params in init_actions:
            self._execute_action_with_params(action_with_params)

    def _execute_action_with_params(self, action_with_params):
        action = str(action_with_params['name'])
        vm = self._vm
        if action == CONVERGENCE_SCHEDULE_SET_DOWNTIME:
            downtime = int(action_with_params['params'][0])
            vm.log.debug('Setting downtime to %d', downtime)
            vm._dom.migrateSetMaxDowntime(downtime, 0)
        elif action == CONVERGENCE_SCHEDULE_POST_COPY:
            if not self._vm.switch_migration_to_post_copy():
                vm.log.warn('Failed to switch to post-copy migration')
        elif action == CONVERGENCE_SCHEDULE_SET_ABORT:
            vm.log.warn('Aborting migration')
            vm._dom.abortJob()
            self.stop()


_Progress = collections.namedtuple('_Progress', [
    'job_type', 'time_elapsed', 'data_total',
    'data_processed', 'data_remaining',
    'mem_total', 'mem_processed', 'mem_remaining',
    'mem_bps', 'mem_constant', 'compression_bytes',
    'dirty_rate', 'mem_iteration'
])


class Progress(_Progress):
    __slots__ = ()

    @classmethod
    def from_job_stats(cls, stats):
        return cls(
            stats['type'],
            stats[libvirt.VIR_DOMAIN_JOB_TIME_ELAPSED],
            stats[libvirt.VIR_DOMAIN_JOB_DATA_TOTAL],
            stats[libvirt.VIR_DOMAIN_JOB_DATA_PROCESSED],
            stats[libvirt.VIR_DOMAIN_JOB_DATA_REMAINING],
            stats[libvirt.VIR_DOMAIN_JOB_MEMORY_TOTAL],
            stats[libvirt.VIR_DOMAIN_JOB_MEMORY_PROCESSED],
            stats[libvirt.VIR_DOMAIN_JOB_MEMORY_REMAINING],
            stats.get(libvirt.VIR_DOMAIN_JOB_MEMORY_BPS, 0),
            stats.get(libvirt.VIR_DOMAIN_JOB_MEMORY_CONSTANT, 0),
            stats.get(libvirt.VIR_DOMAIN_JOB_COMPRESSION_BYTES, 0),
            stats.get('memory_dirty_rate', -1),
            stats.get('memory_iteration', -1),
        )

    def __str__(self):
        return (
            'Migration Progress: %s seconds elapsed,'
            ' %s%% of data processed,'
            ' total data: %iMB,'
            ' processed data: %iMB, remaining data: %iMB,'
            ' transfer speed %iMBps, zero pages: %iMB,'
            ' compressed: %iMB, dirty rate: %i,'
            ' memory iteration: %i' % (
                (self.time_elapsed / 1000),
                self.percentage,
                (self.data_total // Mbytes),
                (self.data_processed // Mbytes),
                (self.data_remaining // Mbytes),
                (self.mem_bps // Mbytes),
                self.mem_constant,
                (self.compression_bytes // Mbytes),
                self.dirty_rate,
                self.mem_iteration,
            )
        )

    @property
    def percentage(self):
        if self.data_remaining == 0 and self.data_total:
            return 100
        progress = 0
        if self.data_total:
            progress = 100 - 100 * self.data_remaining // self.data_total
        if progress < 100:
            return progress
        return 99


def ongoing(stats):
    try:
        job_type = stats['type']
    except KeyError:
        return False
    else:
        return (job_type != libvirt.VIR_DOMAIN_JOB_NONE and
                stats['operation'] ==
                libvirt.VIR_DOMAIN_JOB_OPERATION_MIGRATION_OUT)


def __guess_defaults():
    global ADDRESS, PORT
    try:
        from vdsm.config import config
        PORT = config.getint('addresses', 'management_port')
        ADDRESS = config.get('addresses', 'management_ip')
        if ADDRESS == '::':
            ADDRESS = 'localhost'
    except:
        pass


__guess_defaults()


def _cannonize_host_port(hostPort=None, port=PORT):
    if hostPort is None or hostPort == '0':
        addr = ADDRESS
        if ':' in addr:
            addr = '[%s]' % addr
    else:
        hostPort = re.match(r'(?P<Host>.+?)(:(?P<Port>\d+))?$', hostPort)
        addr = hostPort.group('Host')
        if hostPort.group('Port'):
            port = int(hostPort.group('Port'))
    return '%s:%i' % (addr, port)
