Add Kubernetes Lock support

Depends-On: https://review.opendev.org/c/openstack/requirements/+/935735

Change-Id: I1fc434bd774f940de1684a9bb7dd8ce5686828b4

doc/requirements.txt

@@ -23,3 +23,6 @@ kazoo>=2.2 # Apache-2.0
 pymemcache>=1.2.9 # Apache 2.0 License
 ## ipc
 sysv-ipc>=0.6.8 # BSD License
+## kubernetes
+sherlock>=0.4.1 # MIT License
+kubernetes>=2.8.1 # Apache-2.0

doc/source/reference/index.rst

@@ -38,6 +38,12 @@ IPC
 .. autoclass:: tooz.drivers.ipc.IPCDriver
    :members:
 
+Kubernetes
+~~~~~~~~~~
+
+.. autoclass:: tooz.drivers.kubernetes.SherlockDriver
+   :members:
+
 Memcached
 ~~~~~~~~~
 

doc/source/user/compatibility.rst

@@ -37,6 +37,8 @@ Driver support
      - Yes
    * - :py:class:`~tooz.drivers.ipc.IPCDriver`
      - No
+   * - :py:class:`~tooz.drivers.kubernetes.SherlockDriver`
+     - No
    * - :py:class:`~tooz.drivers.memcached.MemcachedDriver`
      - Yes
    * - :py:class:`~tooz.drivers.mysql.MySQLDriver`
@@ -77,6 +79,8 @@ Driver support
      - No
    * - :py:class:`~tooz.drivers.ipc.IPCDriver`
      - No
+   * - :py:class:`~tooz.drivers.kubernetes.SherlockDriver`
+     - No
    * - :py:class:`~tooz.drivers.memcached.MemcachedDriver`
      - Yes
    * - :py:class:`~tooz.drivers.mysql.MySQLDriver`
@@ -114,6 +118,8 @@ Driver support
      - Yes
    * - :py:class:`~tooz.drivers.ipc.IPCDriver`
      - Yes
+   * - :py:class:`~tooz.drivers.kubernetes.SherlockDriver`
+     - Yes
    * - :py:class:`~tooz.drivers.memcached.MemcachedDriver`
      - Yes
    * - :py:class:`~tooz.drivers.mysql.MySQLDriver`

doc/source/user/drivers.rst

@@ -234,6 +234,22 @@ 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).
 
+Kubernetes
+----------
+
+**Driver:**  :py:class:`tooz.drivers.kubernetes.SherlockDriver`
+
+**Characteristics:**
+
+:py:attr:`tooz.drivers.kubernetes.SherlockDriver.CHARACTERISTICS`
+
+**Entrypoint name:** ``kubernetes``
+
+**Summary:**
+
+The `sherlock`_ driver is a driver providing `kubernetes`_ distributed locking
+that based on Kubernetes Lease API.
+
 Characteristics
 ---------------
 
@@ -249,3 +265,5 @@ Characteristics
 .. _MySQL database server: http://mysql.org
 .. _redis-sentinel: http://redis.io/topics/sentinel
 .. _GRPC Gateway: https://github.com/grpc-ecosystem/grpc-gateway
+.. _kubernetes: https://kubernetes.io/
+.. _sherlock: https://sher-lock.readthedocs.io/en/latest/

releasenotes/notes/add-k8s-lock-support-5c9e5c6d4bbe8405.yaml

@@ -0,0 +1,8 @@
+---
+features:
+  - |
+    Add `kubernetes` driver that support basic lock managements.
+    This is directly using kubernetes config paths from environment,
+    so no need to expose or set extra client settings for
+    authentication in tooz. Please reference [1] for more detail.
+    [1] https://github.com/kubernetes-client/python/blob/master/README.md

setup.cfg

@@ -42,6 +42,7 @@ tooz.backends =
     file = tooz.drivers.file:FileDriver
     zookeeper = tooz.drivers.zookeeper:KazooDriver
     consul = tooz.drivers.consul:ConsulDriver
+    kubernetes = tooz.drivers.kubernetes:SherlockDriver
 
 [extras]
 consul =
@@ -64,3 +65,6 @@ memcached =
   pymemcache>=1.2.9 # Apache 2.0 License
 ipc =
   sysv-ipc>=0.6.8 # BSD License
+kubernetes =
+  kubernetes>=2.8.1 # Apache-2.0
+  sherlock>=0.4.1 # MIT License

tools/compat-matrix.py

