
from __future__ import absolute_import
from __future__ import division

"""
Code to perform periodic maintenance and bookkeeping of the VMs.
"""

import logging
import threading

import libvirt
import six

from vdsm import executor
from vdsm import host
from vdsm import throttledlog
from vdsm.common import errors
from vdsm.common import exception
from vdsm.common import libvirtconnection
from vdsm.config import config
from vdsm.virt import migration
from vdsm.virt import recovery
from vdsm.virt import sampling
from vdsm.virt import virdomain
from vdsm.virt import vmstatus


_WORKERS = config.getint('sampling', 'periodic_workers')
_TASK_PER_WORKER = config.getint('sampling', 'periodic_task_per_worker')
_TASKS = _WORKERS * _TASK_PER_WORKER
_MAX_WORKERS = config.getint('sampling', 'max_workers')

_operations = []
_executor = None


class Error(errors.Base):
    msg = 'Generic error for periodic infrastructure'


class InvalidValue(Error):
    msg = 'Invalid {self.value} for {self.key} for Operation {self.op_desc}'

    def __init__(self, op_desc, key, value):
        self.op_desc = op_desc
        self.key = key
        self.value = value


def _timeout_from(interval):
    """
    Estimate a sensible timeout given a periodic interval.
    """
    return interval / 2.


def start(cif, scheduler):
    global _executor
    global _operations

    _executor = executor.Executor(name="periodic",
                                  workers_count=_WORKERS,
                                  max_tasks=_TASKS,
                                  scheduler=scheduler,
                                  max_workers=_MAX_WORKERS)

    _executor.start()

    _operations = _create(cif, scheduler)

    if config.getboolean('sampling', 'enable'):
        host.stats.start()

    for op in _operations:
        try:
            op.start()
        except Error as e:
            logging.warning('Operation not started: %s', e)


def stop():
    for op in _operations:
        op.stop()

    _executor.stop(wait=False)


class Operation(object):

    _log = logging.getLogger("virt.periodic.Operation")

    def __init__(self, func, period, scheduler, timeout=0, executor=None,
                 exclusive=False, discard=True):
        self._func = func
        self._period = period
        self._timeout = _timeout_from(period) if timeout == 0 else timeout
        self._scheduler = scheduler
        self._executor = _executor if executor is None else executor
        self._exclusive = exclusive
        self._discard = discard
        self._lock = threading.Lock()
        self._running = False
        self._call = None
        self._name = str(self._func)

    def start(self):
        throttledlog.throttle(self._name, _THROTTLING_INTERVAL)
        with self._lock:
            if self._period <= 0:
                raise InvalidValue(repr(self), 'period', self._period)
            if self._running:
                raise AssertionError("Operation already running")
            self._log.debug("starting operation %s", self._func)
            self._running = True
            self._dispatch()

    def stop(self):
        with self._lock:
            if self._running:
                self._log.debug("stopping operation %s", self._func)
                self._running = False
                if self._call:
                    self._call.cancel()
                    self._call = None

    def __call__(self):
        try:
            self._func()
        except Exception:
            self._log.exception("%s operation failed", self._func)
        finally:
            if self._exclusive:
                self._reschedule()

    def _reschedule(self):
        self._call = self._scheduler.schedule(self._period,
                                              self._try_to_dispatch)

    def _try_to_dispatch(self):
        with self._lock:
            if self._running:
                self._dispatch()

    def _dispatch(self):
        state = None
        self._call = None
        dispatched = False
        try:
            self._executor.dispatch(self, self._timeout, discard=self._discard)
            dispatched = True
        except exception.ResourceExhausted:
            self._log.warning('could not run %s, executor queue full',
                              self._func)
            state = repr(self._executor)
        finally:
            if not self._exclusive or not dispatched:
                self._reschedule()
        if state:
            throttledlog.warning(self._name, 'executor state: %s', state)

    def __repr__(self):
        return '<Operation action=%s at 0x%x>' % (
            self._func, id(self)
        )


