Merge pull request #40809 from gtmanfred/develop

Add zookeeper managing module and state

doc/ref/modules/all/salt.modules.zookeeper.rst

@@ -0,0 +1,6 @@
+======================
+salt.modules.zookeeper
+======================
+
+.. automodule:: salt.modules.zookeeper
+    :members:

doc/ref/states/all/salt.states.zookeeper.rst

@@ -0,0 +1,6 @@
+=====================
+salt.states.zookeeper
+=====================
+
+.. automodule:: salt.states.zookeeper
+    :members:

salt/modules/zk_concurrency.py

@@ -1,5 +1,8 @@
 # -*- coding: utf-8 -*-
 '''
+:depends: kazoo
+:configuration: See :py:mod:`salt.modules.zookeeper` for setup instructions.
+
 Concurrency controls in zookeeper
 =========================================================================
 
@@ -13,7 +16,7 @@ import logging
 import sys
 
 try:
-    from kazoo.client import KazooClient
+    import kazoo.client
 
     from kazoo.retry import (
         ForceRetryError
@@ -90,8 +93,6 @@ try:
 except ImportError:
     HAS_DEPS = False
 
-ZK_CONNECTION = None
-SEMAPHORE_MAP = {}
 
 __virtualname__ = 'zk_concurrency'
 
@@ -100,33 +101,67 @@ def __virtual__():
     if not HAS_DEPS:
         return (False, "Module zk_concurrency: dependencies failed")
 
+    __context__['semaphore_map'] = {}
+
     return __virtualname__
 
 
-def _get_zk_conn(hosts):
+def _get_zk_conn(profile=None, **connection_args):
-    global ZK_CONNECTION
+    if profile:
-    if ZK_CONNECTION is None:
+        prefix = 'zookeeper:' + profile
-        ZK_CONNECTION = KazooClient(hosts=hosts)
+    else:
-        ZK_CONNECTION.start()
+        prefix = 'zookeeper'
 
-    return ZK_CONNECTION
+    def get(key, default=None):
+        '''
+        look in connection_args first, then default to config file
+        '''
+        return connection_args.get(key) or __salt__['config.get'](':'.join([prefix, key]), default)
 
+    hosts = get('hosts', '127.0.0.1:2181')
+    scheme = get('scheme', None)
+    username = get('username', None)
+    password = get('password', None)
+    default_acl = get('default_acl', None)
 
-def _close_zk_conn():
+    if isinstance(hosts, list):
-    global ZK_CONNECTION
+        hosts = ','.join(hosts)
-    if ZK_CONNECTION is None:
-        return
 
-    ZK_CONNECTION.stop()
+    if username is not None and password is not None and scheme is None:
-    ZK_CONNECTION = None
+        scheme = 'digest'
+
+    auth_data = None
+    if scheme and username and password:
+        auth_data = [(scheme, ':'.join([username, password]))]
+
+    if default_acl is not None:
+        if isinstance(default_acl, list):
+            default_acl = [__salt__['zookeeper.make_digest_acl'](**acl) for acl in default_acl]
+        else:
+            default_acl = [__salt__['zookeeper.make_digest_acl'](**default_acl)]
+
+    __context__.setdefault('zkconnection', {}).setdefault(profile or hosts,
+                                                          kazoo.client.KazooClient(hosts=hosts,
+                                                                                   default_acl=default_acl,
+                                                                                   auth_data=auth_data))
+
+    if not __context__['zkconnection'][profile or hosts].connected:
+        __context__['zkconnection'][profile or hosts].start()
+
+    return __context__['zkconnection'][profile or hosts]
 
 
 def lock_holders(path,
-                 zk_hosts,
+                 zk_hosts=None,
                  identifier=None,
                  max_concurrency=1,
                  timeout=None,
-                 ephemeral_lease=False):
+                 ephemeral_lease=False,
+                 profile=None,
+                 scheme=None,
+                 username=None,
+                 password=None,
+                 default_acl=None):
     '''
     Return an un-ordered list of lock holders
 
