
from __future__ import absolute_import

import errno
import fcntl
import collections
import logging
import os
import subprocess
import threading

from vdsm import constants
from vdsm import utils

from vdsm.common import concurrent
from vdsm.common import errors
from vdsm.common import osutils
from vdsm.config import config
from vdsm.storage import constants as sc
from vdsm.storage import exception as se
from vdsm.storage import misc
from vdsm.storage.compat import sanlock


MAX_HOST_ID = 250


HOST_STATUS_UNAVAILABLE = "unavailable"

HOST_STATUS_UNKNOWN = "unknown"

HOST_STATUS_FREE = "free"

HOST_STATUS_LIVE = "live"

HOST_STATUS_FAIL = "fail"

HOST_STATUS_DEAD = "dead"


class Error(errors.Base):
    """ Base class for clusterlock errors. """


class InvalidLeaseName(Error):

    msg = ("Sanlock resource name {self.resource} does not match lease name "
           "{self.lease}, this lease must be repaired.")

    def __init__(self, resource, lease):
        self.resource = resource
        self.lease = lease


class MultipleLeasesNotSupported(Error):

    msg = "Mulitple leases not supported, cannot {self.action} {self.lease}"

    def __init__(self, action, lease):
        self.action = action
        self.lease = lease


class TemporaryFailure(Error):
    msg = "Cannot {self.action} {self.lease}: {self.reason}"

    def __init__(self, action, lease, reason):
        self.action = action
        self.lease = lease
        self.reason = reason


Lease = collections.namedtuple("Lease", "name, path, offset")


class SafeLease(object):
    log = logging.getLogger("storage.Safelease")

    lockUtilPath = config.get('irs', 'lock_util_path')
    lockCmd = config.get('irs', 'lock_cmd')
    freeLockCmd = config.get('irs', 'free_lock_cmd')

    def __init__(self, sdUUID, idsPath, lease, lockRenewalIntervalSec,
                 leaseTimeSec, leaseFailRetry, ioOpTimeoutSec, **kwargs):
        self._lock = threading.Lock()
        self._sdUUID = sdUUID
        self._idsPath = idsPath
        self._lease = lease
        self.setParams(lockRenewalIntervalSec, leaseTimeSec, leaseFailRetry,
                       ioOpTimeoutSec)

    @property
    def supports_multiple_leases(self):
        return False

    def initLock(self, lease):
        if lease != self._lease:
            raise MultipleLeasesNotSupported("init", lease)
        lockUtil = constants.EXT_SAFELEASE
        initCommand = [lockUtil, "release", "-f", lease.path, "0"]
        rc, out, err = misc.execCmd(initCommand, cwd=self.lockUtilPath)
        if rc != 0:
            self.log.warn("could not initialise spm lease (%s): %s", rc, out)
            raise se.ClusterLockInitError()

    def setParams(self, lockRenewalIntervalSec, leaseTimeSec, leaseFailRetry,
                  ioOpTimeoutSec):
        self._lockRenewalIntervalSec = lockRenewalIntervalSec
        self._leaseTimeSec = leaseTimeSec
        self._leaseFailRetry = leaseFailRetry
        self._ioOpTimeoutSec = ioOpTimeoutSec

    def getReservedId(self):
        return 1000

    def acquireHostId(self, hostId, wait):
        self.log.debug("Host id for domain %s successfully acquired (id: %s)",
                       self._sdUUID, hostId)

    def releaseHostId(self, hostId, wait, unused):
        self.log.debug("Host id for domain %s released successfully (id: %s)",
                       self._sdUUID, hostId)

    def hasHostId(self, hostId):
        return True

    def getHostStatus(self, hostId):
        return HOST_STATUS_UNAVAILABLE

    def acquire(self, hostID, lease):
        if lease != self._lease:
            raise MultipleLeasesNotSupported("acquire", lease)
        leaseTimeMs = self._leaseTimeSec * 1000
        ioOpTimeoutMs = self._ioOpTimeoutSec * 1000
        with self._lock:
            self.log.debug("Acquiring cluster lock for domain %s" %
                           self._sdUUID)

            lockUtil = self.getLockUtilFullPath()
            acquireLockCommand = subprocess.list2cmdline([
                lockUtil, "start", self._sdUUID, str(hostID),
                str(self._lockRenewalIntervalSec), str(lease.path),
                str(leaseTimeMs), str(ioOpTimeoutMs),
                str(self._leaseFailRetry), str(os.getpid())
            ])

            cmd = [constants.EXT_SU, misc.IOUSER, '-s', constants.EXT_SH, '-c',
                   acquireLockCommand]
            (rc, out, err) = misc.execCmd(cmd, cwd=self.lockUtilPath,
                                          sudo=True,
                                          ioclass=utils.IOCLASS.REALTIME,
                                          ioclassdata=0, setsid=True)
            if rc != 0:
                raise se.AcquireLockFailure(self._sdUUID, rc, out, err)
            self.log.debug("Clustered lock acquired successfully")

    def inquire(self, lease):
        raise se.InquireNotSupportedError()

    def getLockUtilFullPath(self):
        return os.path.join(self.lockUtilPath, self.lockCmd)

    def release(self, lease):
        if lease != self._lease:
            raise MultipleLeasesNotSupported("release", lease)
        with self._lock:
            freeLockUtil = os.path.join(self.lockUtilPath, self.freeLockCmd)
            releaseLockCommand = [freeLockUtil, self._sdUUID]
            self.log.info("Releasing cluster lock for domain %s" %
                          self._sdUUID)
            (rc, out, err) = misc.execCmd(releaseLockCommand, raw=True,
                                          cwd=self.lockUtilPath)
            if rc != 0:
                self.log.error("Could not release cluster lock for domain %s "
                               "(rc=%d, out=%s, err=%s)" %
                               (self._sdUUID, rc, out, err))
                return

            self.log.debug("Cluster lock for domain %s released successfully",
                           self._sdUUID)


