From 9b87f189894f33aa72b3f1bd61c6c2d5d4c3b87b Mon Sep 17 00:00:00 2001 From: chala2001 Date: Sat, 29 Aug 2026 00:25:53 +0530 Subject: [PATCH] Add LeaseLock for leader election Leader election could only use a ConfigMap. client-go moved to the coordination.k8s.io Lease resource and dropped ConfigMap support, so add a LeaseLock next to ConfigMapLock and use it in the example. Based on the implementation from #2314, with four corrections: use the package logger instead of calling logging.basicConfig at import, accept the times str(datetime) produces when the microseconds are zero, keep the created lease so a following update has a reference, and store the times as the real UTC instant rather than local wall clock labeled as UTC. Co-authored-by: Lane Richard --- kubernetes/base/leaderelection/example.py | 8 +- .../leaderelection/resourcelock/leaselock.py | 155 +++++++++++++ .../resourcelock/leaselock_test.py | 208 ++++++++++++++++++ 3 files changed, 370 insertions(+), 1 deletion(-) create mode 100644 kubernetes/base/leaderelection/resourcelock/leaselock.py create mode 100644 kubernetes/base/leaderelection/resourcelock/leaselock_test.py diff --git a/kubernetes/base/leaderelection/example.py b/kubernetes/base/leaderelection/example.py index 3b3336c8e3..038a294862 100644 --- a/kubernetes/base/leaderelection/example.py +++ b/kubernetes/base/leaderelection/example.py @@ -16,6 +16,7 @@ from kubernetes import client, config from kubernetes.leaderelection import leaderelection from kubernetes.leaderelection.resourcelock.configmaplock import ConfigMapLock +from kubernetes.leaderelection.resourcelock.leaselock import LeaseLock from kubernetes.leaderelection import electionconfig @@ -42,8 +43,13 @@ def example_func(): # A user can choose not to provide any callbacks for what to do when a candidate fails to lead - onStoppedLeading() # In that case, a default callback function will be used +lock = LeaseLock(lock_name, lock_namespace, candidate_id) +# Choose the lock to use. LeaseLock uses the coordination.k8s.io Lease +# resource and is the recommended one; ConfigMapLock is kept for existing users. +# lock = ConfigMapLock(lock_name, lock_namespace, candidate_id) + # Create config -config = electionconfig.Config(ConfigMapLock(lock_name, lock_namespace, candidate_id), lease_duration=17, +config = electionconfig.Config(lock, lease_duration=17, renew_deadline=15, retry_period=5, onstarted_leading=example_func, onstopped_leading=None) diff --git a/kubernetes/base/leaderelection/resourcelock/leaselock.py b/kubernetes/base/leaderelection/resourcelock/leaselock.py new file mode 100644 index 0000000000..d7a5c92dc7 --- /dev/null +++ b/kubernetes/base/leaderelection/resourcelock/leaselock.py @@ -0,0 +1,155 @@ +# Copyright 2026 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. + +from kubernetes.client.rest import ApiException +from kubernetes import client +from ..leaderelectionrecord import LeaderElectionRecord +from datetime import datetime, timezone +import logging +logger = logging.getLogger("leaderelection") + +# Formats produced by str(datetime). The microsecond component is omitted +# when it is exactly zero, so both spellings have to be accepted. +TIME_FORMATS = ( + "%Y-%m-%d %H:%M:%S.%f%z", + "%Y-%m-%d %H:%M:%S.%f", + "%Y-%m-%d %H:%M:%S%z", + "%Y-%m-%d %H:%M:%S", +) + + +class LeaseLock: + 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.CoordinationV1Api() + self.name = name + self.namespace = namespace + self.identity = str(identity) + self.lease_reference = None + + # get returns the election record from a Lease spec + def get(self, name, namespace): + """ + :param name: Name of the lease object information to get + :param namespace: Namespace in which the lease object is to be searched + :return: 'True, election record' if object found else 'False, exception response' + """ + try: + lease = self.api_instance.read_namespaced_lease(name, namespace) + except ApiException as e: + return False, e + + self.lease_reference = lease + return True, self.get_lock_object(lease) + + def create(self, name, namespace, election_record): + """ + :param name: Name of the lease object to be created + :param namespace: Namespace in which the lease object is to be created + :param election_record: The election record to store in the lease spec + :return: 'True' if object is created else 'False' if failed + """ + body = client.V1Lease(metadata={"name": name}, + spec=self.get_lease_spec(election_record)) + + try: + # Keep the created lease so that a following update has a + # reference to work from without re-reading it. + self.lease_reference = self.api_instance.create_namespaced_lease( + namespace, body) + return True + except ApiException as e: + logger.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 + """ + if self.lease_reference is None: + logger.info("Lease not initialized, call get or create first") + return False + + try: + self.lease_reference.spec = self.get_lease_spec( + updated_record, self.lease_reference.spec) + self.api_instance.replace_namespaced_lease( + name=name, namespace=namespace, body=self.lease_reference) + return True + except ApiException as e: + logger.info("Failed to update lock as {}".format(e)) + return False + + def get_lease_spec(self, leader_election_record, current_spec=None): + """Build the lease spec that holds the given election record.""" + spec = current_spec if current_spec else client.V1LeaseSpec() + + spec.holder_identity = leader_election_record.holder_identity + spec.lease_duration_seconds = int(leader_election_record.lease_duration) + spec.acquire_time = self.time_str_to_iso( + leader_election_record.acquire_time) + spec.renew_time = self.time_str_to_iso( + leader_election_record.renew_time) + + return spec + + def get_lock_object(self, lease): + """Build the election record held in the given lease spec.""" + leader_election_record = LeaderElectionRecord(None, None, None, None) + + if not lease.spec: + return leader_election_record + + if lease.spec.holder_identity: + leader_election_record.holder_identity = lease.spec.holder_identity + if lease.spec.lease_duration_seconds: + leader_election_record.lease_duration = str( + lease.spec.lease_duration_seconds) + if lease.spec.acquire_time: + leader_election_record.acquire_time = self.time_from_utc( + lease.spec.acquire_time) + if lease.spec.renew_time: + leader_election_record.renew_time = self.time_from_utc( + lease.spec.renew_time) + + return leader_election_record + + def time_str_to_iso(self, str_time): + """Convert an election record time into the instant to store. + + ``leaderelection.py`` builds its times with + ``datetime.fromtimestamp()``, which is local and naive. The Lease is + shared with other clients, so the value has to go on the wire as the + real UTC instant rather than as local wall clock labeled as UTC. + """ + for fmt in TIME_FORMATS: + try: + parsed = datetime.strptime(str_time, fmt) + except ValueError: + continue + if parsed.tzinfo is None: + parsed = parsed.astimezone() + return parsed.astimezone(timezone.utc) + raise ValueError("Failed to parse time string: {}".format(str_time)) + + def time_from_utc(self, value): + """Inverse of :meth:`time_str_to_iso`, back to the record format.""" + return str(value.astimezone().replace(tzinfo=None)) diff --git a/kubernetes/base/leaderelection/resourcelock/leaselock_test.py b/kubernetes/base/leaderelection/resourcelock/leaselock_test.py new file mode 100644 index 0000000000..cb96becdec --- /dev/null +++ b/kubernetes/base/leaderelection/resourcelock/leaselock_test.py @@ -0,0 +1,208 @@ +# Copyright 2026 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. + +import datetime +import os +import time +import unittest +from unittest import mock + +from kubernetes import client +from kubernetes.client.rest import ApiException + +from ..leaderelectionrecord import LeaderElectionRecord +from .leaselock import LeaseLock + + +def make_lock(): + with mock.patch.object(client, 'CoordinationV1Api'): + lock = LeaseLock('lock', 'default', 'candidate') + lock.api_instance = mock.MagicMock() + return lock + + +UTC = datetime.timezone.utc + + +class LeaseLockTest(unittest.TestCase): + + def test_create_writes_the_election_record_to_the_spec(self): + lock = make_lock() + record = LeaderElectionRecord('candidate', '17', + '2026-08-29 01:02:03.456789', + '2026-08-29 01:02:03.456789') + + self.assertTrue(lock.create('lock', 'default', record)) + + namespace, body = lock.api_instance.create_namespaced_lease.call_args[0] + self.assertEqual('default', namespace) + self.assertEqual('lock', body.metadata.name) + self.assertEqual('candidate', body.spec.holder_identity) + self.assertEqual(17, body.spec.lease_duration_seconds) + # stored as UTC on the wire, denoting the same instant as the + # local wall clock the election record carries + self.assertEqual(UTC, body.spec.acquire_time.tzinfo) + self.assertEqual('2026-08-29 01:02:03.456789', + str(body.spec.acquire_time.astimezone() + .replace(tzinfo=None))) + self.assertEqual(body.spec.acquire_time, body.spec.renew_time) + + def test_create_returns_false_when_the_api_fails(self): + lock = make_lock() + lock.api_instance.create_namespaced_lease.side_effect = ApiException( + status=409, reason='Conflict') + record = LeaderElectionRecord('candidate', '17', '2026-08-29 01:02:03', + '2026-08-29 01:02:03') + + self.assertFalse(lock.create('lock', 'default', record)) + + def test_get_returns_the_exception_when_the_lease_is_missing(self): + lock = make_lock() + expected = ApiException(status=404, reason='Not Found') + lock.api_instance.read_namespaced_lease.side_effect = expected + + status, response = lock.get('lock', 'default') + + self.assertFalse(status) + self.assertIs(expected, response) + + def test_get_reads_the_election_record_from_the_lease(self): + lock = make_lock() + acquired = datetime.datetime(2026, 8, 29, 1, 2, 3, 456789) + lock.api_instance.read_namespaced_lease.return_value = client.V1Lease( + metadata={'name': 'lock'}, + spec=client.V1LeaseSpec(holder_identity='candidate', + lease_duration_seconds=17, + acquire_time=acquired, + renew_time=acquired)) + + status, record = lock.get('lock', 'default') + + self.assertTrue(status) + self.assertEqual('candidate', record.holder_identity) + self.assertEqual('17', record.lease_duration) + self.assertEqual('2026-08-29 01:02:03.456789', record.acquire_time) + + def test_get_on_a_lease_without_a_spec_returns_an_empty_record(self): + lock = make_lock() + lock.api_instance.read_namespaced_lease.return_value = client.V1Lease( + metadata={'name': 'lock'}, spec=None) + + status, record = lock.get('lock', 'default') + + self.assertTrue(status) + self.assertIsNone(record.holder_identity) + + def test_record_survives_a_write_and_read_unchanged(self): + """leaderelection.py compares the stored record with the observed one + using __dict__, so a round trip has to come back identical.""" + lock = make_lock() + now = datetime.datetime.fromtimestamp(1787000000.5) + record = LeaderElectionRecord('candidate', str(17), str(now), str(now)) + + spec = lock.get_lease_spec(record) + read_back = lock.get_lock_object( + client.V1Lease(metadata={'name': 'lock'}, spec=spec)) + + self.assertEqual(record.__dict__, read_back.__dict__) + + def test_record_without_microseconds_survives_a_write_and_read(self): + """str(datetime) drops the microseconds when they are exactly zero.""" + lock = make_lock() + now = datetime.datetime(2026, 8, 29, 1, 2, 3) + self.assertEqual('2026-08-29 01:02:03', str(now)) + record = LeaderElectionRecord('candidate', str(17), str(now), str(now)) + + spec = lock.get_lease_spec(record) + read_back = lock.get_lock_object( + client.V1Lease(metadata={'name': 'lock'}, spec=spec)) + + self.assertEqual(record.__dict__, read_back.__dict__) + + def test_update_replaces_the_lease_it_read(self): + lock = make_lock() + acquired = datetime.datetime(2026, 8, 29, 1, 2, 3, 456789) + lock.api_instance.read_namespaced_lease.return_value = client.V1Lease( + metadata={'name': 'lock'}, + spec=client.V1LeaseSpec(holder_identity='other', + lease_duration_seconds=17, + acquire_time=acquired, + renew_time=acquired)) + lock.get('lock', 'default') + + record = LeaderElectionRecord('candidate', '17', + '2026-08-29 01:02:03.456789', + '2026-08-29 02:03:04.567890') + self.assertTrue(lock.update('lock', 'default', record)) + + body = lock.api_instance.replace_namespaced_lease.call_args[1]['body'] + self.assertEqual('candidate', body.spec.holder_identity) + self.assertEqual('2026-08-29 02:03:04.567890', + str(body.spec.renew_time.astimezone() + .replace(tzinfo=None))) + + def test_update_returns_false_when_the_api_fails(self): + lock = make_lock() + lock.lease_reference = client.V1Lease(metadata={'name': 'lock'}, + spec=client.V1LeaseSpec()) + lock.api_instance.replace_namespaced_lease.side_effect = ApiException( + status=409, reason='Conflict') + record = LeaderElectionRecord('candidate', '17', '2026-08-29 01:02:03', + '2026-08-29 01:02:03') + + self.assertFalse(lock.update('lock', 'default', record)) + + @unittest.skipUnless(hasattr(time, 'tzset'), 'requires tzset') + def test_times_are_written_as_the_real_utc_instant(self): + """The election record holds local wall clock. The Lease is shared + with other clients, so it has to carry the real UTC instant.""" + lock = make_lock() + previous = os.environ.get('TZ') + os.environ['TZ'] = 'Asia/Kolkata' # UTC+05:30, no DST + time.tzset() + try: + record = LeaderElectionRecord('candidate', '17', + '2026-08-29 01:02:03.456789', + '2026-08-29 01:02:03.456789') + spec = lock.get_lease_spec(record) + + self.assertEqual( + datetime.datetime(2026, 8, 28, 19, 32, 3, 456789, tzinfo=UTC), + spec.acquire_time) + # and it still round trips back to the local wall clock + read_back = lock.get_lock_object( + client.V1Lease(metadata={'name': 'lock'}, spec=spec)) + self.assertEqual(record.__dict__, read_back.__dict__) + finally: + if previous is None: + os.environ.pop('TZ', None) + else: + os.environ['TZ'] = previous + time.tzset() + + def test_update_before_get_or_create_does_not_raise(self): + lock = make_lock() + record = LeaderElectionRecord('candidate', '17', '2026-08-29 01:02:03', + '2026-08-29 01:02:03') + + self.assertFalse(lock.update('lock', 'default', record)) + + def test_an_unparsable_time_is_reported(self): + lock = make_lock() + with self.assertRaises(ValueError): + lock.time_str_to_iso('not a time') + + +if __name__ == '__main__': + unittest.main()