Browse Source

Implements Group API for the Consul driver

Also fixes tox env for py38-consul tests
Update zuul script to use consul 1.7.4 (2020) vs 0.6.3 (2016)

Adds ACL tokens to all session management calls.

Change-Id: Iaddf21f14c434129541e7c9ec7134e0661f7be52
changes/05/648805/25 2.6.0
Nick Maludy 2 years ago
committed by Jacob Floyd
No known key found for this signature in database GPG Key ID: 1A72D048C9AA75BE
6 changed files with 330 additions and 25 deletions
  1. +1
  2. +1
  3. +2
  4. +7
  5. +1
  6. +318

+ 1
- 1
.zuul.yaml View File

@ -57,7 +57,7 @@
description: |
Run tests using ``py38-consul`` environment.
tox_envlist: mysql-python
tox_envlist: py38-consul
- job:
name: tooz-tox-py38-etcd

+ 1
- 1
doc/source/user/compatibility.rst View File

@ -30,7 +30,7 @@ Driver support
* - Driver
- Supported
* - :py:class:`~tooz.drivers.consul.ConsulDriver`
- No
- Yes
* - :py:class:`~tooz.drivers.etcd.EtcdDriver`
- No
* - :py:class:`~tooz.drivers.file.FileDriver`

+ 2
- 2
doc/source/user/drivers.rst View File

@ -243,8 +243,8 @@ Consul
The `consul`_ driver is a driver providing only distributed locks (for now)
and is based on the consul server key/value storage and/or
The `consul`_ driver is a driver providing distributed locking and group
membership and is based on the consul server key/value storage and/or
primitives. When a lock is acquired it will release either when explicitly
released or automatically when the consul session ends (for example if
the program using the lock crashes).

+ 7
- 1 View File

@ -2,7 +2,13 @@
set -eux
if [ -z "$(which consul)" ]; then
# originally we used 0.6.3 (released in Jan 2016).
# updated to 1.7.4 in Change-Id: Iaddf21f14c434129541e7c9ec7134e0661f7be52
# 1.4.0 (released Nov 2018) has a new ACL system.
# 1.6.1 (released Sep 2019) is the version used by python-consul2 for testing.
# 1.7.0 (released Feb 2020) changes standards enforcement.
# For details see upgrade notes in Change-Id: I98fc96468b21368ce66365e3fc38c495b1f2918a
case `uname -s` in

+ 1
- 1
tools/ View File

