Merge remote-tracking branch 'zalando/master' into multisite

Author: ants

.github/workflows/install_deps.py

@@ -10,7 +10,7 @@
 
 def install_requirements(what):
     subprocess.call([sys.executable, '-m', 'pip', 'install', '--upgrade', 'pip'])
-    s = subprocess.call([sys.executable, '-m', 'pip', 'install', '--upgrade', 'wheel', 'setuptools'])
+    s = subprocess.call([sys.executable, '-m', 'pip', 'install', '--upgrade', 'wheel', 'setuptools', 'distlib'])
     if s != 0:
         return s
 
@@ -25,11 +25,11 @@ def install_requirements(what):
     requirements += ['coverage']
     # try to split tests between psycopg2 and psycopg3
     requirements += ['psycopg[binary]'] if sys.version_info >= (3, 8, 0) and\
-        (sys.platform != 'darwin' or what == 'etcd3') else ['psycopg2-binary==2.9.9' 
+        (sys.platform != 'darwin' or what == 'etcd3') else ['psycopg2-binary==2.9.9'
                                                             if sys.platform == 'darwin' else 'psycopg2-binary']
 
-    from pip._vendor.distlib.markers import evaluator, DEFAULT_CONTEXT
-    from pip._vendor.distlib.util import parse_requirement
+    from distlib.markers import evaluator, DEFAULT_CONTEXT
+    from distlib.util import parse_requirement
 
     for r in read('requirements.txt').split('\n'):
         r = parse_requirement(r)

Dockerfile

@@ -1,6 +1,6 @@
 ## This Dockerfile is meant to aid in the building and debugging patroni whilst developing on your local machine
 ## It has all the necessary components to play/debug with a single node appliance, running etcd
-ARG PG_MAJOR=16
+ARG PG_MAJOR=17
 ARG COMPRESS=false
 ARG PGHOME=/home/postgres
 ARG PGDATA=$PGHOME/data

Dockerfile.citus

@@ -1,6 +1,6 @@
 ## This Dockerfile is meant to aid in the building and debugging patroni whilst developing on your local machine
 ## It has all the necessary components to play/debug with a single node appliance, running etcd
-ARG PG_MAJOR=16
+ARG PG_MAJOR=17
 ARG COMPRESS=false
 ARG PGHOME=/home/postgres
 ARG PGDATA=$PGHOME/data
@@ -40,7 +40,7 @@ RUN set -ex \
         echo "deb [signed-by=/etc/apt/trusted.gpg.d/citusdata_community.gpg] https://packagecloud.io/citusdata/community/debian/ $(lsb_release -cs) main" > /etc/apt/sources.list.d/citusdata_community.list \
         && curl -sL https://packagecloud.io/citusdata/community/gpgkey | gpg --dearmor > /etc/apt/trusted.gpg.d/citusdata_community.gpg \
         && apt-get update -y \
-        && apt-get -y install postgresql-$PG_MAJOR-citus-12.1; \
+        && apt-get -y install postgresql-$PG_MAJOR-citus-13.1; \
     fi \
 \
     # Cleanup all locales but en_US.UTF-8

features/steps/basic_replication.py

@@ -102,7 +102,7 @@ def initiate_load(context, pg_name):
         assert False, "Error loading test data on {0}: {1}".format(pg_name, e)
 
 
-@then('Table {table_name:w} is present on {pg_name:name} after {max_replication_delay:d} seconds')
+@then('table {table_name:w} is present on {pg_name:name} after {max_replication_delay:d} seconds')
 def table_is_present_on(context, table_name, pg_name, max_replication_delay):
     max_replication_delay *= context.timeout_multiplier
     for _ in range(int(max_replication_delay)):
@@ -124,7 +124,7 @@ def check_role(context, pg_name, pg_role, max_promotion_timeout):
 @step('replication works from {primary:name} to {replica:name} after {time_limit:d} seconds')
 @then('replication works from {primary:name} to {replica:name} after {time_limit:d} seconds')
 def replication_works(context, primary, replica, time_limit):
-    context.execute_steps(u"""
+    context.execute_steps("""
         When I add the table test_{0} to {1}
         Then table test_{0} is present on {2} after {3} seconds
     """.format(str(time()).replace('.', '_').replace(',', '_'), primary, replica, time_limit))

features/steps/cascading_replication.py

@@ -9,7 +9,7 @@ def start_patroni_with_a_name_value_tag(context, name, tag_name, tag_value):
     return context.pctl.start(name, custom_config={'tags': {tag_name: tag_value}})
 
 
-@then('There is a {label} with "{content}" in {name:name} data directory')
+@then('there is a {label} with "{content}" in {name:name} data directory')
 def check_label(context, label, content, name):
     value = (context.pctl.read_label(name, label) or '').replace('\n', '\\n')
     assert content in value, "\"{0}\" in {1} doesn't contain {2}".format(value, label, content)

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:
@@ -576,16 +583,21 @@ def run(self) -> None:
                 time.sleep(1)
 
     def kill_stream(self) -> None:
-        sock = None
+        conn_sock: Any = None
         with self._response_lock:
             if isinstance(self._response, urllib3.response.HTTPResponse):
                 try:
-                    sock = self._response.connection.sock if self._response.connection else None
+                    conn_sock = self._response.connection.sock if self._response.connection else None
                 except Exception:
-                    sock = None
+                    conn_sock = None
             else:
                 self._response = False
-        if sock:
+        if conn_sock:
+            # python-etcd forces usage of pyopenssl if the last one is available.
+            # In this case HTTPConnection.socket is not inherited from socket.socket, but urllib3 uses custom
+            # class `WrappedSocket`, which shutdown() method could be incompatible with socket.shutdown().
+            # Therefore we use WrappedSocket.socket, which points to original `socket` object.
+            sock: socket.socket = conn_sock.socket if conn_sock.__class__.__name__ == 'WrappedSocket' else conn_sock
             try:
                 sock.shutdown(socket.SHUT_RDWR)
                 sock.close()

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)