Merge pull request #11 from nzlosh/master

Implementation basic locking using consul keystore/session

Author: nzlosh

CHANGES.md

@@ -1,5 +1,13 @@
 # Change Log
 
+## [0.6.3] 25 Jan 2018
+
+### Added
+  - Implementation for basic locking using consul key store/session.
+
+### Changed
+  - Make acl_list return a list instead of a string.
+
 ## [0.6.2]  8 Jan 2018
 
 ### Added

actions/custom_lock.py

@@ -0,0 +1,13 @@
+from lib import action
+
+
+class ConsulCustomLockAction(action.ConsulBaseAction):
+    """
+    :key_prefix: (string) Prefix the lock name which is the path in consul's kvstore.
+    :max_locks: (integer) Maximum number of concurrent lock holders.
+    :name: (string) Name of the key that will hold the locking information.
+    """
+    def run(self, key_prefix, max_locks=1, acquire_timeout=10, name="CustomLock",
+            node=None, checks=None, behavior="release", ttl=None, token=None, dc=None):
+        self.lock_manager = action.LockManager(self.consul, key_prefix, name, node, token, dc)
+        return self.lock_manager.lock(max_locks, acquire_timeout, checks, behavior, ttl)

actions/custom_lock.yaml

@@ -0,0 +1,52 @@
+name: custom_lock
+runner_type: python-script
+description: "Create a lock or semaphore.  Used for synchronising discret workflows."
+enabled: true
+entry_point: "custom_lock.py"
+parameters:
+    key_prefix:
+        type: string
+        description: KV path to prefix to lock name.
+        required: true
+    max_locks:
+        type: integer
+        description: Maximum number of concurrent holders for the lock.
+        default: 1
+        required: false
+    acquire_timeout:
+        type: integer
+        description: Number of seconds to wait for the lock to be acquired.
+        default: 10
+        required: false
+    name:
+        type: string
+        description: The name of the lock.
+        required: false
+    node:
+        type: string
+        description: The node that will establish and maintain the lock.
+        required: false
+    checks:
+        type: array
+        description: A list of the health checks used to determine if the lock should be invalidated.
+        required: false
+    behavior:
+        type: string
+        enum:
+          - release
+          - delete
+        description: The behavior to use when the lock expires (release or delete).
+        default: release
+        required: false
+    ttl:
+        type: integer
+        description: Time to Live before the lock expires (between 10 and 86400 seconds).
+        required: false
+    token:
+        type: string
+        description: An optional ACL token to apply to this request.
+        required: false
+    dc:
+        type: string
+        description: Optional datacenter that you wish to communicate with.
+        required: false

actions/custom_unlock.py

@@ -0,0 +1,12 @@
+from lib import action
+
+
+class ConsulCustomUnLockAction(action.ConsulBaseAction):
+    """
+    :session_id: (string) the session id that holds the lock to a key.
+    :key_prefix: (string) Prefix the lock name which is the path in consul's kvstore.
+    :name: (string) Name of the key that will hold the locking information.
+    """
+    def run(self, session_id, key_prefix, name, node=None, token=None, dc=None):
+        self.lock_manager = action.LockManager(self.consul, key_prefix, name, node, token, dc)
+        return self.lock_manager.unlock(session_id)

actions/custom_unlock.yaml

@@ -0,0 +1,30 @@
+name: custom_unlock
+runner_type: python-script
+description: "Release a lock or semaphore.  Used for synchronising discret workflows."
+enabled: true
+entry_point: "custom_unlock.py"
+parameters:
+    session_id:
+        type: string
+        description: The session id that currently holds the lock on the key.
+        required: true
+    key_prefix:
+        type: string
+        description: KV path to prefix to lock name.
+        required: true
+    name:
+        type: string
+        description: The name of the lock.
+        required: true
+    node:
+        type: string
+        description: The node that will establish and maintain the lock.
+        required: false
+    token:
+        type: string
+        description: An optional ACL token to apply to this request.
+        required: false
+    dc:
+        type: string
+        description: Optional datacenter that you wish to communicate with.
+        required: false

actions/kv_put.py

