python-tooz 1.40.0-4 / usr / lib / python2.7 / dist-packages / tooz / coordination.py

# -*- coding: utf-8 -*-
#
#    Copyright (C) 2016 Red Hat, Inc.
#    Copyright (C) 2013-2014 eNovance Inc. All Rights Reserved.
#
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
#    not use this file except in compliance with the License. You may obtain
#    a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#    License for the specific language governing permissions and limitations
#    under the License.

import abc
import collections
import enum
import logging
import threading

from oslo_utils import excutils
from oslo_utils import netutils
from oslo_utils import timeutils
import six
from stevedore import driver

import tooz

LOG = logging.getLogger(__name__)


TOOZ_BACKENDS_NAMESPACE = "tooz.backends"


class Characteristics(enum.Enum):
    """Attempts to describe the characteristic that a driver supports."""

    DISTRIBUTED_ACROSS_THREADS = 'DISTRIBUTED_ACROSS_THREADS'
    """Coordinator components when used by multiple **threads** work
       the same as if those components were only used by a single thread."""

    DISTRIBUTED_ACROSS_PROCESSES = 'DISTRIBUTED_ACROSS_PROCESSES'
    """Coordinator components when used by multiple **processes** work
       the same as if those components were only used by a single thread."""

    DISTRIBUTED_ACROSS_HOSTS = 'DISTRIBUTED_ACROSS_HOSTS'
    """Coordinator components when used by multiple **hosts** work
       the same as if those components were only used by a single thread."""

    NON_TIMEOUT_BASED = 'NON_TIMEOUT_BASED'
    """The driver has the following property:

    * Its operations are not based on the timeout of other clients, but on some
    other more robust mechanisms.
    """

    LINEARIZABLE = 'LINEARIZABLE'
    """The driver has the following properties:

    * Ensures each operation must take place before its
      completion time.
    * Any operation invoked subsequently must take place
      after the invocation and by extension, after the original operation
      itself.
    """

    SEQUENTIAL = 'SEQUENTIAL'
    """The driver has the following properties:

    * Operations can take effect before or after completion – but all
      operations retain the constraint that operations from any given process
      must take place in that processes order.
    """

    CAUSAL = 'CAUSAL'
    """The driver has the following properties:

    * Does **not** have to enforce the order of every
      operation from a process, perhaps, only causally related operations
      must occur in order.
    """

    SERIALIZABLE = 'SERIALIZABLE'
    """The driver has the following properties:

    * The history of **all** operations is equivalent to
      one that took place in some single atomic order but with unknown
      invocation and completion times - it places no bounds on
      time or order.
    """

    SAME_VIEW_UNDER_PARTITIONS = 'SAME_VIEW_UNDER_PARTITIONS'
    """When a client is connected to a server and that server is partitioned
    from a group of other servers it will (somehow) have the same view of
    data as a client connected to a server on the other side of the
    partition (typically this is accomplished by write availability being
    lost and therefore nothing can change).
    """

    SAME_VIEW_ACROSS_CLIENTS = 'SAME_VIEW_ACROSS_CLIENTS'
    """A client connected to one server will *always* have the same view
    every other client will have (no matter what server those other
    clients are connected to). Typically this is a sacrifice in
    write availability because before a write can be acknowledged it must
    be acknowledged by *all* servers in a cluster (so that all clients
    that are connected to those servers read the exact *same* thing).
    """


class Hooks(list):
    def run(self, *args, **kwargs):
        return list(map(lambda cb: cb(*args, **kwargs), self))


class Event(object):
    """Base class for events."""


class MemberJoinedGroup(Event):
    """A member joined a group event."""

    def __init__(self, group_id, member_id):
        self.group_id = group_id
        self.member_id = member_id


class MemberLeftGroup(Event):
    """A member left a group event."""

    def __init__(self, group_id, member_id):
        self.group_id = group_id
        self.member_id = member_id


class LeaderElected(Event):
    """A leader as been elected."""

    def __init__(self, group_id, member_id):
        self.group_id = group_id
        self.member_id = member_id


