Adding serializing functions for serializing & de-serializing locks, leader_election_record as a class

Author: Invictus17

leaderelection/electionconfig.py

@@ -16,6 +16,7 @@
 import logging
 logging.basicConfig(level=logging.INFO)
 
+
 class Config:
     # Validate config, exit if an error is detected
     def __init__(self, lock, lease_duration, renew_deadline, retry_period, onstarted_leading, onstopped_leading):

leaderelection/leaderelection.py

@@ -15,9 +15,11 @@
 import datetime
 import sys
 import time
-import ast
+import json
 import threading
+from leaderelectionrecord import LeaderElectionRecord
 import logging
+
 logging.basicConfig(level=logging.INFO)
 
 
@@ -35,16 +37,6 @@ def __init__(self, election_config):
         # Latest update time of the lock
         self.observed_time_milliseconds = 0
 
-
-    # Annotation used in the lock object
-    def leaderelector_record(self, election_config, now):
-        return {
-                    "holderIdentity": str(election_config.lock.identity),
-                    "leaseDurationSeconds": str(election_config.lease_duration),
-                    "acquireTime": str(now),
-                    "renewTime": str(now)
-                }
-
     # Point of entry to Leader election
     def run(self):
         # Try to create/ acquire a lock
@@ -77,13 +69,13 @@ def renew_loop(self):
         logging.info("Leader has entered renew loop and will try to update lease continuously")
 
         retry_period = self.election_config.retry_period
-        renew_deadline = self.election_config.renew_deadline*1000
+        renew_deadline = self.election_config.renew_deadline * 1000
 
         while True:
-            timeout = int(time.time()*1000) + renew_deadline
+            timeout = int(time.time() * 1000) + renew_deadline
             succeeded = False
 
-            while int(time.time()*1000) < timeout:
+            while int(time.time() * 1000) < timeout:
                 succeeded = self.try_acquire_or_renew()
 
                 if succeeded:
@@ -101,82 +93,76 @@ def try_acquire_or_renew(self):
         now = datetime.datetime.now()
 
         # Check if lock is created
-        lock_status, lock_response, lock_record = self.election_config.lock.get(self.election_config.lock.name,
-                                                                                self.election_config.lock.namespace)
+        lock_status, old_election_record = self.election_config.lock.get(self.election_config.lock.name,
+                                                                        self.election_config.lock.namespace)
 
         # create a default Election record for this candidate
-        leader_election_record = self.leaderelector_record(self.election_config, now)
+        leader_election_record = LeaderElectionRecord(self.election_config.lock.identity,
+                                                     str(self.election_config.lease_duration), str(now), str(now))
 
         # A lock is not created with that name, try to create one
         if not lock_status:
-            if ast.literal_eval(lock_response.body)['code'] != 404:
-                logging.info("Error retrieving resource lock {} as {}".format(self.election_config.lock.name, lock_response.reason))
+            if json.loads(old_election_record.body)['code'] != 404:
+                logging.info("Error retrieving resource lock {} as {}".format(self.election_config.lock.name,
+                                                                              old_election_record.reason))
                 return False
 
-
-            logging.info("{} is trying to create a lock".format(leader_election_record['holderIdentity']))
-            create_status, create_response = self.election_config.lock.create(name=self.election_config.lock.name,
-                                                                              namespace=self.election_config.lock.namespace,
-                                                                              election_record=leader_election_record)
+            logging.info("{} is trying to create a lock".format(leader_election_record.holder_identity))
+            create_status = self.election_config.lock.create(name=self.election_config.lock.name,
+                                                             namespace=self.election_config.lock.namespace,
+                                                             election_record=leader_election_record)
 
             if create_status is False:
+                logging.info("{} Failed to create lock".format(leader_election_record.holder_identity))
                 return False
 
             self.observed_record = leader_election_record
             self.observed_time_milliseconds = int(time.time() * 1000)
             return True
 
         # A lock exists with that name
-        # Validate lock_record
-        if lock_record is None:
+        # Validate old_election_record
+        if old_election_record is None:
             # try to update lock with proper annotation and election record
-            return self.update_lock(lock_response, leader_election_record)
-
-        # check for any key, value errors in the record
-        try:
-            old_election_record = ast.literal_eval(lock_record)
-            if (old_election_record['holderIdentity'] == '' or old_election_record['leaseDurationSeconds'] == ''
-                    or old_election_record['acquireTime'] == '' or old_election_record['renewTime'] == ''):
-                # try to update lock with proper annotation and election record
-                return self.update_lock(lock_response, leader_election_record)
-        except:
-            # try to update lock with proper annotation and election record
-            return self.update_lock(lock_response, leader_election_record)
+            return self.update_lock(leader_election_record)
 
