Record leader transitions in leader election

kubernetes/leaderelection/leaderelection.py

@@ -169,6 +169,7 @@ def try_acquire_or_renew(self):
         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
+            leader_election_record.leader_transitions = old_election_record.leader_transitions + 1
             return self.update_lock(leader_election_record)
 
         # Report transitions
@@ -189,6 +190,9 @@ def try_acquire_or_renew(self):
         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
+            leader_election_record.leader_transitions = self.observed_record.leader_transitions
+        else:
+            leader_election_record.leader_transitions = self.observed_record.leader_transitions + 1
 
         return self.update_lock(leader_election_record)
 

kubernetes/leaderelection/leaderelection_test.py

@@ -392,5 +392,82 @@ def update(self, name, namespace, updated_record):
         self.assertTrue(leaderelection.LeaderElection(config).try_acquire_or_renew())
 
 
+class RecordingLock:
+    """Holds a single record and keeps the last one written to it."""
+    def __init__(self, identity, record):
+        self.name = "lock"
+        self.namespace = "ns"
+        self.identity = identity
+        self.record = record
+        self.written = None
+
+    def get(self, name, namespace):
+        if self.record is None:
+            return False, ApiException(status=404, reason="Not Found",
+                                       body=json.dumps({'code': 404}))
+        return True, self.record
+
+    def create(self, name, namespace, election_record):
+        self.written = election_record
+        return True
+
+    def update(self, name, namespace, updated_record):
+        self.written = updated_record
+        return True
+
+
+def make_election(lock):
+    config = electionconfig.Config(
+        lock=lock, lease_duration=4, renew_deadline=3, retry_period=1,
+        onstarted_leading=lambda: None, onstopped_leading=lambda: None)
+    return leaderelection.LeaderElection(config)
+
+
+class LeaderTransitionsTest(unittest.TestCase):
+
+    def test_new_lock_starts_with_zero_transitions(self):
+        lock = RecordingLock("candidate", None)
+        election = make_election(lock)
+
+        self.assertTrue(election.try_acquire_or_renew())
+        self.assertEqual(0, lock.written.leader_transitions)
+
+    def test_renewing_leader_keeps_leader_transitions(self):
+        record = LeaderElectionRecord("candidate", "4", "then", "then", 3)
+        lock = RecordingLock("candidate", record)
+        election = make_election(lock)
+        election.observed_record = record
+
+        self.assertTrue(election.try_acquire_or_renew())
+        self.assertEqual(3, lock.written.leader_transitions)
+        self.assertEqual("then", lock.written.acquire_time)
+
+    def test_taking_over_an_expired_lock_counts_a_transition(self):
+        record = LeaderElectionRecord("other", "4", "then", "then", 3)
+        lock = RecordingLock("candidate", record)
+        election = make_election(lock)
+        election.observed_record = record
+        election.observed_time_milliseconds = 0
+
+        self.assertTrue(election.try_acquire_or_renew())
+        self.assertEqual("candidate", lock.written.holder_identity)
+        self.assertEqual(4, lock.written.leader_transitions)
+
+    def test_lock_held_by_another_candidate_is_left_alone(self):
+        record = LeaderElectionRecord("other", "4", "then", "then", 3)
+        lock = RecordingLock("candidate", record)
+
+        self.assertFalse(make_election(lock).try_acquire_or_renew())
+        self.assertIsNone(lock.written)
+
+    def test_taking_over_a_released_lock_counts_a_transition(self):
+        record = LeaderElectionRecord(None, None, None, None, 2)
+        lock = RecordingLock("candidate", record)
+
+        self.assertTrue(make_election(lock).try_acquire_or_renew())
+        self.assertEqual("candidate", lock.written.holder_identity)
+        self.assertEqual(3, lock.written.leader_transitions)
+
+
 if __name__ == '__main__':
     unittest.main()

kubernetes/leaderelection/leaderelectionrecord.py

@@ -15,8 +15,10 @@
 
 class LeaderElectionRecord:
     # Annotation used in the lock object
-    def __init__(self, holder_identity, lease_duration, acquire_time, renew_time):
+    def __init__(self, holder_identity, lease_duration, acquire_time, renew_time,
+                 leader_transitions=0):
         self.holder_identity = holder_identity
         self.lease_duration = lease_duration
         self.acquire_time = acquire_time
         self.renew_time = renew_time
+        self.leader_transitions = leader_transitions

kubernetes/leaderelection/resourcelock/configmaplock.py

@@ -38,7 +38,8 @@ def __init__(self, name, namespace, identity):
             'holderIdentity': None,
             'leaseDurationSeconds': None,
             'acquireTime': None,