class Heart(object):
    """Coordination drivers main liveness pump (its heart)."""

    def __init__(self, driver, thread_cls=threading.Thread,
                 event_cls=threading.Event):
        self._thread_cls = thread_cls
        self._dead = event_cls()
        self._finished = event_cls()
        self._finished.set()
        self._runner = None
        self._driver = driver
        self._beats = 0

    @property
    def beats(self):
        """How many times the heart has beaten."""
        return self._beats

    def is_alive(self):
        """Returns if the heart is beating."""
        return not (self._runner is None
                    or not self._runner.is_alive()
                    or self._finished.is_set())

    @excutils.forever_retry_uncaught_exceptions
    def _beat_forever_until_stopped(self):
        """Inner beating loop."""
        try:
            while not self._dead.is_set():
                with timeutils.StopWatch() as w:
                    wait_until_next_beat = self._driver.heartbeat()
                ran_for = w.elapsed()
                if ran_for > wait_until_next_beat:
                    LOG.warning(
                        "Heartbeating took too long to execute (it ran for"
                        " %0.2f seconds which is %0.2f seconds longer than"
                        " the next heartbeat idle time). This may cause"
                        " timeouts (in locks, leadership, ...) to"
                        " happen (which will not end well).", ran_for,
                        ran_for - wait_until_next_beat)
                self._beats += 1
                # NOTE(harlowja): use the event object for waiting and
                # not a sleep function since doing that will allow this code
                # to terminate early if stopped via the stop() method vs
                # having to wait until the sleep function returns.
                self._dead.wait(wait_until_next_beat)
        finally:
            self._finished.set()

    def start(self, thread_cls=None):
        """Starts the heart beating thread (noop if already started)."""
        if not self.is_alive():
            self._finished.clear()
            self._dead.clear()
            self._beats = 0
            if thread_cls is None:
                thread_cls = self._thread_cls
            self._runner = thread_cls(target=self._beat_forever_until_stopped)
            self._runner.daemon = True
            self._runner.start()

    def stop(self):
        """Requests the heart beating thread to stop beating."""
        self._dead.set()

    def wait(self, timeout=None):
        """Wait up to given timeout for the heart beating thread to stop."""
        self._finished.wait(timeout)
        return self._finished.is_set()