+        if (old_election_record.holder_identity is None or old_election_record.lease_duration is None
+                or old_election_record.acquire_time is None or old_election_record.renew_time is None):
+            # try to update lock with proper annotation and election record
+            return self.update_lock(leader_election_record)
 
         # Report transitions
-        if self.observed_record and self.observed_record['holderIdentity'] != old_election_record['holderIdentity']:
-            logging.info("Leader has switched to {}".format(old_election_record['holderIdentity']))
+        if self.observed_record and self.observed_record.holder_identity != old_election_record.holder_identity:
+            logging.info("Leader has switched to {}".format(old_election_record.holder_identity))
 
-        if old_election_record != self.observed_record:
+        if self.observed_record is None or old_election_record.__dict__ != self.observed_record.__dict__:
             self.observed_record = old_election_record
             self.observed_time_milliseconds = int(time.time() * 1000)
 
         # If This candidate is not the leader and lease duration is yet to finish
-        if (str(self.election_config.lock.identity) != self.observed_record['holderIdentity']
-                and self.observed_time_milliseconds + self.election_config.lease_duration*1000 > int(now.timestamp()*1000)):
-            logging.info("yet to finish lease_duration, lease held by {} and has not expired".format(old_election_record['holderIdentity']))
+        if (self.election_config.lock.identity != self.observed_record.holder_identity
+                and self.observed_time_milliseconds + self.election_config.lease_duration * 1000 > int(now.timestamp() * 1000)):
+            logging.info("yet to finish lease_duration, lease held by {} and has not expired".format(old_election_record.holder_identity))
             return False
 
         # If this candidate is the Leader
-        if str(self.election_config.lock.identity) == self.observed_record['holderIdentity']:
-            # Leader sets acquireTime
-            leader_election_record['acquireTime'] = self.observed_record['acquireTime']
+        if self.election_config.lock.identity == self.observed_record.holder_identity:
+            # Leader updates renewTime, but keeps acquire_time unchanged
+            leader_election_record.acquire_time = self.observed_record.acquire_time
 
-        return self.update_lock(lock_response, leader_election_record)
+        return self.update_lock(leader_election_record)
 
-    def update_lock(self, lock_response, leader_election_record):
+    def update_lock(self, leader_election_record):
         # Update object with latest election record
-        update_status, update_response = self.election_config.lock.update(self.election_config.lock.name,
-                                                                          self.election_config.lock.namespace,
-                                                                          lock_response, leader_election_record)
+        update_status = self.election_config.lock.update(self.election_config.lock.name,
+                                                         self.election_config.lock.namespace,
+                                                         leader_election_record)
 
         if update_status is False:
-            logging.info("{} failed to acquire lease".format(leader_election_record['holderIdentity']))
+            logging.info("{} failed to acquire lease".format(leader_election_record.holder_identity))
             return False
 
         self.observed_record = leader_election_record
         self.observed_time_milliseconds = int(time.time() * 1000)
-        logging.info("leader {} has successfully acquired lease".format(leader_election_record['holderIdentity']))
+        logging.info("leader {} has successfully acquired lease".format(leader_election_record.holder_identity))
         return True
-

leaderelection/leaderelectionrecord.py

@@ -0,0 +1,22 @@
+# Copyright 2020 The Kubernetes Authors.
+#
+# 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.
+
+
+class LeaderElectionRecord:
+    # Annotation used in the lock object
+    def __init__(self, holder_identity, lease_duration, acquire_time, renew_time):
+        self.holder_identity = holder_identity
+        self.lease_duration = lease_duration
+        self.acquire_time = acquire_time
+        self.renew_time = renew_time

leaderelection/resourcelock/configmaplock.py

@@ -15,6 +15,11 @@
 from kubernetes.client.rest import ApiException
 from kubernetes import client, config
 from kubernetes.client.api_client import ApiClient
+from leaderelectionrecord import LeaderElectionRecord
+import json
+import logging
+logging.basicConfig(level=logging.INFO)
+
 
 class ConfigMapLock:
     def __init__(self, name, namespace, identity):
@@ -27,61 +32,98 @@ def __init__(self, name, namespace, identity):
         self.leader_electionrecord_annotationkey = 'control-plane.alpha.kubernetes.io/leader'
         self.name = name
         self.namespace = namespace
