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/11 20:19:30 UTC
[6/6] git commit: SLIDER-285. Slider Agents to bind and work with
restarted AM (Gour Saha via smohanty)
SLIDER-285. Slider Agents to bind and work with restarted AM (Gour Saha via smohanty)
Project: http://git-wip-us.apache.org/repos/asf/incubator-slider/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-slider/commit/e682b8f0
Tree: http://git-wip-us.apache.org/repos/asf/incubator-slider/tree/e682b8f0
Diff: http://git-wip-us.apache.org/repos/asf/incubator-slider/diff/e682b8f0
Branch: refs/heads/feature/SLIDER-285_Restart_AM
Commit: e682b8f040b5509f09bdb265165e01b9df8cbb55
Parents: e464bba
Author: Sumit Mohanty <sm...@hortonworks.com>
Authored: Mon Aug 11 11:17:46 2014 -0700
Committer: Sumit Mohanty <sm...@hortonworks.com>
Committed: Mon Aug 11 11:17:46 2014 -0700
----------------------------------------------------------------------
slider-agent/pom.xml | 4 +-
.../src/main/python/agent/AgentConfig.py | 2 +
slider-agent/src/main/python/agent/Constants.py | 4 +-
.../src/main/python/agent/Controller.py | 52 +-
slider-agent/src/main/python/agent/Heartbeat.py | 6 +-
slider-agent/src/main/python/agent/Registry.py | 54 +
slider-agent/src/main/python/agent/main.py | 29 +-
slider-agent/src/main/python/kazoo/__init__.py | 1 +
slider-agent/src/main/python/kazoo/client.py | 1412 ++++++++++++++++++
.../src/main/python/kazoo/exceptions.py | 199 +++
.../src/main/python/kazoo/handlers/__init__.py | 1 +
.../src/main/python/kazoo/handlers/gevent.py | 161 ++
.../src/main/python/kazoo/handlers/threading.py | 287 ++++
.../src/main/python/kazoo/handlers/utils.py | 93 ++
slider-agent/src/main/python/kazoo/hosts.py | 26 +
.../src/main/python/kazoo/interfaces.py | 203 +++
.../src/main/python/kazoo/loggingsupport.py | 2 +
.../src/main/python/kazoo/protocol/__init__.py | 1 +
.../main/python/kazoo/protocol/connection.py | 623 ++++++++
.../src/main/python/kazoo/protocol/paths.py | 54 +
.../main/python/kazoo/protocol/serialization.py | 396 +++++
.../src/main/python/kazoo/protocol/states.py | 237 +++
.../src/main/python/kazoo/recipe/__init__.py | 1 +
.../src/main/python/kazoo/recipe/barrier.py | 214 +++
.../src/main/python/kazoo/recipe/counter.py | 94 ++
.../src/main/python/kazoo/recipe/election.py | 79 +
.../src/main/python/kazoo/recipe/lock.py | 520 +++++++
.../src/main/python/kazoo/recipe/partitioner.py | 377 +++++
.../src/main/python/kazoo/recipe/party.py | 118 ++
.../src/main/python/kazoo/recipe/queue.py | 321 ++++
.../src/main/python/kazoo/recipe/watchers.py | 419 ++++++
slider-agent/src/main/python/kazoo/retry.py | 150 ++
slider-agent/src/main/python/kazoo/security.py | 138 ++
.../src/main/python/kazoo/testing/__init__.py | 5 +
.../src/main/python/kazoo/testing/common.py | 283 ++++
.../src/main/python/kazoo/testing/harness.py | 180 +++
.../src/main/python/kazoo/tests/__init__.py | 0
.../src/main/python/kazoo/tests/test_barrier.py | 157 ++
.../src/main/python/kazoo/tests/test_build.py | 29 +
.../src/main/python/kazoo/tests/test_client.py | 1098 ++++++++++++++
.../main/python/kazoo/tests/test_connection.py | 319 ++++
.../src/main/python/kazoo/tests/test_counter.py | 35 +
.../main/python/kazoo/tests/test_election.py | 139 ++
.../main/python/kazoo/tests/test_exceptions.py | 22 +
.../python/kazoo/tests/test_gevent_handler.py | 160 ++
.../src/main/python/kazoo/tests/test_lock.py | 517 +++++++
.../main/python/kazoo/tests/test_partitioner.py | 92 ++
.../src/main/python/kazoo/tests/test_party.py | 84 ++
.../src/main/python/kazoo/tests/test_paths.py | 98 ++
.../src/main/python/kazoo/tests/test_queue.py | 179 +++
.../src/main/python/kazoo/tests/test_retry.py | 77 +
.../main/python/kazoo/tests/test_security.py | 40 +
.../kazoo/tests/test_threading_handler.py | 326 ++++
.../main/python/kazoo/tests/test_watchers.py | 489 ++++++
.../src/main/python/kazoo/tests/util.py | 126 ++
slider-agent/src/test/python/agent/TestMain.py | 6 +-
.../org/apache/slider/common/SliderKeys.java | 5 +
.../providers/AbstractProviderService.java | 18 +
.../slider/providers/ProviderService.java | 22 +
.../slider/providers/agent/AgentKeys.java | 2 +
.../providers/agent/AgentProviderService.java | 126 +-
.../server/appmaster/SliderAppMaster.java | 7 +
.../slider/server/appmaster/state/AppState.java | 8 +
.../appmaster/web/rest/agent/HeartBeat.java | 11 +
.../model/mock/MockProviderService.groovy | 12 +
.../agent/TestAgentProviderService.java | 26 +-
slider-core/src/test/python/agent/main.py | 5 +-
67 files changed, 10918 insertions(+), 63 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/e682b8f0/slider-agent/pom.xml
----------------------------------------------------------------------
diff --git a/slider-agent/pom.xml b/slider-agent/pom.xml
index 4e67836..e230ec1 100644
--- a/slider-agent/pom.xml
+++ b/slider-agent/pom.xml
@@ -69,7 +69,7 @@
<argument>unitTests.py</argument>
</arguments>
<environmentVariables>
- <PYTHONPATH>${project.basedir}/src/main/python/jinja2:${project.basedir}/src/test/python:${project.basedir}/src/main/python:${project.basedir}/src/main/python/agent:${project.basedir}/src/main/python/resource_management:${project.basedir}/src/test/python/agent:${project.basedir}/src/test/python/resource_management</PYTHONPATH>
+ <PYTHONPATH>${project.basedir}/src/main/python/jinja2:${project.basedir}/src/test/python:${project.basedir}/src/main/python:${project.basedir}/src/main/python/agent:${project.basedir}/src/main/python/resource_management:${project.basedir}/src/test/python/agent:${project.basedir}/src/test/python/resource_management:${project.basedir}/src/main/python/kazoo</PYTHONPATH>
</environmentVariables>
<skip>${skipTests}</skip>
</configuration>
@@ -102,6 +102,8 @@
<exclude>src/main/python/jinja2/**</exclude>
<!-- mock files (BSD license) -->
<exclude>src/test/python/mock/**</exclude>
+ <!-- kazoo files (Apache License, Version 2.0) -->
+ <exclude>src/main/python/kazoo/**</exclude>
</excludes>
</configuration>
</plugin>
http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/e682b8f0/slider-agent/src/main/python/agent/AgentConfig.py
----------------------------------------------------------------------
diff --git a/slider-agent/src/main/python/agent/AgentConfig.py b/slider-agent/src/main/python/agent/AgentConfig.py
index 57b1542..9adace1 100644
--- a/slider-agent/src/main/python/agent/AgentConfig.py
+++ b/slider-agent/src/main/python/agent/AgentConfig.py
@@ -33,6 +33,8 @@ content = """
hostname=localhost
port=8440
secured_port=8441
+zk-quorum=localhost:2181
+zk-reg-path=/register/org-apache-slider/cl1
check_path=/ws/v1/slider/agents/
register_path=/ws/v1/slider/agents/{name}/register
heartbeat_path=/ws/v1/slider/agents/{name}/heartbeat
http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/e682b8f0/slider-agent/src/main/python/agent/Constants.py
----------------------------------------------------------------------
diff --git a/slider-agent/src/main/python/agent/Constants.py b/slider-agent/src/main/python/agent/Constants.py
index 88cd564..a120999 100644
--- a/slider-agent/src/main/python/agent/Constants.py
+++ b/slider-agent/src/main/python/agent/Constants.py
@@ -29,4 +29,6 @@ AGENT_WORK_ROOT = "AGENT_WORK_ROOT"
AGENT_LOG_ROOT = "AGENT_LOG_ROOT"
DO_NOT_REGISTER = "DO_NOT_REGISTER"
DO_NOT_HEARTBEAT = "DO_NOT_HEARTBEAT"
-DO_NOT_HEARTBEAT_AFTER_ = "DO_NOT_HEARTBEAT_AFTER_"
\ No newline at end of file
+DO_NOT_HEARTBEAT_AFTER_ = "DO_NOT_HEARTBEAT_AFTER_"
+ZK_QUORUM="zk_quorum"
+ZK_REG_PATH="zk_reg_path"
http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/e682b8f0/slider-agent/src/main/python/agent/Controller.py
----------------------------------------------------------------------
diff --git a/slider-agent/src/main/python/agent/Controller.py b/slider-agent/src/main/python/agent/Controller.py
index 92e9086..4a103cb 100644
--- a/slider-agent/src/main/python/agent/Controller.py
+++ b/slider-agent/src/main/python/agent/Controller.py
@@ -34,6 +34,7 @@ from Heartbeat import Heartbeat
from Register import Register
from ActionQueue import ActionQueue
from NetUtil import NetUtil
+from Registry import Registry
import ssl
import ProcessHelper
import Constants
@@ -43,7 +44,12 @@ import security
logger = logging.getLogger()
AGENT_AUTO_RESTART_EXIT_CODE = 77
+HEART_BEAT_RETRY_THRESHOLD = 2
+WS_AGENT_CONTEXT_ROOT = '/ws'
+SLIDER_PATH_AGENTS = WS_AGENT_CONTEXT_ROOT + '/v1/slider/agents/'
+SLIDER_REL_PATH_REGISTER = '/register'
+SLIDER_REL_PATH_HEARTBEAT = '/heartbeat'
class State:
INIT, INSTALLING, INSTALLED, STARTING, STARTED, FAILED = range(6)
@@ -57,13 +63,12 @@ class Controller(threading.Thread):
self.safeMode = True
self.credential = None
self.config = config
- self.hostname = config.getLabel()
- server_url = 'https://' + config.get(AgentConfig.SERVER_SECTION,
- 'hostname') + \
- ':' + config.get(AgentConfig.SERVER_SECTION,
- 'secured_port')
- self.registerUrl = server_url + '/ws/v1/slider/agents/' + self.hostname + '/register'
- self.heartbeatUrl = server_url + '/ws/v1/slider/agents/' + self.hostname + '/heartbeat'
+ self.label = config.getLabel()
+ self.hostname = config.get(AgentConfig.SERVER_SECTION, 'hostname')
+ self.secured_port = config.get(AgentConfig.SERVER_SECTION, 'secured_port')
+ self.server_url = 'https://' + self.hostname + ':' + self.secured_port
+ self.registerUrl = self.server_url + SLIDER_PATH_AGENTS + self.label + SLIDER_REL_PATH_REGISTER
+ self.heartbeatUrl = self.server_url + SLIDER_PATH_AGENTS + self.label + SLIDER_REL_PATH_HEARTBEAT
self.netutil = NetUtil()
self.responseId = -1
self.repeatRegistration = False
@@ -80,6 +85,7 @@ class Controller(threading.Thread):
self.componentActualState = State.INIT
self.statusCommand = None
self.failureCount = 0
+ self.heartBeatRetryCount = 0
def __del__(self):
@@ -204,8 +210,8 @@ class Controller(threading.Thread):
try:
if not retry:
data = json.dumps(
- self.heartbeat.build(commandResult, self.responseId,
- self.hasMappedComponents))
+ self.heartbeat.build(commandResult, self.componentActualState,
+ self.responseId, self.hasMappedComponents))
self.updateStateBasedOnResult(commandResult)
logger.debug("Sending request: " + data)
pass
@@ -285,9 +291,33 @@ class Controller(threading.Thread):
print(
"Server certificate verify failed. Did you regenerate server certificate?")
certVerifFailed = True
+ self.heartBeatRetryCount += 1
+ logger.error(
+ "Heartbeat retry count = %d" % (self.heartBeatRetryCount))
+ # Re-read zk registry in case AM was restarted and came up with new
+ # host/port, but do this only after heartbeat retry attempts crosses
+ # threshold
+ if self.heartBeatRetryCount > HEART_BEAT_RETRY_THRESHOLD:
+ self.isRegistered = False
+ self.repeatRegistration = True
+ self.heartBeatRetryCount = 0
+ self.cachedconnect = None # Previous connection is broken now
+ zk_quorum = self.config.get(AgentConfig.SERVER_SECTION, Constants.ZK_QUORUM)
+ zk_reg_path = self.config.get(AgentConfig.SERVER_SECTION, Constants.ZK_REG_PATH)
+ registry = Registry(zk_quorum, zk_reg_path)
+ amHost, amSecuredPort = registry.readAMHostPort()
+ logger.info("Read from ZK registry: AM host = %s, AM secured port = %s" % (amHost, amSecuredPort))
+ self.hostname = amHost
+ self.secured_port = amSecuredPort
+ self.config.set(AgentConfig.SERVER_SECTION, "hostname", self.hostname)
+ self.config.set(AgentConfig.SERVER_SECTION, "secured_port", self.secured_port)
+ self.server_url = 'https://' + self.hostname + ':' + self.secured_port
+ self.registerUrl = self.server_url + SLIDER_PATH_AGENTS + self.label + SLIDER_REL_PATH_REGISTER
+ self.heartbeatUrl = self.server_url + SLIDER_PATH_AGENTS + self.label + SLIDER_REL_PATH_HEARTBEAT
+ return
self.cachedconnect = None # Previous connection is broken now
retry = True
- # Sleep for some time
+ # Sleep for some time
timeout = self.netutil.HEARTBEAT_IDDLE_INTERVAL_SEC \
- self.netutil.MINIMUM_INTERVAL_BETWEEN_HEARTBEATS
self.heartbeat_wait_event.wait(timeout=timeout)
@@ -358,8 +388,8 @@ class Controller(threading.Thread):
statusCommand["serviceName"] = command["serviceName"]
statusCommand["taskId"] = "status"
statusCommand['auto_generated'] = True
- return statusCommand
logger.info("Status command: " + pprint.pformat(statusCommand))
+ return statusCommand
pass
http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/e682b8f0/slider-agent/src/main/python/agent/Heartbeat.py
----------------------------------------------------------------------
diff --git a/slider-agent/src/main/python/agent/Heartbeat.py b/slider-agent/src/main/python/agent/Heartbeat.py
index 8192348..4f2207c 100644
--- a/slider-agent/src/main/python/agent/Heartbeat.py
+++ b/slider-agent/src/main/python/agent/Heartbeat.py
@@ -36,7 +36,8 @@ class Heartbeat:
self.config = config
self.reports = []
- def build(self, commandResult, id='-1', componentsMapped=False):
+ def build(self, commandResult, componentActualState, id='-1',
+ componentsMapped=False):
timestamp = int(time.time() * 1000)
queueResult = self.actionQueue.result()
logger.info("Queue result: " + pformat(queueResult))
@@ -48,7 +49,8 @@ class Heartbeat:
'timestamp': timestamp,
'hostname': self.config.getLabel(),
'nodeStatus': nodeStatus,
- 'fqdn': hostname.public_hostname()
+ 'fqdn': hostname.public_hostname(),
+ 'agentState': componentActualState
}
commandsInProgress = False
http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/e682b8f0/slider-agent/src/main/python/agent/Registry.py
----------------------------------------------------------------------
diff --git a/slider-agent/src/main/python/agent/Registry.py b/slider-agent/src/main/python/agent/Registry.py
new file mode 100644
index 0000000..256509f
--- /dev/null
+++ b/slider-agent/src/main/python/agent/Registry.py
@@ -0,0 +1,54 @@
+#!/usr/bin/env python
+'''
+
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+'''
+
+import json
+import logging
+from kazoo.client import KazooClient
+
+logger = logging.getLogger()
+
+class Registry:
+ def __init__(self, zk_quorum, zk_reg_path):
+ self.zk_quorum = zk_quorum
+ self.zk_reg_path = zk_reg_path
+
+ def readAMHostPort(self):
+ amHost = ""
+ amSecuredPort = ""
+ try:
+ zk = KazooClient(hosts=self.zk_quorum, read_only=True)
+ zk.start()
+ data, stat = zk.get(self.zk_reg_path)
+ logger.debug("Registry Data: %s" % (data.decode("utf-8")))
+ sliderRegistry = json.loads(data)
+ amUrl = sliderRegistry["payload"]["internalView"]["endpoints"]["org.apache.slider.agents"]["address"]
+ amHost = amUrl.split("/")[2].split(":")[0]
+ amSecuredPort = amUrl.split(":")[2].split("/")[0]
+ # the port needs to be utf-8 encoded
+ amSecuredPort = amSecuredPort.encode('utf8', 'ignore')
+ except Exception:
+ # log and let empty strings be returned
+ logger.error("Could not connect to zk registry at {} in quorum {}" %
+ (self.zk_reg_path, self.zk_quorum))
+ pass
+ finally:
+ zk.close()
+ logger.info("AM Host = %s, AM Secured Port = %s" % (amHost, amSecuredPort))
+ return amHost, amSecuredPort
http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/e682b8f0/slider-agent/src/main/python/agent/main.py
----------------------------------------------------------------------
diff --git a/slider-agent/src/main/python/agent/main.py b/slider-agent/src/main/python/agent/main.py
index 105df5a..f68db04 100644
--- a/slider-agent/src/main/python/agent/main.py
+++ b/slider-agent/src/main/python/agent/main.py
@@ -34,6 +34,8 @@ import posixpath
from Controller import Controller
from AgentConfig import AgentConfig
from NetUtil import NetUtil
+from Registry import Registry
+import Constants
logger = logging.getLogger()
IS_WINDOWS = platform.system() == "Windows"
@@ -178,9 +180,8 @@ def main():
parser = OptionParser()
parser.add_option("-v", "--verbose", dest="verbose", help="verbose log output", default=False)
parser.add_option("-l", "--label", dest="label", help="label of the agent", default=None)
- parser.add_option("--host", dest="host", help="AppMaster host", default=None)
- parser.add_option("--port", dest="port", help="AppMaster port", default=None)
- parser.add_option("--secured_port", dest="secured_port", help="AppMaster 2 Way port", default=None)
+ parser.add_option("--zk-quorum", dest=Constants.ZK_QUORUM, help="Zookeeper Quorum", default=None)
+ parser.add_option("--zk-reg-path", dest=Constants.ZK_REG_PATH, help="Zookeeper Registry Path", default=None)
parser.add_option("--debug", dest="debug", help="Agent debug hint", default="")
(options, args) = parser.parse_args()
@@ -207,18 +208,24 @@ def main():
update_config_from_file(agentConfig)
# update configurations if needed
- if options.host:
- agentConfig.set(AgentConfig.SERVER_SECTION, "hostname", options.host)
+ if options.zk_quorum:
+ agentConfig.set(AgentConfig.SERVER_SECTION, Constants.ZK_QUORUM, options.zk_quorum)
- if options.port:
- agentConfig.set(AgentConfig.SERVER_SECTION, "port", options.port)
-
- if options.secured_port:
- agentConfig.set(AgentConfig.SERVER_SECTION, "secured_port", options.secured_port)
+ if options.zk_reg_path:
+ agentConfig.set(AgentConfig.SERVER_SECTION, Constants.ZK_REG_PATH, options.zk_reg_path)
if options.debug:
agentConfig.set(AgentConfig.AGENT_SECTION, AgentConfig.APP_DBG_CMD, options.debug)
+ # Extract the AM hostname and secured port from ZK registry
+ registry = Registry(options.zk_quorum, options.zk_reg_path)
+ amHost, amSecuredPort = registry.readAMHostPort()
+ if amHost:
+ agentConfig.set(AgentConfig.SERVER_SECTION, "hostname", amHost)
+
+ if amSecuredPort:
+ agentConfig.set(AgentConfig.SERVER_SECTION, "secured_port", amSecuredPort)
+
# set the security directory to a subdirectory of the run dir
secDir = posixpath.join(agentConfig.getResolvedPath(AgentConfig.RUN_DIR), "security")
logger.info("Security/Keys directory: " + secDir)
@@ -243,7 +250,7 @@ def main():
server_url = SERVER_STATUS_URL.format(
agentConfig.get(AgentConfig.SERVER_SECTION, 'hostname'),
- agentConfig.get(AgentConfig.SERVER_SECTION, 'port'),
+ agentConfig.get(AgentConfig.SERVER_SECTION, 'secured_port'),
agentConfig.get(AgentConfig.SERVER_SECTION, 'check_path'))
print("Connecting to the server at " + server_url + "...")
logger.info('Connecting to the server at: ' + server_url)
http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/e682b8f0/slider-agent/src/main/python/kazoo/__init__.py
----------------------------------------------------------------------
diff --git a/slider-agent/src/main/python/kazoo/__init__.py b/slider-agent/src/main/python/kazoo/__init__.py
new file mode 100644
index 0000000..792d600
--- /dev/null
+++ b/slider-agent/src/main/python/kazoo/__init__.py
@@ -0,0 +1 @@
+#
http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/e682b8f0/slider-agent/src/main/python/kazoo/client.py
----------------------------------------------------------------------
diff --git a/slider-agent/src/main/python/kazoo/client.py b/slider-agent/src/main/python/kazoo/client.py
new file mode 100644
index 0000000..6e3a219
--- /dev/null
+++ b/slider-agent/src/main/python/kazoo/client.py
@@ -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)
http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/e682b8f0/slider-agent/src/main/python/kazoo/exceptions.py
----------------------------------------------------------------------
diff --git a/slider-agent/src/main/python/kazoo/exceptions.py b/slider-agent/src/main/python/kazoo/exceptions.py
new file mode 100644
index 0000000..8d4f0f3
--- /dev/null
+++ b/slider-agent/src/main/python/kazoo/exceptions.py
@@ -0,0 +1,199 @@
+"""Kazoo Exceptions"""
+from collections import defaultdict
+
+
+class KazooException(Exception):
+ """Base Kazoo exception that all other kazoo library exceptions
+ inherit from"""
+
+
+class ZookeeperError(KazooException):
+ """Base Zookeeper exception for errors originating from the
+ Zookeeper server"""
+
+
+class CancelledError(KazooException):
+ """Raised when a process is cancelled by another thread"""
+
+
+class ConfigurationError(KazooException):
+ """Raised if the configuration arguments to an object are
+ invalid"""
+
+
+class ZookeeperStoppedError(KazooException):
+ """Raised when the kazoo client stopped (and thus not connected)"""
+
+
+class ConnectionDropped(KazooException):
+ """Internal error for jumping out of loops"""
+
+
+class LockTimeout(KazooException):
+ """Raised if failed to acquire a lock.
+
+ .. versionadded:: 1.1
+ """
+
+
+class WriterNotClosedException(KazooException):
+ """Raised if the writer is unable to stop closing when requested.
+
+ .. versionadded:: 1.2
+ """
+
+
+def _invalid_error_code():
+ raise RuntimeError('Invalid error code')
+
+
+EXCEPTIONS = defaultdict(_invalid_error_code)
+
+
+def _zookeeper_exception(code):
+ def decorator(klass):
+ def create(*args, **kwargs):
+ return klass(args, kwargs)
+
+ EXCEPTIONS[code] = create
+ klass.code = code
+ return klass
+
+ return decorator
+
+
+@_zookeeper_exception(0)
+class RolledBackError(ZookeeperError):
+ pass
+
+
+@_zookeeper_exception(-1)
+class SystemZookeeperError(ZookeeperError):
+ pass
+
+
+@_zookeeper_exception(-2)
+class RuntimeInconsistency(ZookeeperError):
+ pass
+
+
+@_zookeeper_exception(-3)
+class DataInconsistency(ZookeeperError):
+ pass
+
+
+@_zookeeper_exception(-4)
+class ConnectionLoss(ZookeeperError):
+ pass
+
+
+@_zookeeper_exception(-5)
+class MarshallingError(ZookeeperError):
+ pass
+
+
+@_zookeeper_exception(-6)
+class UnimplementedError(ZookeeperError):
+ pass
+
+
+@_zookeeper_exception(-7)
+class OperationTimeoutError(ZookeeperError):
+ pass
+
+
+@_zookeeper_exception(-8)
+class BadArgumentsError(ZookeeperError):
+ pass
+
+
+@_zookeeper_exception(-100)
+class APIError(ZookeeperError):
+ pass
+
+
+@_zookeeper_exception(-101)
+class NoNodeError(ZookeeperError):
+ pass
+
+
+@_zookeeper_exception(-102)
+class NoAuthError(ZookeeperError):
+ pass
+
+
+@_zookeeper_exception(-103)
+class BadVersionError(ZookeeperError):
+ pass
+
+
+@_zookeeper_exception(-108)
+class NoChildrenForEphemeralsError(ZookeeperError):
+ pass
+
+
+@_zookeeper_exception(-110)
+class NodeExistsError(ZookeeperError):
+ pass
+
+
+@_zookeeper_exception(-111)
+class NotEmptyError(ZookeeperError):
+ pass
+
+
+@_zookeeper_exception(-112)
+class SessionExpiredError(ZookeeperError):
+ pass
+
+
+@_zookeeper_exception(-113)
+class InvalidCallbackError(ZookeeperError):
+ pass
+
+
+@_zookeeper_exception(-114)
+class InvalidACLError(ZookeeperError):
+ pass
+
+
+@_zookeeper_exception(-115)
+class AuthFailedError(ZookeeperError):
+ pass
+
+
+@_zookeeper_exception(-118)
+class SessionMovedError(ZookeeperError):
+ pass
+
+
+@_zookeeper_exception(-119)
+class NotReadOnlyCallError(ZookeeperError):
+ """An API call that is not read-only was used while connected to
+ a read-only server"""
+
+
+class ConnectionClosedError(SessionExpiredError):
+ """Connection is closed"""
+
+
+# BW Compat aliases for C lib style exceptions
+ConnectionLossException = ConnectionLoss
+MarshallingErrorException = MarshallingError
+SystemErrorException = SystemZookeeperError
+RuntimeInconsistencyException = RuntimeInconsistency
+DataInconsistencyException = DataInconsistency
+UnimplementedException = UnimplementedError
+OperationTimeoutException = OperationTimeoutError
+BadArgumentsException = BadArgumentsError
+ApiErrorException = APIError
+NoNodeException = NoNodeError
+NoAuthException = NoAuthError
+BadVersionException = BadVersionError
+NoChildrenForEphemeralsException = NoChildrenForEphemeralsError
+NodeExistsException = NodeExistsError
+InvalidACLException = InvalidACLError
+AuthFailedException = AuthFailedError
+NotEmptyException = NotEmptyError
+SessionExpiredException = SessionExpiredError
+InvalidCallbackException = InvalidCallbackError
http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/e682b8f0/slider-agent/src/main/python/kazoo/handlers/__init__.py
----------------------------------------------------------------------
diff --git a/slider-agent/src/main/python/kazoo/handlers/__init__.py b/slider-agent/src/main/python/kazoo/handlers/__init__.py
new file mode 100644
index 0000000..792d600
--- /dev/null
+++ b/slider-agent/src/main/python/kazoo/handlers/__init__.py
@@ -0,0 +1 @@
+#
http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/e682b8f0/slider-agent/src/main/python/kazoo/handlers/gevent.py
----------------------------------------------------------------------
diff --git a/slider-agent/src/main/python/kazoo/handlers/gevent.py b/slider-agent/src/main/python/kazoo/handlers/gevent.py
new file mode 100644
index 0000000..6e40cae
--- /dev/null
+++ b/slider-agent/src/main/python/kazoo/handlers/gevent.py
@@ -0,0 +1,161 @@
+"""A gevent based handler."""
+from __future__ import absolute_import
+
+import atexit
+import logging
+
+import gevent
+import gevent.event
+import gevent.queue
+import gevent.select
+import gevent.thread
+
+from gevent.queue import Empty
+from gevent.queue import Queue
+from gevent import socket
+try:
+ from gevent.lock import Semaphore, RLock
+except ImportError:
+ from gevent.coros import Semaphore, RLock
+
+from kazoo.handlers.utils import create_tcp_socket, create_tcp_connection
+
+_using_libevent = gevent.__version__.startswith('0.')
+
+log = logging.getLogger(__name__)
+
+_STOP = object()
+
+AsyncResult = gevent.event.AsyncResult
+
+
+class SequentialGeventHandler(object):
+ """Gevent handler for sequentially executing callbacks.
+
+ This handler executes callbacks in a sequential manner. A queue is
+ created for each of the callback events, so that each type of event
+ has its callback type run sequentially.
+
+ Each queue type has a greenlet worker that pulls the callback event
+ off the queue and runs it in the order the client sees it.
+
+ This split helps ensure that watch callbacks won't block session
+ re-establishment should the connection be lost during a Zookeeper
+ client call.
+
+ Watch callbacks should avoid blocking behavior as the next callback
+ of that type won't be run until it completes. If you need to block,
+ spawn a new greenlet and return immediately so callbacks can
+ proceed.
+
+ """
+ name = "sequential_gevent_handler"
+ sleep_func = staticmethod(gevent.sleep)
+
+ def __init__(self):
+ """Create a :class:`SequentialGeventHandler` instance"""
+ self.callback_queue = Queue()
+ self._running = False
+ self._async = None
+ self._state_change = Semaphore()
+ self._workers = []
+
+ class timeout_exception(gevent.event.Timeout):
+ def __init__(self, msg):
+ gevent.event.Timeout.__init__(self, exception=msg)
+
+ def _create_greenlet_worker(self, queue):
+ def greenlet_worker():
+ while True:
+ try:
+ func = queue.get()
+ if func is _STOP:
+ break
+ func()
+ except Empty:
+ continue
+ except Exception as exc:
+ log.warning("Exception in worker greenlet")
+ log.exception(exc)
+ return gevent.spawn(greenlet_worker)
+
+ def start(self):
+ """Start the greenlet workers."""
+ with self._state_change:
+ if self._running:
+ return
+
+ self._running = True
+
+ # Spawn our worker greenlets, we have
+ # - A callback worker for watch events to be called
+ for queue in (self.callback_queue,):
+ w = self._create_greenlet_worker(queue)
+ self._workers.append(w)
+ atexit.register(self.stop)
+
+ def stop(self):
+ """Stop the greenlet workers and empty all queues."""
+ with self._state_change:
+ if not self._running:
+ return
+
+ self._running = False
+
+ for queue in (self.callback_queue,):
+ queue.put(_STOP)
+
+ while self._workers:
+ worker = self._workers.pop()
+ worker.join()
+
+ # Clear the queues
+ self.callback_queue = Queue() # pragma: nocover
+
+ if hasattr(atexit, "unregister"):
+ atexit.unregister(self.stop)
+
+ def select(self, *args, **kwargs):
+ return gevent.select.select(*args, **kwargs)
+
+ def socket(self, *args, **kwargs):
+ return create_tcp_socket(socket)
+
+ def create_connection(self, *args, **kwargs):
+ return create_tcp_connection(socket, *args, **kwargs)
+
+ def event_object(self):
+ """Create an appropriate Event object"""
+ return gevent.event.Event()
+
+ def lock_object(self):
+ """Create an appropriate Lock object"""
+ return gevent.thread.allocate_lock()
+
+ def rlock_object(self):
+ """Create an appropriate RLock object"""
+ return RLock()
+
+ def async_result(self):
+ """Create a :class:`AsyncResult` instance
+
+ The :class:`AsyncResult` instance will have its completion
+ callbacks executed in the thread the
+ :class:`SequentialGeventHandler` is created in (which should be
+ the gevent/main thread).
+
+ """
+ return AsyncResult()
+
+ def spawn(self, func, *args, **kwargs):
+ """Spawn a function to run asynchronously"""
+ return gevent.spawn(func, *args, **kwargs)
+
+ def dispatch_callback(self, callback):
+ """Dispatch to the callback object
+
+ The callback is put on separate queues to run depending on the
+ type as documented for the :class:`SequentialGeventHandler`.
+
+ """
+ self.callback_queue.put(lambda: callback.func(*callback.args))