class VmDispatcher(object):

    _log = logging.getLogger("virt.periodic.VmDispatcher")

    def __init__(self, get_vms, executor, create, timeout):
        self._get_vms = get_vms
        self._executor = executor
        self._create = create
        self._timeout = timeout

    def __call__(self):
        vms = self._get_vms()
        skipped = []

        for vm_id, vm_obj in six.viewitems(vms):
            try:
                op = self._create(vm_obj)

                if not op.required:
                    continue
                if not op.runnable:
                    skipped.append(vm_id)
                    continue

            except Exception:
                self._log.exception("while dispatching %s", op)
            else:
                try:
                    self._executor.dispatch(op, self._timeout)
                except exception.ResourceExhausted:
                    skipped.append(vm_id)

        if skipped:
            self._log.warning('could not run %s on %s',
                              self._create, skipped)

    def __repr__(self):
        return '<VmDispatcher operation=%s at 0x%x>' % (
            self._create, id(self)
        )


class _RunnableOnVm(object):
    def __init__(self, vm):
        self._vm = vm

    @property
    def required(self):
        return self._vm.monitorable

    @property
    def runnable(self):
        return self._vm.isDomainReadyForCommands()

    def __call__(self):
        migrating = self._vm.isMigrating()
        try:
            self._execute()
        except virdomain.NotConnectedError:
            self._vm.log.warning('could not run on %s: domain not connected',
                                 self._vm.id)
        except libvirt.libvirtError as e:
            if self._vm.post_copy != migration.PostCopyPhase.NONE:
                return
            if e.get_error_code() in (
                libvirt.VIR_ERR_NO_DOMAIN,
            ):
                if migrating or self._vm.lastStatus == vmstatus.DOWN:
                    return
            raise

    def _execute(self):
        raise NotImplementedError

    def __repr__(self):
        return '<%s vm=%s at 0x%x>' % (
            self.__class__.__name__, self._vm.id, id(self)
        )


class UpdateVolumes(_RunnableOnVm):

    @property
    def required(self):
        return (super(UpdateVolumes, self).required and
                self._vm.drive_monitor.enabled())

    def _execute(self):
        for drive in self._vm.getDiskDevices():
            if not drive.readonly:
                self._vm.updateDriveVolume(drive)


class BlockjobMonitor(_RunnableOnVm):

    @property
    def required(self):
        return (super(BlockjobMonitor, self).required and self._vm.hasVmJobs)

    def _execute(self):
        self._vm.updateVmJobs()


class DriveWatermarkMonitor(_RunnableOnVm):

    @property
    def required(self):
        return (super(DriveWatermarkMonitor, self).required and
                self._vm.drive_monitor.monitoring_needed())

    def _execute(self):
        self._vm.monitor_drives()


def _kill_long_paused_vms(cif):
    log = logging.getLogger("virt.periodic")
    log.debug("Looking for stale paused VMs")
    for vm in cif.getVMs().values():
        if vm.lastStatus == vmstatus.PAUSED and \
           vm.pause_code in ('EIO', 'EOTHER',):
            vm.maybe_kill_paused()


def _create(cif, scheduler):
    def per_vm_operation(func, period):
        disp = VmDispatcher(
            cif.getVMs, _executor, func, _timeout_from(period))
        return Operation(disp, period, scheduler)

    ops = [
        per_vm_operation(
            UpdateVolumes,
            config.getint('irs', 'vol_size_sample_interval')),

        per_vm_operation(
            BlockjobMonitor,
            config.getint('vars', 'vm_sample_jobs_interval')),

        per_vm_operation(
            DriveWatermarkMonitor,
            config.getint('vars', 'vm_watermark_interval')),

        Operation(
            lambda: recovery.lookup_external_vms(cif),
            config.getint('sampling', 'external_vm_lookup_interval'),
            scheduler,
            exclusive=True,
            discard=False),

        Operation(
            lambda: _kill_long_paused_vms(cif),
            config.getint('vars', 'vm_kill_paused_time') // 2,
            scheduler,
            exclusive=True,
            discard=False),
    ]

    if config.getboolean('sampling', 'enable'):
        ops.extend([
            Operation(
                sampling.VMBulkstatsMonitor(
                    libvirtconnection.get(cif),
                    cif.getVMs,
                    sampling.stats_cache),
                config.getint('vars', 'vm_sample_interval'),
                scheduler),

            Operation(
                sampling.HostMonitor(cif=cif),
                config.getint('vars', 'host_sample_stats_interval'),
                scheduler,
                timeout=config.getint('vars', 'host_sample_stats_interval'),
                exclusive=True,
                discard=False),
        ])

    return ops
