client.py 48 KB

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