initSANLockLog = logging.getLogger("storage.initSANLock")


def initSANLock(
        sdUUID, idsPath, lease, alignment=sc.ALIGNMENT_1M,
        block_size=sc.BLOCK_SIZE_512):
    initSANLockLog.debug("Initializing SANLock for domain %s", sdUUID)

    try:
        sanlock.write_lockspace(
            sdUUID, idsPath, align=alignment, sector=block_size)
        sanlock.write_resource(
            sdUUID,
            lease.name,
            [(lease.path, lease.offset)],
            align=alignment,
            sector=block_size)
    except sanlock.SanlockException:
        initSANLockLog.error("Cannot initialize SANLock for domain %s",
                             sdUUID, exc_info=True)
        raise se.ClusterLockInitError()


class SANLock(object):

    STATUS_NAME = {
        sanlock.HOST_UNKNOWN: HOST_STATUS_UNKNOWN,
        sanlock.HOST_FREE: HOST_STATUS_FREE,
        sanlock.HOST_LIVE: HOST_STATUS_LIVE,
        sanlock.HOST_FAIL: HOST_STATUS_FAIL,
        sanlock.HOST_DEAD: HOST_STATUS_DEAD,
    }

    ACQUIRE_HOST_ID_TIMEOUT = 180

    log = logging.getLogger("storage.SANLock")

    _sanlock_fd = None
    _sanlock_lock = threading.Lock()

    def __init__(self, sdUUID, idsPath, lease, *args, **kwargs):
        self._lock = threading.Lock()
        self._sdUUID = sdUUID
        self._idsPath = idsPath
        self._alignment = kwargs.get("alignment", sc.ALIGNMENT_1M)
        self._block_size = kwargs.get("block_size", sc.BLOCK_SIZE_512)
        self._ready = concurrent.ValidatingEvent()

    @property
    def supports_multiple_leases(self):
        return True

    def initLock(self, lease):
        initSANLock(
            self._sdUUID,
            self._idsPath,
            lease,
            alignment=self._alignment,
            block_size=self._block_size)

    def setParams(self, *args):
        pass

    def getReservedId(self):
        return MAX_HOST_ID

    def acquireHostId(self, hostId, wait):
        self.log.info("Acquiring host id for domain %s (id=%s, wait=%s)",
                      self._sdUUID, hostId, wait)

        self._ready.valid = True

        with self._lock:
            try:
                with utils.stopwatch("sanlock.add_lockspace"):
                    sanlock.add_lockspace(self._sdUUID, hostId, self._idsPath,
                                          **{'async': not wait})
            except sanlock.SanlockException as e:
                if e.errno == errno.EINPROGRESS:
                    if wait:
                        if not sanlock.inq_lockspace(self._sdUUID, hostId,
                                                     self._idsPath, wait=True):
                            raise se.AcquireHostIdFailure(self._sdUUID, e)
                        self.log.info("Host id for domain %s successfully "
                                      "acquired (id=%s, wait=%s)",
                                      self._sdUUID, hostId, wait)
                        self._ready.set()
                elif e.errno == errno.EEXIST:
                    self.log.info("Host id for domain %s already acquired "
                                  "(id=%s, wait=%s)",
                                  self._sdUUID, hostId, wait)
                    self._ready.set()
                else:
                    raise se.AcquireHostIdFailure(self._sdUUID, e)
            else:
                if wait:
                    self.log.info("Host id for domain %s successfully "
                                  "acquired (id=%s, wait=%s)",
                                  self._sdUUID, hostId, wait)
                    self._ready.set()

    def releaseHostId(self, hostId, wait, unused):
        self.log.info("Releasing host id for domain %s (id: %s)",
                      self._sdUUID, hostId)

        self._ready.valid = False

        with self._lock:
            try:
                sanlock.rem_lockspace(self._sdUUID, hostId, self._idsPath,
                                      unused=unused, **{'async': not wait})
            except sanlock.SanlockException as e:
                if e.errno != errno.ENOENT:
                    raise se.ReleaseHostIdFailure(self._sdUUID, e)

        self.log.info("Host id for domain %s released successfully "
                      "(id: %s)", self._sdUUID, hostId)

    def hasHostId(self, hostId):
        with self._lock:
            try:
                has_host_id = sanlock.inq_lockspace(self._sdUUID, hostId,
                                                    self._idsPath)
            except sanlock.SanlockException:
                self.log.debug("Unable to inquire sanlock lockspace "
                               "status, returning False", exc_info=True)
                return False

            if has_host_id:
                self._ready.set()
            else:
                self._ready.clear()

        return has_host_id

    def getHostStatus(self, hostId):
        try:
            hosts = sanlock.get_hosts(self._sdUUID, hostId)
        except sanlock.SanlockException as e:
            self.log.debug("Unable to get host %d status in lockspace %s: %s",
                           hostId, self._sdUUID, e)
            return HOST_STATUS_UNAVAILABLE
        else:
            status = hosts[0]['flags']
            return self.STATUS_NAME[status]

    def acquire(self, hostId, lease):
        self.log.info("Acquiring %s for host id %s", lease, hostId)

        if not self._ready.wait(self.ACQUIRE_HOST_ID_TIMEOUT):
            raise se.AcquireHostIdFailure(
                "Timeout acquiring host id, cannot acquire %s (id=%s)"
                % (lease, hostId))

        with self._lock, SANLock._sanlock_lock:
            while True:
                if SANLock._sanlock_fd is None:
                    try:
                        SANLock._sanlock_fd = sanlock.register()
                    except sanlock.SanlockException as e:
                        raise se.AcquireLockFailure(
                            self._sdUUID, e.errno,
                            "Cannot register to sanlock", str(e))

                try:
                    sanlock.acquire(self._sdUUID, lease.name,
                                    [(lease.path, lease.offset)],
                                    slkfd=SANLock._sanlock_fd)
                except sanlock.SanlockException as e:
                    if e.errno != errno.EPIPE:
                        raise se.AcquireLockFailure(
                            self._sdUUID, e.errno,
                            "Cannot acquire %s" % (lease,), str(e))
                    SANLock._sanlock_fd = None
                    continue

                break

        self.log.info("Successfully acquired %s for host id %s", lease, hostId)

    def inquire(self, lease):
        resource = sanlock.read_resource(
            lease.path,
            lease.offset,
            align=self._alignment,
            sector=self._block_size)
        if resource["resource"] != lease.name:
            raise InvalidLeaseName(resource["resource"], lease)

        owners = sanlock.read_resource_owners(
            self._sdUUID,
            lease.name,
            [(lease.path, lease.offset)],
            align=self._alignment,
            sector=self._block_size)

        if len(owners) > 1:
            self.log.error("Cluster lock is reported to have more than "
                           "one owner: %s", owners)
            raise RuntimeError("Multiple owners for %s" % (lease,))

        elif not owners:
            return None, None

        resource_owner = owners[0]
        resource_version = resource["version"]
        host_id = resource_owner["host_id"]
        try:
            host = sanlock.get_hosts(self._sdUUID, host_id)[0]
        except sanlock.SanlockException as e:
            if e.errno == errno.ENOENT:
                raise TemporaryFailure("inquire", lease, str(e))
            elif e.errno == errno.EAGAIN:
                self.log.debug("host %s status in not available yet, "
                               "it may hold the lease %s",
                               host_id, lease)
                return resource_version, host_id
            else:
                raise

        host_status = self.STATUS_NAME[host["flags"]]

        if resource_owner["generation"] != host["generation"]:
            self.log.debug("host %r generation %r does not match resource "
                           "generation %r, lease %s is free", host_id,
                           host, resource_owner["generation"], lease)
            return resource_version, None

        if host_status in (HOST_STATUS_DEAD, HOST_STATUS_FREE):
            self.log.debug("host %s cannot hold %s is effectively free",
                           host, lease)
            return resource_version, None

        return resource_version, host_id

    def release(self, lease):
        self.log.info("Releasing %s", lease)
        with self._lock:
            try:
                sanlock.release(self._sdUUID, lease.name,
                                [(lease.path, lease.offset)],
                                slkfd=SANLock._sanlock_fd)
            except sanlock.SanlockException as e:
                raise se.ReleaseLockFailure(self._sdUUID, e)

        self.log.info("Successfully released %s", lease)


