Merge branch 'master' into dependabot/pip/pywin32-301

Author: judahrand

README.rst

@@ -199,17 +199,6 @@ Available `here`_.
 
 .. _here: http://sher-lock.readthedocs.org
 
-Roadmap
--------
-
-* Support for `Zookeeper`_ as backend.
-* Support for `Gevent`_, `Multithreading`_ and `Multiprocessing`_.
-
-.. _Zookeeper: http://zookeeper.apache.org/
-.. _Gevent: http://www.gevent.org/
-.. _Multithreading: http://docs.python.org/2/library/multithreading.html
-.. _Multiprocessing: http://docs.python.org/2/library/multiprocessing.html
-
 License
 -------
 

sherlock/lock.py

@@ -11,7 +11,6 @@
 import json
 import pathlib
 import re
-import signal
 import time
 import typing
 import uuid
@@ -207,8 +206,22 @@ def release(self):
 
         return self._release()
 
+    def _renew(self) -> bool:
+        """
+        Implementation of renewing an acquired lock. Must be implemented in
+        the sub-class.
+        """
+        raise NotImplementedError("Must be implemented in the sub-class")
+
+    def renew(self) -> bool:
+        """
+        Renew a lock that is already acquired.
+        """
+        return self._renew()
+
     def __enter__(self):
         self.acquire()
+        return self
 
     def __exit__(self, exc_type, exc_value, traceback):
         self.release()
@@ -331,6 +344,15 @@ def _release(self):
             )
         return self._lock_proxy.release()
 
+    def _renew(self) -> bool:
+        if self._lock_proxy is None:
+            raise LockException(
+                "Lock backend has not been configured and "
+                "lock cannot be acquired or released. "
+                "Configure lock backend first."
+            )
+        return self._lock_proxy.renew()
+
     @property
     def _locked(self):
         if self._lock_proxy is None:
@@ -406,6 +428,19 @@ class RedisLock(BaseLock):
     return result
     """
 
+    _renew_script = """
+    local result = 0
+    if redis.call('GET', KEYS[1]) == KEYS[2] then
+        if KEYS[3] ~= -1 then
+            redis.call('EXPIRE', KEYS[1], KEYS[3])
+        else
+            redis.call('PERSIST', KEYS[1])
+        end
+        result = 1
+    end
+    return result
+    """
+
     def __init__(self, lock_name, **kwargs):
         """
         :param str lock_name: name of the lock to uniquely identify the lock
@@ -434,6 +469,7 @@ def __init__(self, lock_name, **kwargs):
         # Register Lua script
         self._acquire_func = self.client.register_script(self._acquire_script)
         self._release_func = self.client.register_script(self._release_script)
+        self._renew_func = self.client.register_script(self._renew_script)
 
     @property
     def _key_name(self):
@@ -466,6 +502,14 @@ def _release(self):
 
         self._owner = None
 
+    def _renew(self) -> bool:
+        if self._owner is None:
+            raise LockException("Lock was not set by this process.")
+
+        if self._release_func(keys=[self._key_name, self._owner, self.expire]) != 1:
+            return False
+        return True
+
     @property
     def _locked(self):
         if self.client.get(self._key_name) is None:
@@ -555,10 +599,6 @@ def _key_name(self):
     def _acquire(self):
         owner = str(uuid.uuid4())
 
-        _args = [self._key_name, owner]
-        if self.expire is not None:
-            _args.append(self.expire)
-
         try:
             self.client.write(self._key_name, owner, prevExist=False, ttl=self.expire)
             self._owner = owner
@@ -572,7 +612,7 @@ def _release(self):
             raise LockException("Lock was not set by this process.")
 
         try:
-            self.client.delete(self._key_name, prevValue=str(self._owner))
+            self.client.delete(self._key_name, prevValue=self._owner)
             self._owner = None
         except ValueError:
             raise LockException(
@@ -584,6 +624,23 @@ def _release(self):
                 "Lock could not be released as it has not been acquired"
             )
 
