|
|
@@ -0,0 +1,1412 @@
|
|
|
+"""Kazoo Zookeeper Client"""
|
|
|
+import inspect
|
|
|
+import logging
|
|
|
+import os
|
|
|
+import re
|
|
|
+import warnings
|
|
|
+from collections import defaultdict, deque
|
|
|
+from functools import partial
|
|
|
+from os.path import split
|
|
|
+
|
|
|
+from kazoo.exceptions import (
|
|
|
+ AuthFailedError,
|
|
|
+ ConfigurationError,
|
|
|
+ ConnectionClosedError,
|
|
|
+ ConnectionLoss,
|
|
|
+ NoNodeError,
|
|
|
+ NodeExistsError,
|
|
|
+ SessionExpiredError,
|
|
|
+ WriterNotClosedException,
|
|
|
+)
|
|
|
+from kazoo.handlers.threading import SequentialThreadingHandler
|
|
|
+from kazoo.handlers.utils import capture_exceptions, wrap
|
|
|
+from kazoo.hosts import collect_hosts
|
|
|
+from kazoo.loggingsupport import BLATHER
|
|
|
+from kazoo.protocol.connection import ConnectionHandler
|
|
|
+from kazoo.protocol.paths import normpath
|
|
|
+from kazoo.protocol.paths import _prefix_root
|
|
|
+from kazoo.protocol.serialization import (
|
|
|
+ Auth,
|
|
|
+ CheckVersion,
|
|
|
+ CloseInstance,
|
|
|
+ Create,
|
|
|
+ Delete,
|
|
|
+ Exists,
|
|
|
+ GetChildren,
|
|
|
+ GetChildren2,
|
|
|
+ GetACL,
|
|
|
+ SetACL,
|
|
|
+ GetData,
|
|
|
+ SetData,
|
|
|
+ Sync,
|
|
|
+ Transaction
|
|
|
+)
|
|
|
+from kazoo.protocol.states import KazooState
|
|
|
+from kazoo.protocol.states import KeeperState
|
|
|
+from kazoo.retry import KazooRetry
|
|
|
+from kazoo.security import ACL
|
|
|
+from kazoo.security import OPEN_ACL_UNSAFE
|
|
|
+
|
|
|
+# convenience API
|
|
|
+from kazoo.recipe.barrier import Barrier
|
|
|
+from kazoo.recipe.barrier import DoubleBarrier
|
|
|
+from kazoo.recipe.counter import Counter
|
|
|
+from kazoo.recipe.election import Election
|
|
|
+from kazoo.recipe.lock import Lock
|
|
|
+from kazoo.recipe.lock import Semaphore
|
|
|
+from kazoo.recipe.partitioner import SetPartitioner
|
|
|
+from kazoo.recipe.party import Party
|
|
|
+from kazoo.recipe.party import ShallowParty
|
|
|
+from kazoo.recipe.queue import Queue
|
|
|
+from kazoo.recipe.queue import LockingQueue
|
|
|
+from kazoo.recipe.watchers import ChildrenWatch
|
|
|
+from kazoo.recipe.watchers import DataWatch
|
|
|
+
|
|
|
+try: # pragma: nocover
|
|
|
+ basestring
|
|
|
+except NameError: # pragma: nocover
|
|
|
+ basestring = str
|
|
|
+
|
|
|
+LOST_STATES = (KeeperState.EXPIRED_SESSION, KeeperState.AUTH_FAILED,
|
|
|
+ KeeperState.CLOSED)
|
|
|
+ENVI_VERSION = re.compile('[\w\s:.]*=([\d\.]*).*', re.DOTALL)
|
|
|
+log = logging.getLogger(__name__)
|
|
|
+
|
|
|
+
|
|
|
+_RETRY_COMPAT_DEFAULTS = dict(
|
|
|
+ max_retries=None,
|
|
|
+ retry_delay=0.1,
|
|
|
+ retry_backoff=2,
|
|
|
+ retry_jitter=0.8,
|
|
|
+ retry_max_delay=3600,
|
|
|
+)
|
|
|
+
|
|
|
+_RETRY_COMPAT_MAPPING = dict(
|
|
|
+ max_retries='max_tries',
|
|
|
+ retry_delay='delay',
|
|
|
+ retry_backoff='backoff',
|
|
|
+ retry_jitter='max_jitter',
|
|
|
+ retry_max_delay='max_delay',
|
|
|
+)
|
|
|
+
|
|
|
+
|
|
|
+class KazooClient(object):
|
|
|
+ """An Apache Zookeeper Python client supporting alternate callback
|
|
|
+ handlers and high-level functionality.
|
|
|
+
|
|
|
+ Watch functions registered with this class will not get session
|
|
|
+ events, unlike the default Zookeeper watches. They will also be
|
|
|
+ called with a single argument, a
|
|
|
+ :class:`~kazoo.protocol.states.WatchedEvent` instance.
|
|
|
+
|
|
|
+ """
|
|
|
+ def __init__(self, hosts='127.0.0.1:2181',
|
|
|
+ timeout=10.0, client_id=None, handler=None,
|
|
|
+ default_acl=None, auth_data=None, read_only=None,
|
|
|
+ randomize_hosts=True, connection_retry=None,
|
|
|
+ command_retry=None, logger=None, **kwargs):
|
|
|
+ """Create a :class:`KazooClient` instance. All time arguments
|
|
|
+ are in seconds.
|
|
|
+
|
|
|
+ :param hosts: Comma-separated list of hosts to connect to
|
|
|
+ (e.g. 127.0.0.1:2181,127.0.0.1:2182,[::1]:2183).
|
|
|
+ :param timeout: The longest to wait for a Zookeeper connection.
|
|
|
+ :param client_id: A Zookeeper client id, used when
|
|
|
+ re-establishing a prior session connection.
|
|
|
+ :param handler: An instance of a class implementing the
|
|
|
+ :class:`~kazoo.interfaces.IHandler` interface
|
|
|
+ for callback handling.
|
|
|
+ :param default_acl: A default ACL used on node creation.
|
|
|
+ :param auth_data:
|
|
|
+ A list of authentication credentials to use for the
|
|
|
+ connection. Should be a list of (scheme, credential)
|
|
|
+ tuples as :meth:`add_auth` takes.
|
|
|
+ :param read_only: Allow connections to read only servers.
|
|
|
+ :param randomize_hosts: By default randomize host selection.
|
|
|
+ :param connection_retry:
|
|
|
+ A :class:`kazoo.retry.KazooRetry` object to use for
|
|
|
+ retrying the connection to Zookeeper. Also can be a dict of
|
|
|
+ options which will be used for creating one.
|
|
|
+ :param command_retry:
|
|
|
+ A :class:`kazoo.retry.KazooRetry` object to use for
|
|
|
+ the :meth:`KazooClient.retry` method. Also can be a dict of
|
|
|
+ options which will be used for creating one.
|
|
|
+ :param logger: A custom logger to use instead of the module
|
|
|
+ global `log` instance.
|
|
|
+
|
|
|
+ Basic Example:
|
|
|
+
|
|
|
+ .. code-block:: python
|
|
|
+
|
|
|
+ zk = KazooClient()
|
|
|
+ zk.start()
|
|
|
+ children = zk.get_children('/')
|
|
|
+ zk.stop()
|
|
|
+
|
|
|
+ As a convenience all recipe classes are available as attributes
|
|
|
+ and get automatically bound to the client. For example::
|
|
|
+
|
|
|
+ zk = KazooClient()
|
|
|
+ zk.start()
|
|
|
+ lock = zk.Lock('/lock_path')
|
|
|
+
|
|
|
+ .. versionadded:: 0.6
|
|
|
+ The read_only option. Requires Zookeeper 3.4+
|
|
|
+
|
|
|
+ .. versionadded:: 0.6
|
|
|
+ The retry_max_delay option.
|
|
|
+
|
|
|
+ .. versionadded:: 0.6
|
|
|
+ The randomize_hosts option.
|
|
|
+
|
|
|
+ .. versionchanged:: 0.8
|
|
|
+ Removed the unused watcher argument (was second argument).
|
|
|
+
|
|
|
+ .. versionadded:: 1.2
|
|
|
+ The connection_retry, command_retry and logger options.
|
|
|
+
|
|
|
+ """
|
|
|
+ self.logger = logger or log
|
|
|
+
|
|
|
+ # Record the handler strategy used
|
|
|
+ self.handler = handler if handler else SequentialThreadingHandler()
|
|
|
+ if inspect.isclass(self.handler):
|
|
|
+ raise ConfigurationError("Handler must be an instance of a class, "
|
|
|
+ "not the class: %s" % self.handler)
|
|
|
+
|
|
|
+ self.auth_data = auth_data if auth_data else set([])
|
|
|
+ self.default_acl = default_acl
|
|
|
+ self.randomize_hosts = randomize_hosts
|
|
|
+ self.hosts = None
|
|
|
+ self.chroot = None
|
|
|
+ self.set_hosts(hosts)
|
|
|
+
|
|
|
+ # Curator like simplified state tracking, and listeners for
|
|
|
+ # state transitions
|
|
|
+ self._state = KeeperState.CLOSED
|
|
|
+ self.state = KazooState.LOST
|
|
|
+ self.state_listeners = set()
|
|
|
+
|
|
|
+ self._reset()
|
|
|
+ self.read_only = read_only
|
|
|
+
|
|
|
+ if client_id:
|
|
|
+ self._session_id = client_id[0]
|
|
|
+ self._session_passwd = client_id[1]
|
|
|
+ else:
|
|
|
+ self._reset_session()
|
|
|
+
|
|
|
+ # ZK uses milliseconds
|
|
|
+ self._session_timeout = int(timeout * 1000)
|
|
|
+
|
|
|
+ # We use events like twitter's client to track current and
|
|
|
+ # desired state (connected, and whether to shutdown)
|
|
|
+ self._live = self.handler.event_object()
|
|
|
+ self._writer_stopped = self.handler.event_object()
|
|
|
+ self._stopped = self.handler.event_object()
|
|
|
+ self._stopped.set()
|
|
|
+ self._writer_stopped.set()
|
|
|
+
|
|
|
+ self.retry = self._conn_retry = None
|
|
|
+
|
|
|
+ if type(connection_retry) is dict:
|
|
|
+ self._conn_retry = KazooRetry(**connection_retry)
|
|
|
+ elif type(connection_retry) is KazooRetry:
|
|
|
+ self._conn_retry = connection_retry
|
|
|
+
|
|
|
+ if type(command_retry) is dict:
|
|
|
+ self.retry = KazooRetry(**command_retry)
|
|
|
+ elif type(command_retry) is KazooRetry:
|
|
|
+ self.retry = command_retry
|
|
|
+
|
|
|
+
|
|
|
+ if type(self._conn_retry) is KazooRetry:
|
|
|
+ if self.handler.sleep_func != self._conn_retry.sleep_func:
|
|
|
+ raise ConfigurationError("Retry handler and event handler "
|
|
|
+ " must use the same sleep func")
|
|
|
+
|
|
|
+ if type(self.retry) is KazooRetry:
|
|
|
+ if self.handler.sleep_func != self.retry.sleep_func:
|
|
|
+ raise ConfigurationError("Command retry handler and event "
|
|
|
+ "handler must use the same sleep func")
|
|
|
+
|
|
|
+ if self.retry is None or self._conn_retry is None:
|
|
|
+ old_retry_keys = dict(_RETRY_COMPAT_DEFAULTS)
|
|
|
+ for key in old_retry_keys:
|
|
|
+ try:
|
|
|
+ old_retry_keys[key] = kwargs.pop(key)
|
|
|
+ warnings.warn('Passing retry configuration param %s to the'
|
|
|
+ ' client directly is deprecated, please pass a'
|
|
|
+ ' configured retry object (using param %s)' % (
|
|
|
+ key, _RETRY_COMPAT_MAPPING[key]),
|
|
|
+ DeprecationWarning, stacklevel=2)
|
|
|
+ except KeyError:
|
|
|
+ pass
|
|
|
+
|
|
|
+ retry_keys = {}
|
|
|
+ for oldname, value in old_retry_keys.items():
|
|
|
+ retry_keys[_RETRY_COMPAT_MAPPING[oldname]] = value
|
|
|
+
|
|
|
+ if self._conn_retry is None:
|
|
|
+ self._conn_retry = KazooRetry(
|
|
|
+ sleep_func=self.handler.sleep_func,
|
|
|
+ **retry_keys)
|
|
|
+ if self.retry is None:
|
|
|
+ self.retry = KazooRetry(
|
|
|
+ sleep_func=self.handler.sleep_func,
|
|
|
+ **retry_keys)
|
|
|
+
|
|
|
+ self._conn_retry.interrupt = lambda: self._stopped.is_set()
|
|
|
+ self._connection = ConnectionHandler(self, self._conn_retry.copy(),
|
|
|
+ logger=self.logger)
|
|
|
+
|
|
|
+ # Every retry call should have its own copy of the retry helper
|
|
|
+ # to avoid shared retry counts
|
|
|
+ self._retry = self.retry
|
|
|
+ def _retry(*args, **kwargs):
|
|
|
+ return self._retry.copy()(*args, **kwargs)
|
|
|
+ self.retry = _retry
|
|
|
+
|
|
|
+ self.Barrier = partial(Barrier, self)
|
|
|
+ self.Counter = partial(Counter, self)
|
|
|
+ self.DoubleBarrier = partial(DoubleBarrier, self)
|
|
|
+ self.ChildrenWatch = partial(ChildrenWatch, self)
|
|
|
+ self.DataWatch = partial(DataWatch, self)
|
|
|
+ self.Election = partial(Election, self)
|
|
|
+ self.Lock = partial(Lock, self)
|
|
|
+ self.Party = partial(Party, self)
|
|
|
+ self.Queue = partial(Queue, self)
|
|
|
+ self.LockingQueue = partial(LockingQueue, self)
|
|
|
+ self.SetPartitioner = partial(SetPartitioner, self)
|
|
|
+ self.Semaphore = partial(Semaphore, self)
|
|
|
+ self.ShallowParty = partial(ShallowParty, self)
|
|
|
+
|
|
|
+ # If we got any unhandled keywords, complain like python would
|
|
|
+ if kwargs:
|
|
|
+ raise TypeError('__init__() got unexpected keyword arguments: %s'
|
|
|
+ % (kwargs.keys(),))
|
|
|
+
|
|
|
+ def _reset(self):
|
|
|
+ """Resets a variety of client states for a new connection."""
|
|
|
+ self._queue = deque()
|
|
|
+ self._pending = deque()
|
|
|
+
|
|
|
+ self._reset_watchers()
|
|
|
+ self._reset_session()
|
|
|
+ self.last_zxid = 0
|
|
|
+ self._protocol_version = None
|
|
|
+
|
|
|
+ def _reset_watchers(self):
|
|
|
+ self._child_watchers = defaultdict(set)
|
|
|
+ self._data_watchers = defaultdict(set)
|
|
|
+
|
|
|
+ def _reset_session(self):
|
|
|
+ self._session_id = None
|
|
|
+ self._session_passwd = b'\x00' * 16
|
|
|
+
|
|
|
+ @property
|
|
|
+ def client_state(self):
|
|
|
+ """Returns the last Zookeeper client state
|
|
|
+
|
|
|
+ This is the non-simplified state information and is generally
|
|
|
+ not as useful as the simplified KazooState information.
|
|
|
+
|
|
|
+ """
|
|
|
+ return self._state
|
|
|
+
|
|
|
+ @property
|
|
|
+ def client_id(self):
|
|
|
+ """Returns the client id for this Zookeeper session if
|
|
|
+ connected.
|
|
|
+
|
|
|
+ :returns: client id which consists of the session id and
|
|
|
+ password.
|
|
|
+ :rtype: tuple
|
|
|
+ """
|
|
|
+ if self._live.is_set():
|
|
|
+ return (self._session_id, self._session_passwd)
|
|
|
+ return None
|
|
|
+
|
|
|
+ @property
|
|
|
+ def connected(self):
|
|
|
+ """Returns whether the Zookeeper connection has been
|
|
|
+ established."""
|
|
|
+ return self._live.is_set()
|
|
|
+
|
|
|
+ def set_hosts(self, hosts, randomize_hosts=None):
|
|
|
+ """ sets the list of hosts used by this client.
|
|
|
+
|
|
|
+ This function accepts the same format hosts parameter as the init
|
|
|
+ function and sets the client to use the new hosts the next time it
|
|
|
+ needs to look up a set of hosts. This function does not affect the
|
|
|
+ current connected status.
|
|
|
+
|
|
|
+ It is not currently possible to change the chroot with this function,
|
|
|
+ setting a host list with a new chroot will raise a ConfigurationError.
|
|
|
+
|
|
|
+ :param hosts: see description in :meth:`KazooClient.__init__`
|
|
|
+ :param randomize_hosts: override client default for host randomization
|
|
|
+ :raises:
|
|
|
+ :exc:`ConfigurationError` if the hosts argument changes the chroot
|
|
|
+
|
|
|
+ .. versionadded:: 1.4
|
|
|
+
|
|
|
+ .. warning::
|
|
|
+
|
|
|
+ Using this function to point a client to a completely disparate
|
|
|
+ zookeeper server cluster has undefined behavior.
|
|
|
+
|
|
|
+ """
|
|
|
+
|
|
|
+ if randomize_hosts is None:
|
|
|
+ randomize_hosts = self.randomize_hosts
|
|
|
+
|
|
|
+ self.hosts, chroot = collect_hosts(hosts, randomize_hosts)
|
|
|
+
|
|
|
+ if chroot:
|
|
|
+ new_chroot = normpath(chroot)
|
|
|
+ else:
|
|
|
+ new_chroot = ''
|
|
|
+
|
|
|
+ if self.chroot is not None and new_chroot != self.chroot:
|
|
|
+ raise ConfigurationError("Changing chroot at runtime is not "
|
|
|
+ "currently supported")
|
|
|
+
|
|
|
+ self.chroot = new_chroot
|
|
|
+
|
|
|
+ def add_listener(self, listener):
|
|
|
+ """Add a function to be called for connection state changes.
|
|
|
+
|
|
|
+ This function will be called with a
|
|
|
+ :class:`~kazoo.protocol.states.KazooState` instance indicating
|
|
|
+ the new connection state on state transitions.
|
|
|
+
|
|
|
+ .. warning::
|
|
|
+
|
|
|
+ This function must not block. If its at all likely that it
|
|
|
+ might need data or a value that could result in blocking
|
|
|
+ than the :meth:`~kazoo.interfaces.IHandler.spawn` method
|
|
|
+ should be used so that the listener can return immediately.
|
|
|
+
|
|
|
+ """
|
|
|
+ if not (listener and callable(listener)):
|
|
|
+ raise ConfigurationError("listener must be callable")
|
|
|
+ self.state_listeners.add(listener)
|
|
|
+
|
|
|
+ def remove_listener(self, listener):
|
|
|
+ """Remove a listener function"""
|
|
|
+ self.state_listeners.discard(listener)
|
|
|
+
|
|
|
+ def _make_state_change(self, state):
|
|
|
+ # skip if state is current
|
|
|
+ if self.state == state:
|
|
|
+ return
|
|
|
+
|
|
|
+ self.state = state
|
|
|
+
|
|
|
+ # Create copy of listeners for iteration in case one needs to
|
|
|
+ # remove itself
|
|
|
+ for listener in list(self.state_listeners):
|
|
|
+ try:
|
|
|
+ remove = listener(state)
|
|
|
+ if remove is True:
|
|
|
+ self.remove_listener(listener)
|
|
|
+ except Exception:
|
|
|
+ self.logger.exception("Error in connection state listener")
|
|
|
+
|
|
|
+ def _session_callback(self, state):
|
|
|
+ if state == self._state:
|
|
|
+ return
|
|
|
+
|
|
|
+ # Note that we don't check self.state == LOST since that's also
|
|
|
+ # the client's initial state
|
|
|
+ dead_state = self._state in LOST_STATES
|
|
|
+ self._state = state
|
|
|
+
|
|
|
+ # If we were previously closed or had an expired session, and
|
|
|
+ # are now connecting, don't bother with the rest of the
|
|
|
+ # transitions since they only apply after
|
|
|
+ # we've established a connection
|
|
|
+ if dead_state and state == KeeperState.CONNECTING:
|
|
|
+ self.logger.log(BLATHER, "Skipping state change")
|
|
|
+ return
|
|
|
+
|
|
|
+ if state in (KeeperState.CONNECTED, KeeperState.CONNECTED_RO):
|
|
|
+ self.logger.info("Zookeeper connection established, state: %s", state)
|
|
|
+ self._live.set()
|
|
|
+ self._make_state_change(KazooState.CONNECTED)
|
|
|
+ elif state in LOST_STATES:
|
|
|
+ self.logger.info("Zookeeper session lost, state: %s", state)
|
|
|
+ self._live.clear()
|
|
|
+ self._make_state_change(KazooState.LOST)
|
|
|
+ self._notify_pending(state)
|
|
|
+ self._reset()
|
|
|
+ else:
|
|
|
+ self.logger.info("Zookeeper connection lost")
|
|
|
+ # Connection lost
|
|
|
+ self._live.clear()
|
|
|
+ self._notify_pending(state)
|
|
|
+ self._make_state_change(KazooState.SUSPENDED)
|
|
|
+ self._reset_watchers()
|
|
|
+
|
|
|
+ def _notify_pending(self, state):
|
|
|
+ """Used to clear a pending response queue and request queue
|
|
|
+ during connection drops."""
|
|
|
+ if state == KeeperState.AUTH_FAILED:
|
|
|
+ exc = AuthFailedError()
|
|
|
+ elif state == KeeperState.EXPIRED_SESSION:
|
|
|
+ exc = SessionExpiredError()
|
|
|
+ else:
|
|
|
+ exc = ConnectionLoss()
|
|
|
+
|
|
|
+ while True:
|
|
|
+ try:
|
|
|
+ request, async_object, xid = self._pending.popleft()
|
|
|
+ if async_object:
|
|
|
+ async_object.set_exception(exc)
|
|
|
+ except IndexError:
|
|
|
+ break
|
|
|
+
|
|
|
+ while True:
|
|
|
+ try:
|
|
|
+ request, async_object = self._queue.popleft()
|
|
|
+ if async_object:
|
|
|
+ async_object.set_exception(exc)
|
|
|
+ except IndexError:
|
|
|
+ break
|
|
|
+
|
|
|
+ def _safe_close(self):
|
|
|
+ self.handler.stop()
|
|
|
+ timeout = self._session_timeout // 1000
|
|
|
+ if timeout < 10:
|
|
|
+ timeout = 10
|
|
|
+ if not self._connection.stop(timeout):
|
|
|
+ raise WriterNotClosedException(
|
|
|
+ "Writer still open from prior connection "
|
|
|
+ "and wouldn't close after %s seconds" % timeout)
|
|
|
+
|
|
|
+ def _call(self, request, async_object):
|
|
|
+ """Ensure there's an active connection and put the request in
|
|
|
+ the queue if there is.
|
|
|
+
|
|
|
+ Returns False if the call short circuits due to AUTH_FAILED,
|
|
|
+ CLOSED, EXPIRED_SESSION or CONNECTING state.
|
|
|
+
|
|
|
+ """
|
|
|
+
|
|
|
+ if self._state == KeeperState.AUTH_FAILED:
|
|
|
+ async_object.set_exception(AuthFailedError())
|
|
|
+ return False
|
|
|
+ elif self._state == KeeperState.CLOSED:
|
|
|
+ async_object.set_exception(ConnectionClosedError(
|
|
|
+ "Connection has been closed"))
|
|
|
+ return False
|
|
|
+ elif self._state in (KeeperState.EXPIRED_SESSION,
|
|
|
+ KeeperState.CONNECTING):
|
|
|
+ async_object.set_exception(SessionExpiredError())
|
|
|
+ return False
|
|
|
+
|
|
|
+ self._queue.append((request, async_object))
|
|
|
+
|
|
|
+ # wake the connection, guarding against a race with close()
|
|
|
+ write_pipe = self._connection._write_pipe
|
|
|
+ if write_pipe is None:
|
|
|
+ async_object.set_exception(ConnectionClosedError(
|
|
|
+ "Connection has been closed"))
|
|
|
+ try:
|
|
|
+ os.write(write_pipe, b'\0')
|
|
|
+ except:
|
|
|
+ async_object.set_exception(ConnectionClosedError(
|
|
|
+ "Connection has been closed"))
|
|
|
+
|
|
|
+ def start(self, timeout=15):
|
|
|
+ """Initiate connection to ZK.
|
|
|
+
|
|
|
+ :param timeout: Time in seconds to wait for connection to
|
|
|
+ succeed.
|
|
|
+ :raises: :attr:`~kazoo.interfaces.IHandler.timeout_exception`
|
|
|
+ if the connection wasn't established within `timeout`
|
|
|
+ seconds.
|
|
|
+
|
|
|
+ """
|
|
|
+ event = self.start_async()
|
|
|
+ event.wait(timeout=timeout)
|
|
|
+ if not self.connected:
|
|
|
+ # We time-out, ensure we are disconnected
|
|
|
+ self.stop()
|
|
|
+ raise self.handler.timeout_exception("Connection time-out")
|
|
|
+
|
|
|
+ if self.chroot and not self.exists("/"):
|
|
|
+ warnings.warn("No chroot path exists, the chroot path "
|
|
|
+ "should be created before normal use.")
|
|
|
+
|
|
|
+ def start_async(self):
|
|
|
+ """Asynchronously initiate connection to ZK.
|
|
|
+
|
|
|
+ :returns: An event object that can be checked to see if the
|
|
|
+ connection is alive.
|
|
|
+ :rtype: :class:`~threading.Event` compatible object.
|
|
|
+
|
|
|
+ """
|
|
|
+ # If we're already connected, ignore
|
|
|
+ if self._live.is_set():
|
|
|
+ return self._live
|
|
|
+
|
|
|
+ # Make sure we're safely closed
|
|
|
+ self._safe_close()
|
|
|
+
|
|
|
+ # We've been asked to connect, clear the stop and our writer
|
|
|
+ # thread indicator
|
|
|
+ self._stopped.clear()
|
|
|
+ self._writer_stopped.clear()
|
|
|
+
|
|
|
+ # Start the handler
|
|
|
+ self.handler.start()
|
|
|
+
|
|
|
+ # Start the connection
|
|
|
+ self._connection.start()
|
|
|
+ return self._live
|
|
|
+
|
|
|
+ def stop(self):
|
|
|
+ """Gracefully stop this Zookeeper session.
|
|
|
+
|
|
|
+ This method can be called while a reconnection attempt is in
|
|
|
+ progress, which will then be halted.
|
|
|
+
|
|
|
+ Once the connection is closed, its session becomes invalid. All
|
|
|
+ the ephemeral nodes in the ZooKeeper server associated with the
|
|
|
+ session will be removed. The watches left on those nodes (and
|
|
|
+ on their parents) will be triggered.
|
|
|
+
|
|
|
+ """
|
|
|
+ if self._stopped.is_set():
|
|
|
+ return
|
|
|
+
|
|
|
+ self._stopped.set()
|
|
|
+ self._queue.append((CloseInstance, None))
|
|
|
+ os.write(self._connection._write_pipe, b'\0')
|
|
|
+ self._safe_close()
|
|
|
+
|
|
|
+ def restart(self):
|
|
|
+ """Stop and restart the Zookeeper session."""
|
|
|
+ self.stop()
|
|
|
+ self.start()
|
|
|
+
|
|
|
+ def close(self):
|
|
|
+ """Free any resources held by the client.
|
|
|
+
|
|
|
+ This method should be called on a stopped client before it is
|
|
|
+ discarded. Not doing so may result in filehandles being leaked.
|
|
|
+
|
|
|
+ .. versionadded:: 1.0
|
|
|
+ """
|
|
|
+ self._connection.close()
|
|
|
+
|
|
|
+ def command(self, cmd=b'ruok'):
|
|
|
+ """Sent a management command to the current ZK server.
|
|
|
+
|
|
|
+ Examples are `ruok`, `envi` or `stat`.
|
|
|
+
|
|
|
+ :returns: An unstructured textual response.
|
|
|
+ :rtype: str
|
|
|
+
|
|
|
+ :raises:
|
|
|
+ :exc:`ConnectionLoss` if there is no connection open, or
|
|
|
+ possibly a :exc:`socket.error` if there's a problem with
|
|
|
+ the connection used just for this command.
|
|
|
+
|
|
|
+ .. versionadded:: 0.5
|
|
|
+
|
|
|
+ """
|
|
|
+ if not self._live.is_set():
|
|
|
+ raise ConnectionLoss("No connection to server")
|
|
|
+
|
|
|
+ peer = self._connection._socket.getpeername()
|
|
|
+ sock = self.handler.create_connection(
|
|
|
+ peer, timeout=self._session_timeout / 1000.0)
|
|
|
+ sock.sendall(cmd)
|
|
|
+ result = sock.recv(8192)
|
|
|
+ sock.close()
|
|
|
+ return result.decode('utf-8', 'replace')
|
|
|
+
|
|
|
+ def server_version(self):
|
|
|
+ """Get the version of the currently connected ZK server.
|
|
|
+
|
|
|
+ :returns: The server version, for example (3, 4, 3).
|
|
|
+ :rtype: tuple
|
|
|
+
|
|
|
+ .. versionadded:: 0.5
|
|
|
+
|
|
|
+ """
|
|
|
+ data = self.command(b'envi')
|
|
|
+ string = ENVI_VERSION.match(data).group(1)
|
|
|
+ return tuple([int(i) for i in string.split('.')])
|
|
|
+
|
|
|
+ def add_auth(self, scheme, credential):
|
|
|
+ """Send credentials to server.
|
|
|
+
|
|
|
+ :param scheme: authentication scheme (default supported:
|
|
|
+ "digest").
|
|
|
+ :param credential: the credential -- value depends on scheme.
|
|
|
+
|
|
|
+ :returns: True if it was successful.
|
|
|
+ :rtype: bool
|
|
|
+
|
|
|
+ :raises:
|
|
|
+ :exc:`~kazoo.exceptions.AuthFailedError` if it failed though
|
|
|
+ the session state will be set to AUTH_FAILED as well.
|
|
|
+
|
|
|
+ """
|
|
|
+ return self.add_auth_async(scheme, credential).get()
|
|
|
+
|
|
|
+ def add_auth_async(self, scheme, credential):
|
|
|
+ """Asynchronously send credentials to server. Takes the same
|
|
|
+ arguments as :meth:`add_auth`.
|
|
|
+
|
|
|
+ :rtype: :class:`~kazoo.interfaces.IAsyncResult`
|
|
|
+
|
|
|
+ """
|
|
|
+ if not isinstance(scheme, basestring):
|
|
|
+ raise TypeError("Invalid type for scheme")
|
|
|
+ if not isinstance(credential, basestring):
|
|
|
+ raise TypeError("Invalid type for credential")
|
|
|
+
|
|
|
+ # we need this auth data to re-authenticate on reconnect
|
|
|
+ self.auth_data.add((scheme, credential))
|
|
|
+
|
|
|
+ async_result = self.handler.async_result()
|
|
|
+ self._call(Auth(0, scheme, credential), async_result)
|
|
|
+ return async_result
|
|
|
+
|
|
|
+ def unchroot(self, path):
|
|
|
+ """Strip the chroot if applicable from the path."""
|
|
|
+ if not self.chroot:
|
|
|
+ return path
|
|
|
+
|
|
|
+ if path.startswith(self.chroot):
|
|
|
+ return path[len(self.chroot):]
|
|
|
+ else:
|
|
|
+ return path
|
|
|
+
|
|
|
+ def sync_async(self, path):
|
|
|
+ """Asynchronous sync.
|
|
|
+
|
|
|
+ :rtype: :class:`~kazoo.interfaces.IAsyncResult`
|
|
|
+
|
|
|
+ """
|
|
|
+ async_result = self.handler.async_result()
|
|
|
+ self._call(Sync(_prefix_root(self.chroot, path)), async_result)
|
|
|
+ return async_result
|
|
|
+
|
|
|
+ def sync(self, path):
|
|
|
+ """Sync, blocks until response is acknowledged.
|
|
|
+
|
|
|
+ Flushes channel between process and leader.
|
|
|
+
|
|
|
+ :param path: path of node.
|
|
|
+ :returns: The node path that was synced.
|
|
|
+ :raises:
|
|
|
+ :exc:`~kazoo.exceptions.ZookeeperError` if the server
|
|
|
+ returns a non-zero error code.
|
|
|
+
|
|
|
+ .. versionadded:: 0.5
|
|
|
+
|
|
|
+ """
|
|
|
+ return self.sync_async(path).get()
|
|
|
+
|
|
|
+ def create(self, path, value=b"", acl=None, ephemeral=False,
|
|
|
+ sequence=False, makepath=False):
|
|
|
+ """Create a node with the given value as its data. Optionally
|
|
|
+ set an ACL on the node.
|
|
|
+
|
|
|
+ The ephemeral and sequence arguments determine the type of the
|
|
|
+ node.
|
|
|
+
|
|
|
+ An ephemeral node will be automatically removed by ZooKeeper
|
|
|
+ when the session associated with the creation of the node
|
|
|
+ expires.
|
|
|
+
|
|
|
+ A sequential node will be given the specified path plus a
|
|
|
+ suffix `i` where i is the current sequential number of the
|
|
|
+ node. The sequence number is always fixed length of 10 digits,
|
|
|
+ 0 padded. Once such a node is created, the sequential number
|
|
|
+ will be incremented by one.
|
|
|
+
|
|
|
+ If a node with the same actual path already exists in
|
|
|
+ ZooKeeper, a NodeExistsError will be raised. Note that since a
|
|
|
+ different actual path is used for each invocation of creating
|
|
|
+ sequential nodes with the same path argument, the call will
|
|
|
+ never raise NodeExistsError.
|
|
|
+
|
|
|
+ If the parent node does not exist in ZooKeeper, a NoNodeError
|
|
|
+ will be raised. Setting the optional `makepath` argument to
|
|
|
+ `True` will create all missing parent nodes instead.
|
|
|
+
|
|
|
+ An ephemeral node cannot have children. If the parent node of
|
|
|
+ the given path is ephemeral, a NoChildrenForEphemeralsError
|
|
|
+ will be raised.
|
|
|
+
|
|
|
+ This operation, if successful, will trigger all the watches
|
|
|
+ left on the node of the given path by :meth:`exists` and
|
|
|
+ :meth:`get` API calls, and the watches left on the parent node
|
|
|
+ by :meth:`get_children` API calls.
|
|
|
+
|
|
|
+ The maximum allowable size of the node value is 1 MB. Values
|
|
|
+ larger than this will cause a ZookeeperError to be raised.
|
|
|
+
|
|
|
+ :param path: Path of node.
|
|
|
+ :param value: Initial bytes value of node.
|
|
|
+ :param acl: :class:`~kazoo.security.ACL` list.
|
|
|
+ :param ephemeral: Boolean indicating whether node is ephemeral
|
|
|
+ (tied to this session).
|
|
|
+ :param sequence: Boolean indicating whether path is suffixed
|
|
|
+ with a unique index.
|
|
|
+ :param makepath: Whether the path should be created if it
|
|
|
+ doesn't exist.
|
|
|
+ :returns: Real path of the new node.
|
|
|
+ :rtype: str
|
|
|
+
|
|
|
+ :raises:
|
|
|
+ :exc:`~kazoo.exceptions.NodeExistsError` if the node
|
|
|
+ already exists.
|
|
|
+
|
|
|
+ :exc:`~kazoo.exceptions.NoNodeError` if parent nodes are
|
|
|
+ missing.
|
|
|
+
|
|
|
+ :exc:`~kazoo.exceptions.NoChildrenForEphemeralsError` if
|
|
|
+ the parent node is an ephemeral node.
|
|
|
+
|
|
|
+ :exc:`~kazoo.exceptions.ZookeeperError` if the provided
|
|
|
+ value is too large.
|
|
|
+
|
|
|
+ :exc:`~kazoo.exceptions.ZookeeperError` if the server
|
|
|
+ returns a non-zero error code.
|
|
|
+
|
|
|
+ """
|
|
|
+ acl = acl or self.default_acl
|
|
|
+ return self.create_async(path, value, acl=acl, ephemeral=ephemeral,
|
|
|
+ sequence=sequence, makepath=makepath).get()
|
|
|
+
|
|
|
+ def create_async(self, path, value=b"", acl=None, ephemeral=False,
|
|
|
+ sequence=False, makepath=False):
|
|
|
+ """Asynchronously create a ZNode. Takes the same arguments as
|
|
|
+ :meth:`create`.
|
|
|
+
|
|
|
+ :rtype: :class:`~kazoo.interfaces.IAsyncResult`
|
|
|
+
|
|
|
+ .. versionadded:: 1.1
|
|
|
+ The makepath option.
|
|
|
+
|
|
|
+ """
|
|
|
+ if acl is None and self.default_acl:
|
|
|
+ acl = self.default_acl
|
|
|
+
|
|
|
+ if not isinstance(path, basestring):
|
|
|
+ raise TypeError("path must be a string")
|
|
|
+ if acl and (isinstance(acl, ACL) or
|
|
|
+ not isinstance(acl, (tuple, list))):
|
|
|
+ raise TypeError("acl must be a tuple/list of ACL's")
|
|
|
+ if value is not None and not isinstance(value, bytes):
|
|
|
+ raise TypeError("value must be a byte string")
|
|
|
+ if not isinstance(ephemeral, bool):
|
|
|
+ raise TypeError("ephemeral must be a bool")
|
|
|
+ if not isinstance(sequence, bool):
|
|
|
+ raise TypeError("sequence must be a bool")
|
|
|
+ if not isinstance(makepath, bool):
|
|
|
+ raise TypeError("makepath must be a bool")
|
|
|
+
|
|
|
+ flags = 0
|
|
|
+ if ephemeral:
|
|
|
+ flags |= 1
|
|
|
+ if sequence:
|
|
|
+ flags |= 2
|
|
|
+ if acl is None:
|
|
|
+ acl = OPEN_ACL_UNSAFE
|
|
|
+
|
|
|
+ async_result = self.handler.async_result()
|
|
|
+
|
|
|
+ @capture_exceptions(async_result)
|
|
|
+ def do_create():
|
|
|
+ result = self._create_async_inner(path, value, acl, flags, trailing=sequence)
|
|
|
+ result.rawlink(create_completion)
|
|
|
+
|
|
|
+ @capture_exceptions(async_result)
|
|
|
+ def retry_completion(result):
|
|
|
+ result.get()
|
|
|
+ do_create()
|
|
|
+
|
|
|
+ @wrap(async_result)
|
|
|
+ def create_completion(result):
|
|
|
+ try:
|
|
|
+ return self.unchroot(result.get())
|
|
|
+ except NoNodeError:
|
|
|
+ if not makepath:
|
|
|
+ raise
|
|
|
+ if sequence and path.endswith('/'):
|
|
|
+ parent = path.rstrip('/')
|
|
|
+ else:
|
|
|
+ parent, _ = split(path)
|
|
|
+ self.ensure_path_async(parent, acl).rawlink(retry_completion)
|
|
|
+
|
|
|
+ do_create()
|
|
|
+ return async_result
|
|
|
+
|
|
|
+ def _create_async_inner(self, path, value, acl, flags, trailing=False):
|
|
|
+ async_result = self.handler.async_result()
|
|
|
+ call_result = self._call(
|
|
|
+ Create(_prefix_root(self.chroot, path, trailing=trailing),
|
|
|
+ value, acl, flags), async_result)
|
|
|
+ if call_result is False:
|
|
|
+ # We hit a short-circuit exit on the _call. Because we are
|
|
|
+ # not using the original async_result here, we bubble the
|
|
|
+ # exception upwards to the do_create function in
|
|
|
+ # KazooClient.create so that it gets set on the correct
|
|
|
+ # async_result object
|
|
|
+ raise async_result.exception
|
|
|
+ return async_result
|
|
|
+
|
|
|
+ def ensure_path(self, path, acl=None):
|
|
|
+ """Recursively create a path if it doesn't exist.
|
|
|
+
|
|
|
+ :param path: Path of node.
|
|
|
+ :param acl: Permissions for node.
|
|
|
+
|
|
|
+ """
|
|
|
+ return self.ensure_path_async(path, acl).get()
|
|
|
+
|
|
|
+ def ensure_path_async(self, path, acl=None):
|
|
|
+ """Recursively create a path asynchronously if it doesn't
|
|
|
+ exist. Takes the same arguments as :meth:`ensure_path`.
|
|
|
+
|
|
|
+ :rtype: :class:`~kazoo.interfaces.IAsyncResult`
|
|
|
+
|
|
|
+ .. versionadded:: 1.1
|
|
|
+
|
|
|
+ """
|
|
|
+ acl = acl or self.default_acl
|
|
|
+ async_result = self.handler.async_result()
|
|
|
+
|
|
|
+ @wrap(async_result)
|
|
|
+ def create_completion(result):
|
|
|
+ try:
|
|
|
+ return result.get()
|
|
|
+ except NodeExistsError:
|
|
|
+ return True
|
|
|
+
|
|
|
+ @capture_exceptions(async_result)
|
|
|
+ def prepare_completion(next_path, result):
|
|
|
+ result.get()
|
|
|
+ self.create_async(next_path, acl=acl).rawlink(create_completion)
|
|
|
+
|
|
|
+ @wrap(async_result)
|
|
|
+ def exists_completion(path, result):
|
|
|
+ if result.get():
|
|
|
+ return True
|
|
|
+ parent, node = split(path)
|
|
|
+ if node:
|
|
|
+ self.ensure_path_async(parent, acl=acl).rawlink(
|
|
|
+ partial(prepare_completion, path))
|
|
|
+ else:
|
|
|
+ self.create_async(path, acl=acl).rawlink(create_completion)
|
|
|
+
|
|
|
+ self.exists_async(path).rawlink(partial(exists_completion, path))
|
|
|
+
|
|
|
+ return async_result
|
|
|
+
|
|
|
+ def exists(self, path, watch=None):
|
|
|
+ """Check if a node exists.
|
|
|
+
|
|
|
+ If a watch is provided, it will be left on the node with the
|
|
|
+ given path. The watch will be triggered by a successful
|
|
|
+ operation that creates/deletes the node or sets the data on the
|
|
|
+ node.
|
|
|
+
|
|
|
+ :param path: Path of node.
|
|
|
+ :param watch: Optional watch callback to set for future changes
|
|
|
+ to this path.
|
|
|
+ :returns: ZnodeStat of the node if it exists, else None if the
|
|
|
+ node does not exist.
|
|
|
+ :rtype: :class:`~kazoo.protocol.states.ZnodeStat` or `None`.
|
|
|
+
|
|
|
+ :raises:
|
|
|
+ :exc:`~kazoo.exceptions.ZookeeperError` if the server
|
|
|
+ returns a non-zero error code.
|
|
|
+
|
|
|
+ """
|
|
|
+ return self.exists_async(path, watch).get()
|
|
|
+
|
|
|
+ def exists_async(self, path, watch=None):
|
|
|
+ """Asynchronously check if a node exists. Takes the same
|
|
|
+ arguments as :meth:`exists`.
|
|
|
+
|
|
|
+ :rtype: :class:`~kazoo.interfaces.IAsyncResult`
|
|
|
+
|
|
|
+ """
|
|
|
+ if not isinstance(path, basestring):
|
|
|
+ raise TypeError("path must be a string")
|
|
|
+ if watch and not callable(watch):
|
|
|
+ raise TypeError("watch must be a callable")
|
|
|
+
|
|
|
+ async_result = self.handler.async_result()
|
|
|
+ self._call(Exists(_prefix_root(self.chroot, path), watch),
|
|
|
+ async_result)
|
|
|
+ return async_result
|
|
|
+
|
|
|
+ def get(self, path, watch=None):
|
|
|
+ """Get the value of a node.
|
|
|
+
|
|
|
+ If a watch is provided, it will be left on the node with the
|
|
|
+ given path. The watch will be triggered by a successful
|
|
|
+ operation that sets data on the node, or deletes the node.
|
|
|
+
|
|
|
+ :param path: Path of node.
|
|
|
+ :param watch: Optional watch callback to set for future changes
|
|
|
+ to this path.
|
|
|
+ :returns:
|
|
|
+ Tuple (value, :class:`~kazoo.protocol.states.ZnodeStat`) of
|
|
|
+ node.
|
|
|
+ :rtype: tuple
|
|
|
+
|
|
|
+ :raises:
|
|
|
+ :exc:`~kazoo.exceptions.NoNodeError` if the node doesn't
|
|
|
+ exist
|
|
|
+
|
|
|
+ :exc:`~kazoo.exceptions.ZookeeperError` if the server
|
|
|
+ returns a non-zero error code
|
|
|
+
|
|
|
+ """
|
|
|
+ return self.get_async(path, watch).get()
|
|
|
+
|
|
|
+ def get_async(self, path, watch=None):
|
|
|
+ """Asynchronously get the value of a node. Takes the same
|
|
|
+ arguments as :meth:`get`.
|
|
|
+
|
|
|
+ :rtype: :class:`~kazoo.interfaces.IAsyncResult`
|
|
|
+
|
|
|
+ """
|
|
|
+ if not isinstance(path, basestring):
|
|
|
+ raise TypeError("path must be a string")
|
|
|
+ if watch and not callable(watch):
|
|
|
+ raise TypeError("watch must be a callable")
|
|
|
+
|
|
|
+ async_result = self.handler.async_result()
|
|
|
+ self._call(GetData(_prefix_root(self.chroot, path), watch),
|
|
|
+ async_result)
|
|
|
+ return async_result
|
|
|
+
|
|
|
+ def get_children(self, path, watch=None, include_data=False):
|
|
|
+ """Get a list of child nodes of a path.
|
|
|
+
|
|
|
+ If a watch is provided it will be left on the node with the
|
|
|
+ given path. The watch will be triggered by a successful
|
|
|
+ operation that deletes the node of the given path or
|
|
|
+ creates/deletes a child under the node.
|
|
|
+
|
|
|
+ The list of children returned is not sorted and no guarantee is
|
|
|
+ provided as to its natural or lexical order.
|
|
|
+
|
|
|
+ :param path: Path of node to list.
|
|
|
+ :param watch: Optional watch callback to set for future changes
|
|
|
+ to this path.
|
|
|
+ :param include_data:
|
|
|
+ Include the :class:`~kazoo.protocol.states.ZnodeStat` of
|
|
|
+ the node in addition to the children. This option changes
|
|
|
+ the return value to be a tuple of (children, stat).
|
|
|
+
|
|
|
+ :returns: List of child node names, or tuple if `include_data`
|
|
|
+ is `True`.
|
|
|
+ :rtype: list
|
|
|
+
|
|
|
+ :raises:
|
|
|
+ :exc:`~kazoo.exceptions.NoNodeError` if the node doesn't
|
|
|
+ exist.
|
|
|
+
|
|
|
+ :exc:`~kazoo.exceptions.ZookeeperError` if the server
|
|
|
+ returns a non-zero error code.
|
|
|
+
|
|
|
+ .. versionadded:: 0.5
|
|
|
+ The `include_data` option.
|
|
|
+
|
|
|
+ """
|
|
|
+ return self.get_children_async(path, watch, include_data).get()
|
|
|
+
|
|
|
+ def get_children_async(self, path, watch=None, include_data=False):
|
|
|
+ """Asynchronously get a list of child nodes of a path. Takes
|
|
|
+ the same arguments as :meth:`get_children`.
|
|
|
+
|
|
|
+ :rtype: :class:`~kazoo.interfaces.IAsyncResult`
|
|
|
+
|
|
|
+ """
|
|
|
+ if not isinstance(path, basestring):
|
|
|
+ raise TypeError("path must be a string")
|
|
|
+ if watch and not callable(watch):
|
|
|
+ raise TypeError("watch must be a callable")
|
|
|
+ if not isinstance(include_data, bool):
|
|
|
+ raise TypeError("include_data must be a bool")
|
|
|
+
|
|
|
+ async_result = self.handler.async_result()
|
|
|
+ if include_data:
|
|
|
+ req = GetChildren2(_prefix_root(self.chroot, path), watch)
|
|
|
+ else:
|
|
|
+ req = GetChildren(_prefix_root(self.chroot, path), watch)
|
|
|
+ self._call(req, async_result)
|
|
|
+ return async_result
|
|
|
+
|
|
|
+ def get_acls(self, path):
|
|
|
+ """Return the ACL and stat of the node of the given path.
|
|
|
+
|
|
|
+ :param path: Path of the node.
|
|
|
+ :returns: The ACL array of the given node and its
|
|
|
+ :class:`~kazoo.protocol.states.ZnodeStat`.
|
|
|
+ :rtype: tuple of (:class:`~kazoo.security.ACL` list,
|
|
|
+ :class:`~kazoo.protocol.states.ZnodeStat`)
|
|
|
+ :raises:
|
|
|
+ :exc:`~kazoo.exceptions.NoNodeError` if the node doesn't
|
|
|
+ exist.
|
|
|
+
|
|
|
+ :exc:`~kazoo.exceptions.ZookeeperError` if the server
|
|
|
+ returns a non-zero error code
|
|
|
+
|
|
|
+ .. versionadded:: 0.5
|
|
|
+
|
|
|
+ """
|
|
|
+ return self.get_acls_async(path).get()
|
|
|
+
|
|
|
+ def get_acls_async(self, path):
|
|
|
+ """Return the ACL and stat of the node of the given path. Takes
|
|
|
+ the same arguments as :meth:`get_acls`.
|
|
|
+
|
|
|
+ :rtype: :class:`~kazoo.interfaces.IAsyncResult`
|
|
|
+
|
|
|
+ """
|
|
|
+ if not isinstance(path, basestring):
|
|
|
+ raise TypeError("path must be a string")
|
|
|
+
|
|
|
+ async_result = self.handler.async_result()
|
|
|
+ self._call(GetACL(_prefix_root(self.chroot, path)), async_result)
|
|
|
+ return async_result
|
|
|
+
|
|
|
+ def set_acls(self, path, acls, version=-1):
|
|
|
+ """Set the ACL for the node of the given path.
|
|
|
+
|
|
|
+ Set the ACL for the node of the given path if such a node
|
|
|
+ exists and the given version matches the version of the node.
|
|
|
+
|
|
|
+ :param path: Path for the node.
|
|
|
+ :param acls: List of :class:`~kazoo.security.ACL` objects to
|
|
|
+ set.
|
|
|
+ :param version: The expected node version that must match.
|
|
|
+ :returns: The stat of the node.
|
|
|
+ :raises:
|
|
|
+ :exc:`~kazoo.exceptions.BadVersionError` if version doesn't
|
|
|
+ match.
|
|
|
+
|
|
|
+ :exc:`~kazoo.exceptions.NoNodeError` if the node doesn't
|
|
|
+ exist.
|
|
|
+
|
|
|
+ :exc:`~kazoo.exceptions.InvalidACLError` if the ACL is
|
|
|
+ invalid.
|
|
|
+
|
|
|
+ :exc:`~kazoo.exceptions.ZookeeperError` if the server
|
|
|
+ returns a non-zero error code.
|
|
|
+
|
|
|
+ .. versionadded:: 0.5
|
|
|
+
|
|
|
+ """
|
|
|
+ return self.set_acls_async(path, acls, version).get()
|
|
|
+
|
|
|
+ def set_acls_async(self, path, acls, version=-1):
|
|
|
+ """Set the ACL for the node of the given path. Takes the same
|
|
|
+ arguments as :meth:`set_acls`.
|
|
|
+
|
|
|
+ :rtype: :class:`~kazoo.interfaces.IAsyncResult`
|
|
|
+
|
|
|
+ """
|
|
|
+ if not isinstance(path, basestring):
|
|
|
+ raise TypeError("path must be a string")
|
|
|
+ if isinstance(acls, ACL) or not isinstance(acls, (tuple, list)):
|
|
|
+ raise TypeError("acl must be a tuple/list of ACL's")
|
|
|
+ if not isinstance(version, int):
|
|
|
+ raise TypeError("version must be an int")
|
|
|
+
|
|
|
+ async_result = self.handler.async_result()
|
|
|
+ self._call(SetACL(_prefix_root(self.chroot, path), acls, version),
|
|
|
+ async_result)
|
|
|
+ return async_result
|
|
|
+
|
|
|
+ def set(self, path, value, version=-1):
|
|
|
+ """Set the value of a node.
|
|
|
+
|
|
|
+ If the version of the node being updated is newer than the
|
|
|
+ supplied version (and the supplied version is not -1), a
|
|
|
+ BadVersionError will be raised.
|
|
|
+
|
|
|
+ This operation, if successful, will trigger all the watches on
|
|
|
+ the node of the given path left by :meth:`get` API calls.
|
|
|
+
|
|
|
+ The maximum allowable size of the value is 1 MB. Values larger
|
|
|
+ than this will cause a ZookeeperError to be raised.
|
|
|
+
|
|
|
+ :param path: Path of node.
|
|
|
+ :param value: New data value.
|
|
|
+ :param version: Version of node being updated, or -1.
|
|
|
+ :returns: Updated :class:`~kazoo.protocol.states.ZnodeStat` of
|
|
|
+ the node.
|
|
|
+
|
|
|
+ :raises:
|
|
|
+ :exc:`~kazoo.exceptions.BadVersionError` if version doesn't
|
|
|
+ match.
|
|
|
+
|
|
|
+ :exc:`~kazoo.exceptions.NoNodeError` if the node doesn't
|
|
|
+ exist.
|
|
|
+
|
|
|
+ :exc:`~kazoo.exceptions.ZookeeperError` if the provided
|
|
|
+ value is too large.
|
|
|
+
|
|
|
+ :exc:`~kazoo.exceptions.ZookeeperError` if the server
|
|
|
+ returns a non-zero error code.
|
|
|
+
|
|
|
+ """
|
|
|
+ return self.set_async(path, value, version).get()
|
|
|
+
|
|
|
+ def set_async(self, path, value, version=-1):
|
|
|
+ """Set the value of a node. Takes the same arguments as
|
|
|
+ :meth:`set`.
|
|
|
+
|
|
|
+ :rtype: :class:`~kazoo.interfaces.IAsyncResult`
|
|
|
+
|
|
|
+ """
|
|
|
+ if not isinstance(path, basestring):
|
|
|
+ raise TypeError("path must be a string")
|
|
|
+ if value is not None and not isinstance(value, bytes):
|
|
|
+ raise TypeError("value must be a byte string")
|
|
|
+ if not isinstance(version, int):
|
|
|
+ raise TypeError("version must be an int")
|
|
|
+
|
|
|
+ async_result = self.handler.async_result()
|
|
|
+ self._call(SetData(_prefix_root(self.chroot, path), value, version),
|
|
|
+ async_result)
|
|
|
+ return async_result
|
|
|
+
|
|
|
+ def transaction(self):
|
|
|
+ """Create and return a :class:`TransactionRequest` object
|
|
|
+
|
|
|
+ Creates a :class:`TransactionRequest` object. A Transaction can
|
|
|
+ consist of multiple operations which can be committed as a
|
|
|
+ single atomic unit. Either all of the operations will succeed
|
|
|
+ or none of them.
|
|
|
+
|
|
|
+ :returns: A TransactionRequest.
|
|
|
+ :rtype: :class:`TransactionRequest`
|
|
|
+
|
|
|
+ .. versionadded:: 0.6
|
|
|
+ Requires Zookeeper 3.4+
|
|
|
+
|
|
|
+ """
|
|
|
+ return TransactionRequest(self)
|
|
|
+
|
|
|
+ def delete(self, path, version=-1, recursive=False):
|
|
|
+ """Delete a node.
|
|
|
+
|
|
|
+ The call will succeed if such a node exists, and the given
|
|
|
+ version matches the node's version (if the given version is -1,
|
|
|
+ the default, it matches any node's versions).
|
|
|
+
|
|
|
+ This operation, if successful, will trigger all the watches on
|
|
|
+ the node of the given path left by `exists` API calls, and the
|
|
|
+ watches on the parent node left by `get_children` API calls.
|
|
|
+
|
|
|
+ :param path: Path of node to delete.
|
|
|
+ :param version: Version of node to delete, or -1 for any.
|
|
|
+ :param recursive: Recursively delete node and all its children,
|
|
|
+ defaults to False.
|
|
|
+ :type recursive: bool
|
|
|
+
|
|
|
+ :raises:
|
|
|
+ :exc:`~kazoo.exceptions.BadVersionError` if version doesn't
|
|
|
+ match.
|
|
|
+
|
|
|
+ :exc:`~kazoo.exceptions.NoNodeError` if the node doesn't
|
|
|
+ exist.
|
|
|
+
|
|
|
+ :exc:`~kazoo.exceptions.NotEmptyError` if the node has
|
|
|
+ children.
|
|
|
+
|
|
|
+ :exc:`~kazoo.exceptions.ZookeeperError` if the server
|
|
|
+ returns a non-zero error code.
|
|
|
+
|
|
|
+ """
|
|
|
+ if not isinstance(recursive, bool):
|
|
|
+ raise TypeError("recursive must be a bool")
|
|
|
+
|
|
|
+ if recursive:
|
|
|
+ return self._delete_recursive(path)
|
|
|
+ else:
|
|
|
+ return self.delete_async(path, version).get()
|
|
|
+
|
|
|
+ def delete_async(self, path, version=-1):
|
|
|
+ """Asynchronously delete a node. Takes the same arguments as
|
|
|
+ :meth:`delete`, with the exception of `recursive`.
|
|
|
+
|
|
|
+ :rtype: :class:`~kazoo.interfaces.IAsyncResult`
|
|
|
+
|
|
|
+ """
|
|
|
+ if not isinstance(path, basestring):
|
|
|
+ raise TypeError("path must be a string")
|
|
|
+ if not isinstance(version, int):
|
|
|
+ raise TypeError("version must be an int")
|
|
|
+ async_result = self.handler.async_result()
|
|
|
+ self._call(Delete(_prefix_root(self.chroot, path), version),
|
|
|
+ async_result)
|
|
|
+ return async_result
|
|
|
+
|
|
|
+ def _delete_recursive(self, path):
|
|
|
+ try:
|
|
|
+ children = self.get_children(path)
|
|
|
+ except NoNodeError:
|
|
|
+ return True
|
|
|
+
|
|
|
+ if children:
|
|
|
+ for child in children:
|
|
|
+ if path == "/":
|
|
|
+ child_path = path + child
|
|
|
+ else:
|
|
|
+ child_path = path + "/" + child
|
|
|
+
|
|
|
+ self._delete_recursive(child_path)
|
|
|
+ try:
|
|
|
+ self.delete(path)
|
|
|
+ except NoNodeError: # pragma: nocover
|
|
|
+ pass
|
|
|
+
|
|
|
+
|
|
|
+class TransactionRequest(object):
|
|
|
+ """A Zookeeper Transaction Request
|
|
|
+
|
|
|
+ A Transaction provides a builder object that can be used to
|
|
|
+ construct and commit an atomic set of operations. The transaction
|
|
|
+ must be committed before its sent.
|
|
|
+
|
|
|
+ Transactions are not thread-safe and should not be accessed from
|
|
|
+ multiple threads at once.
|
|
|
+
|
|
|
+ .. versionadded:: 0.6
|
|
|
+ Requires Zookeeper 3.4+
|
|
|
+
|
|
|
+ """
|
|
|
+ def __init__(self, client):
|
|
|
+ self.client = client
|
|
|
+ self.operations = []
|
|
|
+ self.committed = False
|
|
|
+
|
|
|
+ def create(self, path, value=b"", acl=None, ephemeral=False,
|
|
|
+ sequence=False):
|
|
|
+ """Add a create ZNode to the transaction. Takes the same
|
|
|
+ arguments as :meth:`KazooClient.create`, with the exception
|
|
|
+ of `makepath`.
|
|
|
+
|
|
|
+ :returns: None
|
|
|
+
|
|
|
+ """
|
|
|
+ if acl is None and self.client.default_acl:
|
|
|
+ acl = self.client.default_acl
|
|
|
+
|
|
|
+ if not isinstance(path, basestring):
|
|
|
+ raise TypeError("path must be a string")
|
|
|
+ if acl and not isinstance(acl, (tuple, list)):
|
|
|
+ raise TypeError("acl must be a tuple/list of ACL's")
|
|
|
+ if not isinstance(value, bytes):
|
|
|
+ raise TypeError("value must be a byte string")
|
|
|
+ if not isinstance(ephemeral, bool):
|
|
|
+ raise TypeError("ephemeral must be a bool")
|
|
|
+ if not isinstance(sequence, bool):
|
|
|
+ raise TypeError("sequence must be a bool")
|
|
|
+
|
|
|
+ flags = 0
|
|
|
+ if ephemeral:
|
|
|
+ flags |= 1
|
|
|
+ if sequence:
|
|
|
+ flags |= 2
|
|
|
+ if acl is None:
|
|
|
+ acl = OPEN_ACL_UNSAFE
|
|
|
+
|
|
|
+ self._add(Create(_prefix_root(self.client.chroot, path), value, acl,
|
|
|
+ flags), None)
|
|
|
+
|
|
|
+ def delete(self, path, version=-1):
|
|
|
+ """Add a delete ZNode to the transaction. Takes the same
|
|
|
+ arguments as :meth:`KazooClient.delete`, with the exception of
|
|
|
+ `recursive`.
|
|
|
+
|
|
|
+ """
|
|
|
+ if not isinstance(path, basestring):
|
|
|
+ raise TypeError("path must be a string")
|
|
|
+ if not isinstance(version, int):
|
|
|
+ raise TypeError("version must be an int")
|
|
|
+ self._add(Delete(_prefix_root(self.client.chroot, path), version))
|
|
|
+
|
|
|
+ def set_data(self, path, value, version=-1):
|
|
|
+ """Add a set ZNode value to the transaction. Takes the same
|
|
|
+ arguments as :meth:`KazooClient.set`.
|
|
|
+
|
|
|
+ """
|
|
|
+ if not isinstance(path, basestring):
|
|
|
+ raise TypeError("path must be a string")
|
|
|
+ if not isinstance(value, bytes):
|
|
|
+ raise TypeError("value must be a byte string")
|
|
|
+ if not isinstance(version, int):
|
|
|
+ raise TypeError("version must be an int")
|
|
|
+ self._add(SetData(_prefix_root(self.client.chroot, path), value,
|
|
|
+ version))
|
|
|
+
|
|
|
+ def check(self, path, version):
|
|
|
+ """Add a Check Version to the transaction.
|
|
|
+
|
|
|
+ This command will fail and abort a transaction if the path
|
|
|
+ does not match the specified version.
|
|
|
+
|
|
|
+ """
|
|
|
+ if not isinstance(path, basestring):
|
|
|
+ raise TypeError("path must be a string")
|
|
|
+ if not isinstance(version, int):
|
|
|
+ raise TypeError("version must be an int")
|
|
|
+ self._add(CheckVersion(_prefix_root(self.client.chroot, path),
|
|
|
+ version))
|
|
|
+
|
|
|
+ def commit_async(self):
|
|
|
+ """Commit the transaction asynchronously.
|
|
|
+
|
|
|
+ :rtype: :class:`~kazoo.interfaces.IAsyncResult`
|
|
|
+
|
|
|
+ """
|
|
|
+ self._check_tx_state()
|
|
|
+ self.committed = True
|
|
|
+ async_object = self.client.handler.async_result()
|
|
|
+ self.client._call(Transaction(self.operations), async_object)
|
|
|
+ return async_object
|
|
|
+
|
|
|
+ def commit(self):
|
|
|
+ """Commit the transaction.
|
|
|
+
|
|
|
+ :returns: A list of the results for each operation in the
|
|
|
+ transaction.
|
|
|
+
|
|
|
+ """
|
|
|
+ return self.commit_async().get()
|
|
|
+
|
|
|
+ def __enter__(self):
|
|
|
+ return self
|
|
|
+
|
|
|
+ def __exit__(self, exc_type, exc_value, exc_tb):
|
|
|
+ """Commit and cleanup accumulated transaction data."""
|
|
|
+ if not exc_type:
|
|
|
+ self.commit()
|
|
|
+
|
|
|
+ def _check_tx_state(self):
|
|
|
+ if self.committed:
|
|
|
+ raise ValueError('Transaction already committed')
|
|
|
+
|
|
|
+ def _add(self, request, post_processor=None):
|
|
|
+ self._check_tx_state()
|
|
|
+ self.client.logger.log(BLATHER, 'Added %r to %r', request, self)
|
|
|
+ self.operations.append(request)
|