Make K8s retriable HTTP status code configurable (#2585)

Configuration parameter is `kubernetes.retriable_http_codes` or `PATRONI_KUBERNETES_RETRIABLE_HTTP_CODES` environment variable.

These status codes are added to the default list of 500, 503, 504.

Close https://github.com/zalando/patroni/issues/2536

docs/ENVIRONMENT.rst

@@ -117,6 +117,7 @@ Kubernetes
 -  **PATRONI\_KUBERNETES\_POD\_IP**: (optional) IP address of the pod Patroni is running in. This value is required when `PATRONI_KUBERNETES_USE_ENDPOINTS` is enabled and is used to populate the leader endpoint subsets when the pod's PostgreSQL is promoted.
 -  **PATRONI\_KUBERNETES\_PORTS**: (optional) if the Service object has the name for the port, the same name must appear in the Endpoint object, otherwise service won't work. For example, if your service is defined as ``{Kind: Service, spec: {ports: [{name: postgresql, port: 5432, targetPort: 5432}]}}``, then you have to set ``PATRONI_KUBERNETES_PORTS='[{"name": "postgresql", "port": 5432}]'`` and Patroni will use it for updating subsets of the leader Endpoint. This parameter is used only if `PATRONI_KUBERNETES_USE_ENDPOINTS` is set.
 -  **PATRONI\_KUBERNETES\_CACERT**: (optional) Specifies the file with the CA_BUNDLE file with certificates of trusted CAs to use while verifying Kubernetes API SSL certs. If not provided, patroni will use the value provided by the ServiceAccount secret.
+-  **PATRONI\_RETRIABLE\_HTTP\_CODES**: (optional) list of HTTP status codes from K8s API to retry on. By default Patroni is retrying on ``500``, ``503``, and ``504``, or if K8s API response has ``retry-after`` HTTP header.
 
 Raft (deprecated)
 -----------------

docs/SETTINGS.rst

@@ -218,6 +218,7 @@ Kubernetes
 -  **pod\_ip**: (optional) IP address of the pod Patroni is running in. This value is required when `use_endpoints` is enabled and is used to populate the leader endpoint subsets when the pod's PostgreSQL is promoted.
 -  **ports**: (optional) if the Service object has the name for the port, the same name must appear in the Endpoint object, otherwise service won't work. For example, if your service is defined as ``{Kind: Service, spec: {ports: [{name: postgresql, port: 5432, targetPort: 5432}]}}``, then you have to set ``kubernetes.ports: [{"name": "postgresql", "port": 5432}]`` and Patroni will use it for updating subsets of the leader Endpoint. This parameter is used only if `kubernetes.use_endpoints` is set.
 -  **cacert**: (optional) Specifies the file with the CA_BUNDLE file with certificates of trusted CAs to use while verifying Kubernetes API SSL certs. If not provided, patroni will use the value provided by the ServiceAccount secret.
+-  **retriable\_http\_codes**: (optional) list of HTTP status codes from K8s API to retry on. By default Patroni is retrying on ``500``, ``503``, and ``504``, or if K8s API response has ``retry-after`` HTTP header.
 
 
 .. _raft_settings:

patroni/config.py

@@ -363,8 +363,8 @@ class Config(object):
                               'CACERT', 'CERT', 'KEY', 'VERIFY', 'TOKEN', 'CHECKS', 'DC', 'CONSISTENCY',
                               'REGISTER_SERVICE', 'SERVICE_CHECK_INTERVAL', 'SERVICE_CHECK_TLS_SERVER_NAME',
                               'SERVICE_TAGS', 'NAMESPACE', 'CONTEXT', 'USE_ENDPOINTS', 'SCOPE_LABEL', 'ROLE_LABEL',
-                              'POD_IP', 'PORTS', 'LABELS', 'BYPASS_API_SERVICE', 'KEY_PASSWORD', 'USE_SSL', 'SET_ACLS',
-                              'GROUP', 'DATABASE') and name:
+                              'POD_IP', 'PORTS', 'LABELS', 'BYPASS_API_SERVICE', 'RETRIABLE_HTTP_CODES', 'KEY_PASSWORD',
+                              'USE_SSL', 'SET_ACLS', 'GROUP', 'DATABASE') and name:
                     value = os.environ.pop(param)
                     if name == 'CITUS':
                         if suffix == 'GROUP':