@@ -16,7 +16,7 @@ def run(self,
         if to_json:
             value = self.to_json(value)
 
-        return (True, self.consul.kv.put(
+        return self.consul.kv.put(
             key,
             value,
             cas=cas,
@@ -25,4 +25,4 @@ def run(self,
             release=release,
             token=token,
             dc=dc
-        ))
+        )

actions/lib/action.py

@@ -1,17 +1,22 @@
+import six
 import json
+import random
+import time
+
 from st2common import log as logging
 from st2common.runners.base_action import Action
 
 # http://python-consul.readthedocs.org/en/latest/#
 import consul
 
+LOG = logging.getLogger(__name__)
+
 
 class ConsulBaseAction(Action):
 
     def __init__(self, config):
         super(ConsulBaseAction, self).__init__(config)
         self.consul = self._get_client()
-        self.logger = logging.getLogger(__name__)
 
     def _get_client(self):
         dc = self.config.get('dc')
@@ -41,3 +46,116 @@ def from_json(self, value):
         except ValueError:
             pass
         return value
+
+
+class LockManager(object):
+    semaphore = {
+        "Limit": 0,
+        "Holders": {}
+    }
+
+    def __init__(self, client, key_prefix, name, node, token, dc):
+        """
+        Initialise object with common variables used to create or destroy a lock.
+        """
+        self.client = client
+        self.key_prefix = key_prefix
+        self.name = name
+        self.node = node
+        self.token = token
+        self.dc = dc
+
+    def lock(self, max_locks, acquire_timeout, checks, behavior, ttl):
+        """
+        Method called by the Lock action.
+        """
+        self.max_locks = max_locks
+        self.wait_timeout = acquire_timeout
+        self.checks = checks
+        self.behavior = behavior
+        self.ttl = ttl
+
+        if self.max_locks > 1:
+            result = self.create_semaphore()
+        else:
+            result = self.create_lock("/".join([self.key_prefix, self.name, '.lock']))
+        return result
+
+    def unlock(self, session_id):
+        """
+        Method called by the Unlock action.
+        """
+        key = "/".join([self.key_prefix, self.name, '.lock'])
+        return self.release_lock(key, session_id)
+
+    def create_semaphore(self):
+        raise NotImplementedError
+
+    def release_lock(self, key, session_id):
+        """
+        Release the lock on the key and destroy the session.
+        """
+        result = (False, "Failed to release lock.")
+        if self.client.kv.put(
+            key=key,
+            value=None,
+            release=session_id,
+            token=self.token,
+            dc=self.dc
+        ) is True:
+            if self.destroy_session(session_id) is True:
+                result = (True, "Lock released and session destroyed.")
+            else:
+                result = (True, "Lock released but session could not be destroyed.")
+        return result
+
+    def destroy_session(self, session_id):
+        self.client.session.destroy(session_id, self.dc)
+
+    def create_lock(self, key_name):
+        result = (False, "")
+        session_id = self.create_session()
+        if isinstance(session_id, six.string_types) and len(session_id) > 0:
+            if self.acquire_lock(session_id, key_name, value=session_id):
+                result = (True, session_id)
+            else:
+                result = (False, "Failed to acquire lock on key.")
+                self.destroy_session(session_id)
+        else:
+            result = (False, "Failed to create a session")
+        return result
+
+    def create_session(self):
+        """
+        https://www.consul.io/docs/internals/sessions.html
+        """
+        return self.client.session.create(
+            name=self.name,
+            node=self.node,
+            checks=self.checks,
+            lock_delay=5,
+            behavior=self.behavior,
+            ttl=self.ttl,
+            dc=self.dc
+        )
+
+    def acquire_lock(self, session_id, key, cas=None, value=""):
+        """
+        wait_timeout is used to determine when to abandon the lock acquisition.
+        """
+        result = False
+        timeout = time.time() + self.wait_timeout
+        LOG.info("Acquire lock timeout: {}".format(timeout))
+        while time.time() < timeout:
+            result = self.client.kv.put(
+                key=key,
+                value=value,
+                cas=cas,
+                acquire=session_id,
+                token=self.token,
+                dc=self.dc
+            )
+            if result is True:
+                break
+            time.sleep(random.random())
+        return result

pack.yaml

@@ -2,6 +2,6 @@
 ref: consul
 name: consul
 description: consul
-version: 0.6.2
+version: 0.6.3
 author: jfryman
 email: [email protected]