client.py 60 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703
  1. """Kazoo Zookeeper Client"""
  2. from collections import defaultdict, deque
  3. from functools import partial
  4. import inspect
  5. import logging
  6. from os.path import split
  7. import re
  8. import warnings
  9. import six
  10. from kazoo.exceptions import (
  11. AuthFailedError,
  12. ConfigurationError,
  13. ConnectionClosedError,
  14. ConnectionLoss,
  15. KazooException,
  16. NoNodeError,
  17. NodeExistsError,
  18. SessionExpiredError,
  19. WriterNotClosedException,
  20. )
  21. from kazoo.handlers.threading import SequentialThreadingHandler
  22. from kazoo.handlers.utils import capture_exceptions, wrap
  23. from kazoo.hosts import collect_hosts
  24. from kazoo.loggingsupport import BLATHER
  25. from kazoo.protocol.connection import ConnectionHandler
  26. from kazoo.protocol.paths import _prefix_root, normpath
  27. from kazoo.protocol.serialization import (
  28. Auth,
  29. CheckVersion,
  30. CloseInstance,
  31. Create,
  32. Create2,
  33. Delete,
  34. Exists,
  35. GetChildren,
  36. GetChildren2,
  37. GetACL,
  38. SetACL,
  39. GetData,
  40. Reconfig,
  41. SetData,
  42. Sync,
  43. Transaction
  44. )
  45. from kazoo.protocol.states import (
  46. Callback,
  47. EventType,
  48. KazooState,
  49. KeeperState,
  50. WatchedEvent
  51. )
  52. from kazoo.retry import KazooRetry
  53. from kazoo.security import ACL, OPEN_ACL_UNSAFE
  54. # convenience API
  55. from kazoo.recipe.barrier import Barrier, DoubleBarrier
  56. from kazoo.recipe.counter import Counter
  57. from kazoo.recipe.election import Election
  58. from kazoo.recipe.lease import NonBlockingLease, MultiNonBlockingLease
  59. from kazoo.recipe.lock import Lock, ReadLock, WriteLock, Semaphore
  60. from kazoo.recipe.partitioner import SetPartitioner
  61. from kazoo.recipe.party import Party, ShallowParty
  62. from kazoo.recipe.queue import Queue, LockingQueue
  63. from kazoo.recipe.watchers import ChildrenWatch, DataWatch
  64. string_types = six.string_types
  65. bytes_types = (six.binary_type,)
  66. CLOSED_STATES = (
  67. KeeperState.EXPIRED_SESSION,
  68. KeeperState.AUTH_FAILED,
  69. KeeperState.CLOSED
  70. )
  71. ENVI_VERSION = re.compile(r'([\d\.]*).*', re.DOTALL)
  72. ENVI_VERSION_KEY = 'zookeeper.version'
  73. log = logging.getLogger(__name__)
  74. _RETRY_COMPAT_DEFAULTS = dict(
  75. max_retries=None,
  76. retry_delay=0.1,
  77. retry_backoff=2,
  78. retry_max_delay=3600,
  79. )
  80. _RETRY_COMPAT_MAPPING = dict(
  81. max_retries='max_tries',
  82. retry_delay='delay',
  83. retry_backoff='backoff',
  84. retry_max_delay='max_delay',
  85. )
  86. class KazooClient(object):
  87. """An Apache Zookeeper Python client supporting alternate callback
  88. handlers and high-level functionality.
  89. Watch functions registered with this class will not get session
  90. events, unlike the default Zookeeper watches. They will also be
  91. called with a single argument, a
  92. :class:`~kazoo.protocol.states.WatchedEvent` instance.
  93. """
  94. def __init__(self, hosts='127.0.0.1:2181',
  95. timeout=10.0, client_id=None, handler=None,
  96. default_acl=None, auth_data=None, sasl_options=None,
  97. read_only=None, randomize_hosts=True, connection_retry=None,
  98. command_retry=None, logger=None, keyfile=None,
  99. keyfile_password=None, certfile=None, ca=None,
  100. use_ssl=False, verify_certs=True, **kwargs):
  101. """Create a :class:`KazooClient` instance. All time arguments
  102. are in seconds.
  103. :param hosts: Comma-separated list of hosts to connect to
  104. (e.g. 127.0.0.1:2181,127.0.0.1:2182,[::1]:2183).
  105. :param timeout: The longest to wait for a Zookeeper connection.
  106. :param client_id: A Zookeeper client id, used when
  107. re-establishing a prior session connection.
  108. :param handler: An instance of a class implementing the
  109. :class:`~kazoo.interfaces.IHandler` interface
  110. for callback handling.
  111. :param default_acl: A default ACL used on node creation.
  112. :param auth_data:
  113. A list of authentication credentials to use for the
  114. connection. Should be a list of (scheme, credential)
  115. tuples as :meth:`add_auth` takes.
  116. :param sasl_options:
  117. SASL options for the connection, if SASL support is to be used.
  118. Should be a dict of SASL options passed to the underlying
  119. `pure-sasl <https://pypi.org/project/pure-sasl>`_ library.
  120. For example using the DIGEST-MD5 mechnism:
  121. .. code-block:: python
  122. sasl_options = {
  123. 'mechanism': 'DIGEST-MD5',
  124. 'username': 'myusername',
  125. 'password': 'mypassword'
  126. }
  127. For GSSAPI, using the running process' ticket cache:
  128. .. code-block:: python
  129. sasl_options = {
  130. 'mechanism': 'GSSAPI',
  131. 'service': 'myzk', # optional
  132. 'principal': 'client@EXAMPLE.COM' # optional
  133. }
  134. :param read_only: Allow connections to read only servers.
  135. :param randomize_hosts: By default randomize host selection.
  136. :param connection_retry:
  137. A :class:`kazoo.retry.KazooRetry` object to use for
  138. retrying the connection to Zookeeper. Also can be a dict of
  139. options which will be used for creating one.
  140. :param command_retry:
  141. A :class:`kazoo.retry.KazooRetry` object to use for
  142. the :meth:`KazooClient.retry` method. Also can be a dict of
  143. options which will be used for creating one.
  144. :param logger: A custom logger to use instead of the module
  145. global `log` instance.
  146. :param keyfile: SSL keyfile to use for authentication
  147. :param keyfile_password: SSL keyfile password
  148. :param certfile: SSL certfile to use for authentication
  149. :param ca: SSL CA file to use for authentication
  150. :param use_ssl: argument to control whether SSL is used or not
  151. :param verify_certs: when using SSL, argument to bypass
  152. certs verification
  153. Basic Example:
  154. .. code-block:: python
  155. zk = KazooClient()
  156. zk.start()
  157. children = zk.get_children('/')
  158. zk.stop()
  159. As a convenience all recipe classes are available as attributes
  160. and get automatically bound to the client. For example::
  161. zk = KazooClient()
  162. zk.start()
  163. lock = zk.Lock('/lock_path')
  164. .. versionadded:: 0.6
  165. The read_only option. Requires Zookeeper 3.4+
  166. .. versionadded:: 0.6
  167. The retry_max_delay option.
  168. .. versionadded:: 0.6
  169. The randomize_hosts option.
  170. .. versionchanged:: 0.8
  171. Removed the unused watcher argument (was second argument).
  172. .. versionadded:: 1.2
  173. The connection_retry, command_retry and logger options.
  174. .. versionadded:: 2.7
  175. The sasl_options option.
  176. """
  177. self.logger = logger or log
  178. # Record the handler strategy used
  179. self.handler = handler if handler else SequentialThreadingHandler()
  180. if inspect.isclass(self.handler):
  181. raise ConfigurationError("Handler must be an instance of a class, "
  182. "not the class: %s" % self.handler)
  183. self.auth_data = auth_data if auth_data else set([])
  184. self.default_acl = default_acl
  185. self.randomize_hosts = randomize_hosts
  186. self.hosts = None
  187. self.chroot = None
  188. self.set_hosts(hosts)
  189. self.use_ssl = use_ssl
  190. self.verify_certs = verify_certs
  191. self.certfile = certfile
  192. self.keyfile = keyfile
  193. self.keyfile_password = keyfile_password
  194. self.ca = ca
  195. # Curator like simplified state tracking, and listeners for
  196. # state transitions
  197. self._state = KeeperState.CLOSED
  198. self.state = KazooState.LOST
  199. self.state_listeners = set()
  200. self._child_watchers = defaultdict(set)
  201. self._data_watchers = defaultdict(set)
  202. self._reset()
  203. self.read_only = read_only
  204. if client_id:
  205. self._session_id = client_id[0]
  206. self._session_passwd = client_id[1]
  207. else:
  208. self._reset_session()
  209. # ZK uses milliseconds
  210. self._session_timeout = int(timeout * 1000)
  211. # We use events like twitter's client to track current and
  212. # desired state (connected, and whether to shutdown)
  213. self._live = self.handler.event_object()
  214. self._writer_stopped = self.handler.event_object()
  215. self._stopped = self.handler.event_object()
  216. self._stopped.set()
  217. self._writer_stopped.set()
  218. self.retry = self._conn_retry = None
  219. if type(connection_retry) is dict:
  220. self._conn_retry = KazooRetry(**connection_retry)
  221. elif type(connection_retry) is KazooRetry:
  222. self._conn_retry = connection_retry
  223. if type(command_retry) is dict:
  224. self.retry = KazooRetry(**command_retry)
  225. elif type(command_retry) is KazooRetry:
  226. self.retry = command_retry
  227. if type(self._conn_retry) is KazooRetry:
  228. if self.handler.sleep_func != self._conn_retry.sleep_func:
  229. raise ConfigurationError("Retry handler and event handler "
  230. " must use the same sleep func")
  231. if type(self.retry) is KazooRetry:
  232. if self.handler.sleep_func != self.retry.sleep_func:
  233. raise ConfigurationError(
  234. "Command retry handler and event handler "
  235. "must use the same sleep func")
  236. if self.retry is None or self._conn_retry is None:
  237. old_retry_keys = dict(_RETRY_COMPAT_DEFAULTS)
  238. for key in old_retry_keys:
  239. try:
  240. old_retry_keys[key] = kwargs.pop(key)
  241. warnings.warn(
  242. 'Passing retry configuration param %s to the '
  243. 'client directly is deprecated, please pass a '
  244. 'configured retry object (using param %s)' % (
  245. key, _RETRY_COMPAT_MAPPING[key]),
  246. DeprecationWarning, stacklevel=2)
  247. except KeyError:
  248. pass
  249. retry_keys = {}
  250. for oldname, value in old_retry_keys.items():
  251. retry_keys[_RETRY_COMPAT_MAPPING[oldname]] = value
  252. if self._conn_retry is None:
  253. self._conn_retry = KazooRetry(
  254. sleep_func=self.handler.sleep_func,
  255. **retry_keys)
  256. if self.retry is None:
  257. self.retry = KazooRetry(
  258. sleep_func=self.handler.sleep_func,
  259. **retry_keys)
  260. # Managing legacy SASL options
  261. for scheme, auth in self.auth_data:
  262. if scheme != 'sasl':
  263. continue
  264. if sasl_options:
  265. raise ConfigurationError(
  266. 'Multiple SASL configurations provided'
  267. )
  268. warnings.warn(
  269. 'Passing SASL configuration as part of the auth_data is '
  270. 'deprecated, please use the sasl_options configuration '
  271. 'instead', DeprecationWarning, stacklevel=2
  272. )
  273. username, password = auth.split(':')
  274. # Generate an equivalent SASL configuration
  275. sasl_options = {
  276. 'username': username,
  277. 'password': password,
  278. 'mechanism': 'DIGEST-MD5',
  279. 'service': 'zookeeper',
  280. 'principal': 'zk-sasl-md5',
  281. }
  282. # Cleanup
  283. self.auth_data = set([
  284. (scheme, auth)
  285. for scheme, auth in self.auth_data
  286. if scheme != 'sasl'
  287. ])
  288. self._conn_retry.interrupt = lambda: self._stopped.is_set()
  289. self._connection = ConnectionHandler(
  290. self, self._conn_retry.copy(), logger=self.logger,
  291. sasl_options=sasl_options)
  292. # Every retry call should have its own copy of the retry helper
  293. # to avoid shared retry counts
  294. self._retry = self.retry
  295. def _retry(*args, **kwargs):
  296. return self._retry.copy()(*args, **kwargs)
  297. self.retry = _retry
  298. self.Barrier = partial(Barrier, self)
  299. self.Counter = partial(Counter, self)
  300. self.DoubleBarrier = partial(DoubleBarrier, self)
  301. self.ChildrenWatch = partial(ChildrenWatch, self)
  302. self.DataWatch = partial(DataWatch, self)
  303. self.Election = partial(Election, self)
  304. self.NonBlockingLease = partial(NonBlockingLease, self)
  305. self.MultiNonBlockingLease = partial(MultiNonBlockingLease, self)
  306. self.Lock = partial(Lock, self)
  307. self.ReadLock = partial(ReadLock, self)
  308. self.WriteLock = partial(WriteLock, self)
  309. self.Party = partial(Party, self)
  310. self.Queue = partial(Queue, self)
  311. self.LockingQueue = partial(LockingQueue, self)
  312. self.SetPartitioner = partial(SetPartitioner, self)
  313. self.Semaphore = partial(Semaphore, self)
  314. self.ShallowParty = partial(ShallowParty, self)
  315. # If we got any unhandled keywords, complain like Python would
  316. if kwargs:
  317. raise TypeError('__init__() got unexpected keyword arguments: %s'
  318. % (kwargs.keys(),))
  319. def _reset(self):
  320. """Resets a variety of client states for a new connection."""
  321. self._queue = deque()
  322. self._pending = deque()
  323. self._reset_watchers()
  324. self._reset_session()
  325. self.last_zxid = 0
  326. self._protocol_version = None
  327. def _reset_watchers(self):
  328. watchers = []
  329. for child_watchers in six.itervalues(self._child_watchers):
  330. watchers.extend(child_watchers)
  331. for data_watchers in six.itervalues(self._data_watchers):
  332. watchers.extend(data_watchers)
  333. self._child_watchers = defaultdict(set)
  334. self._data_watchers = defaultdict(set)
  335. ev = WatchedEvent(EventType.NONE, self._state, None)
  336. for watch in watchers:
  337. self.handler.dispatch_callback(Callback("watch", watch, (ev,)))
  338. def _reset_session(self):
  339. self._session_id = None
  340. self._session_passwd = b'\x00' * 16
  341. @property
  342. def client_state(self):
  343. """Returns the last Zookeeper client state
  344. This is the non-simplified state information and is generally
  345. not as useful as the simplified KazooState information.
  346. """
  347. return self._state
  348. @property
  349. def client_id(self):
  350. """Returns the client id for this Zookeeper session if
  351. connected.
  352. :returns: client id which consists of the session id and
  353. password.
  354. :rtype: tuple
  355. """
  356. if self._live.is_set():
  357. return (self._session_id, self._session_passwd)
  358. return None
  359. @property
  360. def connected(self):
  361. """Returns whether the Zookeeper connection has been
  362. established."""
  363. return self._live.is_set()
  364. def set_hosts(self, hosts, randomize_hosts=None):
  365. """ sets the list of hosts used by this client.
  366. This function accepts the same format hosts parameter as the init
  367. function and sets the client to use the new hosts the next time it
  368. needs to look up a set of hosts. This function does not affect the
  369. current connected status.
  370. It is not currently possible to change the chroot with this function,
  371. setting a host list with a new chroot will raise a ConfigurationError.
  372. :param hosts: see description in :meth:`KazooClient.__init__`
  373. :param randomize_hosts: override client default for host randomization
  374. :raises:
  375. :exc:`ConfigurationError` if the hosts argument changes the chroot
  376. .. versionadded:: 1.4
  377. .. warning::
  378. Using this function to point a client to a completely disparate
  379. zookeeper server cluster has undefined behavior.
  380. """
  381. # Change the client setting for randomization if specified
  382. if randomize_hosts is not None:
  383. self.randomize_hosts = randomize_hosts
  384. # Randomizing the list will be done at connect time
  385. self.hosts, chroot = collect_hosts(hosts)
  386. if chroot:
  387. new_chroot = normpath(chroot)
  388. else:
  389. new_chroot = ''
  390. if self.chroot is not None and new_chroot != self.chroot:
  391. raise ConfigurationError("Changing chroot at runtime is not "
  392. "currently supported")
  393. self.chroot = new_chroot
  394. def add_listener(self, listener):
  395. """Add a function to be called for connection state changes.
  396. This function will be called with a
  397. :class:`~kazoo.protocol.states.KazooState` instance indicating
  398. the new connection state on state transitions.
  399. .. warning::
  400. This function must not block. If its at all likely that it
  401. might need data or a value that could result in blocking
  402. than the :meth:`~kazoo.interfaces.IHandler.spawn` method
  403. should be used so that the listener can return immediately.
  404. """
  405. if not (listener and callable(listener)):
  406. raise ConfigurationError("listener must be callable")
  407. self.state_listeners.add(listener)
  408. def remove_listener(self, listener):
  409. """Remove a listener function"""
  410. self.state_listeners.discard(listener)
  411. def _make_state_change(self, state):
  412. # skip if state is current
  413. if self.state == state:
  414. return
  415. self.state = state
  416. # Create copy of listeners for iteration in case one needs to
  417. # remove itself
  418. for listener in list(self.state_listeners):
  419. try:
  420. remove = listener(state)
  421. if remove is True:
  422. self.remove_listener(listener)
  423. except Exception:
  424. self.logger.exception("Error in connection state listener")
  425. def _session_callback(self, state):
  426. if state == self._state:
  427. return
  428. # Note that we don't check self.state == LOST since that's also
  429. # the client's initial state
  430. closed_state = self._state in CLOSED_STATES
  431. self._state = state
  432. # If we were previously closed or had an expired session, and
  433. # are now connecting, don't bother with the rest of the
  434. # transitions since they only apply after
  435. # we've established a connection
  436. if closed_state and state == KeeperState.CONNECTING:
  437. self.logger.log(BLATHER, "Skipping state change")
  438. return
  439. if state in (KeeperState.CONNECTED, KeeperState.CONNECTED_RO):
  440. self.logger.info("Zookeeper connection established, "
  441. "state: %s", state)
  442. self._live.set()
  443. self._make_state_change(KazooState.CONNECTED)
  444. elif state in CLOSED_STATES:
  445. self.logger.info("Zookeeper session closed, state: %s", state)
  446. self._live.clear()
  447. self._make_state_change(KazooState.LOST)
  448. self._notify_pending(state)
  449. self._reset()
  450. else:
  451. self.logger.info("Zookeeper connection lost")
  452. # Connection lost
  453. self._live.clear()
  454. self._notify_pending(state)
  455. self._make_state_change(KazooState.SUSPENDED)
  456. self._reset_watchers()
  457. def _notify_pending(self, state):
  458. """Used to clear a pending response queue and request queue
  459. during connection drops."""
  460. if state == KeeperState.AUTH_FAILED:
  461. exc = AuthFailedError()
  462. elif state == KeeperState.EXPIRED_SESSION:
  463. exc = SessionExpiredError()
  464. else:
  465. exc = ConnectionLoss()
  466. while True:
  467. try:
  468. request, async_object, xid = self._pending.popleft()
  469. if async_object:
  470. async_object.set_exception(exc)
  471. except IndexError:
  472. break
  473. while True:
  474. try:
  475. request, async_object = self._queue.popleft()
  476. if async_object:
  477. async_object.set_exception(exc)
  478. except IndexError:
  479. break
  480. def _safe_close(self):
  481. self.handler.stop()
  482. timeout = self._session_timeout // 1000
  483. if timeout < 10:
  484. timeout = 10
  485. if not self._connection.stop(timeout):
  486. raise WriterNotClosedException(
  487. "Writer still open from prior connection "
  488. "and wouldn't close after %s seconds" % timeout)
  489. def _call(self, request, async_object):
  490. """Ensure the client is in CONNECTED or SUSPENDED state and put the
  491. request in the queue if it is.
  492. Returns False if the call short circuits due to AUTH_FAILED,
  493. CLOSED, or EXPIRED_SESSION state.
  494. """
  495. if self._state == KeeperState.AUTH_FAILED:
  496. async_object.set_exception(AuthFailedError())
  497. return False
  498. elif self._state == KeeperState.CLOSED:
  499. async_object.set_exception(ConnectionClosedError(
  500. "Connection has been closed"))
  501. return False
  502. elif self._state == KeeperState.EXPIRED_SESSION:
  503. async_object.set_exception(SessionExpiredError())
  504. return False
  505. self._queue.append((request, async_object))
  506. # wake the connection, guarding against a race with close()
  507. write_sock = self._connection._write_sock
  508. if write_sock is None:
  509. async_object.set_exception(ConnectionClosedError(
  510. "Connection has been closed"))
  511. try:
  512. write_sock.send(b'\0')
  513. except: # NOQA
  514. async_object.set_exception(ConnectionClosedError(
  515. "Connection has been closed"))
  516. def start(self, timeout=15):
  517. """Initiate connection to ZK.
  518. :param timeout: Time in seconds to wait for connection to
  519. succeed.
  520. :raises: :attr:`~kazoo.interfaces.IHandler.timeout_exception`
  521. if the connection wasn't established within `timeout`
  522. seconds.
  523. """
  524. event = self.start_async()
  525. event.wait(timeout=timeout)
  526. if not self.connected:
  527. # We time-out, ensure we are disconnected
  528. self.stop()
  529. self.close()
  530. raise self.handler.timeout_exception("Connection time-out")
  531. if self.chroot and not self.exists("/"):
  532. warnings.warn("No chroot path exists, the chroot path "
  533. "should be created before normal use.")
  534. def start_async(self):
  535. """Asynchronously initiate connection to ZK.
  536. :returns: An event object that can be checked to see if the
  537. connection is alive.
  538. :rtype: :class:`~threading.Event` compatible object.
  539. """
  540. # If we're already connected, ignore
  541. if self._live.is_set():
  542. return self._live
  543. # Make sure we're safely closed
  544. self._safe_close()
  545. # We've been asked to connect, clear the stop and our writer
  546. # thread indicator
  547. self._stopped.clear()
  548. self._writer_stopped.clear()
  549. # Start the handler
  550. self.handler.start()
  551. # Start the connection
  552. self._connection.start()
  553. return self._live
  554. def stop(self):
  555. """Gracefully stop this Zookeeper session.
  556. This method can be called while a reconnection attempt is in
  557. progress, which will then be halted.
  558. Once the connection is closed, its session becomes invalid. All
  559. the ephemeral nodes in the ZooKeeper server associated with the
  560. session will be removed. The watches left on those nodes (and
  561. on their parents) will be triggered.
  562. """
  563. if self._stopped.is_set():
  564. return
  565. self._stopped.set()
  566. self._queue.append((CloseInstance, None))
  567. try:
  568. self._connection._write_sock.send(b'\0')
  569. finally:
  570. self._safe_close()
  571. def restart(self):
  572. """Stop and restart the Zookeeper session."""
  573. self.stop()
  574. self.start()
  575. def close(self):
  576. """Free any resources held by the client.
  577. This method should be called on a stopped client before it is
  578. discarded. Not doing so may result in filehandles being leaked.
  579. .. versionadded:: 1.0
  580. """
  581. self._connection.close()
  582. def command(self, cmd=b'ruok'):
  583. """Sent a management command to the current ZK server.
  584. Examples are `ruok`, `envi` or `stat`.
  585. :returns: An unstructured textual response.
  586. :rtype: str
  587. :raises:
  588. :exc:`ConnectionLoss` if there is no connection open, or
  589. possibly a :exc:`socket.error` if there's a problem with
  590. the connection used just for this command.
  591. .. versionadded:: 0.5
  592. """
  593. if not self._live.is_set():
  594. raise ConnectionLoss("No connection to server")
  595. peer = self._connection._socket.getpeername()[:2]
  596. sock = self.handler.create_connection(
  597. peer, timeout=self._session_timeout / 1000.0,
  598. use_ssl=self.use_ssl,
  599. ca=self.ca,
  600. certfile=self.certfile,
  601. keyfile=self.keyfile,
  602. keyfile_password=self.keyfile_password,
  603. verify_certs=self.verify_certs,
  604. )
  605. sock.sendall(cmd)
  606. result = sock.recv(8192)
  607. sock.close()
  608. return result.decode('utf-8', 'replace')
  609. def server_version(self, retries=3):
  610. """Get the version of the currently connected ZK server.
  611. :returns: The server version, for example (3, 4, 3).
  612. :rtype: tuple
  613. .. versionadded:: 0.5
  614. """
  615. def _try_fetch():
  616. data = self.command(b'envi')
  617. data_parsed = {}
  618. for line in data.splitlines():
  619. try:
  620. k, v = line.split("=", 1)
  621. k = k.strip()
  622. v = v.strip()
  623. except ValueError:
  624. pass
  625. else:
  626. if k:
  627. data_parsed[k] = v
  628. version = data_parsed.get(ENVI_VERSION_KEY, '')
  629. version_digits = ENVI_VERSION.match(version).group(1)
  630. try:
  631. return tuple([int(d) for d in version_digits.split('.')])
  632. except ValueError:
  633. return None
  634. def _is_valid(version):
  635. # All zookeeper versions should have at least major.minor
  636. # version numbers; if we get one that doesn't it is likely not
  637. # correct and was truncated...
  638. if version and len(version) > 1:
  639. return True
  640. return False
  641. # Try 1 + retries amount of times to get a version that we know
  642. # will likely be acceptable...
  643. version = _try_fetch()
  644. if _is_valid(version):
  645. return version
  646. for _i in six.moves.range(0, retries):
  647. version = _try_fetch()
  648. if _is_valid(version):
  649. return version
  650. raise KazooException("Unable to fetch useable server"
  651. " version after trying %s times"
  652. % (1 + max(0, retries)))
  653. def add_auth(self, scheme, credential):
  654. """Send credentials to server.
  655. :param scheme: authentication scheme (default supported:
  656. "digest").
  657. :param credential: the credential -- value depends on scheme.
  658. :returns: True if it was successful.
  659. :rtype: bool
  660. :raises:
  661. :exc:`~kazoo.exceptions.AuthFailedError` if it failed though
  662. the session state will be set to AUTH_FAILED as well.
  663. """
  664. return self.add_auth_async(scheme, credential).get()
  665. def add_auth_async(self, scheme, credential):
  666. """Asynchronously send credentials to server. Takes the same
  667. arguments as :meth:`add_auth`.
  668. :rtype: :class:`~kazoo.interfaces.IAsyncResult`
  669. """
  670. if not isinstance(scheme, string_types):
  671. raise TypeError("Invalid type for 'scheme' (string expected)")
  672. if not isinstance(credential, string_types):
  673. raise TypeError("Invalid type for 'credential' (string expected)")
  674. # we need this auth data to re-authenticate on reconnect
  675. self.auth_data.add((scheme, credential))
  676. async_result = self.handler.async_result()
  677. self._call(Auth(0, scheme, credential), async_result)
  678. return async_result
  679. def unchroot(self, path):
  680. """Strip the chroot if applicable from the path."""
  681. if not self.chroot:
  682. return path
  683. if self.chroot == path:
  684. return "/"
  685. if path.startswith(self.chroot):
  686. return path[len(self.chroot):]
  687. else:
  688. return path
  689. def sync_async(self, path):
  690. """Asynchronous sync.
  691. :rtype: :class:`~kazoo.interfaces.IAsyncResult`
  692. """
  693. async_result = self.handler.async_result()
  694. @wrap(async_result)
  695. def _sync_completion(result):
  696. return self.unchroot(result.get())
  697. def _do_sync():
  698. result = self.handler.async_result()
  699. self._call(
  700. Sync(_prefix_root(self.chroot, path)),
  701. result
  702. )
  703. result.rawlink(_sync_completion)
  704. _do_sync()
  705. return async_result
  706. def sync(self, path):
  707. """Sync, blocks until response is acknowledged.
  708. Flushes channel between process and leader.
  709. :param path: path of node.
  710. :returns: The node path that was synced.
  711. :raises:
  712. :exc:`~kazoo.exceptions.ZookeeperError` if the server
  713. returns a non-zero error code.
  714. .. versionadded:: 0.5
  715. """
  716. return self.sync_async(path).get()
  717. def create(self, path, value=b"", acl=None, ephemeral=False,
  718. sequence=False, makepath=False, include_data=False):
  719. """Create a node with the given value as its data. Optionally
  720. set an ACL on the node.
  721. The ephemeral and sequence arguments determine the type of the
  722. node.
  723. An ephemeral node will be automatically removed by ZooKeeper
  724. when the session associated with the creation of the node
  725. expires.
  726. A sequential node will be given the specified path plus a
  727. suffix `i` where i is the current sequential number of the
  728. node. The sequence number is always fixed length of 10 digits,
  729. 0 padded. Once such a node is created, the sequential number
  730. will be incremented by one.
  731. If a node with the same actual path already exists in
  732. ZooKeeper, a NodeExistsError will be raised. Note that since a
  733. different actual path is used for each invocation of creating
  734. sequential nodes with the same path argument, the call will
  735. never raise NodeExistsError.
  736. If the parent node does not exist in ZooKeeper, a NoNodeError
  737. will be raised. Setting the optional `makepath` argument to
  738. `True` will create all missing parent nodes instead.
  739. An ephemeral node cannot have children. If the parent node of
  740. the given path is ephemeral, a NoChildrenForEphemeralsError
  741. will be raised.
  742. This operation, if successful, will trigger all the watches
  743. left on the node of the given path by :meth:`exists` and
  744. :meth:`get` API calls, and the watches left on the parent node
  745. by :meth:`get_children` API calls.
  746. The maximum allowable size of the node value is 1 MB. Values
  747. larger than this will cause a ZookeeperError to be raised.
  748. :param path: Path of node.
  749. :param value: Initial bytes value of node.
  750. :param acl: :class:`~kazoo.security.ACL` list.
  751. :param ephemeral: Boolean indicating whether node is ephemeral
  752. (tied to this session).
  753. :param sequence: Boolean indicating whether path is suffixed
  754. with a unique index.
  755. :param makepath: Whether the path should be created if it
  756. doesn't exist.
  757. :param include_data:
  758. Include the :class:`~kazoo.protocol.states.ZnodeStat` of
  759. the node in addition to its real path. This option changes
  760. the return value to be a tuple of (path, stat).
  761. :returns: Real path of the new node, or tuple if `include_data`
  762. is `True`
  763. :rtype: str
  764. :raises:
  765. :exc:`~kazoo.exceptions.NodeExistsError` if the node
  766. already exists.
  767. :exc:`~kazoo.exceptions.NoNodeError` if parent nodes are
  768. missing.
  769. :exc:`~kazoo.exceptions.NoChildrenForEphemeralsError` if
  770. the parent node is an ephemeral node.
  771. :exc:`~kazoo.exceptions.ZookeeperError` if the provided
  772. value is too large.
  773. :exc:`~kazoo.exceptions.ZookeeperError` if the server
  774. returns a non-zero error code.
  775. .. versionadded:: 1.1
  776. The `makepath` option.
  777. .. versionadded:: 2.7
  778. The `include_data` option.
  779. """
  780. acl = acl or self.default_acl
  781. return self.create_async(
  782. path, value, acl=acl, ephemeral=ephemeral,
  783. sequence=sequence, makepath=makepath, include_data=include_data
  784. ).get()
  785. def create_async(self, path, value=b"", acl=None, ephemeral=False,
  786. sequence=False, makepath=False, include_data=False):
  787. """Asynchronously create a ZNode. Takes the same arguments as
  788. :meth:`create`.
  789. :rtype: :class:`~kazoo.interfaces.IAsyncResult`
  790. .. versionadded:: 1.1
  791. The makepath option.
  792. .. versionadded:: 2.7
  793. The `include_data` option.
  794. """
  795. if acl is None and self.default_acl:
  796. acl = self.default_acl
  797. if not isinstance(path, string_types):
  798. raise TypeError("Invalid type for 'path' (string expected)")
  799. if acl and (isinstance(acl, ACL) or
  800. not isinstance(acl, (tuple, list))):
  801. raise TypeError("Invalid type for 'acl' (acl must be a tuple/list"
  802. " of ACL's")
  803. if value is not None and not isinstance(value, bytes_types):
  804. raise TypeError("Invalid type for 'value' (must be a byte string)")
  805. if not isinstance(ephemeral, bool):
  806. raise TypeError("Invalid type for 'ephemeral' (bool expected)")
  807. if not isinstance(sequence, bool):
  808. raise TypeError("Invalid type for 'sequence' (bool expected)")
  809. if not isinstance(makepath, bool):
  810. raise TypeError("Invalid type for 'makepath' (bool expected)")
  811. if not isinstance(include_data, bool):
  812. raise TypeError("Invalid type for 'include_data' (bool expected)")
  813. flags = 0
  814. if ephemeral:
  815. flags |= 1
  816. if sequence:
  817. flags |= 2
  818. if acl is None:
  819. acl = OPEN_ACL_UNSAFE
  820. async_result = self.handler.async_result()
  821. @capture_exceptions(async_result)
  822. def do_create():
  823. result = self._create_async_inner(
  824. path, value, acl, flags,
  825. trailing=sequence, include_data=include_data
  826. )
  827. result.rawlink(create_completion)
  828. @capture_exceptions(async_result)
  829. def retry_completion(result):
  830. result.get()
  831. do_create()
  832. @wrap(async_result)
  833. def create_completion(result):
  834. try:
  835. if include_data:
  836. new_path, stat = result.get()
  837. return self.unchroot(new_path), stat
  838. else:
  839. return self.unchroot(result.get())
  840. except NoNodeError:
  841. if not makepath:
  842. raise
  843. if sequence and path.endswith('/'):
  844. parent = path.rstrip('/')
  845. else:
  846. parent, _ = split(path)
  847. self.ensure_path_async(parent, acl).rawlink(retry_completion)
  848. do_create()
  849. return async_result
  850. def _create_async_inner(self, path, value, acl, flags,
  851. trailing=False, include_data=False):
  852. async_result = self.handler.async_result()
  853. if include_data:
  854. opcode = Create2
  855. else:
  856. opcode = Create
  857. call_result = self._call(
  858. opcode(_prefix_root(self.chroot, path, trailing=trailing),
  859. value, acl, flags), async_result)
  860. if call_result is False:
  861. # We hit a short-circuit exit on the _call. Because we are
  862. # not using the original async_result here, we bubble the
  863. # exception upwards to the do_create function in
  864. # KazooClient.create so that it gets set on the correct
  865. # async_result object
  866. raise async_result.exception
  867. return async_result
  868. def ensure_path(self, path, acl=None):
  869. """Recursively create a path if it doesn't exist.
  870. :param path: Path of node.
  871. :param acl: Permissions for node.
  872. """
  873. return self.ensure_path_async(path, acl).get()
  874. def ensure_path_async(self, path, acl=None):
  875. """Recursively create a path asynchronously if it doesn't
  876. exist. Takes the same arguments as :meth:`ensure_path`.
  877. :rtype: :class:`~kazoo.interfaces.IAsyncResult`
  878. .. versionadded:: 1.1
  879. """
  880. acl = acl or self.default_acl
  881. async_result = self.handler.async_result()
  882. @wrap(async_result)
  883. def create_completion(result):
  884. try:
  885. return result.get()
  886. except NodeExistsError:
  887. return True
  888. @capture_exceptions(async_result)
  889. def prepare_completion(next_path, result):
  890. result.get()
  891. self.create_async(next_path, acl=acl).rawlink(create_completion)
  892. @wrap(async_result)
  893. def exists_completion(path, result):
  894. if result.get():
  895. return True
  896. parent, node = split(path)
  897. if node:
  898. self.ensure_path_async(parent, acl=acl).rawlink(
  899. partial(prepare_completion, path))
  900. else:
  901. self.create_async(path, acl=acl).rawlink(create_completion)
  902. self.exists_async(path).rawlink(partial(exists_completion, path))
  903. return async_result
  904. def exists(self, path, watch=None):
  905. """Check if a node exists.
  906. If a watch is provided, it will be left on the node with the
  907. given path. The watch will be triggered by a successful
  908. operation that creates/deletes the node or sets the data on the
  909. node.
  910. :param path: Path of node.
  911. :param watch: Optional watch callback to set for future changes
  912. to this path.
  913. :returns: ZnodeStat of the node if it exists, else None if the
  914. node does not exist.
  915. :rtype: :class:`~kazoo.protocol.states.ZnodeStat` or `None`.
  916. :raises:
  917. :exc:`~kazoo.exceptions.ZookeeperError` if the server
  918. returns a non-zero error code.
  919. """
  920. return self.exists_async(path, watch=watch).get()
  921. def exists_async(self, path, watch=None):
  922. """Asynchronously check if a node exists. Takes the same
  923. arguments as :meth:`exists`.
  924. :rtype: :class:`~kazoo.interfaces.IAsyncResult`
  925. """
  926. if not isinstance(path, string_types):
  927. raise TypeError("Invalid type for 'path' (string expected)")
  928. if watch and not callable(watch):
  929. raise TypeError("Invalid type for 'watch' (must be a callable)")
  930. async_result = self.handler.async_result()
  931. self._call(Exists(_prefix_root(self.chroot, path), watch),
  932. async_result)
  933. return async_result
  934. def get(self, path, watch=None):
  935. """Get the value of a node.
  936. If a watch is provided, it will be left on the node with the
  937. given path. The watch will be triggered by a successful
  938. operation that sets data on the node, or deletes the node.
  939. :param path: Path of node.
  940. :param watch: Optional watch callback to set for future changes
  941. to this path.
  942. :returns:
  943. Tuple (value, :class:`~kazoo.protocol.states.ZnodeStat`) of
  944. node.
  945. :rtype: tuple
  946. :raises:
  947. :exc:`~kazoo.exceptions.NoNodeError` if the node doesn't
  948. exist
  949. :exc:`~kazoo.exceptions.ZookeeperError` if the server
  950. returns a non-zero error code
  951. """
  952. return self.get_async(path, watch=watch).get()
  953. def get_async(self, path, watch=None):
  954. """Asynchronously get the value of a node. Takes the same
  955. arguments as :meth:`get`.
  956. :rtype: :class:`~kazoo.interfaces.IAsyncResult`
  957. """
  958. if not isinstance(path, string_types):
  959. raise TypeError("Invalid type for 'path' (string expected)")
  960. if watch and not callable(watch):
  961. raise TypeError("Invalid type for 'watch' (must be a callable)")
  962. async_result = self.handler.async_result()
  963. self._call(GetData(_prefix_root(self.chroot, path), watch),
  964. async_result)
  965. return async_result
  966. def get_children(self, path, watch=None, include_data=False):
  967. """Get a list of child nodes of a path.
  968. If a watch is provided it will be left on the node with the
  969. given path. The watch will be triggered by a successful
  970. operation that deletes the node of the given path or
  971. creates/deletes a child under the node.
  972. The list of children returned is not sorted and no guarantee is
  973. provided as to its natural or lexical order.
  974. :param path: Path of node to list.
  975. :param watch: Optional watch callback to set for future changes
  976. to this path.
  977. :param include_data:
  978. Include the :class:`~kazoo.protocol.states.ZnodeStat` of
  979. the node in addition to the children. This option changes
  980. the return value to be a tuple of (children, stat).
  981. :returns: List of child node names, or tuple if `include_data`
  982. is `True`.
  983. :rtype: list
  984. :raises:
  985. :exc:`~kazoo.exceptions.NoNodeError` if the node doesn't
  986. exist.
  987. :exc:`~kazoo.exceptions.ZookeeperError` if the server
  988. returns a non-zero error code.
  989. .. versionadded:: 0.5
  990. The `include_data` option.
  991. """
  992. return self.get_children_async(path, watch=watch,
  993. include_data=include_data).get()
  994. def get_children_async(self, path, watch=None, include_data=False):
  995. """Asynchronously get a list of child nodes of a path. Takes
  996. the same arguments as :meth:`get_children`.
  997. :rtype: :class:`~kazoo.interfaces.IAsyncResult`
  998. """
  999. if not isinstance(path, string_types):
  1000. raise TypeError("Invalid type for 'path' (string expected)")
  1001. if watch and not callable(watch):
  1002. raise TypeError("Invalid type for 'watch' (must be a callable)")
  1003. if not isinstance(include_data, bool):
  1004. raise TypeError("Invalid type for 'include_data' (bool expected)")
  1005. async_result = self.handler.async_result()
  1006. if include_data:
  1007. req = GetChildren2(_prefix_root(self.chroot, path), watch)
  1008. else:
  1009. req = GetChildren(_prefix_root(self.chroot, path), watch)
  1010. self._call(req, async_result)
  1011. return async_result
  1012. def get_acls(self, path):
  1013. """Return the ACL and stat of the node of the given path.
  1014. :param path: Path of the node.
  1015. :returns: The ACL array of the given node and its
  1016. :class:`~kazoo.protocol.states.ZnodeStat`.
  1017. :rtype: tuple of (:class:`~kazoo.security.ACL` list,
  1018. :class:`~kazoo.protocol.states.ZnodeStat`)
  1019. :raises:
  1020. :exc:`~kazoo.exceptions.NoNodeError` if the node doesn't
  1021. exist.
  1022. :exc:`~kazoo.exceptions.ZookeeperError` if the server
  1023. returns a non-zero error code
  1024. .. versionadded:: 0.5
  1025. """
  1026. return self.get_acls_async(path).get()
  1027. def get_acls_async(self, path):
  1028. """Return the ACL and stat of the node of the given path. Takes
  1029. the same arguments as :meth:`get_acls`.
  1030. :rtype: :class:`~kazoo.interfaces.IAsyncResult`
  1031. """
  1032. if not isinstance(path, string_types):
  1033. raise TypeError("Invalid type for 'path' (string expected)")
  1034. async_result = self.handler.async_result()
  1035. self._call(GetACL(_prefix_root(self.chroot, path)), async_result)
  1036. return async_result
  1037. def set_acls(self, path, acls, version=-1):
  1038. """Set the ACL for the node of the given path.
  1039. Set the ACL for the node of the given path if such a node
  1040. exists and the given version matches the version of the node.
  1041. :param path: Path for the node.
  1042. :param acls: List of :class:`~kazoo.security.ACL` objects to
  1043. set.
  1044. :param version: The expected node version that must match.
  1045. :returns: The stat of the node.
  1046. :raises:
  1047. :exc:`~kazoo.exceptions.BadVersionError` if version doesn't
  1048. match.
  1049. :exc:`~kazoo.exceptions.NoNodeError` if the node doesn't
  1050. exist.
  1051. :exc:`~kazoo.exceptions.InvalidACLError` if the ACL is
  1052. invalid.
  1053. :exc:`~kazoo.exceptions.ZookeeperError` if the server
  1054. returns a non-zero error code.
  1055. .. versionadded:: 0.5
  1056. """
  1057. return self.set_acls_async(path, acls, version).get()
  1058. def set_acls_async(self, path, acls, version=-1):
  1059. """Set the ACL for the node of the given path. Takes the same
  1060. arguments as :meth:`set_acls`.
  1061. :rtype: :class:`~kazoo.interfaces.IAsyncResult`
  1062. """
  1063. if not isinstance(path, string_types):
  1064. raise TypeError("Invalid type for 'path' (string expected)")
  1065. if isinstance(acls, ACL) or not isinstance(acls, (tuple, list)):
  1066. raise TypeError("Invalid type for 'acl' (acl must be a tuple/list"
  1067. " of ACL's)")
  1068. if not isinstance(version, int):
  1069. raise TypeError("Invalid type for 'version' (int expected)")
  1070. async_result = self.handler.async_result()
  1071. self._call(SetACL(_prefix_root(self.chroot, path), acls, version),
  1072. async_result)
  1073. return async_result
  1074. def set(self, path, value, version=-1):
  1075. """Set the value of a node.
  1076. If the version of the node being updated is newer than the
  1077. supplied version (and the supplied version is not -1), a
  1078. BadVersionError will be raised.
  1079. This operation, if successful, will trigger all the watches on
  1080. the node of the given path left by :meth:`get` API calls.
  1081. The maximum allowable size of the value is 1 MB. Values larger
  1082. than this will cause a ZookeeperError to be raised.
  1083. :param path: Path of node.
  1084. :param value: New data value.
  1085. :param version: Version of node being updated, or -1.
  1086. :returns: Updated :class:`~kazoo.protocol.states.ZnodeStat` of
  1087. the node.
  1088. :raises:
  1089. :exc:`~kazoo.exceptions.BadVersionError` if version doesn't
  1090. match.
  1091. :exc:`~kazoo.exceptions.NoNodeError` if the node doesn't
  1092. exist.
  1093. :exc:`~kazoo.exceptions.ZookeeperError` if the provided
  1094. value is too large.
  1095. :exc:`~kazoo.exceptions.ZookeeperError` if the server
  1096. returns a non-zero error code.
  1097. """
  1098. return self.set_async(path, value, version).get()
  1099. def set_async(self, path, value, version=-1):
  1100. """Set the value of a node. Takes the same arguments as
  1101. :meth:`set`.
  1102. :rtype: :class:`~kazoo.interfaces.IAsyncResult`
  1103. """
  1104. if not isinstance(path, string_types):
  1105. raise TypeError("Invalid type for 'path' (string expected)")
  1106. if value is not None and not isinstance(value, bytes_types):
  1107. raise TypeError("Invalid type for 'value' (must be a byte string)")
  1108. if not isinstance(version, int):
  1109. raise TypeError("Invalid type for 'version' (int expected)")
  1110. async_result = self.handler.async_result()
  1111. self._call(SetData(_prefix_root(self.chroot, path), value, version),
  1112. async_result)
  1113. return async_result
  1114. def transaction(self):
  1115. """Create and return a :class:`TransactionRequest` object
  1116. Creates a :class:`TransactionRequest` object. A Transaction can
  1117. consist of multiple operations which can be committed as a
  1118. single atomic unit. Either all of the operations will succeed
  1119. or none of them.
  1120. :returns: A TransactionRequest.
  1121. :rtype: :class:`TransactionRequest`
  1122. .. versionadded:: 0.6
  1123. Requires Zookeeper 3.4+
  1124. """
  1125. return TransactionRequest(self)
  1126. def delete(self, path, version=-1, recursive=False):
  1127. """Delete a node.
  1128. The call will succeed if such a node exists, and the given
  1129. version matches the node's version (if the given version is -1,
  1130. the default, it matches any node's versions).
  1131. This operation, if successful, will trigger all the watches on
  1132. the node of the given path left by `exists` API calls, and the
  1133. watches on the parent node left by `get_children` API calls.
  1134. :param path: Path of node to delete.
  1135. :param version: Version of node to delete, or -1 for any.
  1136. :param recursive: Recursively delete node and all its children,
  1137. defaults to False.
  1138. :type recursive: bool
  1139. :raises:
  1140. :exc:`~kazoo.exceptions.BadVersionError` if version doesn't
  1141. match.
  1142. :exc:`~kazoo.exceptions.NoNodeError` if the node doesn't
  1143. exist.
  1144. :exc:`~kazoo.exceptions.NotEmptyError` if the node has
  1145. children.
  1146. :exc:`~kazoo.exceptions.ZookeeperError` if the server
  1147. returns a non-zero error code.
  1148. """
  1149. if not isinstance(recursive, bool):
  1150. raise TypeError("Invalid type for 'recursive' (bool expected)")
  1151. if recursive:
  1152. return self._delete_recursive(path)
  1153. else:
  1154. return self.delete_async(path, version).get()
  1155. def delete_async(self, path, version=-1):
  1156. """Asynchronously delete a node. Takes the same arguments as
  1157. :meth:`delete`, with the exception of `recursive`.
  1158. :rtype: :class:`~kazoo.interfaces.IAsyncResult`
  1159. """
  1160. if not isinstance(path, string_types):
  1161. raise TypeError("Invalid type for 'path' (string expected)")
  1162. if not isinstance(version, int):
  1163. raise TypeError("Invalid type for 'version' (int expected)")
  1164. async_result = self.handler.async_result()
  1165. self._call(Delete(_prefix_root(self.chroot, path), version),
  1166. async_result)
  1167. return async_result
  1168. def _delete_recursive(self, path):
  1169. try:
  1170. children = self.get_children(path)
  1171. except NoNodeError:
  1172. return True
  1173. if children:
  1174. for child in children:
  1175. if path == "/":
  1176. child_path = path + child
  1177. else:
  1178. child_path = path + "/" + child
  1179. self._delete_recursive(child_path)
  1180. try:
  1181. self.delete(path)
  1182. except NoNodeError: # pragma: nocover
  1183. pass
  1184. def reconfig(self, joining, leaving, new_members, from_config=-1):
  1185. """Reconfig a cluster.
  1186. This call will succeed if the cluster was reconfigured accordingly.
  1187. :param joining: a comma separated list of servers being added
  1188. (see example for format) (incremental reconfiguration)
  1189. :param leaving: a comma separated list of servers being removed
  1190. (see example for format) (incremental reconfiguration)
  1191. :param new_members: a comma separated list of new membership
  1192. (non-incremental reconfiguration)
  1193. :param from_config: version of the current configuration (optional -
  1194. causes reconfiguration to throw an exception if
  1195. configuration is no longer current)
  1196. :type from_config: int
  1197. :returns:
  1198. Tuple (value, :class:`~kazoo.protocol.states.ZnodeStat`) of
  1199. node.
  1200. :rtype: tuple
  1201. Basic Example:
  1202. .. code-block:: python
  1203. zk = KazooClient()
  1204. zk.start()
  1205. # first add an observer (incremental reconfiguration)
  1206. joining = 'server.100=10.0.0.10:2889:3888:observer;0.0.0.0:2181'
  1207. data, _ = zk.reconfig(
  1208. joining=joining, leaving=None, new_members=None)
  1209. # wait and then remove it (just by using its id) (incremental)
  1210. data, _ = zk.reconfig(joining=None, leaving='100',
  1211. new_members=None)
  1212. # now do a full change of the cluster (non-incremental)
  1213. new = [
  1214. 'server.100=10.0.0.10:2889:3888:observer;0.0.0.0:2181',
  1215. 'server.100=10.0.0.11:2889:3888:observer;0.0.0.0:2181',
  1216. 'server.100=10.0.0.12:2889:3888:observer;0.0.0.0:2181',
  1217. ]
  1218. data, _ = zk.reconfig(
  1219. joining=None, leaving=None, new_members=','.join(new))
  1220. zk.stop()
  1221. :raises:
  1222. :exc:`~kazoo.exceptions.UnimplementedError` if not supported.
  1223. :exc:`~kazoo.exceptions.NewConfigNoQuorumError` if no quorum of new
  1224. config is connected and up-to-date with the leader of last
  1225. commmitted config - try invoking reconfiguration after new servers
  1226. are connected and synced.
  1227. :exc:`~kazoo.exceptions.ReconfigInProcessError` if another
  1228. reconfiguration is in progress.
  1229. :exc:`~kazoo.exceptions.BadVersionError` if version doesn't
  1230. match.
  1231. :exc:`~kazoo.exceptions.BadArgumentsError` if any of the given
  1232. lists of servers has a bad format.
  1233. :exc:`~kazoo.exceptions.ZookeeperError` if the server
  1234. returns a non-zero error code.
  1235. """
  1236. result = self.reconfig_async(joining, leaving, new_members,
  1237. from_config)
  1238. return result.get()
  1239. def reconfig_async(self, joining, leaving, new_members, from_config):
  1240. """Asynchronously reconfig a cluster. Takes the same arguments as
  1241. :meth:`reconfig`.
  1242. :rtype: :class:`~kazoo.interfaces.IAsyncResult`
  1243. """
  1244. if joining and not isinstance(joining, string_types):
  1245. raise TypeError("Invalid type for 'joining' (string expected)")
  1246. if leaving and not isinstance(leaving, string_types):
  1247. raise TypeError("Invalid type for 'leaving' (string expected)")
  1248. if new_members and not isinstance(new_members, string_types):
  1249. raise TypeError("Invalid type for 'new_members' (string "
  1250. "expected)")
  1251. if not isinstance(from_config, int):
  1252. raise TypeError("Invalid type for 'from_config' (int expected)")
  1253. async_result = self.handler.async_result()
  1254. reconfig = Reconfig(joining, leaving, new_members, from_config)
  1255. self._call(reconfig, async_result)
  1256. return async_result
  1257. class TransactionRequest(object):
  1258. """A Zookeeper Transaction Request
  1259. A Transaction provides a builder object that can be used to
  1260. construct and commit an atomic set of operations. The transaction
  1261. must be committed before its sent.
  1262. Transactions are not thread-safe and should not be accessed from
  1263. multiple threads at once.
  1264. .. note::
  1265. The ``committed`` attribute only indicates whether this
  1266. transaction has been sent to Zookeeper and is used to prevent
  1267. duplicate commits of the same transaction. The result should be
  1268. checked to determine if the transaction executed as desired.
  1269. .. versionadded:: 0.6
  1270. Requires Zookeeper 3.4+
  1271. """
  1272. def __init__(self, client):
  1273. self.client = client
  1274. self.operations = []
  1275. self.committed = False
  1276. def create(self, path, value=b"", acl=None, ephemeral=False,
  1277. sequence=False):
  1278. """Add a create ZNode to the transaction. Takes the same
  1279. arguments as :meth:`KazooClient.create`, with the exception
  1280. of `makepath`.
  1281. :returns: None
  1282. """
  1283. if acl is None and self.client.default_acl:
  1284. acl = self.client.default_acl
  1285. if not isinstance(path, string_types):
  1286. raise TypeError("Invalid type for 'path' (string expected)")
  1287. if acl and not isinstance(acl, (tuple, list)):
  1288. raise TypeError("Invalid type for 'acl' (acl must be a tuple/list"
  1289. " of ACL's")
  1290. if not isinstance(value, bytes_types):
  1291. raise TypeError("Invalid type for 'value' (must be a byte string)")
  1292. if not isinstance(ephemeral, bool):
  1293. raise TypeError("Invalid type for 'ephemeral' (bool expected)")
  1294. if not isinstance(sequence, bool):
  1295. raise TypeError("Invalid type for 'sequence' (bool expected)")
  1296. flags = 0
  1297. if ephemeral:
  1298. flags |= 1
  1299. if sequence:
  1300. flags |= 2
  1301. if acl is None:
  1302. acl = OPEN_ACL_UNSAFE
  1303. self._add(Create(_prefix_root(self.client.chroot, path), value, acl,
  1304. flags), None)
  1305. def delete(self, path, version=-1):
  1306. """Add a delete ZNode to the transaction. Takes the same
  1307. arguments as :meth:`KazooClient.delete`, with the exception of
  1308. `recursive`.
  1309. """
  1310. if not isinstance(path, string_types):
  1311. raise TypeError("Invalid type for 'path' (string expected)")
  1312. if not isinstance(version, int):
  1313. raise TypeError("Invalid type for 'version' (int expected)")
  1314. self._add(Delete(_prefix_root(self.client.chroot, path), version))
  1315. def set_data(self, path, value, version=-1):
  1316. """Add a set ZNode value to the transaction. Takes the same
  1317. arguments as :meth:`KazooClient.set`.
  1318. """
  1319. if not isinstance(path, string_types):
  1320. raise TypeError("Invalid type for 'path' (string expected)")
  1321. if not isinstance(value, bytes_types):
  1322. raise TypeError("Invalid type for 'value' (must be a byte string)")
  1323. if not isinstance(version, int):
  1324. raise TypeError("Invalid type for 'version' (int expected)")
  1325. self._add(SetData(_prefix_root(self.client.chroot, path), value,
  1326. version))
  1327. def check(self, path, version):
  1328. """Add a Check Version to the transaction.
  1329. This command will fail and abort a transaction if the path
  1330. does not match the specified version.
  1331. """
  1332. if not isinstance(path, string_types):
  1333. raise TypeError("Invalid type for 'path' (string expected)")
  1334. if not isinstance(version, int):
  1335. raise TypeError("Invalid type for 'version' (int expected)")
  1336. self._add(CheckVersion(_prefix_root(self.client.chroot, path),
  1337. version))
  1338. def commit_async(self):
  1339. """Commit the transaction asynchronously.
  1340. :rtype: :class:`~kazoo.interfaces.IAsyncResult`
  1341. """
  1342. self._check_tx_state()
  1343. self.committed = True
  1344. async_object = self.client.handler.async_result()
  1345. self.client._call(Transaction(self.operations), async_object)
  1346. return async_object
  1347. def commit(self):
  1348. """Commit the transaction.
  1349. :returns: A list of the results for each operation in the
  1350. transaction.
  1351. """
  1352. return self.commit_async().get()
  1353. def __enter__(self):
  1354. return self
  1355. def __exit__(self, exc_type, exc_value, exc_tb):
  1356. """Commit and cleanup accumulated transaction data."""
  1357. if not exc_type:
  1358. self.commit()
  1359. def _check_tx_state(self):
  1360. if self.committed:
  1361. raise ValueError('Transaction already committed')
  1362. def _add(self, request, post_processor=None):
  1363. self._check_tx_state()
  1364. self.client.logger.log(BLATHER, 'Added %r to %r', request, self)
  1365. self.operations.append(request)