Add distributed lock to entry_property actions

Signed-off-by: Jacob Floyd <cognifloyd@gmail.com>

Author: cognifloyd

README.md

@@ -49,10 +49,10 @@ You can also use dynamic values from the datastore. See the
 * ``kv.upsert_entry_property`` - Update or insert a property in the named entry
   of a JSON serialized object. Then, serialize and store the updated object.
   The property's value may be any json-serializable type.
-  Fails if the datastore key does not exist.
+  Fails if the datastore key does not exist. A coordination backend is recommended.
 * ``kv.delete_entry_property`` - Delete a property from a named entry of a JSON
   serialized object in the datastore. If the entry is empty, delete it as well.
-  Fails if the datastore key does not exist.
+  Fails if the datastore key does not exist. A coordination backend is recommended.
 
 Note: ``kv.set`` and ``kv.get`` actions support compressing value before
 storing it in a datastore and decompressing it when retrieving it from

actions/kv_delete_entry_property.py

@@ -1,5 +1,7 @@
 import json
 
+from st2common.services import coordination as coordination_service
+
 from lib.action import St2BaseAction
 
 __all__ = [
@@ -10,38 +12,39 @@
 class St2KVPDeleteEntryPropertyAction(St2BaseAction):
     # noinspection PyShadowingBuiltins
     def run(self, key, entry, property):
-        # get and deserialize object or fail.
-        _key = self.client.keys.get_by_name(key, decrypt=False)
-
-        if not _key:
-            raise Exception("Key does not exist in datastore")
+        with coordination_service.get_coordinator().get_lock('st2.kv_entry.' + key):
+            # get and deserialize object or fail.
+            _key = self.client.keys.get_by_name(key, decrypt=False)
 
-        deserialized = json.loads(_key.value)
+            if not _key:
+                raise Exception("Key does not exist in datastore")
 
-        # delete object.entry.property
-        _entry = deserialized.get(entry, {})
-        try:
-            del _entry[property]
-        except KeyError:
-            pass
+            deserialized = json.loads(_key.value)
 
-        # delete object.entry if entry is empty
-        if not _entry:
+            # delete object.entry.property
+            _entry = deserialized.get(entry, {})
             try:
-                del deserialized[entry]
+                del _entry[property]
             except KeyError:
                 pass
 
-        # re-serialize and save
-        serialized = json.dumps(deserialized)
-        kvp = self._kvp(name=key, value=serialized)
-        kvp.id = key
-
-        self.client.keys.update(kvp)
-        response = {
-            'key': key,
-            'entry_name': entry,
-            'entry': _entry,
-            'entry_deleted': not _entry,
-        }
-        return response
+            # delete object.entry if entry is empty
+            if not _entry:
+                try:
+                    del deserialized[entry]
+                except KeyError:
+                    pass
+
+            # re-serialize and save
+            serialized = json.dumps(deserialized)
+            kvp = self._kvp(name=key, value=serialized)
+            kvp.id = key
+
+            self.client.keys.update(kvp)
+            response = {
+                'key': key,
+                'entry_name': entry,
+                'entry': _entry,
+                'entry_deleted': not _entry,
+            }
+            return response

actions/kv_upsert_entry_property.py

@@ -1,5 +1,7 @@
 import json
 
+from st2common.services import coordination as coordination_service
+
 from lib.action import St2BaseAction
 
 __all__ = [
@@ -10,37 +12,38 @@
 class St2KVPUpsertEntryPropertyAction(St2BaseAction):
     # noinspection PyShadowingBuiltins
     def run(self, key, entry, property, value):
-        # get and deserialize object or fail.
-        _key = self.client.keys.get_by_name(key, decrypt=False)
-
-        if not _key:
-            raise Exception("Key does not exist in datastore")
-
-        # optimistically try to decode a json value
-        try:
-            value = json.loads(value)
-        except (TypeError, ValueError):
-            # assume it is either already decoded (TypeError)
-            # or it is a plain string (ValueError)
-            # (malformed JSON objects/arrays will be strings)
-            pass
-
-        deserialized = json.loads(_key.value)
-
-        # update or insert object.entry.property
-        _entry = deserialized.get(entry, {})
-        _entry[property] = value
-        deserialized[entry] = _entry
-
-        # re-serialize and save
-        serialized = json.dumps(deserialized)
-        kvp = self._kvp(name=key, value=serialized)
-        kvp.id = key
-
-        self.client.keys.update(kvp)
-        response = {
-            'key': key,
-            'entry_name': entry,
-            'entry': _entry,
-        }
-        return response
+        with coordination_service.get_coordinator().get_lock('st2.kv_entry.' + key):
+            # get and deserialize object or fail.
+            _key = self.client.keys.get_by_name(key, decrypt=False)
+
+            if not _key:
+                raise Exception("Key does not exist in datastore")
+
+            # optimistically try to decode a json value
+            try:
+                value = json.loads(value)
+            except (TypeError, ValueError):
+                # assume it is either already decoded (TypeError)
+                # or it is a plain string (ValueError)
+                # (malformed JSON objects/arrays will be strings)
+                pass
+
+            deserialized = json.loads(_key.value)
+
+            # update or insert object.entry.property
+            _entry = deserialized.get(entry, {})
+            _entry[property] = value
+            deserialized[entry] = _entry
+
+            # re-serialize and save
+            serialized = json.dumps(deserialized)
+            kvp = self._kvp(name=key, value=serialized)
+            kvp.id = key
+
+            self.client.keys.update(kvp)
+            response = {
+                'key': key,
+                'entry_name': entry,
+                'entry': _entry,
+            }
+            return response