-        self.identity = identity
+        self.identity = str(identity)
+        self.configmap_reference = None
+        self.lock_record = {
+            'holderIdentity': None,
+            'leaseDurationSeconds': None,
+            'acquireTime': None,
+            'renewTime': None
+                            }
 
     # get returns the election record from a ConfigMap Annotation
     def get(self, name, namespace):
         """
         :param name: Name of the configmap object information to get
         :param namespace: Namespace in which the configmap object is to be searched
-        :return: 'True, response, record' if object found else 'False, exception, None'
+        :return: 'True, election record' if object found else 'False, exception response'
         """
         try:
             api_response = self.api_instance.read_namespaced_config_map(name, namespace)
 
             # If an annotation does not exist - add the leader_electionrecord_annotationkey
-            if api_response.metadata.annotations is None or api_response.metadata.annotations == '':
+            annotations = api_response.metadata.annotations
+            if annotations is None or annotations == '':
                 api_response.metadata.annotations = {self.leader_electionrecord_annotationkey: ''}
-                return True, api_response, None
+                self.configmap_reference = api_response
+                return True, None
 
             # If an annotation exists but, the leader_electionrecord_annotationkey does not then add it as a key
-            if not api_response.metadata.annotations.get(self.leader_electionrecord_annotationkey):
+            if not annotations.get(self.leader_electionrecord_annotationkey):
                 api_response.metadata.annotations = {self.leader_electionrecord_annotationkey: ''}
-                return True, api_response, None
+                self.configmap_reference = api_response
+                return True, None
+
+            lock_record = self.get_lock_object(json.loads(annotations[self.leader_electionrecord_annotationkey]))
 
-            return True, api_response, api_response.metadata.annotations[self.leader_electionrecord_annotationkey]
+            self.configmap_reference = api_response
+            return True, lock_record
         except ApiException as e:
-            return False, e, None
+            return False, e
 
     def create(self, name, namespace, election_record):
         """
         :param electionRecord: Annotation string
         :param name: Name of the configmap object to be created
         :param namespace: Namespace in which the configmap object is to be created
-        :return: 'True, None' if object is created else 'False, error' if failed
+        :return: 'True' if object is created else 'False' if failed
         """
         body = client.V1ConfigMap(
-            metadata={"name": name, "annotations": {self.leader_electionrecord_annotationkey: str(election_record)}})
+            metadata={"name": name,
+                      "annotations": {self.leader_electionrecord_annotationkey: json.dumps(self.get_lock_dict(election_record))}})
 
         try:
             api_response = self.api_instance.create_namespaced_config_map(namespace, body, pretty=True)
-            return True, api_response
+            return True
         except ApiException as e:
-            return False, e
+            logging.info("Failed to create lock as {}".format(e))
+            return False
 
-    def update(self, name, namespace, update_object, updated_record):
+    def update(self, name, namespace, updated_record):
         """
         :param name: name of the lock to be updated
         :param namespace: namespace the lock is in
-        :param update_object: the object used to update the lock
         :param updated_record: the updated election record
         :return: True if update is succesful False if it fails
         """
         try:
             # Set the updated record
-            update_object.metadata.annotations[self.leader_electionrecord_annotationkey] = str(updated_record)
-            api_response = self.api_instance.replace_namespaced_config_map(name=name, namespace=namespace, body=update_object)
-            return True, api_response
+            self.configmap_reference.metadata.annotations[self.leader_electionrecord_annotationkey] = json.dumps(self.get_lock_dict(updated_record))
+            api_response = self.api_instance.replace_namespaced_config_map(name=name, namespace=namespace,
+                                                                           body=self.configmap_reference)
+            return True
         except ApiException as e:
-            return False, e
+            logging.info("Failed to update lock as {}".format(e))
+            return False
+
+    def get_lock_object(self, lock_record):
+        leader_election_record = LeaderElectionRecord(None, None, None, None)
+
+        if lock_record.get('holderIdentity'):
+            leader_election_record.holder_identity = lock_record['holderIdentity']
+        if lock_record.get('leaseDurationSeconds'):
+            leader_election_record.lease_duration = lock_record['leaseDurationSeconds']
+        if lock_record.get('acquireTime'):
+            leader_election_record.acquire_time = lock_record['acquireTime']
+        if lock_record.get('renewTime'):
+            leader_election_record.renew_time = lock_record['renewTime']
+
+        return leader_election_record
 
+    def get_lock_dict(self, leader_election_record):
+        self.lock_record['holderIdentity'] = leader_election_record.holder_identity
+        self.lock_record['leaseDurationSeconds'] = leader_election_record.lease_duration
+        self.lock_record['acquireTime'] = leader_election_record.acquire_time
+        self.lock_record['renewTime'] = leader_election_record.renew_time
+        
+        return self.lock_record