@six.add_metaclass(abc.ABCMeta)
class CoordinationDriver(object):

    requires_beating = False
    """
    Usage requirement that if true requires that the :py:meth:`~.heartbeat`
    be called periodically (at a given rate) to avoid locks, sessions and
    other from being automatically closed/discarded by the coordinators
    backing store.
    """

    CHARACTERISTICS = ()
    """
    Tuple of :py:class:`~tooz.coordination.Characteristics` introspectable
    enum member(s) that can be used to interogate how this driver works.
    """

    def __init__(self):
        self._started = False
        self._hooks_join_group = collections.defaultdict(Hooks)
        self._hooks_leave_group = collections.defaultdict(Hooks)
        self._hooks_elected_leader = collections.defaultdict(Hooks)
        # A cache for group members
        self._group_members = collections.defaultdict(set)
        self.requires_beating = (
            CoordinationDriver.heartbeat != self.__class__.heartbeat
        )
        self.heart = Heart(self)

    def _has_hooks_for_group(self, group_id):
        return (len(self._hooks_join_group[group_id]) +
                len(self._hooks_leave_group[group_id]))

    @staticmethod
    def run_watchers(timeout=None):
        """Run the watchers callback.

        This may also activate :py:meth:`.run_elect_coordinator` (depending
        on driver implementation).
        """
        raise tooz.NotImplemented

    @staticmethod
    def run_elect_coordinator():
        """Try to leader elect this coordinator & activate hooks on success."""
        raise tooz.NotImplemented

    @abc.abstractmethod
    def watch_join_group(self, group_id, callback):
        """Call a function when group_id sees a new member joined.

        The callback functions will be executed when `run_watchers` is
        called.

        :param group_id: The group id to watch
        :param callback: The function to execute when a member joins this group

        """
        self._hooks_join_group[group_id].append(callback)

    @abc.abstractmethod
    def unwatch_join_group(self, group_id, callback):
        """Stop executing a function when a group_id sees a new member joined.

        :param group_id: The group id to unwatch
        :param callback: The function that was executed when a member joined
                         this group
        """
        try:
            self._hooks_join_group[group_id].remove(callback)
        except ValueError:
            raise WatchCallbackNotFound(group_id, callback)

        if (not self._has_hooks_for_group(group_id) and
           group_id in self._group_members):
            del self._group_members[group_id]

    @abc.abstractmethod
    def watch_leave_group(self, group_id, callback):
        """Call a function when group_id sees a new member leaving.

        The callback functions will be executed when `run_watchers` is
        called.

        :param group_id: The group id to watch
        :param callback: The function to execute when a member leaves this
                         group

        """
        self._hooks_leave_group[group_id].append(callback)

    @abc.abstractmethod
    def unwatch_leave_group(self, group_id, callback):
        """Stop executing a function when a group_id sees a new member leaving.

        :param group_id: The group id to unwatch
        :param callback: The function that was executed when a member left
                         this group
        """
        try:
            self._hooks_leave_group[group_id].remove(callback)
        except ValueError:
            raise WatchCallbackNotFound(group_id, callback)

        if (not self._has_hooks_for_group(group_id) and
           group_id in self._group_members):
            del self._group_members[group_id]

    @abc.abstractmethod
    def watch_elected_as_leader(self, group_id, callback):
        """Call a function when member gets elected as leader.

        The callback functions will be executed when `run_watchers` is
        called.

        :param group_id: The group id to watch
        :param callback: The function to execute when a member leaves this
                         group

        """
        self._hooks_elected_leader[group_id].append(callback)

    @abc.abstractmethod
    def unwatch_elected_as_leader(self, group_id, callback):
        """Call a function when member gets elected as leader.

        The callback functions will be executed when `run_watchers` is
        called.

        :param group_id: The group id to watch
        :param callback: The function to execute when a member leaves this
                         group

        """
        try:
            self._hooks_elected_leader[group_id].remove(callback)
        except ValueError:
            raise WatchCallbackNotFound(group_id, callback)

        if not self._hooks_elected_leader[group_id]:
            del self._hooks_elected_leader[group_id]

    @staticmethod
    def stand_down_group_leader(group_id):
        """Stand down as the group leader if we are.

        :param group_id: The group where we don't want to be a leader anymore
        """
        raise tooz.NotImplemented

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

    def start(self, start_heart=False):
        """Start the service engine.

        If needed, the establishment of a connection to the servers
        is initiated.
        """
        if self._started:
            raise ToozError(
                "Can not start a driver which has not been stopped")
        self._start()
        if self.requires_beating and start_heart:
            self.heart.start()
        self._started = True

    def _start(self):
        pass

    def stop(self):
        """Stop the service engine.

        If needed, the connection to servers is closed and the client will
        disappear from all joined groups.
        """
        if not self._started:
            raise ToozError("Can not stop a driver which has not been started")
        if self.heart.is_alive():
            self.heart.stop()
            self.heart.wait()
        self._stop()
        self._started = False

    def _stop(self):
        pass

    @staticmethod
    def create_group(group_id):
        """Request the creation of a group asynchronously.

        :param group_id: the id of the group to create
        :type group_id: str
        :returns: None
        :rtype: CoordAsyncResult
        """
        raise tooz.NotImplemented

    @staticmethod
    def get_groups():
        """Return the list composed by all groups ids asynchronously.

        :returns: the list of all created group ids
        :rtype: CoordAsyncResult
        """
        raise tooz.NotImplemented

    @staticmethod
    def join_group(group_id, capabilities=b""):
        """Join a group and establish group membership asynchronously.

        :param group_id: the id of the group to join
        :type group_id: str
        :param capabilities: the capabilities of the joined member
        :type capabilities: object (typically str)
        :returns: None
        :rtype: CoordAsyncResult
        """
        raise tooz.NotImplemented

    @staticmethod
    def leave_group(group_id):
        """Leave a group asynchronously.

        :param group_id: the id of the group to leave
        :type group_id: str
        :returns: None
        :rtype: CoordAsyncResult
        """
        raise tooz.NotImplemented

    @staticmethod
    def delete_group(group_id):
        """Delete a group asynchronously.

        :param group_id: the id of the group to leave
        :type group_id: str
        :returns: Result
        :rtype: CoordAsyncResult
        """
        raise tooz.NotImplemented

    @staticmethod
    def get_members(group_id):
        """Return the list of all members ids of the specified group.

        :returns: list of all created group ids
        :rtype: CoordAsyncResult
        """
        raise tooz.NotImplemented

    @staticmethod
    def get_member_capabilities(group_id, member_id):
        """Return the capabilities of a member asynchronously.

        :param group_id: the id of the group of the member
        :type group_id: str
        :param member_id: the id of the member
        :type member_id: str
        :returns: capabilities of a member
        :rtype: CoordAsyncResult
        """
        raise tooz.NotImplemented

    @staticmethod
    def get_member_info(group_id, member_id):
        """Return the statistics and capabilities of a member asynchronously.

        :param group_id: the id of the group of the member
        :type group_id: str
        :param member_id: the id of the member
        :type member_id: str
        :returns: capabilities and statistics of a member
        :rtype: CoordAsyncResult
        """
        raise tooz.NotImplemented

    @staticmethod
    def update_capabilities(group_id, capabilities):
        """Update member capabilities in the specified group.

        :param group_id: the id of the group of the current member
        :type group_id: str
        :param capabilities: the capabilities of the updated member
        :type capabilities: object (typically str)
        :returns: None
        :rtype: CoordAsyncResult
        """
        raise tooz.NotImplemented

    @staticmethod
    def get_leader(group_id):
        """Return the leader for a group.

        :param group_id: the id of the group:
        :returns: the leader
        :rtype: CoordAsyncResult
        """
        raise tooz.NotImplemented

    @staticmethod
    def get_lock(name):
        """Return a distributed lock.

        This is a exclusive lock, a second call to acquire() will block or
        return False.

        :param name: The lock name that is used to identify it across all
                     nodes.

        """
        raise tooz.NotImplemented

    @staticmethod
    def heartbeat():
        """Update member status to indicate it is still alive.

        Method to run once in a while to be sure that the member is not dead
        and is still an active member of a group.

        :return: The number of seconds to wait before sending a new heartbeat.
        """
        pass


