etcd3: add etcd3 coordination driver

This patch adds support to Tooz for taking advantage of the etcd3 gRPC-based
API (instead of the etcd2 HTTP/REST-based API) via the python-etcd3 library.

Change-Id: Ic7c3d9be42a9912fcb09c43e7637270db4011c4a

setup.cfg

@@ -26,6 +26,7 @@ packages =
 [entry_points]
 tooz.backends =
     etcd = tooz.drivers.etcd:EtcdDriver
+    etcd3 = tooz.drivers.etcd3:Etcd3Driver
     kazoo = tooz.drivers.zookeeper:KazooDriver
     zake = tooz.drivers.zake:ZakeDriver
     memcached = tooz.drivers.memcached:MemcachedDriver
@@ -42,6 +43,8 @@ consul =
   python-consul>=0.4.7 # MIT License
 etcd =
   requests>=2.10.0 # Apache-2.0
+etcd3 =
+  etcd3>=0.5.1 # Apache-2.0
 zake =
   zake>=0.1.6 # Apache-2.0
 redis =

tooz/drivers/etcd3.py

@@ -0,0 +1,121 @@
+# -*- coding: utf-8 -*-
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+
+from __future__ import absolute_import
+
+import etcd3
+from etcd3 import exceptions as etcd3_exc
+from oslo_utils import encodeutils
+import six
+
+from tooz import coordination
+from tooz import locking
+from tooz import utils
+
+
+def _translate_failures(func):
+    """Translates common requests exceptions into tooz exceptions."""
+
+    @six.wraps(func)
+    def wrapper(*args, **kwargs):
+        try:
+            return func(*args, **kwargs)
+        except etcd3_exc.ConnectionFailedError as e:
+            utils.raise_with_cause(coordination.ToozConnectionError,
+                                   encodeutils.exception_to_unicode(e),
+                                   cause=e)
+        except etcd3_exc.ConnectionTimeoutError as e:
+            utils.raise_with_cause(coordination.OperationTimedOut,
+                                   encodeutils.exception_to_unicode(e),
+                                   cause=e)
+        except etcd3_exc.Etcd3Exception as e:
+            utils.raise_with_cause(coordination.ToozError,
+                                   encodeutils.exception_to_unicode(e),
+                                   cause=e)
+
+    return wrapper
+
+
+class Etcd3Lock(locking.Lock):
+    """An etcd3-specific lock.
+
+    Thin wrapper over etcd3's lock object basically to provide the heartbeat()
+    semantics for the coordination driver.
+    """
+
+    def __init__(self, etcd3_lock, coord):
+        super(Etcd3Lock, self).__init__(etcd3_lock.name)
+        self._lock = etcd3_lock
+        self._coord = coord
+
+    @_translate_failures
+    def acquire(self, *args, **kwargs):
+        res = self._lock.acquire()
+        if res:
+            self._coord._acquired_locks.add(self)
+        return res
+
+    @_translate_failures
+    def release(self, *args, **kwargs):
+        if self._lock.is_acquired():
+            res = self._lock.release()
+            if res:
+                self._coord._acquired_locks.remove(self)
+
+    @_translate_failures
+    def break_(self):
+        res = self._lock.release()
+        if res:
+            self._coord._acquired_locks.remove(self)
+
+    @_translate_failures
+    def heartbeat(self):
+        return self._lock.refresh()
+
+
+class Etcd3Driver(coordination.CoordinationDriver):
+    """An etcd based driver.
+
+    This driver uses etcd provide the coordination driver semantics and
+    required API(s).
+    """
+
+    #: Default socket/lock/member/leader timeout used when none is provided.
+    DEFAULT_TIMEOUT = 30
+
+    #: Default hostname used when none is provided.
+    DEFAULT_HOST = "localhost"
+
+    #: Default port used if none provided (4001 or 2379 are the common ones).
+    DEFAULT_PORT = 2379
+
+    def __init__(self, member_id, parsed_url, options):
+        super(Etcd3Driver, self).__init__(member_id)
+        host = parsed_url.hostname or self.DEFAULT_HOST
+        port = parsed_url.port or self.DEFAULT_PORT
+        options = utils.collapse(options)
+        timeout = int(options.get('timeout', self.DEFAULT_TIMEOUT))
+        self.client = etcd3.client(host=host, port=port, timeout=timeout)
+        self.lock_timeout = int(options.get('lock_timeout', timeout))
+        self._acquired_locks = set()
+
+    def get_lock(self, name):
+        etcd3_lock = self.client.lock(name, ttl=self.lock_timeout)
+        return Etcd3Lock(etcd3_lock, self)
+
+    def heartbeat(self):
+        # NOTE(jaypipes): Copying because set can mutate during iteration
+        for lock in self._acquired_locks.copy():
+            lock.heartbeat()
+        return self.lock_timeout

