# tooz.drivers.file

# Source code for tooz.drivers.file

    # -*- coding: utf-8 -*-
    #
    # Copyright © 2015 eNovance
    #
    # 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 contextlib
    import datetime
    import errno
    import functools
    import hashlib
    import logging
    import os
    import re
    import shutil
    import sys
    import tempfile
    import threading
    import weakref

    import fasteners
    from oslo_utils import encodeutils
    from oslo_utils import fileutils
    from oslo_utils import timeutils
    import voluptuous

    import tooz
    from tooz import coordination
    from tooz import locking
    from tooz import utils

    LOG = logging.getLogger(__name__)

    class _Barrier(object):
        def __init__(self):
            self.cond = threading.Condition()
            self.owner = None
            self.shared = False
            self.ref = 0

    @contextlib.contextmanager
    def _translate_failures():
        try:
            yield
        except (EnvironmentError, voluptuous.Invalid) as e:
            utils.raise_with_cause(tooz.ToozError,
                                   encodeutils.exception_to_unicode(e),
                                   cause=e)

    def _convert_from_old_format(data):
        # NOTE(sileht): previous version of the driver was storing str as-is
        # making impossible to read from python3 something written with python2
        # version of the lib.
        # Now everything is stored with explicit type bytes or unicode. This
        # convert the old format to the new one to maintain compat of already
        # deployed file.
        # example of potential old python2 payload:
        # {b"member_id": b"member"}
        # {b"member_id": u"member"}
        # example of potential old python3 payload:
        # {u"member_id": b"member"}
        # {u"member_id": u"member"}
        if b"member_id" in data or b"group_id" in data:
            data = dict((k.decode("utf8"), v) for k, v in data.items())
            # About member_id and group_id valuse if the file have been written
            # with python2 and in the old format, we can't known with python3
            # if we need to decode the value or not. Python3 see bytes blob
            # We keep it as-is and pray, this have a good change to break if
            # the application was using str in python2 and unicode in python3
            # The member file is often overridden so it's should be fine
            # But the group file can be very old, so we
            # now have to update it each time create_group is called
        return data

    def _lock_me(lock):

        def wrapper(func):

            @functools.wraps(func)
            def decorator(*args, **kwargs):
                with lock:
                    return func(*args, **kwargs)

            return decorator

        return wrapper

    class FileLock(locking.Lock):
        """A file based lock."""

        def __init__(self, path, barrier, member_id):
            super(FileLock, self).__init__(path)
            self.acquired = False
            self._lock = fasteners.InterProcessLock(path)
            self._barrier = barrier
            self._member_id = member_id
            self.ref = 0

        def is_still_owner(self):
            return self.acquired

        def acquire(self, blocking=True, shared=False, timeout=None):
            blocking, timeout = utils.convert_blocking(blocking, timeout)
            watch = timeutils.StopWatch(duration=timeout)
            watch.start()

            # Make the shared barrier ours first.
            with self._barrier.cond:
                while self._barrier.owner is not None:
                    if (shared and self._barrier.shared):
                        break
                    if not blocking or watch.expired():
                        return False
                    self._barrier.cond.wait(watch.leftover(return_none=True))
                self._barrier.owner = (threading.current_thread().ident,
                                       os.getpid(), self._member_id)
                self._barrier.shared = shared
                self._barrier.ref += 1
                self.ref += 1

            # Ok at this point we are now working in a thread safe manner,
            # and now we can try to get the actual lock...
            gotten = False
            try:
                gotten = self._lock.acquire(
                    blocking=blocking,
                    # Since the barrier waiting may have
                    # taken a long time, we have to use
                    # the leftover (and not the original).
                    timeout=watch.leftover(return_none=True))
            finally:
                # NOTE(harlowja): do this in a finally block to **ensure** that
                # we release the barrier if something bad happens...
                if not gotten:
                    # Release the barrier to let someone else have a go at it...
                    with self._barrier.cond:
                        self._barrier.owner = None
                        self._barrier.ref = 0
                        self._barrier.shared = False
                        self._barrier.cond.notify_all()

            self.acquired = gotten
            return gotten

        def release(self):
            if not self.acquired:
                return False
            with self._barrier.cond:
                self._barrier.ref -= 1
                self.ref -= 1
                if not self.ref:
                    self.acquired = False
                if not self._barrier.ref:
                    self._barrier.owner = None
                    self._lock.release()
                    self._barrier.cond.notify_all()
            return True

        def __del__(self):
            if self.acquired:
                LOG.warning("Unreleased lock %s garbage collected", self.name)

    class FileDriver(coordination.CoordinationDriverCachedRunWatchers,
                     coordination.CoordinationDriverWithExecutor):
        """A file based driver.

        This driver uses files and directories (and associated file locks) to
        provide the coordination driver semantics and required API(s). It **is**
        missing some functionality but in the future these not implemented API(s)
        will be filled in.

        The File driver connection URI should look like::

          file://DIRECTORY[?timeout=TIMEOUT]

        DIRECTORY is the location that should be used to store lock files.
        TIMEOUT defaults to 10.

        General recommendations/usage considerations:

        - It does **not** automatically delete members from
          groups of processes that have died, manual cleanup will be needed
          for those types of failures.

        - It is **not** distributed (or recommended to be used in those
          situations, so the developer using this should really take that into
          account when applying this driver in there app).
        """

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

        HASH_ROUTINE = 'sha1'
        """This routine is used to hash a member (or group) id into a filesystem
           safe name that can be used for member lookup and group joining."""

        _barriers = weakref.WeakValueDictionary()
        """
        Barriers shared among all file driver locks, this is required
        since interprocess locking is not thread aware, so we must add the
        thread awareness on-top of it instead.
        """

        def __init__(self, member_id, parsed_url, options):
            """Initialize the file driver."""
            super(FileDriver, self).__init__(member_id, parsed_url, options)
            self._dir = self._normalize_path(parsed_url.path)
            self._group_dir = os.path.join(self._dir, 'groups')
            self._tmpdir = os.path.join(self._dir, 'tmp')
            self._driver_lock_path = os.path.join(self._dir, '.driver_lock')
            self._driver_lock = self._get_raw_lock(self._driver_lock_path,
                                                   self._member_id)
            self._reserved_dirs = [self._dir, self._group_dir, self._tmpdir]
            self._reserved_paths = list(self._reserved_dirs)
            self._reserved_paths.append(self._driver_lock_path)
            self._safe_member_id = self._make_filesystem_safe(member_id)
            self._timeout = int(self._options.get('timeout', 10))

        @staticmethod
        def _normalize_path(path):
            if sys.platform == 'win32':
                # Replace slashes with backslashes and make sure we don't
                # have any at the beginning of paths that include drive letters.
                #
                # Expected url format:
                # file:////share_address/share_name
                # file:///C:/path
                return re.sub(r'\\(?=\w:\\)', '',
                              os.path.normpath(path))
            return path

        @classmethod
        def _get_raw_lock(cls, path, member_id):
            lock_barrier = cls._barriers.setdefault(path, _Barrier())
            return FileLock(path, lock_barrier, member_id)

        def get_lock(self, name):
            path = utils.safe_abs_path(self._dir, name.decode())
            if path in self._reserved_paths:
                raise ValueError("Unable to create a lock using"
                                 " reserved path '%s' for lock"
                                 " with name '%s'" % (path, name))
            return self._get_raw_lock(path, self._member_id)

        @classmethod
        def _make_filesystem_safe(cls, item):
            item = utils.to_binary(item, encoding="utf8")
            return hashlib.new(cls.HASH_ROUTINE, item).hexdigest()

        def _start(self):
            super(FileDriver, self)._start()
            for a_dir in self._reserved_dirs:
                try:
                    fileutils.ensure_tree(a_dir)
                except OSError as e:
                    raise coordination.ToozConnectionError(e)

        def _update_group_metadata(self, path, group_id):
            details = {
                u'group_id': utils.to_binary(group_id, encoding="utf8")
            }
            details[u'encoded'] = details[u"group_id"] != group_id
            details_blob = utils.dumps(details)
            fd, name = tempfile.mkstemp("tooz", dir=self._tmpdir)
            with os.fdopen(fd, "wb") as fh:
                fh.write(details_blob)
            os.rename(name, path)

        def create_group(self, group_id):
            safe_group_id = self._make_filesystem_safe(group_id)
            group_dir = os.path.join(self._group_dir, safe_group_id)
            group_meta_path = os.path.join(group_dir, '.metadata')

            def _do_create_group():
                if os.path.exists(os.path.join(group_dir, ".metadata")):
                    # NOTE(sileht): We update the group metadata even
                    # they are already good, so ensure dict key are convert
                    # to unicode in case of the file have been written with
                    # tooz < 1.36
                    self._update_group_metadata(group_meta_path, group_id)
                    raise coordination.GroupAlreadyExist(group_id)
                else:
                    fileutils.ensure_tree(group_dir)
                    self._update_group_metadata(group_meta_path, group_id)
            fut = self._executor.submit(_do_create_group)
            return FileFutureResult(fut)

        def join_group(self, group_id, capabilities=b""):
            safe_group_id = self._make_filesystem_safe(group_id)
            group_dir = os.path.join(self._group_dir, safe_group_id)
            me_path = os.path.join(group_dir, "%s.raw" % self._safe_member_id)

            @_lock_me(self._driver_lock)
            def _do_join_group():
                if not os.path.exists(os.path.join(group_dir, ".metadata")):
                    raise coordination.GroupNotCreated(group_id)
                if os.path.isfile(me_path):
                    raise coordination.MemberAlreadyExist(group_id,
                                                          self._member_id)
                details = {
                    u'capabilities': capabilities,
                    u'joined_on': datetime.datetime.now(),
                    u'member_id': utils.to_binary(self._member_id,
                                                  encoding="utf-8")
                }
                details[u'encoded'] = details[u"member_id"] != self._member_id
                details_blob = utils.dumps(details)
                with open(me_path, "wb") as fh:
                    fh.write(details_blob)
                self._joined_groups.add(group_id)

            fut = self._executor.submit(_do_join_group)
            return FileFutureResult(fut)

        def leave_group(self, group_id):
            safe_group_id = self._make_filesystem_safe(group_id)
            group_dir = os.path.join(self._group_dir, safe_group_id)
            me_path = os.path.join(group_dir, "%s.raw" % self._safe_member_id)

            @_lock_me(self._driver_lock)
            def _do_leave_group():
                if not os.path.exists(os.path.join(group_dir, ".metadata")):
                    raise coordination.GroupNotCreated(group_id)
                try:
                    os.unlink(me_path)
                except EnvironmentError as e:
                    if e.errno != errno.ENOENT:
                        raise
                    else:
                        raise coordination.MemberNotJoined(group_id,
                                                           self._member_id)
                else:
                    self._joined_groups.discard(group_id)

            fut = self._executor.submit(_do_leave_group)
            return FileFutureResult(fut)

        _SCHEMAS = {
            'group': voluptuous.Schema({
                voluptuous.Required('group_id'): voluptuous.Any(str, bytes),
                # NOTE(sileht): tooz <1.36 was creating file without this
                voluptuous.Optional('encoded'): bool,
            }),
            'member': voluptuous.Schema({
                voluptuous.Required('member_id'): voluptuous.Any(str, bytes),
                voluptuous.Required('joined_on'): datetime.datetime,
                # NOTE(sileht): tooz <1.36 was creating file without this
                voluptuous.Optional('encoded'): bool,
            }, extra=voluptuous.ALLOW_EXTRA),
        }

        def _load_and_validate(self, blob, schema_key):
            data = utils.loads(blob)
            data = _convert_from_old_format(data)
            schema = self._SCHEMAS[schema_key]
            return schema(data)

        def _read_member_id(self, path):
            with open(path, 'rb') as fh:
                details = self._load_and_validate(fh.read(), 'member')
                if details.get("encoded"):
                    return details[u'member_id'].decode("utf-8")
                return details[u'member_id']

        def get_members(self, group_id):
            safe_group_id = self._make_filesystem_safe(group_id)
            group_dir = os.path.join(self._group_dir, safe_group_id)

            @_lock_me(self._driver_lock)
            def _do_get_members():
                if not os.path.isdir(group_dir):
                    raise coordination.GroupNotCreated(group_id)
                members = set()
                try:
                    entries = os.listdir(group_dir)
                except EnvironmentError as e:
                    # Did someone manage to remove it before we got here...
                    if e.errno != errno.ENOENT:
                        raise
                else:
                    for entry in entries:
                        if not entry.endswith('.raw'):
                            continue
                        entry_path = os.path.join(group_dir, entry)
                        try:
                            m_time = datetime.datetime.fromtimestamp(
                                os.stat(entry_path).st_mtime)
                            current_time = datetime.datetime.now()
                            delta_time = timeutils.delta_seconds(m_time,
                                                                 current_time)
                            if delta_time >= 0 and delta_time <= self._timeout:
                                member_id = self._read_member_id(entry_path)
                            else:
                                continue
                        except EnvironmentError as e:
                            if e.errno != errno.ENOENT:
                                raise
                        else:
                            members.add(member_id)
                return members

            fut = self._executor.submit(_do_get_members)
            return FileFutureResult(fut)

        def get_member_capabilities(self, group_id, member_id):
            safe_group_id = self._make_filesystem_safe(group_id)
            group_dir = os.path.join(self._group_dir, safe_group_id)
            safe_member_id = self._make_filesystem_safe(member_id)
            member_path = os.path.join(group_dir, "%s.raw" % safe_member_id)

            @_lock_me(self._driver_lock)
            def _do_get_member_capabilities():
                try:
                    with open(member_path, "rb") as fh:
                        contents = fh.read()
                except EnvironmentError as e:
                    if e.errno == errno.ENOENT:
                        if not os.path.isdir(group_dir):
                            raise coordination.GroupNotCreated(group_id)
                        else:
                            raise coordination.MemberNotJoined(group_id,
                                                               member_id)
                    else:
                        raise
                else:
                    details = self._load_and_validate(contents, 'member')
                    return details.get(u"capabilities")

            fut = self._executor.submit(_do_get_member_capabilities)
            return FileFutureResult(fut)

        def delete_group(self, group_id):
            safe_group_id = self._make_filesystem_safe(group_id)
            group_dir = os.path.join(self._group_dir, safe_group_id)

            @_lock_me(self._driver_lock)
            def _do_delete_group():
                try:
                    entries = os.listdir(group_dir)
                except EnvironmentError as e:
                    if e.errno == errno.ENOENT:
                        raise coordination.GroupNotCreated(group_id)
                    else:
                        raise
                else:
                    if len(entries) > 1:
                        raise coordination.GroupNotEmpty(group_id)
                    elif len(entries) == 1 and entries != ['.metadata']:
                        raise tooz.ToozError(
                            "Unexpected path '%s' found in"
                            " group directory '%s' (expected to only find"
                            " a '.metadata' path)" % (entries[0], group_dir))
                    else:
                        try:
                            shutil.rmtree(group_dir)
                        except EnvironmentError as e:
                            if e.errno != errno.ENOENT:
                                raise

            fut = self._executor.submit(_do_delete_group)
            return FileFutureResult(fut)

        def _read_group_id(self, path):
            with open(path, 'rb') as fh:
                details = self._load_and_validate(fh.read(), 'group')
                if details.get("encoded"):
                    return details[u'group_id'].decode("utf-8")
                return details[u'group_id']

        def get_groups(self):

            def _do_get_groups():
                groups = []
                for entry in os.listdir(self._group_dir):
                    path = os.path.join(self._group_dir, entry, '.metadata')
                    try:
                        groups.append(self._read_group_id(path))
                    except EnvironmentError as e:
                        if e.errno != errno.ENOENT:
                            raise
                return groups

            fut = self._executor.submit(_do_get_groups)
            return FileFutureResult(fut)

        def heartbeat(self):
            for group_id in self._joined_groups:
                safe_group_id = self._make_filesystem_safe(group_id)
                group_dir = os.path.join(self._group_dir, safe_group_id)
                member_path = os.path.join(group_dir, "%s.raw" %
                                           self._safe_member_id)

                @_lock_me(self._driver_lock)
                def _do_heartbeat():
                    try:
                        os.utime(member_path, None)
                    except EnvironmentError as err:
                        if err.errno != errno.ENOENT:
                            raise
                _do_heartbeat()
            return self._timeout

        @staticmethod
        def watch_elected_as_leader(group_id, callback):
            raise tooz.NotImplemented

        @staticmethod
        def unwatch_elected_as_leader(group_id, callback):
            raise tooz.NotImplemented

    FileFutureResult = functools.partial(coordination.CoordinatorResult,
                                         failure_translator=_translate_failures)

this page last updated: 2024-03-11 13:06:07

Except where otherwise noted, this document is licensed under Creative Commons Attribution 3.0 License. See all OpenStack Legal Documents.

found an error? report a bug