kubernetes-client · nkbeast · Sep 3, 2026
diff --git a/kubernetes/aio/leaderelection/leaderelection.py b/kubernetes/aio/leaderelection/leaderelection.py
@@ -145,11 +145,22 @@ async def try_acquire_or_renew(self) -> bool:
 
         # A lock is not created with that name, try to create one
         if not lock_status:
-            assert (
+            # The error body comes straight from the API server, but anything
+            # sitting in front of it (ingress, load balancer, proxy) can answer
+            # with an HTML page, an empty payload or some other non-JSON body.
+            # Only a clean 404 means the lock is absent and may be created;
+            # everything else is retried on the next period instead of taking
+            # the whole leader election down.
+            error_code = None
+            if (
                 isinstance(old_election_record, ApiException)
                 and old_election_record.body is not None
-            )
-            if json.loads(old_election_record.body)["code"] != HTTPStatus.NOT_FOUND:
+            ):
+                try:
+                    error_code = json.loads(old_election_record.body)["code"]
+                except (ValueError, TypeError, KeyError, AttributeError):
+                    error_code = None
+            if error_code != HTTPStatus.NOT_FOUND:
                 logger.error(
                     "Error retrieving resource lock %s as %s",
                     self.election_config.lock.name,

diff --git a/kubernetes/aio/leaderelection/leaderelection_test.py b/kubernetes/aio/leaderelection/leaderelection_test.py
@@ -365,5 +365,38 @@ async def update(
             self.lock.release()
 
 
+    def test_acquire_survives_non_json_error_body(self):
+        """A proxy answering with an HTML error page must not kill the elector."""
+
+        class GatewayErrorLock:
+            def __init__(self):
+                self.name = "lock"
+                self.namespace = "ns"
+                self.identity = "candidate"
+
+            async def get(self, name, namespace):
+                return False, ApiException(
+                    status=502,
+                    reason="Bad Gateway",
+                    body="<html><body>502 Bad Gateway</body></html>",
+                )
+
+            async def create(self, name, namespace, election_record):
+                return False
+
+        config = electionconfig.Config(
+            lock=GatewayErrorLock(),
+            lease_duration=4,
+            renew_deadline=3,
+            retry_period=1,
+            onstarted_leading=lambda: None,
+            onstopped_leading=lambda: None,
+        )
+
+        elector = leaderelection.LeaderElection(config)
+        result = asyncio.run(elector.try_acquire_or_renew())
+        self.assertFalse(result)
+
+
 if __name__ == "__main__":
     unittest.main()
diff --git a/kubernetes/aio/leaderelection/resourcelock/configmaplock.py b/kubernetes/aio/leaderelection/resourcelock/configmaplock.py
@@ -80,9 +80,26 @@ async def get(
                 self.configmap_reference = api_response
                 return True, None
 
-            lock_record = self.get_lock_object(
-                json.loads(annotations[self.leader_electionrecord_annotationkey])
-            )
+            # A corrupted annotation must not take the elector down: treat it
+            # like a missing one so the next update rewrites a clean record.
+            try:
+                annotation_record = json.loads(
+                    annotations[self.leader_electionrecord_annotationkey]
+                )
+            except ValueError:
+                logger.warning(
+                    "Leader election annotation on ConfigMap %s/%s is not valid "
+                    "JSON; treating the lock as unheld",
+                    name,
+                    namespace,
+                )
+                api_response.metadata.annotations = {
+                    self.leader_electionrecord_annotationkey: ""
+                }
+                self.configmap_reference = api_response
+                return True, None
+
+            lock_record = self.get_lock_object(annotation_record)
 
             self.configmap_reference = api_response
             return True, lock_record

diff --git a/kubernetes/leaderelection/leaderelection.py b/kubernetes/leaderelection/leaderelection.py
@@ -130,8 +130,17 @@ def try_acquire_or_renew(self):
 
         # A lock is not created with that name, try to create one
         if not lock_status:
-            if json.loads(old_election_record.body)[
-                    'code'] != HTTPStatus.NOT_FOUND:
+            # The error body comes straight from the API server, but anything
+            # sitting in front of it (ingress, load balancer, proxy) can answer
+            # with an HTML page, an empty payload or some other non-JSON body.
+            # Only a clean 404 means the lock is absent and may be created;
+            # everything else is retried on the next period instead of taking
+            # the whole leader election down.
+            try:
+                error_code = json.loads(old_election_record.body)['code']
+            except (ValueError, TypeError, KeyError, AttributeError):
+                error_code = None
+            if error_code != HTTPStatus.NOT_FOUND:
                 logger.info(
                     "Error retrieving resource lock {} as {}".format(
                         self.election_config.lock.name,

diff --git a/kubernetes/leaderelection/leaderelection_test.py b/kubernetes/leaderelection/leaderelection_test.py
@@ -321,5 +321,76 @@ def update(self, name, namespace, updated_record):
             self.lock.release()
 
 
+    def test_acquire_survives_non_json_error_body(self):
+        """A proxy answering with an HTML error page must not kill the elector."""
+        class GatewayErrorLock:
+            def __init__(self):
+                self.name = "lock"
+                self.namespace = "ns"
+                self.identity = "candidate"
+
+            def get(self, name, namespace):
+                return False, ApiException(
+                    status=502, reason="Bad Gateway",
+                    body="<html><body>502 Bad Gateway</body></html>")
+
+        config = electionconfig.Config(
+            lock=GatewayErrorLock(), lease_duration=4, renew_deadline=3,
+            retry_period=1, onstarted_leading=lambda: None,
+            onstopped_leading=lambda: None)
+
+        result = leaderelection.LeaderElection(config).try_acquire_or_renew()
+        self.assertFalse(result)
+
+    def test_acquire_survives_empty_error_body(self):
+        class EmptyErrorLock:
+            def __init__(self):
+                self.name = "lock"
+                self.namespace = "ns"
+                self.identity = "candidate"
+
+            def get(self, name, namespace):
+                return False, ApiException(status=500, reason="Server Error",
+                                           body=None)
+
+        config = electionconfig.Config(
+            lock=EmptyErrorLock(), lease_duration=4, renew_deadline=3,
+            retry_period=1, onstarted_leading=lambda: None,
+            onstopped_leading=lambda: None)
+
+        result = leaderelection.LeaderElection(config).try_acquire_or_renew()
+        self.assertFalse(result)
+
+    def test_acquire_still_creates_on_clean_404(self):
+        class NotFoundLock:
+            def __init__(self):
+                self.name = "lock"
+                self.namespace = "ns"
+                self.identity = "candidate"
+                self.created = False
+
+            def get(self, name, namespace):
+                if self.created:
+                    return True, LeaderElectionRecord(
+                        "candidate", "4", "now", "now")
+                return False, ApiException(
+                    status=404, reason="Not Found",
+                    body=json.dumps({'code': 404}))
+
+            def create(self, name, namespace, election_record):
+                self.created = True
+                return True
+
+            def update(self, name, namespace, updated_record):
+                return True
+
+        config = electionconfig.Config(
+            lock=NotFoundLock(), lease_duration=4, renew_deadline=3,
+            retry_period=1, onstarted_leading=lambda: None,
+            onstopped_leading=lambda: None)
+
+        self.assertTrue(leaderelection.LeaderElection(config).try_acquire_or_renew())
+
+
 if __name__ == '__main__':
     unittest.main()
diff --git a/kubernetes/leaderelection/resourcelock/configmaplock.py b/kubernetes/leaderelection/resourcelock/configmaplock.py
@@ -64,7 +64,20 @@ def get(self, name, namespace):
                 self.configmap_reference = api_response
                 return True, None
 
-            lock_record = self.get_lock_object(json.loads(annotations[self.leader_electionrecord_annotationkey]))
+            # A corrupted annotation must not take the elector down: treat it
+            # like a missing one so the next update rewrites a clean record.
+            try:
+                annotation_record = json.loads(
+                    annotations[self.leader_electionrecord_annotationkey])
+            except ValueError:
+                logger.warning(
+                    "Leader election annotation on ConfigMap {}/{} is not valid "
+                    "JSON; treating the lock as unheld".format(name, namespace))
+                api_response.metadata.annotations = {self.leader_electionrecord_annotationkey: ''}
+                self.configmap_reference = api_response
+                return True, None
+
+            lock_record = self.get_lock_object(annotation_record)
 
             self.configmap_reference = api_response
             return True, lock_record