You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@slider.apache.org by sm...@apache.org on 2014/08/12 22:47:30 UTC
[11/20] SLIDER-285. Slider Agents to bind and work with restarted AM
(Gour Saha via smohanty)
http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/bbdb6bab/slider-agent/src/main/python/kazoo/recipe/lock.py
----------------------------------------------------------------------
diff --git a/slider-agent/src/main/python/kazoo/recipe/lock.py b/slider-agent/src/main/python/kazoo/recipe/lock.py
new file mode 100644
index 0000000..8c944b7
--- /dev/null
+++ b/slider-agent/src/main/python/kazoo/recipe/lock.py
@@ -0,0 +1,520 @@
+"""Zookeeper Locking Implementations
+
+:Maintainer: None
+:Status: Unknown
+
+Error Handling
+==============
+
+It's highly recommended to add a state listener with
+:meth:`~KazooClient.add_listener` and watch for
+:attr:`~KazooState.LOST` and :attr:`~KazooState.SUSPENDED` state
+changes and re-act appropriately. In the event that a
+:attr:`~KazooState.LOST` state occurs, its certain that the lock
+and/or the lease has been lost.
+
+"""
+import uuid
+
+from kazoo.retry import (
+ KazooRetry,
+ RetryFailedError,
+ ForceRetryError
+)
+from kazoo.exceptions import CancelledError
+from kazoo.exceptions import KazooException
+from kazoo.exceptions import LockTimeout
+from kazoo.exceptions import NoNodeError
+from kazoo.protocol.states import KazooState
+
+
+class Lock(object):
+ """Kazoo Lock
+
+ Example usage with a :class:`~kazoo.client.KazooClient` instance:
+
+ .. code-block:: python
+
+ zk = KazooClient()
+ lock = zk.Lock("/lockpath", "my-identifier")
+ with lock: # blocks waiting for lock acquisition
+ # do something with the lock
+
+ Note: This lock is re-entrant. Repeat calls after acquired will
+ continue to return ''True''.
+
+ """
+ _NODE_NAME = '__lock__'
+
+ def __init__(self, client, path, identifier=None):
+ """Create a Kazoo lock.
+
+ :param client: A :class:`~kazoo.client.KazooClient` instance.
+ :param path: The lock path to use.
+ :param identifier: Name to use for this lock contender. This
+ can be useful for querying to see who the
+ current lock contenders are.
+
+ """
+ self.client = client
+ self.path = path
+
+ # some data is written to the node. this can be queried via
+ # contenders() to see who is contending for the lock
+ self.data = str(identifier or "").encode('utf-8')
+
+ self.wake_event = client.handler.event_object()
+
+ # props to Netflix Curator for this trick. It is possible for our
+ # create request to succeed on the server, but for a failure to
+ # prevent us from getting back the full path name. We prefix our
+ # lock name with a uuid and can check for its presence on retry.
+ self.prefix = uuid.uuid4().hex + self._NODE_NAME
+ self.create_path = self.path + "/" + self.prefix
+
+ self.create_tried = False
+ self.is_acquired = False
+ self.assured_path = False
+ self.cancelled = False
+ self._retry = KazooRetry(max_tries=None,
+ sleep_func=client.handler.sleep_func)
+
+ def _ensure_path(self):
+ self.client.ensure_path(self.path)
+ self.assured_path = True
+
+ def cancel(self):
+ """Cancel a pending lock acquire."""
+ self.cancelled = True
+ self.wake_event.set()
+
+ def acquire(self, blocking=True, timeout=None):
+ """
+ Acquire the lock. By defaults blocks and waits forever.
+
+ :param blocking: Block until lock is obtained or return immediately.
+ :type blocking: bool
+ :param timeout: Don't wait forever to acquire the lock.
+ :type timeout: float or None
+
+ :returns: Was the lock acquired?
+ :rtype: bool
+
+ :raises: :exc:`~kazoo.exceptions.LockTimeout` if the lock
+ wasn't acquired within `timeout` seconds.
+
+ .. versionadded:: 1.1
+ The timeout option.
+ """
+ try:
+ retry = self._retry.copy()
+ retry.deadline = timeout
+ self.is_acquired = retry(self._inner_acquire,
+ blocking=blocking, timeout=timeout)
+ except RetryFailedError:
+ self._best_effort_cleanup()
+ except KazooException:
+ # if we did ultimately fail, attempt to clean up
+ self._best_effort_cleanup()
+ self.cancelled = False
+ raise
+
+ if not self.is_acquired:
+ self._delete_node(self.node)
+
+ return self.is_acquired
+
+ def _watch_session(self, state):
+ self.wake_event.set()
+ return True
+
+ def _inner_acquire(self, blocking, timeout):
+ # make sure our election parent node exists
+ if not self.assured_path:
+ self._ensure_path()
+
+ node = None
+ if self.create_tried:
+ node = self._find_node()
+ else:
+ self.create_tried = True
+
+ if not node:
+ node = self.client.create(self.create_path, self.data,
+ ephemeral=True, sequence=True)
+ # strip off path to node
+ node = node[len(self.path) + 1:]
+
+ self.node = node
+
+ while True:
+ self.wake_event.clear()
+
+ # bail out with an exception if cancellation has been requested
+ if self.cancelled:
+ raise CancelledError()
+
+ children = self._get_sorted_children()
+
+ try:
+ our_index = children.index(node)
+ except ValueError: # pragma: nocover
+ # somehow we aren't in the children -- probably we are
+ # recovering from a session failure and our ephemeral
+ # node was removed
+ raise ForceRetryError()
+
+ if self.acquired_lock(children, our_index):
+ return True
+
+ if not blocking:
+ return False
+
+ # otherwise we are in the mix. watch predecessor and bide our time
+ predecessor = self.path + "/" + children[our_index - 1]
+ self.client.add_listener(self._watch_session)
+ try:
+ if self.client.exists(predecessor, self._watch_predecessor):
+ self.wake_event.wait(timeout)
+ if not self.wake_event.isSet():
+ raise LockTimeout("Failed to acquire lock on %s after %s "
+ "seconds" % (self.path, timeout))
+ finally:
+ self.client.remove_listener(self._watch_session)
+
+ def acquired_lock(self, children, index):
+ return index == 0
+
+ def _watch_predecessor(self, event):
+ self.wake_event.set()
+
+ def _get_sorted_children(self):
+ children = self.client.get_children(self.path)
+
+ # can't just sort directly: the node names are prefixed by uuids
+ lockname = self._NODE_NAME
+ children.sort(key=lambda c: c[c.find(lockname) + len(lockname):])
+ return children
+
+ def _find_node(self):
+ children = self.client.get_children(self.path)
+ for child in children:
+ if child.startswith(self.prefix):
+ return child
+ return None
+
+ def _delete_node(self, node):
+ self.client.delete(self.path + "/" + node)
+
+ def _best_effort_cleanup(self):
+ try:
+ node = self._find_node()
+ if node:
+ self._delete_node(node)
+ except KazooException: # pragma: nocover
+ pass
+
+ def release(self):
+ """Release the lock immediately."""
+ return self.client.retry(self._inner_release)
+
+ def _inner_release(self):
+ if not self.is_acquired:
+ return False
+
+ try:
+ self._delete_node(self.node)
+ except NoNodeError: # pragma: nocover
+ pass
+
+ self.is_acquired = False
+ self.node = None
+
+ return True
+
+ def contenders(self):
+ """Return an ordered list of the current contenders for the
+ lock.
+
+ .. note::
+
+ If the contenders did not set an identifier, it will appear
+ as a blank string.
+
+ """
+ # make sure our election parent node exists
+ if not self.assured_path:
+ self._ensure_path()
+
+ children = self._get_sorted_children()
+
+ contenders = []
+ for child in children:
+ try:
+ data, stat = self.client.get(self.path + "/" + child)
+ contenders.append(data.decode('utf-8'))
+ except NoNodeError: # pragma: nocover
+ pass
+ return contenders
+
+ def __enter__(self):
+ self.acquire()
+
+ def __exit__(self, exc_type, exc_value, traceback):
+ self.release()
+
+
+class Semaphore(object):
+ """A Zookeeper-based Semaphore
+
+ This synchronization primitive operates in the same manner as the
+ Python threading version only uses the concept of leases to
+ indicate how many available leases are available for the lock
+ rather than counting.
+
+ Example:
+
+ .. code-block:: python
+
+ zk = KazooClient()
+ semaphore = zk.Semaphore("/leasepath", "my-identifier")
+ with semaphore: # blocks waiting for lock acquisition
+ # do something with the semaphore
+
+ .. warning::
+
+ This class stores the allowed max_leases as the data on the
+ top-level semaphore node. The stored value is checked once
+ against the max_leases of each instance. This check is
+ performed when acquire is called the first time. The semaphore
+ node needs to be deleted to change the allowed leases.
+
+ .. versionadded:: 0.6
+ The Semaphore class.
+
+ .. versionadded:: 1.1
+ The max_leases check.
+
+ """
+ def __init__(self, client, path, identifier=None, max_leases=1):
+ """Create a Kazoo Lock
+
+ :param client: A :class:`~kazoo.client.KazooClient` instance.
+ :param path: The semaphore path to use.
+ :param identifier: Name to use for this lock contender. This
+ can be useful for querying to see who the
+ current lock contenders are.
+ :param max_leases: The maximum amount of leases available for
+ the semaphore.
+
+ """
+ # Implementation notes about how excessive thundering herd
+ # and watches are avoided
+ # - A node (lease pool) holds children for each lease in use
+ # - A lock is acquired for a process attempting to acquire a
+ # lease. If a lease is available, the ephemeral node is
+ # created in the lease pool and the lock is released.
+ # - Only the lock holder watches for children changes in the
+ # lease pool
+ self.client = client
+ self.path = path
+
+ # some data is written to the node. this can be queried via
+ # contenders() to see who is contending for the lock
+ self.data = str(identifier or "").encode('utf-8')
+ self.max_leases = max_leases
+ self.wake_event = client.handler.event_object()
+
+ self.create_path = self.path + "/" + uuid.uuid4().hex
+ self.lock_path = path + '-' + '__lock__'
+ self.is_acquired = False
+ self.assured_path = False
+ self.cancelled = False
+ self._session_expired = False
+
+ def _ensure_path(self):
+ result = self.client.ensure_path(self.path)
+ self.assured_path = True
+ if result is True:
+ # node did already exist
+ data, _ = self.client.get(self.path)
+ try:
+ leases = int(data.decode('utf-8'))
+ except (ValueError, TypeError):
+ # ignore non-numeric data, maybe the node data is used
+ # for other purposes
+ pass
+ else:
+ if leases != self.max_leases:
+ raise ValueError(
+ "Inconsistent max leases: %s, expected: %s" %
+ (leases, self.max_leases)
+ )
+ else:
+ self.client.set(self.path, str(self.max_leases).encode('utf-8'))
+
+ def cancel(self):
+ """Cancel a pending semaphore acquire."""
+ self.cancelled = True
+ self.wake_event.set()
+
+ def acquire(self, blocking=True, timeout=None):
+ """Acquire the semaphore. By defaults blocks and waits forever.
+
+ :param blocking: Block until semaphore is obtained or
+ return immediately.
+ :type blocking: bool
+ :param timeout: Don't wait forever to acquire the semaphore.
+ :type timeout: float or None
+
+ :returns: Was the semaphore acquired?
+ :rtype: bool
+
+ :raises:
+ ValueError if the max_leases value doesn't match the
+ stored value.
+
+ :exc:`~kazoo.exceptions.LockTimeout` if the semaphore
+ wasn't acquired within `timeout` seconds.
+
+ .. versionadded:: 1.1
+ The blocking, timeout arguments and the max_leases check.
+ """
+ # If the semaphore had previously been canceled, make sure to
+ # reset that state.
+ self.cancelled = False
+
+ try:
+ self.is_acquired = self.client.retry(
+ self._inner_acquire, blocking=blocking, timeout=timeout)
+ except KazooException:
+ # if we did ultimately fail, attempt to clean up
+ self._best_effort_cleanup()
+ self.cancelled = False
+ raise
+
+ return self.is_acquired
+
+ def _inner_acquire(self, blocking, timeout=None):
+ """Inner loop that runs from the top anytime a command hits a
+ retryable Zookeeper exception."""
+ self._session_expired = False
+ self.client.add_listener(self._watch_session)
+
+ if not self.assured_path:
+ self._ensure_path()
+
+ # Do we already have a lease?
+ if self.client.exists(self.create_path):
+ return True
+
+ with self.client.Lock(self.lock_path, self.data):
+ while True:
+ self.wake_event.clear()
+
+ # Attempt to grab our lease...
+ if self._get_lease():
+ return True
+
+ if blocking:
+ # If blocking, wait until self._watch_lease_change() is
+ # called before returning
+ self.wake_event.wait(timeout)
+ if not self.wake_event.isSet():
+ raise LockTimeout(
+ "Failed to acquire semaphore on %s "
+ "after %s seconds" % (self.path, timeout))
+ else:
+ # If not blocking, register another watch that will trigger
+ # self._get_lease() as soon as the children change again.
+ self.client.get_children(self.path, self._get_lease)
+ return False
+
+ def _watch_lease_change(self, event):
+ self.wake_event.set()
+
+ def _get_lease(self, data=None):
+ # Make sure the session is still valid
+ if self._session_expired:
+ raise ForceRetryError("Retry on session loss at top")
+
+ # Make sure that the request hasn't been canceled
+ if self.cancelled:
+ raise CancelledError("Semaphore cancelled")
+
+ # Get a list of the current potential lock holders. If they change,
+ # notify our wake_event object. This is used to unblock a blocking
+ # self._inner_acquire call.
+ children = self.client.get_children(self.path,
+ self._watch_lease_change)
+
+ # If there are leases available, acquire one
+ if len(children) < self.max_leases:
+ self.client.create(self.create_path, self.data, ephemeral=True)
+
+ # Check if our acquisition was successful or not. Update our state.
+ if self.client.exists(self.create_path):
+ self.is_acquired = True
+ else:
+ self.is_acquired = False
+
+ # Return current state
+ return self.is_acquired
+
+ def _watch_session(self, state):
+ if state == KazooState.LOST:
+ self._session_expired = True
+ self.wake_event.set()
+
+ # Return true to de-register
+ return True
+
+ def _best_effort_cleanup(self):
+ try:
+ self.client.delete(self.create_path)
+ except KazooException: # pragma: nocover
+ pass
+
+ def release(self):
+ """Release the lease immediately."""
+ return self.client.retry(self._inner_release)
+
+ def _inner_release(self):
+ if not self.is_acquired:
+ return False
+
+ try:
+ self.client.delete(self.create_path)
+ except NoNodeError: # pragma: nocover
+ pass
+ self.is_acquired = False
+ return True
+
+ def lease_holders(self):
+ """Return an unordered list of the current lease holders.
+
+ .. note::
+
+ If the lease holder did not set an identifier, it will
+ appear as a blank string.
+
+ """
+ if not self.client.exists(self.path):
+ return []
+
+ children = self.client.get_children(self.path)
+
+ lease_holders = []
+ for child in children:
+ try:
+ data, stat = self.client.get(self.path + "/" + child)
+ lease_holders.append(data.decode('utf-8'))
+ except NoNodeError: # pragma: nocover
+ pass
+ return lease_holders
+
+ def __enter__(self):
+ self.acquire()
+
+ def __exit__(self, exc_type, exc_value, traceback):
+ self.release()
http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/bbdb6bab/slider-agent/src/main/python/kazoo/recipe/partitioner.py
----------------------------------------------------------------------
diff --git a/slider-agent/src/main/python/kazoo/recipe/partitioner.py b/slider-agent/src/main/python/kazoo/recipe/partitioner.py
new file mode 100644
index 0000000..8de4231
--- /dev/null
+++ b/slider-agent/src/main/python/kazoo/recipe/partitioner.py
@@ -0,0 +1,377 @@
+"""Zookeeper Partitioner Implementation
+
+:Maintainer: None
+:Status: Unknown
+
+:class:`SetPartitioner` implements a partitioning scheme using
+Zookeeper for dividing up resources amongst members of a party.
+
+This is useful when there is a set of resources that should only be
+accessed by a single process at a time that multiple processes
+across a cluster might want to divide up.
+
+Example Use-Case
+----------------
+
+- Multiple workers across a cluster need to divide up a list of queues
+ so that no two workers own the same queue.
+
+"""
+import logging
+import os
+import socket
+from functools import partial
+
+from kazoo.exceptions import KazooException
+from kazoo.protocol.states import KazooState
+from kazoo.recipe.watchers import PatientChildrenWatch
+
+log = logging.getLogger(__name__)
+
+
+class PartitionState(object):
+ """High level partition state values
+
+ .. attribute:: ALLOCATING
+
+ The set needs to be partitioned, and may require an existing
+ partition set to be released before acquiring a new partition
+ of the set.
+
+ .. attribute:: ACQUIRED
+
+ The set has been partitioned and acquired.
+
+ .. attribute:: RELEASE
+
+ The set needs to be repartitioned, and the current partitions
+ must be released before a new allocation can be made.
+
+ .. attribute:: FAILURE
+
+ The set partition has failed. This occurs when the maximum
+ time to partition the set is exceeded or the Zookeeper session
+ is lost. The partitioner is unusable after this state and must
+ be recreated.
+
+ """
+ ALLOCATING = "ALLOCATING"
+ ACQUIRED = "ACQUIRED"
+ RELEASE = "RELEASE"
+ FAILURE = "FAILURE"
+
+
+class SetPartitioner(object):
+ """Partitions a set amongst members of a party
+
+ This class will partition a set amongst members of a party such
+ that each member will be given zero or more items of the set and
+ each set item will be given to a single member. When new members
+ enter or leave the party, the set will be re-partitioned amongst
+ the members.
+
+ When the :class:`SetPartitioner` enters the
+ :attr:`~PartitionState.FAILURE` state, it is unrecoverable
+ and a new :class:`SetPartitioner` should be created.
+
+ Example:
+
+ .. code-block:: python
+
+ from kazoo.client import KazooClient
+ client = KazooClient()
+
+ qp = client.SetPartitioner(
+ path='/work_queues', set=('queue-1', 'queue-2', 'queue-3'))
+
+ while 1:
+ if qp.failed:
+ raise Exception("Lost or unable to acquire partition")
+ elif qp.release:
+ qp.release_set()
+ elif qp.acquired:
+ for partition in qp:
+ # Do something with each partition
+ elif qp.allocating:
+ qp.wait_for_acquire()
+
+ **State Transitions**
+
+ When created, the :class:`SetPartitioner` enters the
+ :attr:`PartitionState.ALLOCATING` state.
+
+ :attr:`~PartitionState.ALLOCATING` ->
+ :attr:`~PartitionState.ACQUIRED`
+
+ Set was partitioned successfully, the partition list assigned
+ is accessible via list/iter methods or calling list() on the
+ :class:`SetPartitioner` instance.
+
+ :attr:`~PartitionState.ALLOCATING` ->
+ :attr:`~PartitionState.FAILURE`
+
+ Allocating the set failed either due to a Zookeeper session
+ expiration, or failure to acquire the items of the set within
+ the timeout period.
+
+ :attr:`~PartitionState.ACQUIRED` ->
+ :attr:`~PartitionState.RELEASE`
+
+ The members of the party have changed, and the set needs to be
+ repartitioned. :meth:`SetPartitioner.release` should be called
+ as soon as possible.
+
+ :attr:`~PartitionState.ACQUIRED` ->
+ :attr:`~PartitionState.FAILURE`
+
+ The current partition was lost due to a Zookeeper session
+ expiration.
+
+ :attr:`~PartitionState.RELEASE` ->
+ :attr:`~PartitionState.ALLOCATING`
+
+ The current partition was released and is being re-allocated.
+
+ """
+ def __init__(self, client, path, set, partition_func=None,
+ identifier=None, time_boundary=30):
+ """Create a :class:`~SetPartitioner` instance
+
+ :param client: A :class:`~kazoo.client.KazooClient` instance.
+ :param path: The partition path to use.
+ :param set: The set of items to partition.
+ :param partition_func: A function to use to decide how to
+ partition the set.
+ :param identifier: An identifier to use for this member of the
+ party when participating. Defaults to the
+ hostname + process id.
+ :param time_boundary: How long the party members must be stable
+ before allocation can complete.
+
+ """
+ self.state = PartitionState.ALLOCATING
+
+ self._client = client
+ self._path = path
+ self._set = set
+ self._partition_set = []
+ self._partition_func = partition_func or self._partitioner
+ self._identifier = identifier or '%s-%s' % (
+ socket.getfqdn(), os.getpid())
+ self._locks = []
+ self._lock_path = '/'.join([path, 'locks'])
+ self._party_path = '/'.join([path, 'party'])
+ self._time_boundary = time_boundary
+
+ self._acquire_event = client.handler.event_object()
+
+ # Create basic path nodes
+ client.ensure_path(path)
+ client.ensure_path(self._lock_path)
+ client.ensure_path(self._party_path)
+
+ # Join the party
+ self._party = client.ShallowParty(self._party_path,
+ identifier=self._identifier)
+ self._party.join()
+
+ self._was_allocated = False
+ self._state_change = client.handler.rlock_object()
+ client.add_listener(self._establish_sessionwatch)
+
+ # Now watch the party and set the callback on the async result
+ # so we know when we're ready
+ self._children_updated = False
+ self._child_watching(self._allocate_transition, async=True)
+
+ def __iter__(self):
+ """Return the partitions in this partition set"""
+ for partition in self._partition_set:
+ yield partition
+
+ @property
+ def failed(self):
+ """Corresponds to the :attr:`PartitionState.FAILURE` state"""
+ return self.state == PartitionState.FAILURE
+
+ @property
+ def release(self):
+ """Corresponds to the :attr:`PartitionState.RELEASE` state"""
+ return self.state == PartitionState.RELEASE
+
+ @property
+ def allocating(self):
+ """Corresponds to the :attr:`PartitionState.ALLOCATING`
+ state"""
+ return self.state == PartitionState.ALLOCATING
+
+ @property
+ def acquired(self):
+ """Corresponds to the :attr:`PartitionState.ACQUIRED` state"""
+ return self.state == PartitionState.ACQUIRED
+
+ def wait_for_acquire(self, timeout=30):
+ """Wait for the set to be partitioned and acquired
+
+ :param timeout: How long to wait before returning.
+ :type timeout: int
+
+ """
+ self._acquire_event.wait(timeout)
+
+ def release_set(self):
+ """Call to release the set
+
+ This method begins the step of allocating once the set has
+ been released.
+
+ """
+ self._release_locks()
+ if self._locks: # pragma: nocover
+ # This shouldn't happen, it means we couldn't release our
+ # locks, abort
+ self._fail_out()
+ return
+ else:
+ with self._state_change:
+ if self.failed:
+ return
+ self.state = PartitionState.ALLOCATING
+ self._child_watching(self._allocate_transition, async=True)
+
+ def finish(self):
+ """Call to release the set and leave the party"""
+ self._release_locks()
+ self._fail_out()
+
+ def _fail_out(self):
+ with self._state_change:
+ self.state = PartitionState.FAILURE
+ if self._party.participating:
+ try:
+ self._party.leave()
+ except KazooException: # pragma: nocover
+ pass
+
+ def _allocate_transition(self, result):
+ """Called when in allocating mode, and the children settled"""
+ # Did we get an exception waiting for children to settle?
+ if result.exception: # pragma: nocover
+ self._fail_out()
+ return
+
+ children, async_result = result.get()
+ self._children_updated = False
+
+ # Add a callback when children change on the async_result
+ def updated(result):
+ with self._state_change:
+ if self.acquired:
+ self.state = PartitionState.RELEASE
+ self._children_updated = True
+
+ async_result.rawlink(updated)
+
+ # Split up the set
+ self._partition_set = self._partition_func(
+ self._identifier, list(self._party), self._set)
+
+ # Proceed to acquire locks for the working set as needed
+ for member in self._partition_set:
+ if self._children_updated or self.failed:
+ # Still haven't settled down, release locks acquired
+ # so far and go back
+ return self._abort_lock_acquisition()
+
+ lock = self._client.Lock(self._lock_path + '/' +
+ str(member))
+ try:
+ lock.acquire()
+ except KazooException: # pragma: nocover
+ return self.finish()
+ self._locks.append(lock)
+
+ # All locks acquired! Time for state transition, make sure
+ # we didn't inadvertently get lost thus far
+ with self._state_change:
+ if self.failed: # pragma: nocover
+ return self.finish()
+ self.state = PartitionState.ACQUIRED
+ self._acquire_event.set()
+
+ def _release_locks(self):
+ """Attempt to completely remove all the locks"""
+ self._acquire_event.clear()
+ for lock in self._locks[:]:
+ try:
+ lock.release()
+ except KazooException: # pragma: nocover
+ # We proceed to remove as many as possible, and leave
+ # the ones we couldn't remove
+ pass
+ else:
+ self._locks.remove(lock)
+
+ def _abort_lock_acquisition(self):
+ """Called during lock acquisition if a party change occurs"""
+ self._partition_set = []
+ self._release_locks()
+ if self._locks:
+ # This shouldn't happen, it means we couldn't release our
+ # locks, abort
+ self._fail_out()
+ return
+ return self._child_watching(self._allocate_transition)
+
+ def _child_watching(self, func=None, async=False):
+ """Called when children are being watched to stabilize
+
+ This actually returns immediately, child watcher spins up a
+ new thread/greenlet and waits for it to stabilize before
+ any callbacks might run.
+
+ """
+ watcher = PatientChildrenWatch(self._client, self._party_path,
+ self._time_boundary)
+ asy = watcher.start()
+ if func is not None:
+ # We spin up the function in a separate thread/greenlet
+ # to ensure that the rawlink's it might use won't be
+ # blocked
+ if async:
+ func = partial(self._client.handler.spawn, func)
+ asy.rawlink(func)
+ return asy
+
+ def _establish_sessionwatch(self, state):
+ """Register ourself to listen for session events, we shut down
+ if we become lost"""
+ with self._state_change:
+ # Handle network partition: If connection gets suspended,
+ # change state to ALLOCATING if we had already ACQUIRED. This way
+ # the caller does not process the members since we could eventually
+ # lose session get repartitioned. If we got connected after a suspension
+ # it means we've not lost the session and still have our members. Hence,
+ # restore to ACQUIRED
+ if state == KazooState.SUSPENDED:
+ if self.state == PartitionState.ACQUIRED:
+ self._was_allocated = True
+ self.state = PartitionState.ALLOCATING
+ elif state == KazooState.CONNECTED:
+ if self._was_allocated:
+ self._was_allocated = False
+ self.state = PartitionState.ACQUIRED
+
+ if state == KazooState.LOST:
+ self._client.handler.spawn(self._fail_out)
+ return True
+
+ def _partitioner(self, identifier, members, partitions):
+ # Ensure consistent order of partitions/members
+ all_partitions = sorted(partitions)
+ workers = sorted(members)
+
+ i = workers.index(identifier)
+ # Now return the partition list starting at our location and
+ # skipping the other workers
+ return all_partitions[i::len(workers)]
http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/bbdb6bab/slider-agent/src/main/python/kazoo/recipe/party.py
----------------------------------------------------------------------
diff --git a/slider-agent/src/main/python/kazoo/recipe/party.py b/slider-agent/src/main/python/kazoo/recipe/party.py
new file mode 100644
index 0000000..7186c10
--- /dev/null
+++ b/slider-agent/src/main/python/kazoo/recipe/party.py
@@ -0,0 +1,118 @@
+"""Party
+
+:Maintainer: Ben Bangert <be...@groovie.org>
+:Status: Production
+
+A Zookeeper pool of party members. The :class:`Party` object can be
+used for determining members of a party.
+
+"""
+import uuid
+
+from kazoo.exceptions import NodeExistsError, NoNodeError
+
+
+class BaseParty(object):
+ """Base implementation of a party."""
+ def __init__(self, client, path, identifier=None):
+ """
+ :param client: A :class:`~kazoo.client.KazooClient` instance.
+ :param path: The party path to use.
+ :param identifier: An identifier to use for this member of the
+ party when participating.
+
+ """
+ self.client = client
+ self.path = path
+ self.data = str(identifier or "").encode('utf-8')
+ self.ensured_path = False
+ self.participating = False
+
+ def _ensure_parent(self):
+ if not self.ensured_path:
+ # make sure our parent node exists
+ self.client.ensure_path(self.path)
+ self.ensured_path = True
+
+ def join(self):
+ """Join the party"""
+ return self.client.retry(self._inner_join)
+
+ def _inner_join(self):
+ self._ensure_parent()
+ try:
+ self.client.create(self.create_path, self.data, ephemeral=True)
+ self.participating = True
+ except NodeExistsError:
+ # node was already created, perhaps we are recovering from a
+ # suspended connection
+ self.participating = True
+
+ def leave(self):
+ """Leave the party"""
+ self.participating = False
+ return self.client.retry(self._inner_leave)
+
+ def _inner_leave(self):
+ try:
+ self.client.delete(self.create_path)
+ except NoNodeError:
+ return False
+ return True
+
+ def __len__(self):
+ """Return a count of participating clients"""
+ self._ensure_parent()
+ return len(self._get_children())
+
+ def _get_children(self):
+ return self.client.retry(self.client.get_children, self.path)
+
+
+class Party(BaseParty):
+ """Simple pool of participating processes"""
+ _NODE_NAME = "__party__"
+
+ def __init__(self, client, path, identifier=None):
+ BaseParty.__init__(self, client, path, identifier=identifier)
+ self.node = uuid.uuid4().hex + self._NODE_NAME
+ self.create_path = self.path + "/" + self.node
+
+ def __iter__(self):
+ """Get a list of participating clients' data values"""
+ self._ensure_parent()
+ children = self._get_children()
+ for child in children:
+ try:
+ d, _ = self.client.retry(self.client.get, self.path +
+ "/" + child)
+ yield d.decode('utf-8')
+ except NoNodeError: # pragma: nocover
+ pass
+
+ def _get_children(self):
+ children = BaseParty._get_children(self)
+ return [c for c in children if self._NODE_NAME in c]
+
+
+class ShallowParty(BaseParty):
+ """Simple shallow pool of participating processes
+
+ This differs from the :class:`Party` as the identifier is used in
+ the name of the party node itself, rather than the data. This
+ places some restrictions on the length as it must be a valid
+ Zookeeper node (an alphanumeric string), but reduces the overhead
+ of getting a list of participants to a single Zookeeper call.
+
+ """
+ def __init__(self, client, path, identifier=None):
+ BaseParty.__init__(self, client, path, identifier=identifier)
+ self.node = '-'.join([uuid.uuid4().hex, self.data.decode('utf-8')])
+ self.create_path = self.path + "/" + self.node
+
+ def __iter__(self):
+ """Get a list of participating clients' identifiers"""
+ self._ensure_parent()
+ children = self._get_children()
+ for child in children:
+ yield child[child.find('-') + 1:]
http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/bbdb6bab/slider-agent/src/main/python/kazoo/recipe/queue.py
----------------------------------------------------------------------
diff --git a/slider-agent/src/main/python/kazoo/recipe/queue.py b/slider-agent/src/main/python/kazoo/recipe/queue.py
new file mode 100644
index 0000000..e2e30dd
--- /dev/null
+++ b/slider-agent/src/main/python/kazoo/recipe/queue.py
@@ -0,0 +1,321 @@
+"""Zookeeper based queue implementations.
+
+:Maintainer: None
+:Status: Unknown
+
+"""
+
+import uuid
+from kazoo.exceptions import NoNodeError, NodeExistsError
+from kazoo.retry import ForceRetryError
+from kazoo.protocol.states import EventType
+
+
+class BaseQueue(object):
+ """A common base class for queue implementations."""
+
+ def __init__(self, client, path):
+ """
+ :param client: A :class:`~kazoo.client.KazooClient` instance.
+ :param path: The queue path to use in ZooKeeper.
+ """
+ self.client = client
+ self.path = path
+ self._entries_path = path
+ self.structure_paths = (self.path, )
+ self.ensured_path = False
+
+ def _check_put_arguments(self, value, priority=100):
+ if not isinstance(value, bytes):
+ raise TypeError("value must be a byte string")
+ if not isinstance(priority, int):
+ raise TypeError("priority must be an int")
+ elif priority < 0 or priority > 999:
+ raise ValueError("priority must be between 0 and 999")
+
+ def _ensure_paths(self):
+ if not self.ensured_path:
+ # make sure our parent / internal structure nodes exists
+ for path in self.structure_paths:
+ self.client.ensure_path(path)
+ self.ensured_path = True
+
+ def __len__(self):
+ self._ensure_paths()
+ _, stat = self.client.retry(self.client.get, self._entries_path)
+ return stat.children_count
+
+
+class Queue(BaseQueue):
+ """A distributed queue with optional priority support.
+
+ This queue does not offer reliable consumption. An entry is removed
+ from the queue prior to being processed. So if an error occurs, the
+ consumer has to re-queue the item or it will be lost.
+
+ """
+
+ prefix = "entry-"
+
+ def __init__(self, client, path):
+ """
+ :param client: A :class:`~kazoo.client.KazooClient` instance.
+ :param path: The queue path to use in ZooKeeper.
+ """
+ super(Queue, self).__init__(client, path)
+ self._children = []
+
+ def __len__(self):
+ """Return queue size."""
+ return super(Queue, self).__len__()
+
+ def get(self):
+ """
+ Get item data and remove an item from the queue.
+
+ :returns: Item data or None.
+ :rtype: bytes
+ """
+ self._ensure_paths()
+ return self.client.retry(self._inner_get)
+
+ def _inner_get(self):
+ if not self._children:
+ self._children = self.client.retry(self.client.get_children, self.path)
+ self._children = sorted(self._children)
+ if not self._children:
+ return None
+ name = self._children[0]
+ try:
+ data, stat = self.client.get(self.path + "/" + name)
+ except NoNodeError: # pragma: nocover
+ # the first node has vanished in the meantime, try to
+ # get another one
+ raise ForceRetryError()
+ try:
+ self.client.delete(self.path + "/" + name)
+ except NoNodeError: # pragma: nocover
+ # we were able to get the data but someone else has removed
+ # the node in the meantime. consider the item as processed
+ # by the other process
+ raise ForceRetryError()
+ self._children.pop(0)
+ return data
+
+ def put(self, value, priority=100):
+ """Put an item into the queue.
+
+ :param value: Byte string to put into the queue.
+ :param priority:
+ An optional priority as an integer with at most 3 digits.
+ Lower values signify higher priority.
+ """
+ self._check_put_arguments(value, priority)
+ self._ensure_paths()
+ path = '{path}/{prefix}{priority:03d}-'.format(
+ path=self.path, prefix=self.prefix, priority=priority)
+ self.client.create(path, value, sequence=True)
+
+
+class LockingQueue(BaseQueue):
+ """A distributed queue with priority and locking support.
+
+ Upon retrieving an entry from the queue, the entry gets locked with an
+ ephemeral node (instead of deleted). If an error occurs, this lock gets
+ released so that others could retake the entry. This adds a little penalty
+ as compared to :class:`Queue` implementation.
+
+ The user should call the :meth:`LockingQueue.get` method first to lock and
+ retrieve the next entry. When finished processing the entry, a user should
+ call the :meth:`LockingQueue.consume` method that will remove the entry
+ from the queue.
+
+ This queue will not track connection status with ZooKeeper. If a node locks
+ an element, then loses connection with ZooKeeper and later reconnects, the
+ lock will probably be removed by Zookeeper in the meantime, but a node
+ would still think that it holds a lock. The user should check the
+ connection status with Zookeeper or call :meth:`LockingQueue.holds_lock`
+ method that will check if a node still holds the lock.
+
+ .. note::
+ :class:`LockingQueue` requires ZooKeeper 3.4 or above, since it is
+ using transactions.
+ """
+ lock = "/taken"
+ entries = "/entries"
+ entry = "entry"
+
+ def __init__(self, client, path):
+ """
+ :param client: A :class:`~kazoo.client.KazooClient` instance.
+ :param path: The queue path to use in ZooKeeper.
+ """
+ super(LockingQueue, self).__init__(client, path)
+ self.id = uuid.uuid4().hex.encode()
+ self.processing_element = None
+ self._lock_path = self.path + self.lock
+ self._entries_path = self.path + self.entries
+ self.structure_paths = (self._lock_path, self._entries_path)
+
+ def __len__(self):
+ """Returns the current length of the queue.
+
+ :returns: queue size (includes locked entries count).
+ """
+ return super(LockingQueue, self).__len__()
+
+ def put(self, value, priority=100):
+ """Put an entry into the queue.
+
+ :param value: Byte string to put into the queue.
+ :param priority:
+ An optional priority as an integer with at most 3 digits.
+ Lower values signify higher priority.
+
+ """
+ self._check_put_arguments(value, priority)
+ self._ensure_paths()
+
+ self.client.create(
+ "{path}/{prefix}-{priority:03d}-".format(
+ path=self._entries_path,
+ prefix=self.entry,
+ priority=priority),
+ value, sequence=True)
+
+ def put_all(self, values, priority=100):
+ """Put several entries into the queue. The action only succeeds
+ if all entries where put into the queue.
+
+ :param values: A list of values to put into the queue.
+ :param priority:
+ An optional priority as an integer with at most 3 digits.
+ Lower values signify higher priority.
+
+ """
+ if not isinstance(values, list):
+ raise TypeError("values must be a list of byte strings")
+ if not isinstance(priority, int):
+ raise TypeError("priority must be an int")
+ elif priority < 0 or priority > 999:
+ raise ValueError("priority must be between 0 and 999")
+ self._ensure_paths()
+
+ with self.client.transaction() as transaction:
+ for value in values:
+ if not isinstance(value, bytes):
+ raise TypeError("value must be a byte string")
+ transaction.create(
+ "{path}/{prefix}-{priority:03d}-".format(
+ path=self._entries_path,
+ prefix=self.entry,
+ priority=priority),
+ value, sequence=True)
+
+ def get(self, timeout=None):
+ """Locks and gets an entry from the queue. If a previously got entry
+ was not consumed, this method will return that entry.
+
+ :param timeout:
+ Maximum waiting time in seconds. If None then it will wait
+ untill an entry appears in the queue.
+ :returns: A locked entry value or None if the timeout was reached.
+ :rtype: bytes
+ """
+ self._ensure_paths()
+ if not self.processing_element is None:
+ return self.processing_element[1]
+ else:
+ return self._inner_get(timeout)
+
+ def holds_lock(self):
+ """Checks if a node still holds the lock.
+
+ :returns: True if a node still holds the lock, False otherwise.
+ :rtype: bool
+ """
+ if self.processing_element is None:
+ return False
+ lock_id, _ = self.processing_element
+ lock_path = "{path}/{id}".format(path=self._lock_path, id=lock_id)
+ self.client.sync(lock_path)
+ value, stat = self.client.retry(self.client.get, lock_path)
+ return value == self.id
+
+ def consume(self):
+ """Removes a currently processing entry from the queue.
+
+ :returns: True if element was removed successfully, False otherwise.
+ :rtype: bool
+ """
+ if not self.processing_element is None and self.holds_lock:
+ id_, value = self.processing_element
+ with self.client.transaction() as transaction:
+ transaction.delete("{path}/{id}".format(
+ path=self._entries_path,
+ id=id_))
+ transaction.delete("{path}/{id}".format(
+ path=self._lock_path,
+ id=id_))
+ self.processing_element = None
+ return True
+ else:
+ return False
+
+ def _inner_get(self, timeout):
+ flag = self.client.handler.event_object()
+ lock = self.client.handler.lock_object()
+ canceled = False
+ value = []
+
+ def check_for_updates(event):
+ if not event is None and event.type != EventType.CHILD:
+ return
+ with lock:
+ if canceled or flag.isSet():
+ return
+ values = self.client.retry(self.client.get_children,
+ self._entries_path,
+ check_for_updates)
+ taken = self.client.retry(self.client.get_children,
+ self._lock_path,
+ check_for_updates)
+ available = self._filter_locked(values, taken)
+ if len(available) > 0:
+ ret = self._take(available[0])
+ if not ret is None:
+ # By this time, no one took the task
+ value.append(ret)
+ flag.set()
+
+ check_for_updates(None)
+ retVal = None
+ flag.wait(timeout)
+ with lock:
+ canceled = True
+ if len(value) > 0:
+ # We successfully locked an entry
+ self.processing_element = value[0]
+ retVal = value[0][1]
+ return retVal
+
+ def _filter_locked(self, values, taken):
+ taken = set(taken)
+ available = sorted(values)
+ return (available if len(taken) == 0 else
+ [x for x in available if x not in taken])
+
+ def _take(self, id_):
+ try:
+ self.client.create(
+ "{path}/{id}".format(
+ path=self._lock_path,
+ id=id_),
+ self.id,
+ ephemeral=True)
+ value, stat = self.client.retry(self.client.get,
+ "{path}/{id}".format(path=self._entries_path, id=id_))
+ except (NoNodeError, NodeExistsError):
+ # Item is already consumed or locked
+ return None
+ return (id_, value)
http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/bbdb6bab/slider-agent/src/main/python/kazoo/recipe/watchers.py
----------------------------------------------------------------------
diff --git a/slider-agent/src/main/python/kazoo/recipe/watchers.py b/slider-agent/src/main/python/kazoo/recipe/watchers.py
new file mode 100644
index 0000000..0624153
--- /dev/null
+++ b/slider-agent/src/main/python/kazoo/recipe/watchers.py
@@ -0,0 +1,419 @@
+"""Higher level child and data watching API's.
+
+:Maintainer: Ben Bangert <be...@groovie.org>
+:Status: Production
+
+.. note::
+
+ :ref:`DataWatch` and :ref:`ChildrenWatch` may only handle a single
+ function, attempts to associate a single instance with multiple functions
+ will result in an exception being thrown.
+
+"""
+import logging
+import time
+import warnings
+from functools import partial, wraps
+
+from kazoo.retry import KazooRetry
+from kazoo.exceptions import (
+ ConnectionClosedError,
+ NoNodeError,
+ KazooException
+)
+from kazoo.protocol.states import KazooState
+
+log = logging.getLogger(__name__)
+
+
+_STOP_WATCHING = object()
+
+
+def _ignore_closed(func):
+ @wraps(func)
+ def wrapper(*args, **kwargs):
+ try:
+ return func(*args, **kwargs)
+ except ConnectionClosedError:
+ pass
+ return wrapper
+
+
+class DataWatch(object):
+ """Watches a node for data updates and calls the specified
+ function each time it changes
+
+ The function will also be called the very first time its
+ registered to get the data.
+
+ Returning `False` from the registered function will disable future
+ data change calls. If the client connection is closed (using the
+ close command), the DataWatch will no longer get updates.
+
+ If the function supplied takes three arguments, then the third one
+ will be a :class:`~kazoo.protocol.states.WatchedEvent`. It will
+ only be set if the change to the data occurs as a result of the
+ server notifying the watch that there has been a change. Events
+ like reconnection or the first call will not include an event.
+
+ If the node does not exist, then the function will be called with
+ ``None`` for all values.
+
+ .. tip::
+
+ Because :class:`DataWatch` can watch nodes that don't exist, it
+ can be used alternatively as a higher-level Exists watcher that
+ survives reconnections and session loss.
+
+ Example with client:
+
+ .. code-block:: python
+
+ @client.DataWatch('/path/to/watch')
+ def my_func(data, stat):
+ print("Data is %s" % data)
+ print("Version is %s" % stat.version)
+
+ # Above function is called immediately and prints
+
+ # Or if you want the event object
+ @client.DataWatch('/path/to/watch')
+ def my_func(data, stat, event):
+ print("Data is %s" % data)
+ print("Version is %s" % stat.version)
+ print("Event is %s" % event)
+
+ .. versionchanged:: 1.2
+
+ DataWatch now ignores additional arguments that were previously
+ passed to it and warns that they are no longer respected.
+
+ """
+ def __init__(self, client, path, func=None, *args, **kwargs):
+ """Create a data watcher for a path
+
+ :param client: A zookeeper client.
+ :type client: :class:`~kazoo.client.KazooClient`
+ :param path: The path to watch for data changes on.
+ :type path: str
+ :param func: Function to call initially and every time the
+ node changes. `func` will be called with a
+ tuple, the value of the node and a
+ :class:`~kazoo.client.ZnodeStat` instance.
+ :type func: callable
+
+ """
+ self._client = client
+ self._path = path
+ self._func = func
+ self._stopped = False
+ self._run_lock = client.handler.lock_object()
+ self._version = None
+ self._retry = KazooRetry(max_tries=None,
+ sleep_func=client.handler.sleep_func)
+ self._include_event = None
+ self._ever_called = False
+ self._used = False
+
+ if args or kwargs:
+ warnings.warn('Passing additional arguments to DataWatch is'
+ ' deprecated. ignore_missing_node is now assumed '
+ ' to be True by default, and the event will be '
+ ' sent if the function can handle receiving it',
+ DeprecationWarning, stacklevel=2)
+
+ # Register our session listener if we're going to resume
+ # across session losses
+ if func is not None:
+ self._used = True
+ self._client.add_listener(self._session_watcher)
+ self._get_data()
+
+ def __call__(self, func):
+ """Callable version for use as a decorator
+
+ :param func: Function to call initially and every time the
+ data changes. `func` will be called with a
+ tuple, the value of the node and a
+ :class:`~kazoo.client.ZnodeStat` instance.
+ :type func: callable
+
+ """
+ if self._used:
+ raise KazooException(
+ "A function has already been associated with this "
+ "DataWatch instance.")
+
+ self._func = func
+
+ self._used = True
+ self._client.add_listener(self._session_watcher)
+ self._get_data()
+ return func
+
+ def _log_func_exception(self, data, stat, event=None):
+ try:
+ # For backwards compatibility, don't send event to the
+ # callback unless the send_event is set in constructor
+ if not self._ever_called:
+ self._ever_called = True
+ try:
+ result = self._func(data, stat, event)
+ except TypeError:
+ result = self._func(data, stat)
+ if result is False:
+ self._stopped = True
+ self._client.remove_listener(self._session_watcher)
+ except Exception as exc:
+ log.exception(exc)
+ raise
+
+ @_ignore_closed
+ def _get_data(self, event=None):
+ # Ensure this runs one at a time, possible because the session
+ # watcher may trigger a run
+ with self._run_lock:
+ if self._stopped:
+ return
+
+ initial_version = self._version
+
+ try:
+ data, stat = self._retry(self._client.get,
+ self._path, self._watcher)
+ except NoNodeError:
+ data = None
+
+ # This will set 'stat' to None if the node does not yet
+ # exist.
+ stat = self._retry(self._client.exists, self._path,
+ self._watcher)
+ if stat:
+ self._client.handler.spawn(self._get_data)
+ return
+
+ # No node data, clear out version
+ if stat is None:
+ self._version = None
+ else:
+ self._version = stat.mzxid
+
+ # Call our function if its the first time ever, or if the
+ # version has changed
+ if initial_version != self._version or not self._ever_called:
+ self._log_func_exception(data, stat, event)
+
+ def _watcher(self, event):
+ self._get_data(event=event)
+
+ def _set_watch(self, state):
+ with self._run_lock:
+ self._watch_established = state
+
+ def _session_watcher(self, state):
+ if state == KazooState.CONNECTED:
+ self._client.handler.spawn(self._get_data)
+
+
+class ChildrenWatch(object):
+ """Watches a node for children updates and calls the specified
+ function each time it changes
+
+ The function will also be called the very first time its
+ registered to get children.
+
+ Returning `False` from the registered function will disable future
+ children change calls. If the client connection is closed (using
+ the close command), the ChildrenWatch will no longer get updates.
+
+ if send_event=True in __init__, then the function will always be
+ called with second parameter, ``event``. Upon initial call or when
+ recovering a lost session the ``event`` is always ``None``.
+ Otherwise it's a :class:`~kazoo.prototype.state.WatchedEvent`
+ instance.
+
+ Example with client:
+
+ .. code-block:: python
+
+ @client.ChildrenWatch('/path/to/watch')
+ def my_func(children):
+ print "Children are %s" % children
+
+ # Above function is called immediately and prints children
+
+ """
+ def __init__(self, client, path, func=None,
+ allow_session_lost=True, send_event=False):
+ """Create a children watcher for a path
+
+ :param client: A zookeeper client.
+ :type client: :class:`~kazoo.client.KazooClient`
+ :param path: The path to watch for children on.
+ :type path: str
+ :param func: Function to call initially and every time the
+ children change. `func` will be called with a
+ single argument, the list of children.
+ :type func: callable
+ :param allow_session_lost: Whether the watch should be
+ re-registered if the zookeeper
+ session is lost.
+ :type allow_session_lost: bool
+ :type send_event: bool
+ :param send_event: Whether the function should be passed the
+ event sent by ZooKeeper or None upon
+ initialization (see class documentation)
+
+ The path must already exist for the children watcher to
+ run.
+
+ """
+ self._client = client
+ self._path = path
+ self._func = func
+ self._send_event = send_event
+ self._stopped = False
+ self._watch_established = False
+ self._allow_session_lost = allow_session_lost
+ self._run_lock = client.handler.lock_object()
+ self._prior_children = None
+ self._used = False
+
+ # Register our session listener if we're going to resume
+ # across session losses
+ if func is not None:
+ self._used = True
+ if allow_session_lost:
+ self._client.add_listener(self._session_watcher)
+ self._get_children()
+
+ def __call__(self, func):
+ """Callable version for use as a decorator
+
+ :param func: Function to call initially and every time the
+ children change. `func` will be called with a
+ single argument, the list of children.
+ :type func: callable
+
+ """
+ if self._used:
+ raise KazooException(
+ "A function has already been associated with this "
+ "ChildrenWatch instance.")
+
+ self._func = func
+
+ self._used = True
+ if self._allow_session_lost:
+ self._client.add_listener(self._session_watcher)
+ self._get_children()
+ return func
+
+ @_ignore_closed
+ def _get_children(self, event=None):
+ with self._run_lock: # Ensure this runs one at a time
+ if self._stopped:
+ return
+
+ children = self._client.retry(self._client.get_children,
+ self._path, self._watcher)
+ if not self._watch_established:
+ self._watch_established = True
+
+ if self._prior_children is not None and \
+ self._prior_children == children:
+ return
+
+ self._prior_children = children
+
+ try:
+ if self._send_event:
+ result = self._func(children, event)
+ else:
+ result = self._func(children)
+ if result is False:
+ self._stopped = True
+ except Exception as exc:
+ log.exception(exc)
+ raise
+
+ def _watcher(self, event):
+ self._get_children(event)
+
+ def _session_watcher(self, state):
+ if state in (KazooState.LOST, KazooState.SUSPENDED):
+ self._watch_established = False
+ elif state == KazooState.CONNECTED and \
+ not self._watch_established and not self._stopped:
+ self._client.handler.spawn(self._get_children)
+
+
+class PatientChildrenWatch(object):
+ """Patient Children Watch that returns values after the children
+ of a node don't change for a period of time
+
+ A separate watcher for the children of a node, that ignores
+ changes within a boundary time and sets the result only when the
+ boundary time has elapsed with no children changes.
+
+ Example::
+
+ watcher = PatientChildrenWatch(client, '/some/path',
+ time_boundary=5)
+ async_object = watcher.start()
+
+ # Blocks until the children have not changed for time boundary
+ # (5 in this case) seconds, returns children list and an
+ # async_result that will be set if the children change in the
+ # future
+ children, child_async = async_object.get()
+
+ .. note::
+
+ This Watch is different from :class:`DataWatch` and
+ :class:`ChildrenWatch` as it only returns once, does not take
+ a function that is called, and provides an
+ :class:`~kazoo.interfaces.IAsyncResult` object that can be
+ checked to see if the children have changed later.
+
+ """
+ def __init__(self, client, path, time_boundary=30):
+ self.client = client
+ self.path = path
+ self.children = []
+ self.time_boundary = time_boundary
+ self.children_changed = client.handler.event_object()
+
+ def start(self):
+ """Begin the watching process asynchronously
+
+ :returns: An :class:`~kazoo.interfaces.IAsyncResult` instance
+ that will be set when no change has occurred to the
+ children for time boundary seconds.
+
+ """
+ self.asy = asy = self.client.handler.async_result()
+ self.client.handler.spawn(self._inner_start)
+ return asy
+
+ def _inner_start(self):
+ try:
+ while True:
+ async_result = self.client.handler.async_result()
+ self.children = self.client.retry(
+ self.client.get_children, self.path,
+ partial(self._children_watcher, async_result))
+ self.client.handler.sleep_func(self.time_boundary)
+
+ if self.children_changed.is_set():
+ self.children_changed.clear()
+ else:
+ break
+
+ self.asy.set((self.children, async_result))
+ except Exception as exc:
+ self.asy.set_exception(exc)
+
+ def _children_watcher(self, async, event):
+ self.children_changed.set()
+ async.set(time.time())
http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/bbdb6bab/slider-agent/src/main/python/kazoo/retry.py
----------------------------------------------------------------------
diff --git a/slider-agent/src/main/python/kazoo/retry.py b/slider-agent/src/main/python/kazoo/retry.py
new file mode 100644
index 0000000..152e9a6
--- /dev/null
+++ b/slider-agent/src/main/python/kazoo/retry.py
@@ -0,0 +1,150 @@
+import logging
+import random
+import time
+
+from kazoo.exceptions import (
+ ConnectionClosedError,
+ ConnectionLoss,
+ KazooException,
+ OperationTimeoutError,
+ SessionExpiredError,
+)
+
+log = logging.getLogger(__name__)
+
+
+class ForceRetryError(Exception):
+ """Raised when some recipe logic wants to force a retry."""
+
+
+class RetryFailedError(KazooException):
+ """Raised when retrying an operation ultimately failed, after
+ retrying the maximum number of attempts.
+ """
+
+
+class InterruptedError(RetryFailedError):
+ """Raised when the retry is forcibly interrupted by the interrupt
+ function"""
+
+
+class KazooRetry(object):
+ """Helper for retrying a method in the face of retry-able
+ exceptions"""
+ RETRY_EXCEPTIONS = (
+ ConnectionLoss,
+ OperationTimeoutError,
+ ForceRetryError
+ )
+
+ EXPIRED_EXCEPTIONS = (
+ SessionExpiredError,
+ )
+
+ def __init__(self, max_tries=1, delay=0.1, backoff=2, max_jitter=0.8,
+ max_delay=3600, ignore_expire=True, sleep_func=time.sleep,
+ deadline=None, interrupt=None):
+ """Create a :class:`KazooRetry` instance for retrying function
+ calls
+
+ :param max_tries: How many times to retry the command. -1 means
+ infinite tries.
+ :param delay: Initial delay between retry attempts.
+ :param backoff: Backoff multiplier between retry attempts.
+ Defaults to 2 for exponential backoff.
+ :param max_jitter: Additional max jitter period to wait between
+ retry attempts to avoid slamming the server.
+ :param max_delay: Maximum delay in seconds, regardless of other
+ backoff settings. Defaults to one hour.
+ :param ignore_expire:
+ Whether a session expiration should be ignored and treated
+ as a retry-able command.
+ :param interrupt:
+ Function that will be called with no args that may return
+ True if the retry should be ceased immediately. This will
+ be called no more than every 0.1 seconds during a wait
+ between retries.
+
+ """
+ self.max_tries = max_tries
+ self.delay = delay
+ self.backoff = backoff
+ self.max_jitter = int(max_jitter * 100)
+ self.max_delay = float(max_delay)
+ self._attempts = 0
+ self._cur_delay = delay
+ self.deadline = deadline
+ self._cur_stoptime = None
+ self.sleep_func = sleep_func
+ self.retry_exceptions = self.RETRY_EXCEPTIONS
+ self.interrupt = interrupt
+ if ignore_expire:
+ self.retry_exceptions += self.EXPIRED_EXCEPTIONS
+
+ def reset(self):
+ """Reset the attempt counter"""
+ self._attempts = 0
+ self._cur_delay = self.delay
+ self._cur_stoptime = None
+
+ def copy(self):
+ """Return a clone of this retry manager"""
+ obj = KazooRetry(max_tries=self.max_tries,
+ delay=self.delay,
+ backoff=self.backoff,
+ max_jitter=self.max_jitter / 100.0,
+ max_delay=self.max_delay,
+ sleep_func=self.sleep_func,
+ deadline=self.deadline,
+ interrupt=self.interrupt)
+ obj.retry_exceptions = self.retry_exceptions
+ return obj
+
+ def __call__(self, func, *args, **kwargs):
+ """Call a function with arguments until it completes without
+ throwing a Kazoo exception
+
+ :param func: Function to call
+ :param args: Positional arguments to call the function with
+ :params kwargs: Keyword arguments to call the function with
+
+ The function will be called until it doesn't throw one of the
+ retryable exceptions (ConnectionLoss, OperationTimeout, or
+ ForceRetryError), and optionally retrying on session
+ expiration.
+
+ """
+ self.reset()
+
+ while True:
+ try:
+ if self.deadline is not None and self._cur_stoptime is None:
+ self._cur_stoptime = time.time() + self.deadline
+ return func(*args, **kwargs)
+ except ConnectionClosedError:
+ raise
+ except self.retry_exceptions:
+ # Note: max_tries == -1 means infinite tries.
+ if self._attempts == self.max_tries:
+ raise RetryFailedError("Too many retry attempts")
+ self._attempts += 1
+ sleeptime = self._cur_delay + (random.randint(0, self.max_jitter) / 100.0)
+
+ if self._cur_stoptime is not None and time.time() + sleeptime >= self._cur_stoptime:
+ raise RetryFailedError("Exceeded retry deadline")
+
+ if self.interrupt:
+ while sleeptime > 0:
+ # Break the time period down and sleep for no longer than
+ # 0.1 before calling the interrupt
+ if sleeptime < 0.1:
+ self.sleep_func(sleeptime)
+ sleeptime -= sleeptime
+ else:
+ self.sleep_func(0.1)
+ sleeptime -= 0.1
+ if self.interrupt():
+ raise InterruptedError()
+ else:
+ self.sleep_func(sleeptime)
+ self._cur_delay = min(self._cur_delay * self.backoff, self.max_delay)
http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/bbdb6bab/slider-agent/src/main/python/kazoo/security.py
----------------------------------------------------------------------
diff --git a/slider-agent/src/main/python/kazoo/security.py b/slider-agent/src/main/python/kazoo/security.py
new file mode 100644
index 0000000..1bd989d
--- /dev/null
+++ b/slider-agent/src/main/python/kazoo/security.py
@@ -0,0 +1,138 @@
+"""Kazoo Security"""
+from base64 import b64encode
+from collections import namedtuple
+import hashlib
+
+
+# Represents a Zookeeper ID and ACL object
+Id = namedtuple('Id', 'scheme id')
+
+
+class ACL(namedtuple('ACL', 'perms id')):
+ """An ACL for a Zookeeper Node
+
+ An ACL object is created by using an :class:`Id` object along with
+ a :class:`Permissions` setting. For convenience,
+ :meth:`make_digest_acl` should be used to create an ACL object with
+ the desired scheme, id, and permissions.
+
+ """
+ @property
+ def acl_list(self):
+ perms = []
+ if self.perms & Permissions.ALL == Permissions.ALL:
+ perms.append('ALL')
+ return perms
+ if self.perms & Permissions.READ == Permissions.READ:
+ perms.append('READ')
+ if self.perms & Permissions.WRITE == Permissions.WRITE:
+ perms.append('WRITE')
+ if self.perms & Permissions.CREATE == Permissions.CREATE:
+ perms.append('CREATE')
+ if self.perms & Permissions.DELETE == Permissions.DELETE:
+ perms.append('DELETE')
+ if self.perms & Permissions.ADMIN == Permissions.ADMIN:
+ perms.append('ADMIN')
+ return perms
+
+ def __repr__(self):
+ return 'ACL(perms=%r, acl_list=%s, id=%r)' % (
+ self.perms, self.acl_list, self.id)
+
+
+class Permissions(object):
+ READ = 1
+ WRITE = 2
+ CREATE = 4
+ DELETE = 8
+ ADMIN = 16
+ ALL = 31
+
+
+# Shortcuts for common Ids
+ANYONE_ID_UNSAFE = Id('world', 'anyone')
+AUTH_IDS = Id('auth', '')
+
+# Shortcuts for common ACLs
+OPEN_ACL_UNSAFE = [ACL(Permissions.ALL, ANYONE_ID_UNSAFE)]
+CREATOR_ALL_ACL = [ACL(Permissions.ALL, AUTH_IDS)]
+READ_ACL_UNSAFE = [ACL(Permissions.READ, ANYONE_ID_UNSAFE)]
+
+
+def make_digest_acl_credential(username, password):
+ """Create a SHA1 digest credential"""
+ credential = username.encode('utf-8') + b":" + password.encode('utf-8')
+ cred_hash = b64encode(hashlib.sha1(credential).digest()).strip()
+ return username + ":" + cred_hash.decode('utf-8')
+
+
+def make_acl(scheme, credential, read=False, write=False,
+ create=False, delete=False, admin=False, all=False):
+ """Given a scheme and credential, return an :class:`ACL` object
+ appropriate for use with Kazoo.
+
+ :param scheme: The scheme to use. I.e. `digest`.
+ :param credential:
+ A colon separated username, password. The password should be
+ hashed with the `scheme` specified. The
+ :meth:`make_digest_acl_credential` method will create and
+ return a credential appropriate for use with the `digest`
+ scheme.
+ :param write: Write permission.
+ :type write: bool
+ :param create: Create permission.
+ :type create: bool
+ :param delete: Delete permission.
+ :type delete: bool
+ :param admin: Admin permission.
+ :type admin: bool
+ :param all: All permissions.
+ :type all: bool
+
+ :rtype: :class:`ACL`
+
+ """
+ if all:
+ permissions = Permissions.ALL
+ else:
+ permissions = 0
+ if read:
+ permissions |= Permissions.READ
+ if write:
+ permissions |= Permissions.WRITE
+ if create:
+ permissions |= Permissions.CREATE
+ if delete:
+ permissions |= Permissions.DELETE
+ if admin:
+ permissions |= Permissions.ADMIN
+ return ACL(permissions, Id(scheme, credential))
+
+
+def make_digest_acl(username, password, read=False, write=False,
+ create=False, delete=False, admin=False, all=False):
+ """Create a digest ACL for Zookeeper with the given permissions
+
+ This method combines :meth:`make_digest_acl_credential` and
+ :meth:`make_acl` to create an :class:`ACL` object appropriate for
+ use with Kazoo's ACL methods.
+
+ :param username: Username to use for the ACL.
+ :param password: A plain-text password to hash.
+ :param write: Write permission.
+ :type write: bool
+ :param create: Create permission.
+ :type create: bool
+ :param delete: Delete permission.
+ :type delete: bool
+ :param admin: Admin permission.
+ :type admin: bool
+ :param all: All permissions.
+ :type all: bool
+
+ :rtype: :class:`ACL`
+
+ """
+ cred = make_digest_acl_credential(username, password)
+ return make_acl("digest", cred, read=read, write=write, create=create,
+ delete=delete, admin=admin, all=all)
http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/bbdb6bab/slider-agent/src/main/python/kazoo/testing/__init__.py
----------------------------------------------------------------------
diff --git a/slider-agent/src/main/python/kazoo/testing/__init__.py b/slider-agent/src/main/python/kazoo/testing/__init__.py
new file mode 100644
index 0000000..bf8f149
--- /dev/null
+++ b/slider-agent/src/main/python/kazoo/testing/__init__.py
@@ -0,0 +1,5 @@
+from kazoo.testing.harness import KazooTestCase
+from kazoo.testing.harness import KazooTestHarness
+
+
+__all__ = ('KazooTestHarness', 'KazooTestCase', )
http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/bbdb6bab/slider-agent/src/main/python/kazoo/testing/common.py
----------------------------------------------------------------------
diff --git a/slider-agent/src/main/python/kazoo/testing/common.py b/slider-agent/src/main/python/kazoo/testing/common.py
new file mode 100644
index 0000000..8e1e9b4
--- /dev/null
+++ b/slider-agent/src/main/python/kazoo/testing/common.py
@@ -0,0 +1,283 @@
+#
+# Copyright (C) 2010-2011, 2011 Canonical Ltd. All Rights Reserved
+#
+# This file was originally taken from txzookeeper and modified later.
+#
+# Authors:
+# Kapil Thangavelu and the Kazoo team
+#
+# txzookeeper is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Lesser General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# txzookeeper is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public License
+# along with txzookeeper. If not, see <http://www.gnu.org/licenses/>.
+
+
+import code
+import os
+import os.path
+import shutil
+import signal
+import subprocess
+import tempfile
+import traceback
+
+from itertools import chain
+from collections import namedtuple
+from glob import glob
+
+
+def debug(sig, frame):
+ """Interrupt running process, and provide a python prompt for
+ interactive debugging."""
+ d = {'_frame': frame} # Allow access to frame object.
+ d.update(frame.f_globals) # Unless shadowed by global
+ d.update(frame.f_locals)
+
+ i = code.InteractiveConsole(d)
+ message = "Signal recieved : entering python shell.\nTraceback:\n"
+ message += ''.join(traceback.format_stack(frame))
+ i.interact(message)
+
+
+def listen():
+ if os.name != 'nt': # SIGUSR1 is not supported on Windows
+ signal.signal(signal.SIGUSR1, debug) # Register handler
+listen()
+
+
+def to_java_compatible_path(path):
+ if os.name == 'nt':
+ path = path.replace('\\', '/')
+ return path
+
+ServerInfo = namedtuple(
+ "ServerInfo", "server_id client_port election_port leader_port")
+
+
+class ManagedZooKeeper(object):
+ """Class to manage the running of a ZooKeeper instance for testing.
+
+ Note: no attempt is made to probe the ZooKeeper instance is
+ actually available, or that the selected port is free. In the
+ future, we may want to do that, especially when run in a
+ Hudson/Buildbot context, to ensure more test robustness."""
+
+ def __init__(self, software_path, server_info, peers=(), classpath=None):
+ """Define the ZooKeeper test instance.
+
+ @param install_path: The path to the install for ZK
+ @param port: The port to run the managed ZK instance
+ """
+ self.install_path = software_path
+ self._classpath = classpath
+ self.server_info = server_info
+ self.host = "127.0.0.1"
+ self.peers = peers
+ self.working_path = tempfile.mkdtemp()
+ self._running = False
+
+ def run(self):
+ """Run the ZooKeeper instance under a temporary directory.
+
+ Writes ZK log messages to zookeeper.log in the current directory.
+ """
+ if self.running:
+ return
+ config_path = os.path.join(self.working_path, "zoo.cfg")
+ log_path = os.path.join(self.working_path, "log")
+ log4j_path = os.path.join(self.working_path, "log4j.properties")
+ data_path = os.path.join(self.working_path, "data")
+
+ # various setup steps
+ if not os.path.exists(self.working_path):
+ os.mkdir(self.working_path)
+ if not os.path.exists(log_path):
+ os.mkdir(log_path)
+ if not os.path.exists(data_path):
+ os.mkdir(data_path)
+
+ with open(config_path, "w") as config:
+ config.write("""
+tickTime=2000
+dataDir=%s
+clientPort=%s
+maxClientCnxns=0
+""" % (to_java_compatible_path(data_path), self.server_info.client_port))
+
+ # setup a replicated setup if peers are specified
+ if self.peers:
+ servers_cfg = []
+ for p in chain((self.server_info,), self.peers):
+ servers_cfg.append("server.%s=localhost:%s:%s" % (
+ p.server_id, p.leader_port, p.election_port))
+
+ with open(config_path, "a") as config:
+ config.write("""
+initLimit=4
+syncLimit=2
+%s
+""" % ("\n".join(servers_cfg)))
+
+ # Write server ids into datadir
+ with open(os.path.join(data_path, "myid"), "w") as myid_file:
+ myid_file.write(str(self.server_info.server_id))
+
+ with open(log4j_path, "w") as log4j:
+ log4j.write("""
+# DEFAULT: console appender only
+log4j.rootLogger=INFO, ROLLINGFILE
+log4j.appender.ROLLINGFILE.layout=org.apache.log4j.PatternLayout
+log4j.appender.ROLLINGFILE.layout.ConversionPattern=%d{ISO8601} [myid:%X{myid}] - %-5p [%t:%C{1}@%L] - %m%n
+log4j.appender.ROLLINGFILE=org.apache.log4j.RollingFileAppender
+log4j.appender.ROLLINGFILE.Threshold=DEBUG
+log4j.appender.ROLLINGFILE.File=""" + to_java_compatible_path(
+ self.working_path + os.sep + "zookeeper.log\n"))
+
+ self.process = subprocess.Popen(
+ args=["java",
+ "-cp", self.classpath,
+ "-Dreadonlymode.enabled=true",
+ "-Dzookeeper.log.dir=%s" % log_path,
+ "-Dzookeeper.root.logger=INFO,CONSOLE",
+ "-Dlog4j.configuration=file:%s" % log4j_path,
+ # "-Dlog4j.debug",
+ "org.apache.zookeeper.server.quorum.QuorumPeerMain",
+ config_path])
+ self._running = True
+
+ @property
+ def classpath(self):
+ """Get the classpath necessary to run ZooKeeper."""
+
+ if self._classpath:
+ return self._classpath
+
+ # Two possibilities, as seen in zkEnv.sh:
+ # Check for a release - top-level zookeeper-*.jar?
+ jars = glob((os.path.join(
+ self.install_path, 'zookeeper-*.jar')))
+ if jars:
+ # Release build (`ant package`)
+ jars.extend(glob(os.path.join(
+ self.install_path,
+ "lib/*.jar")))
+ # support for different file locations on Debian/Ubuntu
+ jars.extend(glob(os.path.join(
+ self.install_path,
+ "log4j-*.jar")))
+ jars.extend(glob(os.path.join(
+ self.install_path,
+ "slf4j-api-*.jar")))
+ jars.extend(glob(os.path.join(
+ self.install_path,
+ "slf4j-log4j-*.jar")))
+ else:
+ # Development build (plain `ant`)
+ jars = glob((os.path.join(
+ self.install_path, 'build/zookeeper-*.jar')))
+ jars.extend(glob(os.path.join(
+ self.install_path,
+ "build/lib/*.jar")))
+
+ return os.pathsep.join(jars)
+
+ @property
+ def address(self):
+ """Get the address of the ZooKeeper instance."""
+ return "%s:%s" % (self.host, self.client_port)
+
+ @property
+ def running(self):
+ return self._running
+
+ @property
+ def client_port(self):
+ return self.server_info.client_port
+
+ def reset(self):
+ """Stop the zookeeper instance, cleaning out its on disk-data."""
+ self.stop()
+ shutil.rmtree(os.path.join(self.working_path, "data"))
+ os.mkdir(os.path.join(self.working_path, "data"))
+ with open(os.path.join(self.working_path, "data", "myid"), "w") as fh:
+ fh.write(str(self.server_info.server_id))
+
+ def stop(self):
+ """Stop the Zookeeper instance, retaining on disk state."""
+ if not self.running:
+ return
+ self.process.terminate()
+ self.process.wait()
+ self._running = False
+
+ def destroy(self):
+ """Stop the ZooKeeper instance and destroy its on disk-state"""
+ # called by at exit handler, reimport to avoid cleanup race.
+ import shutil
+ self.stop()
+
+ shutil.rmtree(self.working_path)
+
+
+class ZookeeperCluster(object):
+
+ def __init__(self, install_path=None, classpath=None, size=3, port_offset=20000):
+ self._install_path = install_path
+ self._classpath = classpath
+ self._servers = []
+
+ # Calculate ports and peer group
+ port = port_offset
+ peers = []
+
+ for i in range(size):
+ info = ServerInfo(i + 1, port, port + 1, port + 2)
+ peers.append(info)
+ port += 10
+
+ # Instantiate Managed ZK Servers
+ for i in range(size):
+ server_peers = list(peers)
+ server_info = server_peers.pop(i)
+ self._servers.append(
+ ManagedZooKeeper(
+ self._install_path, server_info, server_peers, classpath=self._classpath))
+
+ def __getitem__(self, k):
+ return self._servers[k]
+
+ def __iter__(self):
+ return iter(self._servers)
+
+ def start(self):
+ # Zookeeper client expresses a preference for either lower ports or
+ # lexicographical ordering of hosts, to ensure that all servers have a
+ # chance to startup, start them in reverse order.
+ for server in reversed(list(self)):
+ server.run()
+ # Giving the servers a moment to start, decreases the overall time
+ # required for a client to successfully connect (2s vs. 4s without
+ # the sleep).
+ import time
+ time.sleep(2)
+
+ def stop(self):
+ for server in self:
+ server.stop()
+ self._servers = []
+
+ def terminate(self):
+ for server in self:
+ server.destroy()
+
+ def reset(self):
+ for server in self:
+ server.reset()