-            'renewTime': None
+            'renewTime': None,
+            'leaderTransitions': 0
                             }
 
     # get returns the election record from a ConfigMap Annotation
@@ -130,6 +131,8 @@ def get_lock_object(self, lock_record):
             leader_election_record.acquire_time = lock_record['acquireTime']
         if lock_record.get('renewTime'):
             leader_election_record.renew_time = lock_record['renewTime']
+        if lock_record.get('leaderTransitions'):
+            leader_election_record.leader_transitions = lock_record['leaderTransitions']
 
         return leader_election_record
 
@@ -138,5 +141,6 @@ def get_lock_dict(self, leader_election_record):
         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
+        self.lock_record['leaderTransitions'] = leader_election_record.leader_transitions
 
         return self.lock_record

kubernetes/leaderelection/resourcelock/configmaplock_test.py

@@ -0,0 +1,53 @@
+# 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 unittest
+from unittest import mock
+
+from kubernetes import client
+
+from ..leaderelectionrecord import LeaderElectionRecord
+from .configmaplock import ConfigMapLock
+
+
+def make_lock():
+    with mock.patch.object(client, 'CoreV1Api'):
+        return ConfigMapLock('lock', 'default', 'candidate')
+
+
+class ConfigMapLockTest(unittest.TestCase):
+
+    def test_leader_transitions_survive_a_write_and_read(self):
+        lock = make_lock()
+        record = LeaderElectionRecord('candidate', '17', '2026-08-29 01:02:03',
+                                      '2026-08-29 01:02:03', 5)
+
+        lock_dict = lock.get_lock_dict(record)
+        self.assertEqual(5, lock_dict['leaderTransitions'])
+
+        read_back = lock.get_lock_object(dict(lock_dict))
+        self.assertEqual(record.__dict__, read_back.__dict__)
+
+    def test_annotation_without_transitions_reads_as_zero(self):
+        lock = make_lock()
+        annotation = {'holderIdentity': 'candidate',
+                      'leaseDurationSeconds': '17',
+                      'acquireTime': '2026-08-29 01:02:03',
+                      'renewTime': '2026-08-29 01:02:03'}
+
+        self.assertEqual(0, lock.get_lock_object(annotation).leader_transitions)
+
+
+if __name__ == '__main__':
+    unittest.main()

kubernetes/leaderelection/resourcelock/leaselock.py

@@ -108,6 +108,7 @@ def get_lease_spec(self, leader_election_record, current_spec=None):
             leader_election_record.acquire_time)
         spec.renew_time = self.time_str_to_iso(
             leader_election_record.renew_time)
+        spec.lease_transitions = leader_election_record.leader_transitions
 
         return spec
 
@@ -129,6 +130,8 @@ def get_lock_object(self, lease):
         if lease.spec.renew_time:
             leader_election_record.renew_time = self.time_from_utc(
                 lease.spec.renew_time)
+        if lease.spec.lease_transitions:
+            leader_election_record.leader_transitions = lease.spec.lease_transitions
 
         return leader_election_record
 

kubernetes/leaderelection/resourcelock/leaselock_test.py

@@ -130,6 +130,30 @@ def test_record_without_microseconds_survives_a_write_and_read(self):
 
         self.assertEqual(record.__dict__, read_back.__dict__)
 
+    def test_leader_transitions_survive_a_write_and_read(self):
+        lock = make_lock()
+        record = LeaderElectionRecord('candidate', '17', '2026-08-29 01:02:03',
+                                      '2026-08-29 01:02:03', 5)
+
+        spec = lock.get_lease_spec(record)
+        self.assertEqual(5, spec.lease_transitions)
+
+        read_back = lock.get_lock_object(
+            client.V1Lease(metadata={'name': 'lock'}, spec=spec))
+        self.assertEqual(record.__dict__, read_back.__dict__)
+
+    def test_lease_without_transitions_reads_as_zero(self):
+        lock = make_lock()
+        acquired = datetime.datetime(2026, 8, 29, 1, 2, 3)
+        lease = client.V1Lease(
+            metadata={'name': 'lock'},
+            spec=client.V1LeaseSpec(holder_identity='candidate',
+                                    lease_duration_seconds=17,
+                                    acquire_time=acquired,
+                                    renew_time=acquired))
+
+        self.assertEqual(0, lock.get_lock_object(lease).leader_transitions)
+
     def test_update_replaces_the_lease_it_read(self):
         lock = make_lock()
         acquired = datetime.datetime(2026, 8, 29, 1, 2, 3, 456789)