class LocalLock(object):
    log = logging.getLogger("storage.LocalLock")

    LVER = 0

    _globalLockMap = {}
    _globalLockMapSync = threading.Lock()

    def __init__(self, sdUUID, idsPath, lease, *args, **kwargs):
        self._sdUUID = sdUUID
        self._idsPath = idsPath
        self._lease = lease

    @property
    def supports_multiple_leases(self):
        return False

    def initLock(self, lease):
        if lease != self._lease:
            raise MultipleLeasesNotSupported("init", lease)

    def setParams(self, *args):
        pass

    def getReservedId(self):
        return MAX_HOST_ID

    def _getLease(self):
        return self._globalLockMap.get(self._sdUUID, (None, None))

    def acquireHostId(self, hostId, wait):
        with self._globalLockMapSync:
            currentHostId, lockFile = self._getLease()

            if currentHostId is not None and currentHostId != hostId:
                self.log.error("Different host id already acquired (id: %s)",
                               currentHostId)
                raise se.AcquireHostIdFailure(self._sdUUID)

            self._globalLockMap[self._sdUUID] = (hostId, lockFile)

        self.log.debug("Host id for domain %s successfully acquired (id: %s)",
                       self._sdUUID, hostId)

    def releaseHostId(self, hostId, wait, unused):
        with self._globalLockMapSync:
            currentHostId, lockFile = self._getLease()

            if currentHostId is not None and currentHostId != hostId:
                self.log.error("Different host id acquired (id: %s)",
                               currentHostId)
                raise se.ReleaseHostIdFailure(self._sdUUID)

            if lockFile is not None:
                self.log.error("Cannot release host id when lock is acquired")
                raise se.ReleaseHostIdFailure(self._sdUUID)

            del self._globalLockMap[self._sdUUID]

        self.log.debug("Host id for domain %s released successfully (id: %s)",
                       self._sdUUID, hostId)

    def hasHostId(self, hostId):
        with self._globalLockMapSync:
            currentHostId, lockFile = self._getLease()
            return currentHostId == hostId

    def getHostStatus(self, hostId):
        return HOST_STATUS_UNAVAILABLE

    def acquire(self, hostId, lease):
        if lease != self._lease:
            raise MultipleLeasesNotSupported("acquire", lease)
        with self._globalLockMapSync:
            self.log.info("Acquiring local lock for domain %s (id: %s)",
                          self._sdUUID, hostId)

            hostId, lockFile = self._getLease()

            if lockFile:
                try:
                    osutils.uninterruptible(fcntl.fcntl, lockFile,
                                            fcntl.F_GETFD)
                except IOError as e:
                    del self._globalLockMap[self._sdUUID]

                    if e.errno != errno.EBADF:
                        raise
                else:
                    self.log.debug("Local lock already acquired for domain "
                                   "%s (id: %s)", self._sdUUID, hostId)

            lockFile = osutils.uninterruptible(os.open, self._idsPath,
                                               os.O_RDONLY)

            try:
                osutils.uninterruptible(fcntl.flock, lockFile,
                                        fcntl.LOCK_EX | fcntl.LOCK_NB)
            except IOError as e:
                osutils.close_fd(lockFile)
                if e.errno in (errno.EACCES, errno.EAGAIN):
                    raise se.AcquireLockFailure(
                        self._sdUUID, e.errno, "Cannot acquire local lock",
                        str(e))
                raise
            else:
                self._globalLockMap[self._sdUUID] = (hostId, lockFile)

        self.log.debug("Local lock for domain %s successfully acquired "
                       "(id: %s)", self._sdUUID, hostId)

    def inquire(self, lease):
        if lease != self._lease:
            raise MultipleLeasesNotSupported("inquire", lease)
        with self._globalLockMapSync:
            hostId, lockFile = self._getLease()
            return self.LVER, (hostId if lockFile else None)

    def release(self, lease):
        if lease != self._lease:
            raise MultipleLeasesNotSupported("release", lease)
        with self._globalLockMapSync:
            self.log.info("Releasing local lock for domain %s", self._sdUUID)

            hostId, lockFile = self._getLease()

            if not lockFile:
                self.log.debug("Local lock already released for domain %s",
                               self._sdUUID)
                return

            osutils.close_fd(lockFile)
            self._globalLockMap[self._sdUUID] = (hostId, None)

            self.log.debug("Local lock for domain %s successfully released",
                           self._sdUUID)


def alignment(block_size, max_hosts):
    if max_hosts < 1 or max_hosts > sc.HOSTS_512_1M:
        raise se.InvalidParameterException('max_hosts', max_hosts)
    if block_size not in (sc.BLOCK_SIZE_512, sc.BLOCK_SIZE_4K):
        raise se.InvalidParameterException('block_size', block_size)

    if block_size == sc.BLOCK_SIZE_512:
        return sc.ALIGNMENT_1M

    if max_hosts > sc.HOSTS_4K_4M:
        return sc.ALIGNMENT_8M

    if max_hosts > sc.HOSTS_4K_2M:
        return sc.ALIGNMENT_4M

    if max_hosts > sc.HOSTS_4K_1M:
        return sc.ALIGNMENT_2M

    return sc.ALIGNMENT_1M
