Your advanced code editor is loading.

Please wait a moment.

#editor-loading { display: none; }

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.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): “”” :param name: name of the lock :param namespace: namespace :param identity: A unique identifier that the candidate is using “”” self.api_instance = client.CoreV1Api() self.leader_electionrecord_annotationkey = ‘control-plane.alpha.kubernetes.io/leader’ self.name = name self.namespace = namespace 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, 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
          annotations = api_response.metadata.annotations
          if annotations is None or annotations == '':
          api_response.metadata.annotations = {self.leader_electionrecord_annotationkey: ''}
          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 annotations.get(self.leader_electionrecord_annotationkey):
          api_response.metadata.annotations = {self.leader_electionrecord_annotationkey: ''}
          self.configmap_reference = api_response
          return True, None
          lock_record = self.get_lock_object(json.loads(annotations[self.leader_electionrecord_annotationkey]))
          self.configmap_reference = api_response
          return True, lock_record
          except ApiException as e:
          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' if object is created else 'False' if failed
          """
          body = client.V1ConfigMap(
          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
          except ApiException as e:
          logging.info("Failed to create lock as {}".format(e))
          return False
          def update(self, name, namespace, updated_record):
          """
          :param name: name of the lock to be updated
          :param namespace: namespace the lock is in
          :param updated_record: the updated election record
          :return: True if update is successful False if it fails
          """
          try:
          Set the updated record
          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:
          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

Explanation

The ConfigMapLock class implements a leader election mechanism using Kubernetes ConfigMaps. It provides methods for creating, getting, and updating ConfigMaps with leader election records.

Initialization (__init__)

Takes the name of the lock, namespace where it resides, and a unique identity of the candidate.
Initializes an instance of the CoreV1Api from the Kubernetes client for interacting with ConfigMaps.
Sets the leader_electionrecord_annotationkey used for storing leader election information in ConfigMap annotations.
Initializes a dictionary lock_record to store the leader election record data.

Get (get)

Takes the name and namespace of the ConfigMap to read.
Attempts to read the ConfigMap using the read_namespaced_config_map method.
Checks for the presence of annotations in the ConfigMap:
If no annotations exist, it creates a new annotation with the leader_electionrecord_annotationkey and an empty value.
If the leader_electionrecord_annotationkey is not present, it adds it with an empty value.
If the annotation exists, it extracts the leader election record JSON string from the annotation and parses it using json.loads.
It then converts the parsed JSON data into a LeaderElectionRecord object.
Stores the retrieved ConfigMap object in the configmap_reference for future updates.
Returns True and the LeaderElectionRecord object if successful, otherwise returns False and the exception caught.

Create (create)

Takes the name, namespace, and an election_record (a LeaderElectionRecord object) as arguments.
Creates a new V1ConfigMap object with the provided name, namespace, and an annotation containing the encoded JSON string of the election_record.
Attempts to create the ConfigMap using the create_namespaced_config_map method.
Returns True if successful, otherwise False and logs the exception.

Update (update)

Takes the name, namespace, and an updated_record (a LeaderElectionRecord object) as arguments.
Encodes the updated_record as JSON and updates the leader_electionrecord_annotationkey in the configmap_reference object.
Attempts to update the ConfigMap using the replace_namespaced_config_map method.
Returns True if successful, otherwise False and logs the exception.

Helper Methods

get_lock_object: Converts a JSON-parsed leader election record (dict) into a LeaderElectionRecord object.
get_lock_dict: Converts a LeaderElectionRecord object into a dictionary representation suitable for encoding into JSON.

Overall

This class provides a convenient way to implement leader election in Kubernetes using ConfigMaps. It handles creating, reading, and updating ConfigMaps, as well as encoding and decoding leader election records. The helper methods simplify the process of converting between JSON data and LeaderElectionRecord objects.

Your advanced code editor is loading.

Please wait a moment.

Explanation

Graph

Symbols

We couldn't identify any entrypoints. If you believe this to be incorrect then please contact support.