@@ -35,6 +35,7 @@ driver_class_names = [
     "etcd.EtcdDriver",
     "file.FileDriver",
     "ipc.IPCDriver",
+    "kubernetes.SherlockDriver",
     "memcached.MemcachedDriver",
     "mysql.MySQLDriver",
     "pgsql.PostgresDriver",
@@ -75,6 +76,7 @@ grouping_table = [
         "Yes",  # Etcd
         "Yes",  # File
         "No",  # IPC
+        "No",  # Kubernetes
         "Yes",  # Memcached
         "No",  # MySQL
         "No",  # PostgreSQL
@@ -107,6 +109,7 @@ leader_table = [
         "No",  # Etcd
         "No",  # File
         "No",  # IPC
+        "No",  # Kubernetes
         "Yes",  # Memcached
         "No",  # MySQL
         "No",  # PostgreSQL
@@ -136,6 +139,7 @@ lock_table = [
         "Yes",  # Etcd
         "Yes",  # File
         "Yes",  # IPC
+        "Yes",  # Kubernetes
         "Yes",  # Memcached
         "Yes",  # MySQL
         "Yes",  # PostgreSQL

tooz/drivers/kubernetes.py

@@ -0,0 +1,117 @@
+#
+#    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 kubernetes.client import exceptions as k8s_exc
+import sherlock
+
+import tooz
+from tooz import coordination
+from tooz import locking
+from tooz import utils
+
+
+class KubernetesLock(locking.Lock):
+    def __init__(self, name, namespace, lock):
+        super().__init__(name)
+        self._name = name
+        self._namespace = namespace
+        self._lock = lock
+        self._client = lock.client
+
+    def is_still_owner(self):
+        if not self._lock.locked():
+            return False
+        try:
+            holder = self._client.read_namespaced_lease(
+                self._name, self._namespace
+            ).spec.holder_identity
+            if holder == self._lock._owner:
+                return True
+        except k8s_exc.ApiException as e:
+            if "Reason: Not Found" not in str(e):
+                utils.raise_with_cause(
+                    tooz.ToozError,
+                    f"operation error: {str(e)}",
+                    cause=e)
+        return False
+
+    def acquire(self, blocking=True, shared=False, expire=None):
+        if shared:
+            raise tooz.NotImplemented
+        blocking, timeout = utils.convert_blocking(blocking)
+        sherlock.configure(
+            expire=expire,
+            timeout=int(timeout) if timeout else timeout
+        )
+        return self._lock.acquire(blocking=blocking)
+
+    def release(self):
+        if self._lock.locked():
+            try:
+                self._lock.release()
+            except sherlock.lock.LockException as le:
+                msg = "Lock was not set by this process."
+                if msg in str(le):
+                    return True
+                else:
+                    raise
+            return True
+        else:
+            return False
+
+    @property
+    def acquired(self):
+        return (self._lock.locked() and self.is_still_owner())
+
+
+class SherlockDriver(coordination.CoordinationDriverCachedRunWatchers):
+    """This driver uses the `sherlock`_ client against `kubernetes`_ servers.
+
+    The Kubernetes coordinator url should look like::
+
+      kubernetes://[[?OPTION1=VALUE1[&OPTION2=VALUE2[&...]]]
+
+    Currently the following options will be proxied to the contained client:
+
+    ================  ===============================  ====================
+    Name              Source                           Default
+    ================  ===============================  ====================
+    namespace         'namespace' options key          openstack
+    ================  ===============================  ====================
+
+    .. _kubernetes: https://kubernetes.io/
+    .. _sherlock: https://sher-lock.readthedocs.io/en/latest/
+    """
+    #: Default namespace when none is provided.
+    K8S_NAMESPACE = "openstack"
+
+    CHARACTERISTICS = (
+        coordination.Characteristics.NON_TIMEOUT_BASED,
+        coordination.Characteristics.DISTRIBUTED_ACROSS_THREADS,
+        coordination.Characteristics.DISTRIBUTED_ACROSS_PROCESSES,
+        coordination.Characteristics.DISTRIBUTED_ACROSS_HOSTS,
+    )
+    """
+    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().__init__(member_id, parsed_url, options)
+        options = utils.collapse(options)
+        self._namespace = options.get('namespace', self.K8S_NAMESPACE)
+
+    def get_lock(self, name):
+        lock = sherlock.KubernetesLock(
+            lock_name=name, k8s_namespace=self._namespace)
+        return KubernetesLock(name=name, namespace=self._namespace, lock=lock)

tooz/tests/test_kubernetes.py

@@ -0,0 +1,80 @@
+# Copyright (c) 2015 OpenStack Foundation
+# All Rights Reserved.
+#
+#    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 unittest import mock
+
+from testtools import testcase
+
+from tooz import coordination
+from tooz import tests
+
+
+class TestSherlockDriver(testcase.TestCase):
+
+    def _create_coordinator(self, url="kubernetes://?namespace=fake_name"):
+        return coordination.get_coordinator(
+            url, tests.get_random_uuid())
+
+    def test_connect_k8s_driver(self):
+        c = self._create_coordinator()
+        self.assertIsNone(c.start())
+
+    @mock.patch("sherlock.KubernetesLock")
+    def test_parsing_timeout_settings(self, k8s_mock):
+        c = self._create_coordinator()
+
+        name = tests.get_random_uuid()
+        blocking_value = False
+        timeout = 10.1
+        lock = c.get_lock(name)
+        with mock.patch.object(
+            lock, 'acquire', wraps=True, autospec=True,
+            return_value=mock.Mock()
+        ) as mock_acquire:
+            with lock(blocking_value, timeout):
+                mock_acquire.assert_called_once_with(blocking_value, timeout)
+        k8s_mock.assert_called_once_with(
+            lock_name=mock.ANY, k8s_namespace='fake_name')
+
+    @mock.patch("sherlock.KubernetesLock")
+    def test_parsing_blocking_settings(self, k8s_mock):
+        c = self._create_coordinator()
+
+        name = tests.get_random_uuid()
+        blocking_value = True
+        lock = c.get_lock(name)
+        with mock.patch.object(
+            lock, 'acquire', wraps=True, autospec=True,
+            return_value=mock.Mock()
+        ) as mock_acquire:
+            with lock(blocking_value):
+                mock_acquire.assert_called_once_with(blocking_value)
+        k8s_mock.assert_called_once_with(
+            lock_name=mock.ANY, k8s_namespace='fake_name')
+
+    @mock.patch("sherlock.KubernetesLock")
+    @mock.patch("sherlock.configure")
+    def test_parsing_expire_settings(self, conf_mock, k8s_mock):
+        c = self._create_coordinator()
+
+        name = tests.get_random_uuid()
+        blocking_value = 20
+        expire_value = 10
+        lock = c.get_lock(name)
+        lock.acquire(blocking=blocking_value, expire=expire_value)
+        k8s_mock.assert_called_once_with(
+            lock_name=mock.ANY, k8s_namespace='fake_name')
+        conf_mock.assert_called_once_with(
+            expire=expire_value,
+            timeout=blocking_value)

tox.ini

@@ -1,13 +1,13 @@
 [tox]
 minversion = 3.18.0
-envlist = py3,py{39,312}-{zookeeper,redis,sentinel,memcached,postgresql,mysql,consul,etcd,etcd3gw},pep8
+envlist = py3,py{39,312}-{zookeeper,redis,sentinel,memcached,postgresql,mysql,consul,etcd,etcd3gw,kubernetes},pep8
 ignore_basepython_conflict = True
 
 [testenv]
 basepython = python3
 # We need to install a bit more than just `test-requirements' because those drivers have
 # custom tests that we always run
-deps = .[zake,ipc,memcached,mysql,etcd,etcd3gw]
+deps = .[zake,ipc,memcached,mysql,etcd,etcd3gw,kubernetes]
     zookeeper: .[zookeeper]
     redis: .[redis]
     sentinel: .[redis]
@@ -17,6 +17,7 @@ deps = .[zake,ipc,memcached,mysql,etcd,etcd3gw]
     etcd: .[etcd]
     etcd3gw: .[etcd3gw]
     consul: .[consul]
+    kubernetes: .[kubernetes]
     -r{toxinidir}/test-requirements.txt
 setenv =
     TOOZ_TEST_URLS = file:///tmp zake:// ipc://
@@ -31,6 +32,7 @@ setenv =
     etcd3gw: TOOZ_TEST_DRIVERS = etcd
     etcd3gw: TOOZ_TEST_ETCD3GW = 1
     consul: TOOZ_TEST_DRIVERS = consul
+    kubernetes: TOOZ_TEST_DRIVERS = kubernetes
 allowlist_externals =
     {toxinidir}/run-tests.sh
     {toxinidir}/run-examples.sh