@six.add_metaclass(abc.ABCMeta)
class CoordAsyncResult(object):
    """Representation of an asynchronous task.

    Every call API returns an CoordAsyncResult object on which the result or
    the status of the task can be requested.

    """

    @abc.abstractmethod
    def get(self, timeout=10):
        """Retrieve the result of the corresponding asynchronous call.

        :param timeout: block until the timeout expire.
        :type timeout: float
        """

    @abc.abstractmethod
    def done(self):
        """Returns True if the task is done, False otherwise."""


class _RunWatchersMixin(object):
    """Mixin to share the *mostly* common ``run_watchers`` implementation."""

    def run_watchers(self, timeout=None):
        with timeutils.StopWatch(duration=timeout) as w:
            known_groups = self.get_groups().get(
                timeout=w.leftover(return_none=True))
            result = []
            for group_id in known_groups:
                try:
                    group_members_fut = self.get_members(group_id)
                    group_members = group_members_fut.get(
                        timeout=w.leftover(return_none=True))
                except GroupNotCreated:
                    group_members = set()
                else:
                    group_members = set(group_members)
                if (group_id in self._joined_groups and
                        self._member_id not in group_members):
                    self._joined_groups.discard(group_id)
                old_group_members = self._group_members.get(group_id, set())
                for member_id in (group_members - old_group_members):
                    result.extend(
                        self._hooks_join_group[group_id].run(
                            MemberJoinedGroup(group_id, member_id)))
                for member_id in (old_group_members - group_members):
                    result.extend(
                        self._hooks_leave_group[group_id].run(
                            MemberLeftGroup(group_id, member_id)))
                self._group_members[group_id] = group_members
            return result


def get_coordinator(backend_url, member_id,
                    characteristics=frozenset(), **kwargs):
    """Initialize and load the backend.

    :param backend_url: the backend URL to use
    :type backend: str
    :param member_id: the id of the member
    :type member_id: str
    :param characteristics: set
    :type characteristics: set of :py:class:`.Characteristics` that will
                           be matched to the requested driver (this **will**
                           become a **required** parameter in a future tooz
                           version)
    :param kwargs: additional coordinator options (these take precedence over
                   options of the **same** name found in the ``backend_url``
                   arguments query string)
    """
    parsed_url = netutils.urlsplit(backend_url)
    parsed_qs = six.moves.urllib.parse.parse_qs(parsed_url.query)
    if kwargs:
        options = {}
        for (k, v) in six.iteritems(kwargs):
            options[k] = [v]
        for (k, v) in six.iteritems(parsed_qs):
            if k not in options:
                options[k] = v
    else:
        options = parsed_qs
    d = driver.DriverManager(
        namespace=TOOZ_BACKENDS_NAMESPACE,
        name=parsed_url.scheme,
        invoke_on_load=True,
        invoke_args=(member_id, parsed_url, options)).driver
    characteristics = set(characteristics)
    driver_characteristics = set(getattr(d, 'CHARACTERISTICS', set()))
    missing_characteristics = characteristics - driver_characteristics
    if missing_characteristics:
        raise ToozDriverChosenPoorly("Desired characteristics %s"
                                     " is not a strict subset of driver"
                                     " characteristics %s, %s"
                                     " characteristics were not found"
                                     % (characteristics,
                                        driver_characteristics,
                                        missing_characteristics))
    return d


