Recheck annotations when doing attempt_to_acquire_leader on 409 (patroni#3348)

We already doing it in update_leader() code path. However,
attempt_to_acquire_leader() is also affected by the same problem when
for example we are trying to reacquire the leader lock after K8s API
outage in DCS failsafe mode.

Ref patroni#3174
Close patroni#3344

Author: CyberDem0n

patroni/dcs/kubernetes.py

@@ -1236,6 +1236,10 @@ def _retry(*args: Any, **kwargs: Any) -> Any:
         return bool(_run_and_handle_exceptions(self._patch_or_create, self.leader_path, annotations,
                                                kind_resource_version, ips=ips, retry=_retry))
 
+    @staticmethod
+    def _isotime() -> str:
+        return datetime.datetime.now(tzutc).isoformat()
+
     def update_leader(self, cluster: Cluster, last_lsn: Optional[int],
                       slots: Optional[Dict[str, int]] = None, failsafe: Optional[Dict[str, str]] = None) -> bool:
         kind = self._kinds.get(self.leader_path)
@@ -1244,7 +1248,7 @@ def update_leader(self, cluster: Cluster, last_lsn: Optional[int],
         if kind and kind_annotations.get(self._LEADER) != self._name:
             return False
 
-        now = datetime.datetime.now(tzutc).isoformat()
+        now = self._isotime()
         leader_observed_record = kind_annotations or self._leader_observed_record
         annotations = {self._LEADER: self._name, 'ttl': str(self._ttl), 'renewTime': now,
                        'acquireTime': leader_observed_record.get('acquireTime') or now,
@@ -1262,7 +1266,7 @@ def update_leader(self, cluster: Cluster, last_lsn: Optional[int],
         return self._update_leader_with_retry(annotations, resource_version, self.__ips)
 
     def attempt_to_acquire_leader(self) -> bool:
-        now = datetime.datetime.now(tzutc).isoformat()
+        now = self._isotime()
         annotations = {self._LEADER: self._name, 'ttl': str(self._ttl),
                        'renewTime': now, 'acquireTime': now, 'transitions': '0'}
         if self._leader_observed_record:
@@ -1277,18 +1281,55 @@ def attempt_to_acquire_leader(self) -> bool:
                 annotations['acquireTime'] = self._leader_observed_record.get('acquireTime') or now
             annotations['transitions'] = str(transitions)
 
+        resource_version = self._leader_resource_version
+        if resource_version:
+            kind = self._kinds.get(self.leader_path)
+            # If leader object in cache was updated we should better use fresh resource_version
+            if kind and kind.metadata.resource_version != resource_version:
+                kind_annotations = kind and kind.metadata.annotations or EMPTY_DICT
+                # But, only in case if leader annotations didn't change
+                if all(kind_annotations.get(k) == self._leader_observed_record.get(k) for k in annotations.keys()):
+                    resource_version = kind.metadata.resource_version
+
+        retry = self._retry.copy()
+
+        def _retry(*args: Any, **kwargs: Any) -> Any:
+            kwargs['_retry'] = retry
+            return retry(*args, **kwargs)
+
+        handle_conflict = False
         try:
             ret = bool(self._patch_or_create(self.leader_path, annotations,
-                                             self._leader_resource_version, retry=self.retry, ips=self.__ips))
+                                             resource_version, retry=_retry, ips=self.__ips))
         except k8s_client.rest.ApiException as e:
-            if e.status == 409 and self._leader_resource_version:  # Conflict in resource_version
+            if e.status == 409 and resource_version:  # Conflict in resource_version
                 # Terminate watchers, it could be a sign that K8s API is in a failed state
                 self._kinds.kill_stream()
                 self._pods.kill_stream()
+                handle_conflict = True
             ret = False
         except (RetryFailedError, K8sException) as e:
             raise KubernetesError(e)
 
+        if handle_conflict and retry.ensure_deadline(1):
+            # if we are here, that means update failed with 409
+            # Try to get the latest version directly from K8s API instead of relying on async cache
+            try:
+                kind = _retry(self._api.read_namespaced_kind, self.leader_path, self._namespace)
+            except (RetryFailedError, K8sException) as e:
+                raise KubernetesError(e)
+            except Exception as e:
+                logger.error('Failed to get the leader object "%s": %r', self.leader_path, e)
+            else:
+                kind_annotations = kind and kind.metadata.annotations or EMPTY_DICT
+                kind_resource_version = kind and kind.metadata.resource_version
+
+                # We can get 409 because we do at least one retry, and the first update might have succeeded,
+                # therefore we will check if annotations on the read object match expectations.
+                if kind and kind_resource_version != resource_version and\
+                        all(kind_annotations.get(k) == v for k, v in annotations.items()):
+                    ret = True
+
         if not ret:
             logger.info('Could not take out TTL lock')
         return ret

tests/test_kubernetes.py

@@ -294,13 +294,27 @@ def test_get_citus_coordinator(self, mock_logger):
             self.assertEqual(mock_logger.call_args[0][1], 'Citus')
             self.assertIsInstance(mock_logger.call_args[0][2], KubernetesError)
 
-    def test_attempt_to_acquire_leader(self):
+    @patch.object(k8s_client.CoreV1Api, 'read_namespaced_config_map', create=True)
+    def test_attempt_to_acquire_leader(self, mock_read):
+        metadata = k8s_client.V1ObjectMeta(resource_version='2', labels={'f': 'b'}, name='test',
+                                           annotations={'optime': '1234', 'leader': 'p-0', 'transitions': '0',
+                                                        'renewTime': 'now', 'acquireTime': 'now', 'ttl': '30'})
+        mock_read.return_value = k8s_client.V1ConfigMap(metadata=metadata)
         with patch.object(k8s_client.CoreV1Api, 'patch_namespaced_config_map', create=True) as mock_patch:
             mock_patch.side_effect = K8sException
             self.assertRaises(KubernetesError, self.k.attempt_to_acquire_leader)
+
             mock_patch.side_effect = k8s_client.rest.ApiException(409, '')
+            self.k._leader_resource_version = '0'
+            self.k._isotime = Mock(return_value='now')
+            self.assertTrue(self.k.attempt_to_acquire_leader())
+
+            mock_read.side_effect = Exception
             self.assertFalse(self.k.attempt_to_acquire_leader())
 
+            mock_read.side_effect = RetryFailedError('')
+            self.assertRaises(KubernetesError, self.k.attempt_to_acquire_leader)
+
     def test_take_leader(self):
         self.k.take_leader()
         self.k._leader_observed_record['leader'] = 'test'