Add LeaseLock for leader election

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)
 

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))

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()