class ToozError(Exception):
    """Exception raised when an internal error occurs.

    Raised for instance in case of server internal error.

    :ivar cause: the cause of the exception being raised, when not none this
                 will itself be an exception instance, this is useful for
                 creating a chain of exceptions for versions of python where
                 this is not yet implemented/supported natively.

    """

    def __init__(self, message, cause=None):
        super(ToozError, self).__init__(message)
        self.cause = cause


class ToozDriverChosenPoorly(ToozError):
    """Raised when a driver does not match desired characteristics."""


class ToozConnectionError(ToozError):
    """Exception raised when the client cannot connect to the server."""


class OperationTimedOut(ToozError):
    """Exception raised when an operation times out."""


class LockAcquireFailed(ToozError):
    """Exception raised when a lock acquire fails in a context manager."""


class GroupNotCreated(ToozError):
    """Exception raised when the caller request an nonexistent group."""
    def __init__(self, group_id):
        self.group_id = group_id
        super(GroupNotCreated, self).__init__(
            "Group %s does not exist" % group_id)


class GroupAlreadyExist(ToozError):
    """Exception raised trying to create an already existing group."""
    def __init__(self, group_id):
        self.group_id = group_id
        super(GroupAlreadyExist, self).__init__(
            "Group %s already exists" % group_id)


class MemberAlreadyExist(ToozError):
    """Exception raised trying to join a group already joined."""
    def __init__(self, group_id, member_id):
        self.group_id = group_id
        self.member_id = member_id
        super(MemberAlreadyExist, self).__init__(
            "Member %s has already joined %s" %
            (member_id, group_id))


class MemberNotJoined(ToozError):
    """Exception raised trying to access a member not in a group."""
    def __init__(self, group_id, member_id):
        self.group_id = group_id
        self.member_id = member_id
        super(MemberNotJoined, self).__init__("Member %s has not joined %s" %
                                              (member_id, group_id))


class GroupNotEmpty(ToozError):
    "Exception raised when the caller try to delete a group with members."
    def __init__(self, group_id):
        self.group_id = group_id
        super(GroupNotEmpty, self).__init__("Group %s is not empty" % group_id)


class WatchCallbackNotFound(ToozError):
    """Exception raised when unwatching a group.

    Raised when the caller tries to unwatch a group with a callback that
    does not exist.

    """
    def __init__(self, group_id, callback):
        self.group_id = group_id
        self.callback = callback
        super(WatchCallbackNotFound, self).__init__(
            'Callback %s is not registered on group %s' %
            (callback.__name__, group_id))


class SerializationError(ToozError):
    "Exception raised when serialization or deserialization breaks."


def raise_with_cause(exc_cls, message, *args, **kwargs):
    """Helper to raise + chain exceptions (when able) and associate a *cause*.

    **For internal usage only.**

    NOTE(harlowja): Since in py3.x exceptions can be chained (due to
    :pep:`3134`) we should try to raise the desired exception with the given
    *cause*.

    :param exc_cls: the :py:class:`~tooz.coordination.ToozError` class
                    to raise.
    :param message: the text/str message that will be passed to
                    the exceptions constructor as its first positional
                    argument.
    :param args: any additional positional arguments to pass to the
                 exceptions constructor.
    :param kwargs: any additional keyword arguments to pass to the
                   exceptions constructor.
    """
    if not issubclass(exc_cls, ToozError):
        raise ValueError("Subclass of tooz error is required")
    excutils.raise_with_cause(exc_cls, message, *args, **kwargs)