Repository: zalando/patroni
Pull request #2895 (2023-10-04 12:43:17)

Take IP from the pod if `kubernetes.pod_ip` is missing
It used to work before #2652

Besides that fix a couple of more problems:
- make sure `_patch_or_create()` method isn't instantiating the `k8s_client.V1ConfigMap` object instead of `k8s_client.V1Endpoints` for non leader endpoints. The only reason it worked is that the JSON serialization for both object types is the same and doesn't include the object type name.
- `attempt_to_acquire_leader()` should immediately put the IP address of the primary to the leader endpoint. It didn't happen because of the oversight in the https://github.com/zalando/patroni/pull/1820.

coveralls: ## Pull Request Test Coverage Report for [Build 6406141346](https://coveralls.io/builds/63079342)

* **1** of **1**   **(100.0%)**  changed or added relevant line in **1** file are covered.
* No unchanged relevant lines lost coverage.
* Overall coverage remained the same at **99.863%**

---



|  Totals | [![Coverage Status](https://coveralls.io/builds/63079342/badge)](https://coveralls.io/builds/63079342) |
| :-- | --: |
| Change from base [Build 6404304255](https://coveralls.io/builds/63075492): |  0.0% |
| Covered Lines: | 13143 |
| Relevant Lines: | 13161 |

---
##### 💛  - [Coveralls](https://coveralls.io)

Jan-M: :+1:
hughcapet: I think we need to document the subsets definition logic in the user docs. i.e. fix the `kubernetes.pod_ip` description 

Patch:
diff --git a/features/standby_cluster.feature b/features/standby_cluster.feature
index 7ff07679..ac6bbb85 100644
--- a/features/standby_cluster.feature
+++ b/features/standby_cluster.feature
@@ -53,15 +53,22 @@ Feature: standby cluster
     And I receive a response replication_state streaming
     And postgres1 does not have a logical replication slot named test_logical
 
+  Scenario: check switchover
+    When I run patronictl.py switchover batman1 --force
+    And I issue a GET request to http://127.0.0.1:8010/standby_leader
+    Then I receive a response code 200
+    And I receive a response role standby_leader
+    And postgres1 is replicating from postgres2 after 32 seconds
+
   Scenario: check failover
-    When I kill postgres1
-    And I kill postmaster on postgres1
-    Then postgres2 is replicating from postgres0 after 32 seconds
-    When I issue a GET request to http://127.0.0.1:8010/primary
+    When I kill postgres2
+    And I kill postmaster on postgres2
+    Then postgres1 is replicating from postgres0 after 32 seconds
+    When I issue a GET request to http://127.0.0.1:8009/primary
     Then I receive a response code 503
     And I sleep for 3 seconds
-    When I issue a GET request to http://127.0.0.1:8010/standby_leader
+    When I issue a GET request to http://127.0.0.1:8009/standby_leader
     Then I receive a response code 200
     And I receive a response role standby_leader
-    And replication works from postgres0 to postgres2 after 15 seconds
-    And there is a postgres2_cb.log with "on_start replica batman1\non_role_change standby_leader batman1" in postgres2 data directory
+    And replication works from postgres0 to postgres1 after 15 seconds
+    And there is a postgres1_cb.log with "on_start replica batman1\non_role_change standby_leader batman1" in postgres1 data directory
diff --git a/features/steps/standby_cluster.py b/features/steps/standby_cluster.py
index 17d635b8..a428130c 100644
--- a/features/steps/standby_cluster.py
+++ b/features/steps/standby_cluster.py
@@ -34,6 +34,7 @@ def start_patroni_standby_cluster(context, name, cluster_name, name2):
                 "ttl": 20,
                 "loop_wait": 2,
                 "retry_timeout": 5,
+                "synchronous_mode": True,  # should be completely ignored
                 "standby_cluster": {
                     "host": "localhost",
                     "port": port,
diff --git a/patroni/config.py b/patroni/config.py
index a0384504..5779aaac 100644
--- a/patroni/config.py
+++ b/patroni/config.py
@@ -96,8 +96,8 @@ class GlobalConfig(object):
 
     @property
     def is_synchronous_mode(self) -> bool:
-        """``True`` if synchronous replication is requested."""
-        return self.check_mode('synchronous_mode')
+        """``True`` if synchronous replication is requested and it is not a standby cluster config."""
+        return self.check_mode('synchronous_mode') and not self.is_standby_cluster
 
     @property
     def is_synchronous_mode_strict(self) -> bool:
diff --git a/patroni/dcs/kubernetes.py b/patroni/dcs/kubernetes.py
index d4158368..aee87bd3 100644
--- a/patroni/dcs/kubernetes.py
+++ b/patroni/dcs/kubernetes.py
@@ -771,8 +771,7 @@ class Kubernetes(AbstractDCS):
         except k8s_config.ConfigException:
             k8s_config.load_kube_config(context=config.get('context', 'kind-kind'))
 
-        pod_ip = config.get('pod_ip')
-        self.__ips: List[str] = [] if self._ctl or not isinstance(pod_ip, str) else [pod_ip]
+        self.__ips: List[str] = [] if self._ctl else [config.get('pod_ip', '')]
         self.__ports: List[K8sObject] = []
         ports: List[Dict[str, Any]] = config.get('ports', [{}])
         for p in ports:
@@ -1059,6 +1058,27 @@ class Kubernetes(AbstractDCS):
     def _patch_or_create(self, name: str, annotations: Dict[str, Any],
                          resource_version: Optional[str] = None, patch: bool = False,
                          retry: Optional[Callable[..., Any]] = None, ips: Optional[List[str]] = None) -> K8sObject:
+        """Patch or create K8s object, Endpoint or ConfigMap.
+
+        :param name: the name of the object.
+        :param annotations: mapping of annotations that we want to create/update.
+        :param resource_version: object should be updated only if the ``resource_version`` matches provided value.
+        :param patch: ``True`` if we know in advance that the object already exists and we should patch it.
+        :param retry: a callable that will take care of retries
+        :param ips: IP address that we want to put to the subsets of the endpoint. Could have following values:
+
+                    * ``None`` - when we don't need to touch subset;
+                    * ``[]`` - to set subsets to the empty list, when :meth:`delete_leader` method is called;
+
+                    * ``['ip.add.re.ss']`` - when we want to make sure that the subsets of the leader endpoint
+                      contains the IP address of the leader, that we get from the ``kubernetes.pod_ip``;
+
+                    * ``['']`` - when we want to make sure that the subsets of the leader endpoint contains the IP
+                      address of the leader, but ``kubernetes.pod_ip`` configuration is missing. In this case we will
+                      try to take the IP address of the Pod which name matches ``name`` from the config file.
+
+        :returns: the new :class:`V1Endpoints` or :class:`V1ConfigMap` object, that was created or updated.
+        """
         metadata = {'namespace': self._namespace, 'name': name, 'labels': self._labels, 'annotations': annotations}
         if patch or resource_version:
             if resource_version is not None:
@@ -1071,9 +1091,10 @@ class Kubernetes(AbstractDCS):
             metadata['annotations'] = {k: v for k, v in annotations.items() if v is not None}
 
         metadata = k8s_client.V1ObjectMeta(**metadata)
-        if ips is not None and self._api.use_endpoints:
+        if self._api.use_endpoints:
             endpoints = {'metadata': metadata}
-            self._map_subsets(endpoints, ips)
+            if ips is not None:
+                self._map_subsets(endpoints, ips)
             body = k8s_client.V1Endpoints(**endpoints)
         else:
             body = k8s_client.V1ConfigMap(metadata=metadata)
@@ -1222,11 +1243,10 @@ class Kubernetes(AbstractDCS):
             else:
                 annotations['acquireTime'] = self._leader_observed_record.get('acquireTime') or now
             annotations['transitions'] = str(transitions)
-        ips: Optional[List[str]] = [] if self._api.use_endpoints else None
 
         try:
             ret = bool(self._patch_or_create(self.leader_path, annotations,
-                                             self._leader_resource_version, retry=self.retry, ips=ips))
+                                             self._leader_resource_version, retry=self.retry, ips=self.__ips))
         except k8s_client.rest.ApiException as e:
             if e.status == 409 and self._leader_resource_version:  # Conflict in resource_version
                 # Terminate watchers, it could be a sign that K8s API is in a failed state
diff --git a/tox.ini b/tox.ini
index bf66caef..21688937 100644
--- a/tox.ini
+++ b/tox.ini
@@ -125,10 +125,11 @@ commands =
         --file features/Dockerfile
 allowlist_externals = docker
 
-[testenv:pg{12,13,14,15,16}-docker-behave-{etcd}-{lin,mac}]
+[testenv:pg{12,13,14,15,16}-docker-behave-{etcd,etcd3}-{lin,mac}]
 description = Run behaviour tests in patroni-dev docker container
 setenv =
     etcd: DCS=etcd
+    etcd3: DCS=etcd3
     {[common]postgres_matrix}
     CONTAINER_NAME = tox-{env_name}-{env:PYTHONHASHSEED}
 labels =
@@ -149,7 +150,7 @@ commands =
         --tty \
         {env:PATRONI_DEV_IMAGE:patroni-dev:{env:PG_MAJOR}} \
         tox run -x 'tox.env_list=py{[common]python_matrix}-behave-{env:DCS}-lin' \
-        -- --format plain {posargs}
+        -- {posargs}
 
 allowlist_externals =
     docker
@@ -159,7 +160,7 @@ platform =
 ;    win: win32
     mac: darwin
 
-[testenv:py{36,38,39,310,311}-behave-{etcd}-{lin,win,mac}]
+[testenv:py{36,38,39,310,311}-behave-{etcd,etcd3}-{lin,win,mac}]
 description = Run behaviour tests (locally with tox)
 deps =
     -r requirements.txt
@@ -167,11 +168,15 @@ deps =
     coverage
     {[common]psycopg_deps}
 setenv =
-    DCS = {env:DCS:etcd}
+    etcd: DCS = {env:DCS:etcd}
+    etcd3: DCS = {env:DCS:etcd3}
 passenv =
     ETCD_UNSUPPORTED_ARCH
 commands =
-    python3 -m behave {posargs}
+    python3 -m behave --format json --format plain --outfile result.json {posargs}
+    mv result.json features/output
+allowlist_externals =
+    mv
 platform =
     {[common]platforms}
 


Test patch:
diff --git a/tests/test_config.py b/tests/test_config.py
index 8515476a..2b21f589 100644
--- a/tests/test_config.py
+++ b/tests/test_config.py
@@ -5,7 +5,7 @@ import io
 
 from copy import deepcopy
 from mock import MagicMock, Mock, patch
-from patroni.config import Config, ConfigParseError
+from patroni.config import Config, ConfigParseError, GlobalConfig
 
 
 class TestConfig(unittest.TestCase):
@@ -197,3 +197,8 @@ class TestConfig(unittest.TestCase):
             self.assertEqual(mock_logger.call_args_list[0][0],
                              ('Violated the rule "loop_wait + 2*retry_timeout <= ttl", where ttl=%d. Adjusting'
                               ' loop_wait from %d to %d and retry_timeout from %d to %d', 20, 10, 1, 10, 9))
+
+    def test_global_config_is_synchronous_mode(self):
+        # we should ignore synchronous_mode setting in a standby cluster
+        config = {'standby_cluster': {'host': 'some_host'}, 'synchronous_mode': True}
+        self.assertFalse(GlobalConfig(config).is_synchronous_mode)
diff --git a/tests/test_kubernetes.py b/tests/test_kubernetes.py
index 4f9f418c..b6db7fb3 100644
--- a/tests/test_kubernetes.py
+++ b/tests/test_kubernetes.py
@@ -63,7 +63,7 @@ def mock_list_namespaced_pod(*args, **kwargs):
     metadata = k8s_client.V1ObjectMeta(resource_version='1', labels={'f': 'b', Kubernetes._CITUS_LABEL: '1'},
                                        name='p-0', annotations={'status': '{}'},
                                        uid='964dfeae-e79b-4476-8a5a-1920b5c2a69d')
-    status = k8s_client.V1PodStatus(pod_ip='10.0.0.0')
+    status = k8s_client.V1PodStatus(pod_ip='10.0.0.1')
     spec = k8s_client.V1PodSpec(hostname='p-0', node_name='kind-control-plane', containers=[])
     items = [k8s_client.V1Pod(metadata=metadata, status=status, spec=spec)]
     return k8s_client.V1PodList(items=items, kind='PodList')
@@ -356,6 +356,20 @@ class TestKubernetesConfigMaps(BaseTestKubernetes):
         mock_warning.assert_called_once()
 
 
+class TestKubernetesEndpointsNoPodIP(BaseTestKubernetes):
+    @patch.object(k8s_client.CoreV1Api, 'list_namespaced_endpoints', mock_list_namespaced_endpoints, create=True)
+    def setUp(self, config=None):
+        super(TestKubernetesEndpointsNoPodIP, self).setUp({'use_endpoints': True})
+
+    @patch.object(k8s_client.CoreV1Api, 'patch_namespaced_endpoints', create=True)
+    def test_update_leader(self, mock_patch_namespaced_endpoints):
+        leader = self.k.get_cluster().leader
+        self.assertIsNotNone(self.k.update_leader(leader, '123', failsafe={'foo': 'bar'}))
+        args = mock_patch_namespaced_endpoints.call_args[0]
+        self.assertEqual(args[2].subsets[0].addresses[0].target_ref.resource_version, '1')
+        self.assertEqual(args[2].subsets[0].addresses[0].ip, '10.0.0.1')
+
+
 class TestKubernetesEndpoints(BaseTestKubernetes):
 
     @patch.object(k8s_client.CoreV1Api, 'list_namespaced_endpoints', mock_list_namespaced_endpoints, create=True)
@@ -368,6 +382,7 @@ class TestKubernetesEndpoints(BaseTestKubernetes):
         self.assertIsNotNone(self.k.update_leader(leader, '123', failsafe={'foo': 'bar'}))
         args = mock_patch_namespaced_endpoints.call_args[0]
         self.assertEqual(args[2].subsets[0].addresses[0].target_ref.resource_version, '10')
+        self.assertEqual(args[2].subsets[0].addresses[0].ip, '10.0.0.0')
         self.k._kinds._object_cache['test'].subsets[:] = []
         self.assertIsNotNone(self.k.update_leader(leader, '123'))
         self.k._kinds._object_cache['test'].metadata.annotations['leader'] = 'p-1'