tooz/tests/test_etcd3.py

@@ -0,0 +1,50 @@
+# -*- coding: utf-8 -*-
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+import mock
+import testtools
+
+try:
+    from etcd3 import exceptions as etcd3_exc
+    ETCD3_AVAILABLE = True
+except ImportError:
+    ETCD3_AVAILABLE = False
+
+from tooz import coordination
+
+
+@testtools.skipUnless(ETCD3_AVAILABLE, 'etcd3 is not available')
+class TestEtcd3(testtools.TestCase):
+    FAKE_URL = "etcd3://mocked-not-really-localhost:2379"
+    FAKE_MEMBER_ID = "mocked-not-really-member"
+
+    def setUp(self):
+        super(TestEtcd3, self).setUp()
+        self._coord = coordination.get_coordinator(self.FAKE_URL,
+                                                   self.FAKE_MEMBER_ID)
+
+    def test_error_translation(self):
+        lock = self._coord.get_lock('mocked-not-really-random')
+
+        exc = etcd3_exc.ConnectionFailedError()
+        with mock.patch.object(lock._lock, 'acquire', side_effect=exc):
+            self.assertRaises(coordination.ToozConnectionError, lock.acquire)
+
+        exc = etcd3_exc.ConnectionTimeoutError()
+        with mock.patch.object(lock._lock, 'acquire', side_effect=exc):
+            self.assertRaises(coordination.OperationTimedOut, lock.acquire)
+
+        exc = etcd3_exc.InternalServerError()
+        with mock.patch.object(lock._lock, 'acquire', side_effect=exc):
+            self.assertRaises(coordination.ToozError, lock.acquire)

tox.ini

@@ -1,11 +1,11 @@
 [tox]
 minversion = 1.8
-envlist = py27,py35,py27-zookeeper,py35-zookeeper,py27-redis,py35-redis,py27-sentinel,py35-sentinel,py27-memcached,py35-memcached,py27-postgresql,py35-postgresql,py27-mysql,py35-mysql,py27-consul,py35-consul,pep8
+envlist = py27,py35,py27-zookeeper,py35-zookeeper,py27-redis,py35-redis,py27-sentinel,py35-sentinel,py27-memcached,py35-memcached,py27-postgresql,py35-postgresql,py27-mysql,py35-mysql,py27-consul,py35-consul,py27-etcd3,py35-etcd3,pep8
 
 [testenv]
 # We need to install a bit more than just `test' because those drivers have
 # custom tests that we always run
-deps = .[test,zake,ipc,memcached,mysql,etcd]
+deps = .[test,zake,ipc,memcached,mysql,etcd,etcd3]
     zookeeper: .[zookeeper]
     redis: .[redis]
     sentinel: .[redis]
@@ -13,6 +13,7 @@ deps = .[test,zake,ipc,memcached,mysql,etcd]
     postgresql: .[postgresql]
     mysql: .[mysql]
     etcd: .[etcd]
+    etcd3: .[etcd3]
     consul: .[consul]
 setenv =
     TOOZ_TEST_URLS = file:///tmp zake:// ipc://
@@ -70,6 +71,14 @@ commands = {toxinidir}/setup-etcd-env.sh pifpaf -e TOOZ_TEST run etcd --  {toxin
 commands = {toxinidir}/setup-etcd-env.sh pifpaf -e TOOZ_TEST run etcd -- {toxinidir}/tools/pretty_tox.sh "{posargs}"
            {toxinidir}/setup-etcd-env.sh pifpaf -e TOOZ_TEST run etcd --cluster -- {toxinidir}/tools/pretty_tox.sh "{posargs}"
 
+[testenv:py27-etcd3]
+commands = {toxinidir}/setup-etcd-env.sh pifpaf -e TOOZ_TEST run etcd --  {toxinidir}/tools/pretty_tox.sh "{posargs}"
+           {toxinidir}/setup-etcd-env.sh pifpaf -e TOOZ_TEST run etcd --cluster --  {toxinidir}/tools/pretty_tox.sh "{posargs}"
+
+[testenv:py35-etcd3]
+commands = {toxinidir}/setup-etcd-env.sh pifpaf -e TOOZ_TEST run etcd -- {toxinidir}/tools/pretty_tox.sh "{posargs}"
+           {toxinidir}/setup-etcd-env.sh pifpaf -e TOOZ_TEST run etcd --cluster -- {toxinidir}/tools/pretty_tox.sh "{posargs}"
+
 [testenv:py27-consul]
 commands = {toxinidir}/setup-consul-env.sh pifpaf -e TOOZ_TEST run consul -- {toxinidir}/tools/pretty_tox.sh "{posargs}"