@@ -154,24 +189,27 @@ def lock_holders(path,
 
         salt minion zk_concurrency.lock_holders /lock/path host1:1234,host2:1234
     '''
-
+    zk = _get_zk_conn(profile=profile, hosts=zk_hosts, scheme=scheme,
-    zk = _get_zk_conn(zk_hosts)
+                      username=username, password=password, default_acl=default_acl)
-    if path not in SEMAPHORE_MAP:
+    if path not in __context__['semaphore_map']:
-        SEMAPHORE_MAP[path] = _Semaphore(zk,
+        __context__['semaphore_map'][path] = _Semaphore(zk, path, identifier,
-                                        path,
+                                                        max_leases=max_concurrency,
-                                        identifier,
+                                                        ephemeral_lease=ephemeral_lease)
-                                        max_leases=max_concurrency,
+    return __context__['semaphore_map'][path].lease_holders()
-                                        ephemeral_lease=ephemeral_lease)
-    return SEMAPHORE_MAP[path].lease_holders()
 
 
 def lock(path,
-         zk_hosts,
+         zk_hosts=None,
          identifier=None,
          max_concurrency=1,
          timeout=None,
          ephemeral_lease=False,
          force=False,  # foricble get the lock regardless of open slots
+         profile=None,
+         scheme=None,
+         username=None,
+         password=None,
+         default_acl=None,
          ):
     '''
     Get lock (with optional timeout)
@@ -203,35 +241,39 @@ def lock(path,
 
         salt minion zk_concurrency.lock /lock/path host1:1234,host2:1234
     '''
-    zk = _get_zk_conn(zk_hosts)
+    zk = _get_zk_conn(profile=profile, hosts=zk_hosts, scheme=scheme,
-    if path not in SEMAPHORE_MAP:
+                      username=username, password=password, default_acl=default_acl)
-        SEMAPHORE_MAP[path] = _Semaphore(zk,
+    if path not in __context__['semaphore_map']:
-                                        path,
+        __context__['semaphore_map'][path] = _Semaphore(zk, path, identifier,
-                                        identifier,
+                                                        max_leases=max_concurrency,
-                                        max_leases=max_concurrency,
+                                                        ephemeral_lease=ephemeral_lease)
-                                        ephemeral_lease=ephemeral_lease)
 
     # forcibly get the lock regardless of max_concurrency
     if force:
-        SEMAPHORE_MAP[path].assured_path = True
+        __context__['semaphore_map'][path].assured_path = True
-        SEMAPHORE_MAP[path].max_leases = sys.maxint
+        __context__['semaphore_map'][path].max_leases = sys.maxint
 
     # block waiting for lock acquisition
     if timeout:
         logging.info('Acquiring lock {0} with timeout={1}'.format(path, timeout))
-        SEMAPHORE_MAP[path].acquire(timeout=timeout)
+        __context__['semaphore_map'][path].acquire(timeout=timeout)
     else:
         logging.info('Acquiring lock {0} with no timeout'.format(path))
-        SEMAPHORE_MAP[path].acquire()
+        __context__['semaphore_map'][path].acquire()
 
-    return SEMAPHORE_MAP[path].is_acquired
+    return __context__['semaphore_map'][path].is_acquired
 
 
 def unlock(path,
            zk_hosts=None,  # in case you need to unlock without having run lock (failed execution for example)
            identifier=None,
            max_concurrency=1,
-           ephemeral_lease=False
+           ephemeral_lease=False,
+           scheme=None,
+           profile=None,
+           username=None,
+           password=None,
+           default_acl=None
            ):
     '''
     Remove lease from semaphore
@@ -260,19 +302,18 @@ def unlock(path,
 
         salt minion zk_concurrency.unlock /lock/path host1:1234,host2:1234
     '''
-    # if someone passed in zk_hosts, and the path isn't in SEMAPHORE_MAP, lets
+    # if someone passed in zk_hosts, and the path isn't in __context__['semaphore_map'], lets
     # see if we can find it
-    if zk_hosts is not None and path not in SEMAPHORE_MAP:
+    zk = _get_zk_conn(profile=profile, hosts=zk_hosts, scheme=scheme,
-        zk = _get_zk_conn(zk_hosts)
+                      username=username, password=password, default_acl=default_acl)
-        SEMAPHORE_MAP[path] = _Semaphore(zk,
+    if path not in __context__['semaphore_map']:
-                                        path,
+        __context__['semaphore_map'][path] = _Semaphore(zk, path, identifier,
-                                        identifier,
+                                                        max_leases=max_concurrency,
-                                        max_leases=max_concurrency,
+                                                        ephemeral_lease=ephemeral_lease)
-                                        ephemeral_lease=ephemeral_lease)
 
-    if path in SEMAPHORE_MAP:
+    if path in __context__['semaphore_map']:
-        SEMAPHORE_MAP[path].release()
+        __context__['semaphore_map'][path].release()
-        del SEMAPHORE_MAP[path]
+        del __context__['semaphore_map'][path]
         return True
     else:
         logging.error('Unable to find lease for path {0}'.format(path))
@@ -280,9 +321,14 @@ def unlock(path,
 
 
 def party_members(path,
-                  zk_hosts,
+                  zk_hosts=None,
                   min_nodes=1,
-                  blocking=False
+                  blocking=False,
+                  profile=None,
+                  scheme=None,
+                  username=None,
+                  password=None,
+                  default_acl=None,
                   ):
     '''
     Get the List of identifiers in a particular party, optionally waiting for the
@@ -307,7 +353,8 @@ def party_members(path,
         salt minion zk_concurrency.party_members /lock/path host1:1234,host2:1234
         salt minion zk_concurrency.party_members /lock/path host1:1234,host2:1234 min_nodes=3 blocking=True
     '''
-    zk = _get_zk_conn(zk_hosts)
+    zk = _get_zk_conn(profile=profile, hosts=zk_hosts, scheme=scheme,
+                      username=username, password=password, default_acl=default_acl)
     party = kazoo.recipe.party.ShallowParty(zk, path)
     if blocking:
         barrier = kazoo.recipe.barrier.DoubleBarrier(zk, path, min_nodes)

salt/modules/zookeeper.py

@@ -0,0 +1,516 @@
+# -*- coding: utf-8 -*-
+'''
+Zookeeper Module
+~~~~~~~~~~~~~~~~
+:maintainer:    SaltStack
+:maturity:      new
+:platform:      all
+:depends:       kazoo
+
+Configuration
+=============
+
+:configuration: This module is not usable until the following are specified
+    either in a pillar or in the minion's config file:
+
+    .. code-block:: yaml
+
+        zookeeper:
+          hosts: zoo1,zoo2,zoo3
+          default_acl:
+            - username: daniel
+              password: test
+              read: true
+              write: true
+              create: true
+              delete: true
+              admin: true
+          username: daniel
+          password: test
+
+    If configuration for multiple zookeeper environments is required, they can
+    be set up as different configuration profiles. For example:
+
+    .. code-block:: yaml
+        zookeeper:
+          prod:
+            hosts: zoo1,zoo2,zoo3
+            default_acl:
+              - username: daniel
+                password: test
+                read: true
+                write: true
+                create: true
+                delete: true
+                admin: true
+            username: daniel
+            password: test
+          dev:
+            hosts:
+              - dev1
+              - dev2
+              - dev3
+            default_acl:
+              - username: daniel
+                password: test
+                read: true
+                write: true
+                create: true
+                delete: true
+                admin: true
+            username: daniel
+            password: test
+'''
+from __future__ import absolute_import
+
+# Import python libraries
+try:
+    import kazoo.client
+    import kazoo.security
+    HAS_KAZOO = True
+except ImportError:
+    HAS_KAZOO = False
+
+# Import Salt libraries
+
+__virtualname__ = 'zookeeper'
+
+
+def __virtual__():
+    if HAS_KAZOO:
+        return __virtualname__
+    return False
+
+
+def _get_zk_conn(profile=None, **connection_args):
+    if profile:
+        prefix = 'zookeeper:' + profile
+    else:
+        prefix = 'zookeeper'
+
+    def get(key, default=None):
+        '''
+        look in connection_args first, then default to config file
+        '''
+        return connection_args.get(key) or __salt__['config.get'](':'.join([prefix, key]), default)
+
+    hosts = get('hosts', '127.0.0.1:2181')
+    scheme = get('scheme', None)
+    username = get('username', None)
+    password = get('password', None)
+    default_acl = get('default_acl', None)
+
+    if isinstance(hosts, list):
+        hosts = ','.join(hosts)
+
+    if username is not None and password is not None and scheme is None:
+        scheme = 'digest'
+
+    auth_data = None
+    if scheme and username and password:
+        auth_data = [(scheme, ':'.join([username, password]))]
+
+    if default_acl is not None:
+        if isinstance(default_acl, list):
+            default_acl = [make_digest_acl(**acl) for acl in default_acl]
+        else:
+            default_acl = [make_digest_acl(**default_acl)]
+
+    __context__.setdefault('zkconnection', {}).setdefault(profile or hosts,
+                                                          kazoo.client.KazooClient(hosts=hosts,
+                                                                                   default_acl=default_acl,
+                                                                                   auth_data=auth_data))
+
+    if not __context__['zkconnection'][profile or hosts].connected:
+        __context__['zkconnection'][profile or hosts].start()
+
+    return __context__['zkconnection'][profile or hosts]
+
+
+def create(path, value='', acls=None, ephemeral=False, sequence=False, makepath=False, profile=None,
+           hosts=None, scheme=None, username=None, password=None, default_acl=None):
+    '''
+    Create Znode
+
+    path
+        path of znode to create
+
+    value
+        value to assign to znode (Default: '')
+
+    acls
+        list of acl dictionaries to be assigned (Default: None)
+
+    ephemeral
+        indicate node is ephemeral (Default: False)
+
+    sequence
+        indicate node is suffixed with a unique index (Default: False)
+
+    makepath
+        Create parent paths if they do not exist (Default: False)
+
+    profile
+        Configured Zookeeper profile to authenticate with (Default: None)
+
+    hosts
+        Lists of Zookeeper Hosts (Default: '127.0.0.1:2181)
+
+    scheme
+        Scheme to authenticate with (Default: 'digest')
+
+    username
+        Username to authenticate (Default: None)
+
+    password
+        Password to authenticate (Default: None)
+
+    default_acl
+        Default acls to assign if a node is created in this connection (Default: None)
+
+    CLI Example:
+
+    .. code-block:: bash
+
+        salt minion1 zookeeper.create /test/name daniel profile=prod
+
+    '''
+    if acls is None:
+        acls = []
+    acls = [make_digest_acl(**acl) for acl in acls]
+    conn = _get_zk_conn(profile=profile, hosts=hosts, scheme=scheme,
+                        username=username, password=password, default_acl=default_acl)
+    return conn.create(path, value, acls, ephemeral, sequence, makepath)
+
+
+def ensure_path(path, acls=None, profile=None, hosts=None, scheme=None,
+                username=None, password=None, default_acl=None):
+    '''
+    Ensure Znode path exists
+
+    path
+        Parent path to create
+
+    acls
+        list of acls dictionaries to be assigned (Default: None)
+
+    profile
+        Configured Zookeeper profile to authenticate with (Default: None)
+
+    hosts
+        Lists of Zookeeper Hosts (Default: '127.0.0.1:2181)
+
+    scheme
+        Scheme to authenticate with (Default: 'digest')
+
+    username
+        Username to authenticate (Default: None)
+
+    password
+        Password to authenticate (Default: None)
+
+    default_acl
+        Default acls to assign if a node is created in this connection (Default: None)
+
+    CLI Example:
+
+    .. code-block:: bash
+
+        salt minion1 zookeeper.ensure_path /test/name profile=prod
+
+    '''
+    if acls is None:
+        acls = []
+    acls = [make_digest_acl(**acl) for acl in acls]
+    conn = _get_zk_conn(profile=profile, hosts=hosts, scheme=scheme,
+                        username=username, password=password, default_acl=default_acl)
+    return conn.ensure_path(path, acls)
+
+
+def exists(path, profile=None, hosts=None, scheme=None, username=None, password=None, default_acl=None):
+    '''
+    Check if path exists
+
+    path
+        path to check
+
+    profile
+        Configured Zookeeper profile to authenticate with (Default: None)
+
+    hosts
+        Lists of Zookeeper Hosts (Default: '127.0.0.1:2181)
+
+    scheme
+        Scheme to authenticate with (Default: 'digest')
+
+    username
+        Username to authenticate (Default: None)
+
+    password
+        Password to authenticate (Default: None)
+
+    default_acl
+        Default acls to assign if a node is created in this connection (Default: None)
+
+    CLI Example:
+
+    .. code-block:: bash
+
+        salt minion1 zookeeper.exists /test/name profile=prod
+
+    '''
+    conn = _get_zk_conn(profile=profile, hosts=hosts, scheme=scheme,
+                        username=username, password=password, default_acl=default_acl)
+    return bool(conn.exists(path))
+
+
+def get(path, profile=None, hosts=None, scheme=None, username=None, password=None, default_acl=None):
+    '''
+    Get value saved in znode
+
+    path
+        path to check
+
+    profile
+        Configured Zookeeper profile to authenticate with (Default: None)
+
+    hosts
+        Lists of Zookeeper Hosts (Default: '127.0.0.1:2181)
+
+    scheme
+        Scheme to authenticate with (Default: 'digest')
+
+    username
+        Username to authenticate (Default: None)
+
+    password
+        Password to authenticate (Default: None)
+
+    default_acl
+        Default acls to assign if a node is created in this connection (Default: None)
+
+    CLI Example:
+
+    .. code-block:: bash
+
+        salt minion1 zookeeper.get /test/name profile=prod
+
+    '''
+    conn = _get_zk_conn(profile=profile, hosts=hosts, scheme=scheme,
+                        username=username, password=password, default_acl=default_acl)
+    ret, _ = conn.get(path)
+    return ret
+
+
+def get_children(path, profile=None, hosts=None, scheme=None, username=None, password=None, default_acl=None):
+    '''
+    Get children in znode path
+
+    path
+        path to check
+
+    profile
+        Configured Zookeeper profile to authenticate with (Default: None)
+
+    hosts
+        Lists of Zookeeper Hosts (Default: '127.0.0.1:2181)
+
+    scheme
+        Scheme to authenticate with (Default: 'digest')
+
+    username
+        Username to authenticate (Default: None)
+
+    password
+        Password to authenticate (Default: None)
+
+    default_acl
+        Default acls to assign if a node is created in this connection (Default: None)
+
+    CLI Example:
+
+    .. code-block:: bash
+
+        salt minion1 zookeeper.get_children /test profile=prod
+
+    '''
+    conn = _get_zk_conn(profile=profile, hosts=hosts, scheme=scheme,
+                        username=username, password=password, default_acl=default_acl)
+    ret = conn.get_children(path)
+    return ret or []
+
+
+def set(path, value, version=-1, profile=None, hosts=None, scheme=None,
+        username=None, password=None, default_acl=None):
+    '''
+    Update znode with new value
+
+    path
+        znode to update
+
+    value
+        value to set in znode
+
+    version
+        only update znode if version matches (Default: -1 (always matches))
+
+    profile
+        Configured Zookeeper profile to authenticate with (Default: None)
+
+    hosts
+        Lists of Zookeeper Hosts (Default: '127.0.0.1:2181)
+
+    scheme
+        Scheme to authenticate with (Default: 'digest')
+
+    username
+        Username to authenticate (Default: None)
+
+    password
+        Password to authenticate (Default: None)
+
+    default_acl
+        Default acls to assign if a node is created in this connection (Default: None)
+
+    CLI Example:
+
+    .. code-block:: bash
+
+        salt minion1 zookeeper.set /test/name gtmanfred profile=prod
+
+    '''
+    conn = _get_zk_conn(profile=profile, hosts=hosts, scheme=scheme,
+                        username=username, password=password, default_acl=default_acl)
+    return conn.set(path, value, version=version)
+
+
+def get_acls(path, profile=None, hosts=None, scheme=None, username=None, password=None, default_acl=None):
+    '''
+    Get acls on a znode
+
+    path
+        path to znode
+
+    profile
+        Configured Zookeeper profile to authenticate with (Default: None)
+
+    hosts
+        Lists of Zookeeper Hosts (Default: '127.0.0.1:2181)
+
+    scheme
+        Scheme to authenticate with (Default: 'digest')
+
+    username
+        Username to authenticate (Default: None)
+
+    password
+        Password to authenticate (Default: None)
+
+    default_acl
+        Default acls to assign if a node is created in this connection (Default: None)
+
+    CLI Example:
+
+    .. code-block:: bash
+
+        salt minion1 zookeeper.get_acls /test/name profile=prod
+
+    '''
+    conn = _get_zk_conn(profile=profile, hosts=hosts, scheme=scheme,
+                        username=username, password=password, default_acl=default_acl)
+    return conn.get_acls(path)[0]
+
+
+def set_acls(path, acls, version=-1, profile=None, hosts=None, scheme=None,
+             username=None, password=None, default_acl=None):
+    '''
+    Set acls on a znode
+
+    path
+        path to znode
+
+    acls
+        list of acl dictionaries to set on the znode
+
+    version
+        only set acls if version matches (Default: -1 (always matches))
+
+    profile
+        Configured Zookeeper profile to authenticate with (Default: None)
+
+    hosts
+        Lists of Zookeeper Hosts (Default: '127.0.0.1:2181)
+
+    scheme
+        Scheme to authenticate with (Default: 'digest')
+
+    username
+        Username to authenticate (Default: None)
+
+    password
+        Password to authenticate (Default: None)
+
+    default_acl
+        Default acls to assign if a node is created in this connection (Default: None)
+
+    CLI Example:
+
+    .. code-block:: bash
+
+        salt minion1 zookeeper.set_acls /test/name acls='[{"username": "gtmanfred", "password": "test", "all": True}]' profile=prod
+
+    '''
+    conn = _get_zk_conn(profile=profile, hosts=hosts, scheme=scheme,
+                        username=username, password=password, default_acl=default_acl)
+    if acls is None:
+        acls = []
+    acls = [make_digest_acl(**acl) for acl in acls]
+    conn = _get_zk_conn(profile=profile, hosts=hosts, scheme=scheme,
+                        username=username, password=password, default_acl=default_acl)
+    return conn.set_acls(path, acls, version)
+
+
+def delete(path, version=-1, recursive=False, profile=None, hosts=None, scheme=None,
+           username=None, password=None, default_acl=None):
+    '''
+    Delete znode
+
+    path
+        path to znode
+
+    version
+        only delete if version matches (Default: -1 (always matches))
+
+    profile
+        Configured Zookeeper profile to authenticate with (Default: None)
+
+    hosts
+        Lists of Zookeeper Hosts (Default: '127.0.0.1:2181)
+
+    scheme
+        Scheme to authenticate with (Default: 'digest')
+
+    username
+        Username to authenticate (Default: None)
+
+    password
+        Password to authenticate (Default: None)
+
+    default_acl
+        Default acls to assign if a node is created in this connection (Default: None)
+
+    CLI Example:
+
+    .. code-block:: bash
+
+        salt minion1 zookeeper.delete /test/name profile=prod
+
+    '''
+    conn = _get_zk_conn(profile=profile, hosts=hosts, scheme=scheme,
+                        username=username, password=password, default_acl=default_acl)
+    return conn.delete(path, version, recursive)
+
+
+def make_digest_acl(username, password, read=False, write=False, create=False, delete=False, admin=False,
+                    allperms=False):
+    return kazoo.security.make_digest_acl(username, password, read, write, create, delete, admin, allperms)

salt/states/zk_concurrency.py

@@ -1,5 +1,8 @@
 # -*- coding: utf-8 -*-
 '''
+:depends: kazoo
+:configuration: See :py:mod:`salt.modules.zookeeper` for setup instructions.
+
 Control concurrency of steps within state execution using zookeeper
 ===================================================================
 
@@ -60,12 +63,16 @@ def __virtual__():
 
 
 def lock(name,
-         zk_hosts,
+         zk_hosts=None,
          identifier=None,
          max_concurrency=1,
          timeout=None,
          ephemeral_lease=False,
-         ):
+         profile=None,
+         scheme=None,
+         username=None,
+         password=None,
+         default_acl=None):
     '''
     Block state execution until you are able to get the lock (or hit the timeout)
 
@@ -74,6 +81,8 @@ def lock(name,
            'changes': {},
            'result': False,
            'comment': ''}
+    conn_kwargs = {'profile': profile, 'scheme': scheme,
+                   'username': username, 'password': password, 'default_acl': default_acl}
 
     if __opts__['test']:
         ret['result'] = None
@@ -88,7 +97,8 @@ def lock(name,
                                              identifier=identifier,
                                              max_concurrency=max_concurrency,
                                              timeout=timeout,
-                                             ephemeral_lease=ephemeral_lease)
+                                             ephemeral_lease=ephemeral_lease,
+                                             **conn_kwargs)
     if locked:
         ret['result'] = True
         ret['comment'] = 'lock acquired'
@@ -102,8 +112,12 @@ def unlock(name,
            zk_hosts=None,  # in case you need to unlock without having run lock (failed execution for example)
            identifier=None,
            max_concurrency=1,
-           ephemeral_lease=False
+           ephemeral_lease=False,
-           ):
+           profile=None,
+           scheme=None,
+           username=None,
+           password=None,
+           default_acl=None):
     '''
     Remove lease from semaphore.
     '''
@@ -111,6 +125,8 @@ def unlock(name,
            'changes': {},
            'result': False,
            'comment': ''}
+    conn_kwargs = {'profile': profile, 'scheme': scheme,
+                   'username': username, 'password': password, 'default_acl': default_acl}
 
     if __opts__['test']:
         ret['result'] = None
@@ -124,7 +140,8 @@ def unlock(name,
                                                  zk_hosts=zk_hosts,
                                                  identifier=identifier,
                                                  max_concurrency=max_concurrency,
-                                                 ephemeral_lease=ephemeral_lease)
+                                                 ephemeral_lease=ephemeral_lease,
+                                                 **conn_kwargs)
 
     if unlocked:
         ret['result'] = True
@@ -137,8 +154,12 @@ def unlock(name,
 def min_party(name,
               zk_hosts,
               min_nodes,
-              blocking=False
+              blocking=False,
-              ):
+              profile=None,
+              scheme=None,
+              username=None,
+              password=None,
+              default_acl=None):
     '''
     Ensure that there are `min_nodes` in the party at `name`, optionally blocking if not available.
     '''
@@ -146,13 +167,15 @@ def min_party(name,
            'changes': {},
            'result': False,
            'comment': ''}
+    conn_kwargs = {'profile': profile, 'scheme': scheme,
+                   'username': username, 'password': password, 'default_acl': default_acl}
 
     if __opts__['test']:
         ret['result'] = None
         ret['comment'] = 'Attempt to ensure min_party'
         return ret
 
-    nodes = __salt__['zk_concurrency.party_members'](name, zk_hosts, min_nodes, blocking=blocking)
+    nodes = __salt__['zk_concurrency.party_members'](name, zk_hosts, min_nodes, blocking=blocking, **conn_kwargs)
     if not isinstance(nodes, list):
         raise Exception('Error from zk_concurrency.party_members, return was not a list: {0}'.format(nodes))
 

salt/states/zookeeper.py

@@ -0,0 +1,358 @@
+# -*- coding: utf-8 -*-
+'''
+:depends:  kazoo
+:configuration: See :py:mod:`salt.modules.zookeeper` for setup instructions.
+
+ACLS
+~~~~
+
+For more information about acls, please checkout the kazoo documentation.
+
+http://kazoo.readthedocs.io/en/latest/api/security.html#kazoo.security.make_digest_acl
+
+The following options can be included in the acl dictionary:
+
+    :param username: Username to use for the ACL.
+    :param password: A plain-text password to hash.
+    :param write: Write permission.
+    :type write: bool
+    :param create: Create permission.
+    :type create: bool
+    :param delete: Delete permission.
+    :type delete: bool
+    :param admin: Admin permission.
+    :type admin: bool
+    :param all: All permissions.
+    :type all: bool
+'''
+__virtualname__ = 'zookeeper'
+
+
+def __virtual__():
+    if 'zookeeper.create' in __salt__:
+        return __virtualname__
+    return False
+
+
+def _check_acls(left, right):
+    first = not bool(set(left) - set(right))
+    second = not bool(set(right) - set(left))
+    return first and second
+
+
+def present(name, value, acls=None, ephemeral=False, sequence=False, makepath=False, version=-1,
+            profile=None, hosts=None, scheme=None, username=None, password=None, default_acl=None):
+    '''
+    Make sure znode is present in the correct state with the correct acls
+
+    name
+        path to znode
+
+    value
+        value znode should be set to
+
+    acls
+        list of acl dictionaries to set on znode (make sure the ones salt is connected with are included)
+        Default: None
+
+    ephemeral
+        Boolean to indicate if ephemeral znode should be created
+        Default: False
+
+    sequence
+        Boolean to indicate if znode path is suffixed with a unique index
+        Default: False
+
+    makepath
+        Boolean to indicate if the parent paths should be created
+        Default: False
+
+    version
+        For updating, specify the version which should be updated
+        Default: -1 (always match)
+
+    profile
+        Configured Zookeeper profile to authenticate with (Default: None)
+
+    hosts
+        Lists of Zookeeper Hosts (Default: '127.0.0.1:2181)
+
+    scheme
+        Scheme to authenticate with (Default: 'digest')
+
+    username
+        Username to authenticate (Default: None)
+
+    password
+        Password to authenticate (Default: None)
+
+    default_acl
+        Default acls to assign if a node is created in this connection (Default: None)
+
+    .. code-block:: yaml
+
+        add znode:
+          zookeeper.present:
+            - name: /test/name
+            - value: gtmanfred
+            - makepath: True
+
+        update znode:
+          zookeeper.present:
+            - name: /test/name
+            - value: daniel
+            - acls:
+              - username: daniel
+                password: test
+                read: true
+              - username: gtmanfred
+                password: test
+                read: true
+                write: true
+                create: true
+                delete: true
+                admin: true
+            - makepath: True
+    '''
+
+    ret = {'name': name,
+           'result': False,
+           'comment': 'Failed to setup znode {0}'.format(name),
+           'changes': {}}
+    connkwargs = {'profile': profile, 'hosts': hosts, 'scheme': scheme,
+                  'username': username, 'password': password,
+                  'default_acl': default_acl}
+    if acls is None:
+        chk_acls = []
+    else:
+        chk_acls = [__salt__['zookeeper.make_digest_acl'](**acl) for acl in acls]
+    if __salt__['zookeeper.exists'](name, **connkwargs):
+        cur_value = __salt__['zookeeper.get'](name, **connkwargs)
+        cur_acls = __salt__['zookeeper.get_acls'](name, **connkwargs)
+        if cur_value == value and _check_acls(cur_acls, chk_acls):
+            ret['result'] = True
+            ret['comment'] = 'Znode {0} is already set to the correct value with the correct acls'.format(name)
+            return ret
+        elif __opts__['test'] is True:
+            ret['result'] = None
+            ret['comment'] = 'Znode {0} is will be updated'.format(name)
+            ret['changes']['old'] = {}
+            ret['changes']['new'] = {}
+            if value != cur_value:
+                ret['changes']['old']['value'] = cur_value
+                ret['changes']['new']['value'] = value
+            if not _check_acls(chk_acls, cur_acls):
+                ret['changes']['old']['acls'] = cur_acls
+                ret['changes']['new']['acls'] = chk_acls
+            return ret
+        else:
+            value_result, acl_result = True, True
+            changes = {}
+            if value != cur_value:
+                __salt__['zookeeper.set'](name, value, version, **connkwargs)
+                new_value = __salt__['zookeeper.get'](name, **connkwargs)
+                value_result = new_value == value
+                changes.setdefault('new', {}).setdefault('value', new_value)
+                changes.setdefault('old', {}).setdefault('value', cur_value)
+            if not _check_acls(chk_acls, cur_acls):
+                __salt__['zookeeper.set_acls'](name, acls, version, **connkwargs)
+                new_acls = __salt__['zookeeper.get_acls'](name, **connkwargs)
+                acl_result = _check_acls(new_acls, chk_acls)
+                changes.setdefault('new', {}).setdefault('acls', new_acls)
+                changes.setdefault('old', {}).setdefault('value', cur_acls)
+            ret['changes'] = changes
+            if value_result and acl_result:
+                ret['result'] = True
+                ret['comment'] = 'Znode {0} successfully updated'.format(name)
+            return ret
+
+    if __opts__['test'] is True:
+        ret['result'] = None
+        ret['comment'] = '{0} is will be created'.format(name)
+        ret['changes']['old'] = {}
+        ret['changes']['new'] = {}
+        ret['changes']['new']['acls'] = chk_acls
+        ret['changes']['new']['value'] = value
+        return ret
+
+    __salt__['zookeeper.create'](name, value, acls, ephemeral, sequence, makepath, **connkwargs)
+
+    value_result, acl_result = True, True
+    changes = {'old': {}}
+
+    new_value = __salt__['zookeeper.get'](name, **connkwargs)
+    value_result = new_value == value
+    changes.setdefault('new', {}).setdefault('value', new_value)
+
+    new_acls = __salt__['zookeeper.get_acls'](name, **connkwargs)
+    acl_result = acls is None or _check_acls(new_acls, chk_acls)
+    changes.setdefault('new', {}).setdefault('acls', new_acls)
+
+    ret['changes'] = changes
+    if value_result and acl_result:
+        ret['result'] = True
+        ret['comment'] = 'Znode {0} successfully created'.format(name)
+
+    return ret
+
+
+def absent(name, version=-1, recursive=False, profile=None, hosts=None, scheme=None,
+           username=None, password=None, default_acl=None):
+    '''
+    Make sure znode is absent
+
+    name
+        path to znode
+
+    version
+        Specify the version which should be deleted
+        Default: -1 (always match)
+
+    recursive
+        Boolean to indicate if children should be recursively deleted
+        Default: False
+
+    profile
+        Configured Zookeeper profile to authenticate with (Default: None)
+
+    hosts
+        Lists of Zookeeper Hosts (Default: '127.0.0.1:2181)
+
+    scheme
+        Scheme to authenticate with (Default: 'digest')
+
+    username
+        Username to authenticate (Default: None)
+
+    password
+        Password to authenticate (Default: None)
+
+    default_acl
+        Default acls to assign if a node is created in this connection (Default: None)
+
+    .. code-block:: yaml
+
+        delete znode:
+          zookeeper.absent:
+            - name: /test
+            - recursive: True
+    '''
+    ret = {'name': name,
+           'result': False,
+           'comment': 'Failed to delete znode {0}'.format(name),
+           'changes': {}}
+    connkwargs = {'profile': profile, 'hosts': hosts, 'scheme': scheme,
+                  'username': username, 'password': password,
+                  'default_acl': default_acl}
+
+    if __salt__['zookeeper.exists'](name, **connkwargs) is False:
+        ret['result'] = True
+        ret['comment'] = 'Znode {0} does not exist'.format(name)
+        return ret
+
+    changes = {}
+    changes['value'] = __salt__['zookeeper.get'](name, **connkwargs)
+    changes['acls'] = __salt__['zookeeper.get_acls'](name, **connkwargs)
+    if recursive is True:
+        changes['children'] = __salt__['zookeeper.get_children'](name, **connkwargs)
+
+    if __opts__['test'] is True:
+        ret['result'] = None
+        ret['comment'] = 'Znode {0} will be removed'.format(name)
+        ret['changes']['old'] = changes
+        return ret
+
+    __salt__['zookeeper.delete'](name, version, recursive, **connkwargs)
+
+    if __salt__['zookeeper.exists'](name, **connkwargs) is False:
+        ret['result'] = True
+        ret['comment'] = 'Znode {0} has been removed'.format(name)
+        ret['changes']['old'] = changes
+
+    return ret
+
+
+def acls(name, acls, version=-1, profile=None, hosts=None, scheme=None,
+         username=None, password=None, default_acl=None):
+    '''
+    Update acls on a znode
+
+    name
+        path to znode
+
+    acls
+        list of acl dictionaries to set on znode
+
+    version
+        Specify the version which should be deleted
+        Default: -1 (always match)
+
+    profile
+        Configured Zookeeper profile to authenticate with (Default: None)
+
+    hosts
+        Lists of Zookeeper Hosts (Default: '127.0.0.1:2181)
+
+    scheme
+        Scheme to authenticate with (Default: 'digest')
+
+    username
+        Username to authenticate (Default: None)
+
+    password
+        Password to authenticate (Default: None)
+
+    default_acl
+        Default acls to assign if a node is created in this connection (Default: None)
+
+    .. code-block:: yaml
+
+        update acls:
+          zookeeper.acls:
+            - name: /test/name
+            - acls:
+              - username: daniel
+                password: test
+                all: True
+              - username: gtmanfred
+                password: test
+                all: True
+    '''
+    ret = {'name': name,
+           'result': False,
+           'comment': 'Failed to set acls on znode {0}'.format(name),
+           'changes': {}}
+    connkwargs = {'profile': profile, 'hosts': hosts, 'scheme': scheme,
+                  'username': username, 'password': password,
+                  'default_acl': default_acl}
+    if isinstance(acls, dict):
+        acls = [acls]
+    chk_acls = [__salt__['zookeeper.make_digest_acl'](**acl) for acl in acls]
+
+    if not __salt__['zookeeper.exists'](name, **connkwargs):
+        ret['comment'] += ': Znode does not exist'
+        return ret
+
+    cur_acls = __salt__['zookeeper.get_acls'](name, **connkwargs)
+    if _check_acls(cur_acls, chk_acls):
+        ret['result'] = True
+        ret['comment'] = 'Znode {0} acls already set'.format(name)
+        return ret
+
+    if __opts__['test'] is True:
+        ret['result'] = None
+        ret['comment'] = 'Znode {0} acls will be updated'.format(name)
+        ret['changes']['old'] = cur_acls
+        ret['changes']['new'] = chk_acls
+        return ret
+
+    __salt__['zookeeper.set_acls'](name, acls, version, **connkwargs)
+
+    new_acls = __salt__['zookeeper.get_acls'](name, **connkwargs)
+    ret['changes'] = {'old': cur_acls, 'new': new_acls}
+    if _check_acls(new_acls, chk_acls):
+        ret['result'] = True
+        ret['comment'] = 'Znode {0} acls updated'.format(name)
+        return ret
+    ret['comment'] = 'Znode {0} acls failed to update'.format(name)
+    return ret