Handle watch cancelation messages in Etcd3 (patroni#3426)

Etcd3 sends cancelation message to watch channel when for example revision was compacted, however it doesn't close connection on their side. We solve it by breaking a loop of reading chunked response and closing connection on Patroni side.

Close patroni#3396

Author: CyberDem0n

patroni/dcs/etcd3.py

@@ -21,7 +21,7 @@
 from ..collections import EMPTY_DICT
 from ..exceptions import DCSError, PatroniException
 from ..postgresql.mpp import AbstractMPP
-from ..utils import deep_compare, enable_keepalive, iter_response_objects, RetryFailedError, USER_AGENT
+from ..utils import deep_compare, enable_keepalive, iter_response_objects, parse_bool, RetryFailedError, USER_AGENT
 from . import catch_return_false_exception, Cluster, ClusterConfig, \
     Failover, Leader, Member, Status, SyncState, TimelineHistory
 from .etcd import AbstractEtcd, AbstractEtcdClientWithFailover, catch_etcd_errors, \
@@ -66,6 +66,10 @@ class Etcd3Exception(etcd.EtcdException):
     pass
 
 
+class Etcd3WatchCanceled(Etcd3Exception):
+    pass
+
+
 class Etcd3ClientError(Etcd3Exception):
 
     def __init__(self, code: Optional[int] = None, error: Optional[str] = None, status: Optional[int] = None) -> None:
@@ -356,7 +360,6 @@ def handle_auth_errors(self: 'Etcd3Client', func: Callable[..., Any], *args: Any
                 exc = e
             self._reauthenticate = True
             if retry:
-                logger.error('retry = %s', retry)
                 retry.ensure_deadline(0.5, exc)
             elif reauthenticated:
                 raise exc
@@ -508,6 +511,8 @@ def _process_message(self, message: Dict[str, Any]) -> None:
         if 'error' in message:
             raise _raise_for_data(message)
         result = message.get('result', EMPTY_DICT)
+        if parse_bool(result.get('canceled')):
+            raise Etcd3WatchCanceled('Watch canceled')
         header = result.get('header', EMPTY_DICT)
         self._check_cluster_raft_term(header.get('cluster_id'), header.get('raft_term'))
         events: List[Dict[str, Any]] = result.get('events', [])
@@ -555,8 +560,10 @@ def _build_cache(self) -> None:
 
         try:
             self._do_watch(result['header']['revision'])
+        except Etcd3WatchCanceled:
+            logger.info('Watch request canceled')
         except Exception as e:
-            # Following exceptions are expected on Windows because the /watch request  is done with `read_timeout`
+            # Following exceptions are expected on Windows because the /watch request is done with `read_timeout`
             if not (os.name == 'nt' and isinstance(e, (ReadTimeoutError, ProtocolError))):
                 logger.error('watchprefix failed: %r', e)
         finally:

tests/test_etcd3.py

@@ -96,8 +96,12 @@ class TestKVCache(BaseTestEtcd3):
 
     @patch.object(urllib3.PoolManager, 'urlopen', mock_urlopen)
     @patch.object(Etcd3Client, 'watchprefix', Mock(return_value=urllib3.response.HTTPResponse()))
+    @patch.object(urllib3.response.HTTPResponse, 'read_chunked',
+                  Mock(return_value=[b'{"result":{"canceled":true}}']))
     def test__build_cache(self):
-        self.kv_cache._build_cache()
+        with patch('patroni.dcs.etcd3.logger') as mock_logger:
+            self.kv_cache._build_cache()
+            mock_logger.info.assert_called_once_with('Watch request canceled')
 
     def test__do_watch(self):
         self.client.watchprefix = Mock(return_value=False)