@@ -373,7 +373,7 @@ class Config(object):
                             continue
                     elif suffix == 'PORT':
                         value = value and parse_int(value)
-                    elif suffix in ('HOSTS', 'PORTS', 'CHECKS', 'SERVICE_TAGS'):
+                    elif suffix in ('HOSTS', 'PORTS', 'CHECKS', 'SERVICE_TAGS', 'RETRIABLE_HTTP_CODES'):
                         value = value and _parse_list(value)
                     elif suffix in ('LABELS', 'SET_ACLS'):
                         value = _parse_dict(value)

patroni/dcs/kubernetes.py

@@ -19,6 +19,7 @@ from urllib3 import Timeout
 from urllib3.exceptions import HTTPError
 from six.moves.http_client import HTTPException
 from threading import Condition, Lock, Thread
+from typing import Any, Dict, List, Optional
 
 from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, SyncState,\
         TimelineHistory, CITUS_COORDINATOR_GROUP_ID, citus_group_re
@@ -220,7 +221,7 @@ class K8sClient(object):
 
         _API_URL_PREFIX = '/api/v1/namespaces/'
 
-        def __init__(self, bypass_api_service=False):
+        def __init__(self, bypass_api_service: Optional[bool] = False) -> None:
             self._bypass_api_service = bypass_api_service
             self.pool_manager = urllib3.PoolManager(**k8s_config.pool_config)
             self._base_uri = k8s_config.server
@@ -488,11 +489,15 @@ class KubernetesRetriableException(k8s_client.rest.ApiException):
 
 
 class CoreV1ApiProxy(object):
+    """Proxy class to work with k8s_client.CoreV1Api() object"""
 
-    def __init__(self, use_endpoints=False, bypass_api_service=False):
+    _DEFAULT_RETRIABLE_HTTP_CODES = frozenset([500, 503, 504])
+
+    def __init__(self, use_endpoints: Optional[bool] = False, bypass_api_service: Optional[bool] = False) -> None:
         self._api_client = k8s_client.ApiClient(bypass_api_service)
         self._core_v1_api = k8s_client.CoreV1Api(self._api_client)
         self._use_endpoints = bool(use_endpoints)
+        self._retriable_http_codes = set(self._DEFAULT_RETRIABLE_HTTP_CODES)
 
     def configure_timeouts(self, loop_wait, retry_timeout, ttl):
         # Normally every loop_wait seconds we should have receive something from the socket.
@@ -504,10 +509,21 @@ class CoreV1ApiProxy(object):
         self._api_client.set_read_timeout(retry_timeout)
         self._api_client.set_api_servers_cache_ttl(loop_wait)
 
+    def configure_retriable_http_codes(self, retriable_http_codes: List[int]) -> None:
+        self._retriable_http_codes = self._DEFAULT_RETRIABLE_HTTP_CODES | set(retriable_http_codes)
+
     def refresh_api_servers_cache(self):
         self._api_client.refresh_api_servers_cache()
 
-    def __getattr__(self, func):
+    def __getattr__(self, func: str):
+        """Intercepts calls to `CoreV1Api` methods.
+
+        Handles two important cases:
+        1. Depending on whether Patroni is configured to work with `ConfigMaps` or `Endpoints`
+           it remaps "virtual" method names from `*_kind` to `*_endpoints` or `*_config_map`.
+        2. It handles HTTP error codes and raises `KubernetesRetriableException`
+           if the given error is supposed to be handled with retry."""
+
         if func.endswith('_kind'):
             func = func[:-4] + ('endpoints' if self._use_endpoints else 'config_map')
 
@@ -515,7 +531,7 @@ class CoreV1ApiProxy(object):
             try:
                 return getattr(self._core_v1_api, func)(*args, **kwargs)
             except k8s_client.rest.ApiException as e:
-                if e.status in (500, 503, 504) or e.headers and 'retry-after' in e.headers:  # XXX
+                if e.status in self._retriable_http_codes or e.headers and 'retry-after' in e.headers:
                     raise KubernetesRetriableException(e)
                 raise
         return wrapper