+    def _renew(self) -> bool:
+        if self._owner is None:
+            raise LockException("Lock was not set by this process.")
+
+        try:
+            self.client.write(
+                self._key_name,
+                None,
+                ttl=self.expire,
+                prevValue=self._owner,
+                prevExist=True,
+                refresh=True,
+            )
+            return True
+        except (etcd.EtcdCompareFailed, etcd.EtcdKeyNotFound):
+            return False
+
     @property
     def _locked(self):
         try:
@@ -682,7 +739,7 @@ def _acquire(self):
         if self.expire is not None:
             _args.append(self.expire)
         # Set key only if it does not exist
-        if self.client.add(*tuple(_args)) is True:
+        if self.client.add(*_args) is True:
             self._owner = owner
             return True
         else:
@@ -707,6 +764,21 @@ def _release(self):
                 "Lock could not be released as it has not " "been acquired"
             )
 
+    def _renew(self) -> bool:
+        if self._owner is None:
+            raise LockException("Lock was not set by this process.")
+
+        resp = self.client.get(self._key_name)
+        if resp is not None:
+            if resp == str(self._owner):
+                _args = [self._key_name, self._owner]
+                if self.expire is not None:
+                    _args.append(self.expire)
+                # Update key with new TTL
+                self.client.set(*_args)
+                return True
+        return False
+
     @property
     def _locked(self):
         return True if self.client.get(self._key_name) is not None else False
@@ -809,10 +881,7 @@ def _key_name(self):
             key = self.lock_name
         return key
 
-    def _has_expired(
-        self, lease: kubernetes.client.V1Lease, now: datetime.datetime
-    ) -> bool:
-        # Determine whether the Lease has expired.
+    def _expiry_time(self, lease: kubernetes.client.V1Lease) -> datetime.datetime:
         expiry_time = datetime.datetime.min
         if (
             lease.spec.renew_time is not None
@@ -823,7 +892,13 @@ def _has_expired(
             )
         elif lease.spec.lease_duration_seconds is None:
             expiry_time = datetime.datetime.max
-        return now > expiry_time.astimezone(tz=datetime.timezone.utc)
+        return expiry_time.astimezone(tz=datetime.timezone.utc)
+
+    def _has_expired(
+        self, lease: kubernetes.client.V1Lease, now: datetime.datetime
+    ) -> bool:
+        # Determine whether the Lease has expired.
+        return now > self._expiry_time(lease)
 
     def _create_lease(
         self,
@@ -955,6 +1030,32 @@ def _release(self) -> None:
             # protects us from race conditions in deleting the Lease.
             self._delete_lease(lease)
 
+    def _renew(self) -> bool:
+        if self._owner is None:
+            raise LockException("Lock was not set by this process.")
+
+        lease = self._get_lease()
+        if lease is not None and self._owner == lease.spec.holder_identity:
+            now = self._now()
+            has_expired = self._has_expired(lease, now)
+
+            if has_expired:
+                return False
+
+            lease.spec.acquire_time = now
+            lease.spec.renew_time = now
+            lease.spec.lease_duration_seconds = self.expire
+            # The Lease object contains a `.metadata.resource_version` which
+            # protects us from race conditions in updating the Lease as described:
+            # https://blog.atomist.com/kubernetes-apply-replace-patch/.
+            if self._replace_lease(lease) is None:
+                # Someone else has modified the Lease before we renewed it so it
+                # must've expired.
+                raise False
+            return True
+
+        return False
+
     @property
     def _locked(self):
         lease = self._get_lease()
@@ -1109,6 +1210,25 @@ def _release(self) -> None:
                 if self._owner == data["owner"]:
                     self._data_file.unlink()
 
+    def _renew(self) -> bool:
+        if self._owner is None:
+            raise LockException("Lock was not set by this process.")
+
+        if self._data_file.exists():
+            with self._lock_file:
+                data = json.loads(self._data_file.read_text())
+
+                now = self._now()
+                has_expired = self._has_expired(data, now)
+                if self._owner == data["owner"]:
+                    if has_expired:
+                        return False
+                    # Refresh expiry time.
+                    data["expiry_time"] = self._expiry_time()
+                    self._data_file.write_text(json.dumps(data))
+                    return True
+        return False
+
     @property
     def _locked(self):
         if not self._data_file.exists():