diff --git a/kubernetes/aio/leaderelection/leaderelection.py b/kubernetes/aio/leaderelection/leaderelection.py index 1289692b92..fb8938bd6f 100644 --- 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 index 3d1e35dc0e..c1e3656227 100644 --- 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="502 Bad Gateway", + ) + + 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 index 53a46c09bf..aa4c4862da 100644 --- 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 index fc72a1d95e..951511f513 100644 --- 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 index ad9c7e7d1e..ba12b49222 100644 --- 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="502 Bad Gateway") + + 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 index c2f1e1cc67..bb694a9b93 100644 --- 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