@ -73,7 +73,7 @@ print_header("Driver support", delim="-")
grouping_table = [
"No", # Consul
"Yes", # Consul
"No", # Etcd
"Yes", # File
"No", # IPC

+ 318
- 19
tooz/drivers/ View File

@ -14,8 +14,12 @@
# License for the specific language governing permissions and limitations
# under the License.
import contextlib
import functools
import consul
from oslo_utils import encodeutils
import requests
import tooz
from tooz import _retry
@ -24,6 +28,33 @@ from tooz import locking
from tooz import utils
def _failure_translator():
"""Translates common consul exceptions into tooz exceptions."""
except (consul.Timeout, requests.exceptions.RequestException) as e:
except (consul.ConsulException, ValueError) as e:
# ValueError = Typically json decoding failed for some reason.
def _translate_failures(func):
def wrapper(*args, **kwargs):
with _failure_translator():
return func(*args, **kwargs)
return wrapper
class ConsulLock(locking.Lock):
def __init__(self, name, node, address, session_id, client, token=None):
super(ConsulLock, self).__init__(name)
@ -40,6 +71,7 @@ class ConsulLock(locking.Lock):
raise tooz.NotImplemented
def _acquire():
# Check if we are the owner and if we are simulate
# blocking (because consul will not block a second
@ -54,7 +86,7 @@ class ConsulLock(locking.Lock):
# The value can be anything.
gotten = self._client.kv.put(key=self._name,
value=u"I got it!",
value="I got it!",
if gotten:
@ -87,11 +119,12 @@ class ConsulLock(locking.Lock):
return False
class ConsulDriver(coordination.CoordinationDriver):
class ConsulDriver(coordination.CoordinationDriverCachedRunWatchers,
"""This driver uses `python-consul`_ client against `consul`_ servers.
The ConsulDriver implements a minimal set of coordination driver API(s)
needed to make Consul being used as an option for Distributed Locking. The
The ConsulDriver implements the coordination driver API(s) so that Consul
can be used as an option for Distributed Locking and Group Membership. The
data is stored in Consul's key-value store.
The Consul driver connection URI should look like::
@ -110,14 +143,43 @@ class ConsulDriver(coordination.CoordinationDriver):
================== =======
For details on the available options, refer to
.. _python-consul:
The following Key/Value paths are utilized in Consul to implement the
coordination APIs:
| Key | Value | Description |
| <namespace>/groups/<group_id> | None | Group of |
| | | members. |
| <namespace>/groups/<group_id>/<member_id> | Member | Member in a |
| | capabilities | group. |
| | encoded as | |
| | msgpack | |
| <namespace>/group_locks/<group_id> | Consul | Lock for |
| | session ID | group |
| | | membership |
| <namespace>/locks/<name> | Consul | Each key is |
| | session ID | a distributed|
| | | lock. |
NOTE: A group in tooz is NOT the same as a Consul service, so tooz groups
are implemented using Consul Key/Values and not with native services.
Groups in tooz do not have host:port details or health checks that report
to consul to verify that the service is still alive and listening on that
host:port. If you need to use native Consul services, configure the Consul
agent directly (not via tooz).
.. _python-consul:
.. _consul:
#: Default namespace when none is provided
#: Default TTL
@ -128,6 +190,31 @@ class ConsulDriver(coordination.CoordinationDriver):
#: Consul ACL Token if not provided
# client liveness is based on more than just timeouts
# multiple service instances (group members) could register
# with different ports per thread/proc
# but the agent is always on locahost: so not DISTRIBUTED_ACROSS_HOSTS
# The consul consistency mode determines the history characteristics.
# Writes *always* go through a single leader process into raft log.
# Reads *may* use all servers depending on the request's mode:
# - 'consistent' = LINEARIZABLE
# - 'default' = SEQUENTIAL
# - 'stale' = CAUSAL
coordination.Characteristics.SEQUENTIAL, # 'default' consistency mode
# raft log of servers is snapshotted + compacted
Tuple of :py:class:`~tooz.coordination.Characteristics` introspectable
enum member(s) that can be used to interogate how this driver works.
def __init__(self, member_id, parsed_url, options):
super(ConsulDriver, self).__init__(member_id, parsed_url, options)
options = utils.collapse(options)
@ -139,6 +226,8 @@ class ConsulDriver(coordination.CoordinationDriver):
namespace = options.get('namespace', self.TOOZ_NAMESPACE)
self._namespace = encodeutils.safe_decode(namespace)
self._acl_token = options.get('acl_token', self.ACL_TOKEN)
# the empty group name adds a trailing / needed for lookups
self._groups_prefix = self._paths_join(self._namespace, "groups", "")
self._client = None
def _start(self):
@ -151,6 +240,7 @@ class ConsulDriver(coordination.CoordinationDriver):
local_agent = self._client.agent.self()
self._node = local_agent['Member']['Name']
self._address = local_agent['Member']['Addr']
# implicitly uses the agent's datacenter (set in consul agent config)
# Register a Node
@ -167,7 +257,8 @@ class ConsulDriver(coordination.CoordinationDriver):
def _stop(self):
if self._client is not None:
if self._session_id is not None:
self._session_id = None
self._client = None
@ -179,32 +270,240 @@ class ConsulDriver(coordination.CoordinationDriver):
# session has died or is unreachable. When a session expires all locks
# are released and any services that were registered with that session
# are marked as no longer active.
self._client.session.renew(self._session_id, token=self._acl_token)
# renew the session every half-TTL or 1 second, whatever is larger
sleep_sec = max(self._ttl / 2, 1)
return sleep_sec
def get_lock(self, name):
real_name = self._paths_join(self._namespace, u"locks", name)
def _get_lock(self, real_name):
return ConsulLock(real_name, self._node, self._address,
client=self._client, token=self._acl_token)
def get_lock(self, name):
real_name = self._path_lock(name)
return self._get_lock(real_name)
def _get_group_lock(self, group_id):
real_name = self._path_group_lock(group_id)
return self._get_lock(real_name)
def _paths_join(*args):
pieces = []
for arg in args:
return u"/".join(pieces)
return "/".join(pieces)
def _path_lock(self, name):
return self._paths_join(self._namespace, "locks", name)
def _path_group_lock(self, name):
return self._paths_join(self._namespace, "group_locks", name)
def _path_group(self, group_id):
# add an extra '/' at the end so that searches with this path
# will go down and find their children
return self._paths_join(self._namespace, "groups", group_id) + "/"
def _path_member(self, group_id, member_id):
return self._paths_join(
self._namespace, "groups", group_id, member_id
def _group_path_to_id(self, base_path, group_path):
"""Translates a path into a group name.
The group name is the last part of the path. So, we simply split on
the path separator '/' and return the last element.
group_id = self._path_to_group_id("tooz/groups/helloworld")
print(group_id) # "helloworld"
if group_path.startswith(base_path):
group_id = group_path[len(base_path):]
group_id = group_path
# if a group has members (sub-keys) it will contain a trailing /
# we need to strip this to get just the name
# if a group has no members there is no trailing / (for some reason)
group_id = group_id.strip("/")
return utils.to_binary(group_id)
def _get_group_members(self, group_path):
index, data = self._client.kv.get(group_path, recurse=True)
group = None
members = []
for kv in (data or []):
if kv["Key"] == group_path:
group = kv
return (group, members)
def get_groups(self):
def _get_groups():
groups = []
index, data = self._client.kv.get(self._groups_prefix, keys=True,
for key in (data or []):
if key != self._groups_prefix:
group_id = self._group_path_to_id(self._groups_prefix, key)
return groups
return ConsulFutureResult(self._executor.submit(_get_groups))
def create_group(self, group_id):
def _create_group():
group_path = self._path_group(group_id)
# create with Check-And-Set index 0 will only succeed if the key
# doesn't exit
result = self._client.kv.put(group_path, "", cas=0)
if not result:
raise coordination.GroupAlreadyExist(group_id)
return result
return ConsulFutureResult(self._executor.submit(_create_group))
def _destroy_group(self, group_id):
"""Should only be used in tests..."""
with self._get_group_lock(group_id) as lock:
group_path = self._path_group(group_id)
self._client.kv.delete(group_path, recurse=True)
def delete_group(self, group_id):
def _delete_group():
# create a lock for the group so that other operations on this
# group do not conflict while the group is being deleted
with self._get_group_lock(group_id) as lock:
group_path = self._path_group(group_id)
group, members = self._get_group_members(group_path)
if not group:
raise coordination.GroupNotCreated(group_id)
if members:
raise coordination.GroupNotEmpty(group_id)
# delete the group recursively
result = self._client.kv.delete(group_path, recurse=True)
# delete the lock for the group
return result
return ConsulFutureResult(self._executor.submit(_delete_group))
def join_group(self, group_id, capabilities=b""):
def _join_group():
# lock the group so that it doesn't get deleted while we join
with self._get_group_lock(group_id):
group_path = self._path_group(group_id)
member_path = self._path_member(group_id, self._member_id)
group, members = self._get_group_members(group_path)
if not group:
raise coordination.GroupNotCreated(group_id)
for m in members:
if m["Key"] == member_path:
raise coordination.MemberAlreadyExist(group_id,
# create with Check-And-Set index 0 will only succeed if the
# key doesn't exit
self._client.kv.put(member_path, utils.dumps(capabilities),
return ConsulFutureResult(self._executor.submit(_join_group))
def leave_group(self, group_id):
def _leave_group():
# NOTE: We do NOT have to lock the group here because deletes in
# Consul are atomic and succeed even if the key doesn't exist
# This means that there is no race condition between checking
# if the member exists and then deleting it.
group_path = self._path_group(group_id)
member_path = self._path_member(group_id, self._member_id)
group, members = self._get_group_members(group_path)
member = None
for m in members:
if m["Key"] == member_path:
member = m
raise coordination.MemberNotJoined(group_id, self._member_id)
def watch_join_group(self, group_id, callback):
raise tooz.NotImplemented
# delete the member key with Check-And-Set semantics based on index
# we read above
self._client.kv.delete(member_path, cas=member["ModifyIndex"])
def unwatch_join_group(self, group_id, callback):
raise tooz.NotImplemented
return ConsulFutureResult(self._executor.submit(_leave_group))
def get_members(self, group_id):
def _get_members():
group_path = self._path_group(group_id)
group, members = self._get_group_members(group_path)
if not group:
raise coordination.GroupNotCreated(group_id)
result = set()
for m in members:
member_id = self._group_path_to_id(group_path, m["Key"])
return result
return ConsulFutureResult(self._executor.submit(_get_members))
def get_member_capabilities(self, group_id, member_id):
def _get_member_capabilities():
member_path = self._path_member(group_id, member_id)
index, data = self._client.kv.get(member_path)
if not data:
raise coordination.MemberNotJoined(group_id, member_id)
return utils.loads(data["Value"])
def watch_leave_group(self, group_id, callback):
return ConsulFutureResult(
def update_capabilities(self, group_id, capabilities):
def _update_capabilities():
member_path = self._path_member(group_id, self._member_id)
index, data = self._client.kv.get(member_path)
if not data:
raise coordination.MemberNotJoined(group_id, self._member_id)
# no need to Check-And-Set here, latest write wins
self._client.kv.put(member_path, utils.dumps(capabilities))
return ConsulFutureResult(self._executor.submit(_update_capabilities))
def watch_elected_as_leader(group_id, callback):
raise tooz.NotImplemented
def unwatch_leave_group(self, group_id, callback):
def unwatch_elected_as_leader(group_id, callback):
raise tooz.NotImplemented
ConsulFutureResult = functools.partial(coordination.CoordinatorResult,