@@ -775,10 +791,24 @@ class Kubernetes(AbstractDCS):
     def set_retry_timeout(self, retry_timeout):
         self._retry.deadline = retry_timeout
 
-    def reload_config(self, config):
+    def reload_config(self, config: Dict[str, Any]) -> None:
+        """Handles dynamic config changes.
+
+        Either cause by changes in the local configuration file + SIGHUP or by changes of dynamic configuration"""
+
         super(Kubernetes, self).reload_config(config)
         self._api.configure_timeouts(self.loop_wait, self._retry.deadline, self.ttl)
 
+        # retriable_http_codes supposed to be either int, list of integers or comma-separated string with integers.
+        retriable_http_codes = config.get('retriable_http_codes', [])
+        if not isinstance(retriable_http_codes, list):
+            retriable_http_codes = [c.strip() for c in str(retriable_http_codes).split(',')]
+
+        try:
+            self._api.configure_retriable_http_codes([int(c) for c in retriable_http_codes])
+        except Exception as e:
+            logger.warning('Invalid value of retriable_http_codes = %s: %r', config['retriable_http_codes'], e)
+
     @staticmethod
     def member(pod):
         annotations = pod.metadata.annotations or {}

patroni/validator.py

@@ -365,6 +365,7 @@ schema = Schema({
           Optional("use_endpoints"): bool,
           Optional("pod_ip"): Or(is_ipv4_address, is_ipv6_address),
           Optional("ports"): [{"name": str, "port": int}],
+          Optional("retriable_http_codes"): Or(int, [int]),
           },
       }),
   Optional("citus"): {

tests/test_config.py

@@ -61,6 +61,7 @@ class TestConfig(unittest.TestCase):
             'PATRONI_KUBERNETES_LABELS': 'a: b: c',
             'PATRONI_KUBERNETES_SCOPE_LABEL': 'a',
             'PATRONI_KUBERNETES_PORTS': '[{"name": "postgresql"}]',
+            'PATRONI_KUBERNETES_RETRIABLE_HTTP_CODES': '401',
             'PATRONI_ZOOKEEPER_HOSTS': "'host1:2181','host2:2181'",
             'PATRONI_EXHIBITOR_HOSTS': 'host1,host2',
             'PATRONI_EXHIBITOR_PORT': '8181',

tests/test_kubernetes.py

@@ -317,6 +317,17 @@ class TestKubernetesConfigMaps(BaseTestKubernetes):
     def test_set_history_value(self):
         self.k.set_history_value('{}')
 
+    @patch('patroni.dcs.kubernetes.logger.warning')
+    def test_reload_config(self, mock_warning):
+        self.k.reload_config({'loop_wait': 10, 'ttl': 30, 'retry_timeout': 10, 'retriable_http_codes': '401, 403 '})
+        self.assertEqual(self.k._api._retriable_http_codes, self.k._api._DEFAULT_RETRIABLE_HTTP_CODES | set([401, 403]))
+        self.k.reload_config({'loop_wait': 10, 'ttl': 30, 'retry_timeout': 10, 'retriable_http_codes': 402})
+        self.assertEqual(self.k._api._retriable_http_codes, self.k._api._DEFAULT_RETRIABLE_HTTP_CODES | set([402]))
+        self.k.reload_config({'loop_wait': 10, 'ttl': 30, 'retry_timeout': 10, 'retriable_http_codes': [405, 406]})
+        self.assertEqual(self.k._api._retriable_http_codes, self.k._api._DEFAULT_RETRIABLE_HTTP_CODES | set([405, 406]))
+        self.k.reload_config({'loop_wait': 10, 'ttl': 30, 'retry_timeout': 10, 'retriable_http_codes': True})
+        mock_warning.assert_called_once()
+
 
 class TestKubernetesEndpoints(BaseTestKubernetes):
 

tests/test_validator.py

@@ -59,6 +59,7 @@ config = {
         "use_endpoints": False,
         "pod_ip": "127.0.0.1",
         "ports": [{"name": "string", "port": 1000}],
+        "retriable_http_codes": [401],
     },
     "postgresql": {
         "listen": "127.0.0.2,::1:543",