Failover logical slots (#1820)

Effectively, this PR consists of a few changes:

1. The easy part:
  In case of permanent logical slots are defined in the global configuration, Patroni on the primary will not only create them, but also periodically update DCS with the current values of `confirmed_flush_lsn` for all these slots.
  In order to reduce the number of interactions with DCS the new `/status` key was introduced. It will contain the json object with `optime` and `slots` keys. For backward compatibility the `/optime/leader` will be updated if there are members with old Patroni in the cluster.

2. The tricky part:
  On replicas that are eligible for a failover, Patroni creates the logical replication slot by copying the slot file from the primary and restarting the replica. In order to copy the slot file Patroni opens a connection to the primary with `rewind` or `superuser` credentials and calls `pg_read_binary_file()`  function.
  When the logical slot already exists on the replica Patroni periodically calls `pg_replication_slot_advance()` function, which allows moving the slot forward.

3. Additional requirements:
  In order to ensure that primary doesn't cleanup tuples from pg_catalog that are required for logical decoding, Patroni enables `hot_standby_feedback` on replicas with logical slots and on cascading replicas if they are used for streaming by replicas with logical slots.

4. When logical slots are copied from to the replica there is a timeframe when it could be not safe to use them after promotion. Right now there is no protection from promoting such a replica. But, Patroni will show the warning with names of the slots that might be not safe to use.

Compatibility.
The `pg_replication_slot_advance()` function is only available starting from PostgreSQL 11. For older Postgres versions Patroni will refuse to create the logical slot on the primary.

The old "permanent slots" feature, which creates logical slots right after promotion and before allowing connections, was removed.

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

tests/__init__.py

@@ -90,10 +90,14 @@ class MockCursor(object):
             raise psycopg2.OperationalError()
         elif sql.startswith('RetryFailedError'):
             raise RetryFailedError('retry')
+        elif sql.startswith('SELECT catalog_xmin'):
+            self.results = [(100, 501)]
+        elif sql.startswith('SELECT slot_name, catalog_xmin'):
+            self.results = [('ls', 100, 500, b'123456')]
         elif sql.startswith('SELECT slot_name'):
-            self.results = [('blabla', 'physical'), ('foobar', 'physical'), ('ls', 'logical', 'a', 'b')]
+            self.results = [('blabla', 'physical'), ('foobar', 'physical'), ('ls', 'logical', 'a', 'b', 5, 100, 500)]
         elif sql.startswith('SELECT CASE WHEN pg_catalog.pg_is_in_recovery()'):
-            self.results = [(1, 2, 1, 0, False, 1, 1, None, None)]
+            self.results = [(1, 2, 1, 0, False, 1, 1, None, None, [{"slot_name": "ls", "confirmed_flush_lsn": 12345}])]
         elif sql.startswith('SELECT pg_catalog.pg_is_in_recovery()'):
             self.results = [(False, 2)]
         elif sql.startswith('SELECT pg_catalog.pg_postmaster_start_time'):

tests/test_api.py

@@ -162,7 +162,7 @@ class TestRestApiHandler(unittest.TestCase):
     _authorization = '\nAuthorization: Basic dGVzdDp0ZXN0'
 
     def test_do_GET(self):
-        MockPatroni.dcs.cluster.last_leader_operation = 20
+        MockPatroni.dcs.cluster.last_lsn = 20
         MockRestApiServer(RestApiHandler, 'GET /replica')
         MockRestApiServer(RestApiHandler, 'GET /replica?lag=1M')
         MockRestApiServer(RestApiHandler, 'GET /replica?lag=10MB')

tests/test_consul.py

@@ -15,8 +15,7 @@ def kv_get(self, key, **kwargs):
         return None, None
     if key == 'service/good/leader':
         return '1', None
-    if key == 'service/good/':
-        return ('6429',
+    good_cls = ('6429',
                 [{'CreateIndex': 1334, 'Flags': 0, 'Key': key + 'failover', 'LockIndex': 0,
                   'ModifyIndex': 1334, 'Value': b''},
                  {'CreateIndex': 1334, 'Flags': 0, 'Key': key + 'initialize', 'LockIndex': 0,
@@ -34,7 +33,17 @@ def kv_get(self, key, **kwargs):
                  {'CreateIndex': 1085, 'Flags': 0, 'Key': key + 'optime/leader', 'LockIndex': 0,
                   'ModifyIndex': 6429, 'Value': b'4496294792'},
                  {'CreateIndex': 1085, 'Flags': 0, 'Key': key + 'sync', 'LockIndex': 0,
-                  'ModifyIndex': 6429, 'Value': b'{"leader": "leader", "sync_standby": null}'}])
+                  'ModifyIndex': 6429, 'Value': b'{"leader": "leader", "sync_standby": null}'},
+                 {'CreateIndex': 1085, 'Flags': 0, 'Key': key + 'status', 'LockIndex': 0,
+                  'ModifyIndex': 6429, 'Value': b'{"optime":4496294792, "slots":{"ls":12345}}'}])
+    if key == 'service/good/':
+        return good_cls
+    if key == 'service/broken/':
+        good_cls[1][-1]['Value'] = b'{'
+        return good_cls
+    if key == 'service/legacy/':
+        good_cls[1].pop()
+        return good_cls
     raise ConsulException
 
 
@@ -109,6 +118,10 @@ class TestConsul(unittest.TestCase):
         self.assertIsInstance(self.c.get_cluster(), Cluster)
         self.c._base_path = '/service/fail'
         self.assertRaises(ConsulError, self.c.get_cluster)
+        self.c._base_path = '/service/broken'
+        self.assertIsInstance(self.c.get_cluster(), Cluster)
+        self.c._base_path = '/service/legacy'
+        self.assertIsInstance(self.c.get_cluster(), Cluster)
         self.c._base_path = '/service/good'
         self.c._session = 'fd4f44fe-2cac-bba5-a60b-304b51ff39b8'
         self.assertIsInstance(self.c.get_cluster(), Cluster)
@@ -146,7 +159,7 @@ class TestConsul(unittest.TestCase):
 
     @patch.object(consul.Consul.Session, 'renew', Mock())
     def test_update_leader(self):
-        self.c.update_leader(None)
+        self.c.update_leader(12345)
 
     @patch.object(consul.Consul.KV, 'delete', Mock(return_value=True))
     def test_delete_leader(self):

tests/test_etcd.py

@@ -66,7 +66,13 @@ def etcd_read(self, key, **kwargs):
                         "?application_name=http://127.0.0.1:8008/patroni",
                      "expiration": "2015-05-15T09:11:09.611860899Z", "ttl": 30,
                      "modifiedIndex": 20730, "createdIndex": 20730}],
-                 "modifiedIndex": 1581, "createdIndex": 1581}], "modifiedIndex": 1581, "createdIndex": 1581}}
+                 "modifiedIndex": 1581, "createdIndex": 1581},
+                {"key": "/service/batman5/status", "value": '{"optime":2164261704,"slots":{"ls":12345}}',
+                 "modifiedIndex": 1582, "createdIndex": 1582}], "modifiedIndex": 1581, "createdIndex": 1581}}
+    if key == '/service/legacy/':
+        response['node']['nodes'].pop()
+    if key == '/service/broken/':
+        response['node']['nodes'][-1]['value'] = '{'
     result = etcd.EtcdResult(**response)
     result.etcd_index = 0
     return result
@@ -246,6 +252,10 @@ class TestEtcd(unittest.TestCase):
         cluster = self.etcd.get_cluster()
         self.assertIsInstance(cluster, Cluster)
         self.assertFalse(cluster.is_synchronous_mode())
+        self.etcd._base_path = '/service/legacy'
+        self.assertIsInstance(self.etcd.get_cluster(), Cluster)
+        self.etcd._base_path = '/service/broken'
+        self.assertIsInstance(self.etcd.get_cluster(), Cluster)
         self.etcd._base_path = '/service/nocluster'
         cluster = self.etcd.get_cluster()
         self.assertIsInstance(cluster, Cluster)

tests/test_etcd3.py

@@ -33,6 +33,8 @@ def mock_urlopen(self, method, url, **kwargs):
                  "value": base64_encode('foo'), "lease": "bla", "mod_revision": '1'},
                 {"key": base64_encode('/patroni/test/members/foo'),
                  "value": base64_encode('{}'), "lease": "123", "mod_revision": '1'},
+                {"key": base64_encode('/patroni/test/members/bar'),
+                 "value": base64_encode('{"version":"1.6.5"}'), "lease": "123", "mod_revision": '1'},
                 {"key": base64_encode('/patroni/test/failover'), "value": base64_encode('{}'), "mod_revision": '1'}
             ]
         })
@@ -172,6 +174,22 @@ class TestEtcd3(BaseTestEtcd3):
         self.assertIsInstance(self.etcd3.get_cluster(), Cluster)
         self.client._kv_cache = None
         with patch.object(urllib3.PoolManager, 'urlopen') as mock_urlopen:
+            mock_urlopen.return_value = MockResponse()
+            mock_urlopen.return_value.content = json.dumps({
+                "header": {"revision": "1"},
+                "kvs": [
+                    {"key": base64_encode('/patroni/test/status'),
+                     "value": base64_encode('{"optime":1234567,"slots":{"ls":12345}}'), "mod_revision": '1'}
+                ]
+            })
+            self.assertIsInstance(self.etcd3.get_cluster(), Cluster)
+            mock_urlopen.return_value.content = json.dumps({
+                "header": {"revision": "1"},
+                "kvs": [
+                    {"key": base64_encode('/patroni/test/status'), "value": base64_encode('{'), "mod_revision": '1'}
+                ]
+            })
+            self.assertIsInstance(self.etcd3.get_cluster(), Cluster)
             mock_urlopen.side_effect = UnsupportedEtcdVersion('')
             self.assertRaises(UnsupportedEtcdVersion, self.etcd3.get_cluster)
             mock_urlopen.side_effect = SleepException()

tests/test_ha.py

@@ -38,7 +38,7 @@ def get_cluster(initialize, leader, members, failover, sync, cluster_config=None
     history = TimelineHistory(1, '[[1,67197376,"no recovery target specified","' + t + '"]]',
                               [(1, 67197376, 'no recovery target specified', t)])
     cluster_config = cluster_config or ClusterConfig(1, {'check_timeline': True}, 1)
-    return Cluster(initialize, cluster_config, leader, 10, members, failover, sync, history)
+    return Cluster(initialize, cluster_config, leader, 10, members, failover, sync, history, None)
 
 
 def get_cluster_not_initialized_without_leader(cluster_config=None):
@@ -66,7 +66,7 @@ def get_cluster_initialized_with_leader(failover=None, sync=None):
 
 def get_cluster_initialized_with_only_leader(failover=None, cluster_config=None):
     leader = get_cluster_initialized_without_leader(leader=True, failover=failover).leader
-    return get_cluster(True, leader, [leader], failover, None, cluster_config)
+    return get_cluster(True, leader, [leader.member], failover, None, cluster_config)
 
 
 def get_standby_cluster_initialized_with_only_leader(failover=None, sync=None):
@@ -152,13 +152,13 @@ def run_async(self, func, args=()):
 @patch.object(Postgresql, 'is_running', Mock(return_value=MockPostmaster()))
 @patch.object(Postgresql, 'is_leader', Mock(return_value=True))
 @patch.object(Postgresql, 'timeline_wal_position', Mock(return_value=(1, 10, 1)))
-@patch.object(Postgresql, '_cluster_info_state_get', Mock(return_value=3))
+@patch.object(Postgresql, '_cluster_info_state_get', Mock(return_value=10))
 @patch.object(Postgresql, 'data_directory_empty', Mock(return_value=False))
 @patch.object(Postgresql, 'controldata', Mock(return_value={
     'Database system identifier': SYSID,
     'Database cluster state': 'shut down',
     'Latest checkpoint location': '0/12345678'}))
-@patch.object(SlotsHandler, 'sync_replication_slots', Mock())
+@patch.object(SlotsHandler, 'load_replication_slots', Mock(side_effect=Exception))
 @patch.object(ConfigHandler, 'append_pg_hba', Mock())
 @patch.object(ConfigHandler, 'write_pgpass', Mock(return_value={}))
 @patch.object(ConfigHandler, 'write_recovery_conf', Mock())
@@ -384,13 +384,21 @@ class TestHa(PostgresInit):
         self.p.is_leader = false
         self.assertEqual(self.ha.run_cycle(), 'not promoting because failed to update leader lock in DCS')
 
+    @patch.object(Postgresql, 'major_version', PropertyMock(return_value=130000))
     def test_follow(self):
         self.ha.cluster.is_unlocked = false
         self.p.is_leader = false
         self.assertEqual(self.ha.run_cycle(), 'no action.  i am a secondary and i am following a leader')
         self.ha.patroni.replicatefrom = "foo"
         self.p.config.check_recovery_conf = Mock(return_value=(True, False))
+        self.ha.cluster.config.data.update({'slots': {'l': {'database': 'a', 'plugin': 'b'}}})
+        self.ha.cluster.members[1].data['tags']['replicatefrom'] = 'postgresql0'
+        self.ha.patroni.nofailover = True
         self.assertEqual(self.ha.run_cycle(), 'no action.  i am a secondary and i am following a leader')
+        del self.ha.cluster.config.data['slots']
+        self.ha.cluster.config.data.update({'postgresql': {'use_slots': False}})
+        self.assertEqual(self.ha.run_cycle(), 'no action.  i am a secondary and i am following a leader')
+        del self.ha.cluster.config.data['postgresql']['use_slots']
 
     def test_follow_in_pause(self):
         self.ha.cluster.is_unlocked = false
@@ -634,7 +642,7 @@ class TestHa(PostgresInit):
         # in synchronous_mode consider itself healthy if the former leader is accessible in read-only and ahead of us
         with patch.object(Ha, 'is_synchronous_mode', Mock(return_value=True)):
             self.assertTrue(self.ha._is_healthiest_node(self.ha.old_cluster.members))
-        with patch('patroni.postgresql.Postgresql.timeline_wal_position', return_value=(1, 1, 1)):
+        with patch('patroni.postgresql.Postgresql.last_operation', return_value=1):
             self.assertFalse(self.ha._is_healthiest_node(self.ha.old_cluster.members))
         with patch('patroni.postgresql.Postgresql.replica_cached_timeline', return_value=1):
             self.assertFalse(self.ha._is_healthiest_node(self.ha.old_cluster.members))
@@ -1110,6 +1118,7 @@ class TestHa(PostgresInit):
     @patch('psycopg2.connect', psycopg2_connect)
     def test_permanent_logical_slots_after_promote(self):
         config = ClusterConfig(1, {'slots': {'l': {'database': 'postgres', 'plugin': 'test_decoding'}}}, 1)
+        self.p.name = 'other'
         self.ha.cluster = get_cluster_initialized_without_leader(cluster_config=config)
         self.assertEqual(self.ha.run_cycle(), 'acquired session lock as a leader')
         self.ha.cluster = get_cluster_initialized_without_leader(leader=True, cluster_config=config)
@@ -1137,3 +1146,19 @@ class TestHa(PostgresInit):
 
         self.ha.has_lock = true
         self.assertEqual(self.ha.run_cycle(), 'PAUSE: released leader key voluntarily due to the system ID mismatch')
+
+    @patch('psycopg2.connect', psycopg2_connect)
+    @patch('os.path.exists', Mock(return_value=True))
+    @patch('shutil.rmtree', Mock())
+    @patch('os.makedirs', Mock())
+    @patch('os.open', Mock())
+    @patch('os.fsync', Mock())
+    @patch('os.close', Mock())
+    @patch('os.rename', Mock())
+    @patch('patroni.postgresql.Postgresql.is_starting', Mock(return_value=False))
+    @patch.object(builtins, 'open', mock_open())
+    @patch.object(SlotsHandler, 'sync_replication_slots', Mock(return_value=['foo']))
+    def test_follow_copy(self):
+        self.ha.cluster.is_unlocked = false
+        self.p.is_leader = false
+        self.assertTrue(self.ha.run_cycle().startswith('Copying logical slots'))

tests/test_kubernetes.py

@@ -16,7 +16,8 @@ def mock_list_namespaced_config_map(*args, **kwargs):
     metadata = {'resource_version': '1', 'labels': {'f': 'b'}, 'name': 'test-config',
                 'annotations': {'initialize': '123', 'config': '{}'}}
     items = [k8s_client.V1ConfigMap(metadata=k8s_client.V1ObjectMeta(**metadata))]
-    metadata.update({'name': 'test-leader', 'annotations': {'optime': '1234', 'leader': 'p-0', 'ttl': '30s'}})
+    metadata.update({'name': 'test-leader',
+                     'annotations': {'optime': '1234x', 'leader': 'p-0', 'ttl': '30s', 'slots': '{'}})
     items.append(k8s_client.V1ConfigMap(metadata=k8s_client.V1ObjectMeta(**metadata)))
     metadata.update({'name': 'test-failover', 'annotations': {'leader': 'p-0'}})
     items.append(k8s_client.V1ConfigMap(metadata=k8s_client.V1ObjectMeta(**metadata)))
@@ -260,10 +261,6 @@ class TestKubernetesEndpoints(BaseTestKubernetes):
         self.k._kinds._object_cache['test'].metadata.annotations['leader'] = 'p-1'
         self.assertFalse(self.k.update_leader('123'))
 
-    @patch.object(k8s_client.CoreV1Api, 'patch_namespaced_endpoints', mock_namespaced_kind, create=True)
-    def test_update_leader_with_restricted_access(self):
-        self.assertIsNotNone(self.k.update_leader('123', True))
-
     @patch.object(k8s_client.CoreV1Api, 'read_namespaced_endpoints', create=True)
     @patch.object(k8s_client.CoreV1Api, 'patch_namespaced_endpoints', create=True)
     def test__update_leader_with_retry(self, mock_patch, mock_read):

tests/test_postgresql.py

@@ -1,5 +1,4 @@
 import datetime
-import mock  # for the mock.call method, importing it without a namespace breaks python3
 import os
 import psutil
 import psycopg2
@@ -9,11 +8,10 @@ import time
 
 from mock import Mock, MagicMock, PropertyMock, patch, mock_open
 from patroni.async_executor import CriticalTask
-from patroni.dcs import Cluster, ClusterConfig, Member, RemoteMember, SyncState
+from patroni.dcs import Cluster, RemoteMember, SyncState
 from patroni.exceptions import PostgresConnectionException, PatroniException
 from patroni.postgresql import Postgresql, STATE_REJECT, STATE_NO_RESPONSE
 from patroni.postgresql.postmaster import PostmasterProcess
-from patroni.postgresql.slots import SlotsHandler
 from patroni.utils import RetryFailedError
 from six.moves import builtins
 from threading import Thread, current_thread
@@ -303,29 +301,6 @@ class TestPostgresql(BaseTestPostgresql):
         m = RemoteMember('1', {'restore_command': '2', 'primary_slot_name': 'foo', 'conn_kwargs': {'host': 'bar'}})
         self.p.follow(m)
 
-    @patch.object(Postgresql, 'is_running', Mock(return_value=True))
-    def test_sync_replication_slots(self):
-        self.p.start()
-        config = ClusterConfig(1, {'slots': {'test_3': {'database': 'a', 'plugin': 'b'},
-                                             'A': 0, 'ls': 0, 'b': {'type': 'logical', 'plugin': '1'}},
-                                   'ignore_slots': [{'name': 'blabla'}]}, 1)
-        cluster = Cluster(True, config, self.leader, 0, [self.me, self.other, self.leadermem], None, None, None)
-        with mock.patch('patroni.postgresql.Postgresql._query', Mock(side_effect=psycopg2.OperationalError)):
-            self.p.slots_handler.sync_replication_slots(cluster)
-        self.p.slots_handler.sync_replication_slots(cluster)
-        with mock.patch('patroni.postgresql.Postgresql.role', new_callable=PropertyMock(return_value='replica')):
-            self.p.slots_handler.sync_replication_slots(cluster)
-        with patch.object(SlotsHandler, 'drop_replication_slot', Mock(return_value=True)),\
-                patch('patroni.dcs.logger.error', new_callable=Mock()) as errorlog_mock:
-            alias1 = Member(0, 'test-3', 28, {'conn_url': 'postgres://replicator:[email protected]:5436/postgres'})
-            alias2 = Member(0, 'test.3', 28, {'conn_url': 'postgres://replicator:[email protected]:5436/postgres'})
-            cluster.members.extend([alias1, alias2])
-            self.p.slots_handler.sync_replication_slots(cluster)
-            self.assertEqual(errorlog_mock.call_count, 5)
-            ca = errorlog_mock.call_args_list[0][0][1]
-            self.assertTrue("test-3" in ca, "non matching {0}".format(ca))
-            self.assertTrue("test.3" in ca, "non matching {0}".format(ca))
-
     @patch.object(MockCursor, 'execute', Mock(side_effect=psycopg2.OperationalError))
     def test__query(self):
         self.assertRaises(PostgresConnectionException, self.p._query, 'blabla')
@@ -340,7 +315,7 @@ class TestPostgresql(BaseTestPostgresql):
     @patch.object(Postgresql, 'pg_isready', Mock(return_value=STATE_REJECT))
     def test_is_leader(self):
         self.assertTrue(self.p.is_leader())
-        self.p.reset_cluster_info_state()
+        self.p.reset_cluster_info_state(None)
         with patch.object(Postgresql, '_query', Mock(side_effect=RetryFailedError(''))):
             self.assertRaises(PostgresConnectionException, self.p.is_leader)
 
@@ -611,7 +586,7 @@ class TestPostgresql(BaseTestPostgresql):
 
     def test_pick_sync_standby(self):
         cluster = Cluster(True, None, self.leader, 0, [self.me, self.other, self.leadermem], None,
-                          SyncState(0, self.me.name, self.leadermem.name), None)
+                          SyncState(0, self.me.name, self.leadermem.name), None, None)
         mock_cursor = Mock()
         mock_cursor.fetchone.return_value = ('remote_apply',)
 
@@ -727,10 +702,14 @@ class TestPostgresql(BaseTestPostgresql):
     @patch.object(Postgresql, '_query', Mock(side_effect=RetryFailedError('')))
     def test_received_timeline(self):
         self.p.set_role('standby_leader')
-        self.p.reset_cluster_info_state()
+        self.p.reset_cluster_info_state(None)
         self.assertRaises(PostgresConnectionException, self.p.received_timeline)
 
     def test__write_recovery_params(self):
         self.p.config._write_recovery_params(Mock(), {'pause_at_recovery_target': 'false'})
         with patch.object(Postgresql, 'major_version', PropertyMock(return_value=90400)):
             self.p.config._write_recovery_params(Mock(), {'recovery_target_action': 'PROMOTE'})
+
+    @patch.object(Postgresql, 'is_running', Mock(return_value=True))
+    def test_set_enforce_hot_standby_feedback(self):
+        self.p.set_enforce_hot_standby_feedback(True)

tests/test_raft.py

@@ -129,14 +129,19 @@ class TestRaft(unittest.TestCase):
                      'retry_timeout': 10, 'data_dir': self._TMP})
         raft.set_retry_timeout(20)
         raft.set_ttl(60)
+        self.assertTrue(raft._sync_obj.set(raft.members_path + 'legacy', '{"version":"2.0.0"}'))
         self.assertTrue(raft.touch_member(''))
         self.assertTrue(raft.initialize())
         self.assertTrue(raft.cancel_initialization())
         self.assertTrue(raft.set_config_value('{}'))
         self.assertTrue(raft.write_sync_state('foo', 'bar'))
-        self.assertTrue(raft.update_leader('1'))
         self.assertTrue(raft.manual_failover('foo', 'bar'))
         raft.get_cluster()
+        self.assertTrue(raft._sync_obj.set(raft.status_path, '{"optime":1234567,"slots":{"ls":12345}}'))
+        raft.get_cluster()
+        self.assertTrue(raft.update_leader('1'))
+        self.assertTrue(raft._sync_obj.set(raft.status_path, '{'))
+        raft.get_cluster()
         self.assertTrue(raft.delete_sync_state())
         self.assertTrue(raft.delete_leader())
         self.assertTrue(raft.set_history_value(''))

tests/test_rewind.py

@@ -93,7 +93,7 @@ class TestRewind(BaseTestPostgresql):
             self.r.rewind_or_reinitialize_needed_and_possible(self.leader)
 
         with patch.object(Postgresql, 'is_running', Mock(return_value=True)):
-            with patch.object(MockCursor, 'fetchone', Mock(side_effect=[(0, 0, 1, 1,), Exception])):
+            with patch.object(MockCursor, 'fetchone', Mock(side_effect=[(0, 0, 1, 1, 0, 0, 0, 0, 0, None), Exception])):
                 self.r.rewind_or_reinitialize_needed_and_possible(self.leader)
 
     @patch.object(CancellableSubprocess, 'call', mock_cancellable_call)

tests/test_slots.py

@@ -0,0 +1,124 @@
+import mock
+import os
+import psycopg2
+import unittest
+
+
+from mock import Mock, PropertyMock, patch
+
+from patroni.dcs import Cluster, ClusterConfig, Member
+from patroni.postgresql import Postgresql
+from patroni.postgresql.slots import SlotsHandler, fsync_dir
+
+from . import BaseTestPostgresql, psycopg2_connect, MockCursor
+
+
+@patch('subprocess.call', Mock(return_value=0))
+@patch('psycopg2.connect', psycopg2_connect)
+@patch.object(Postgresql, 'is_running', Mock(return_value=True))
+class TestSlotsHandler(BaseTestPostgresql):
+
+    @patch('subprocess.call', Mock(return_value=0))
+    @patch('os.rename', Mock())
+    @patch('patroni.postgresql.CallbackExecutor', Mock())
+    @patch.object(Postgresql, 'get_major_version', Mock(return_value=130000))
+    @patch.object(Postgresql, 'is_running', Mock(return_value=True))
+    def setUp(self):
+        super(TestSlotsHandler, self).setUp()
+        self.s = self.p.slots_handler
+        self.p.start()
+
+    def test_sync_replication_slots(self):
+        config = ClusterConfig(1, {'slots': {'test_3': {'database': 'a', 'plugin': 'b'},
+                                             'A': 0, 'ls': 0, 'b': {'type': 'logical', 'plugin': '1'}},
+                                   'ignore_slots': [{'name': 'blabla'}]}, 1)
+        cluster = Cluster(True, config, self.leader, 0,
+                          [self.me, self.other, self.leadermem], None, None, None, {'test_3': 10})
+        with mock.patch('patroni.postgresql.Postgresql._query', Mock(side_effect=psycopg2.OperationalError)):
+            self.s.sync_replication_slots(cluster, False)
+        self.p.set_role('standby_leader')
+        self.s.sync_replication_slots(cluster, False)
+        self.p.set_role('replica')
+        with patch.object(Postgresql, 'is_leader', Mock(return_value=False)):
+            self.s.sync_replication_slots(cluster, False)
+        self.p.set_role('master')
+        with mock.patch('patroni.postgresql.Postgresql.role', new_callable=PropertyMock(return_value='replica')):
+            self.s.sync_replication_slots(cluster, False)
+        with patch.object(SlotsHandler, 'drop_replication_slot', Mock(return_value=True)),\
+                patch('patroni.dcs.logger.error', new_callable=Mock()) as errorlog_mock:
+            alias1 = Member(0, 'test-3', 28, {'conn_url': 'postgres://replicator:[email protected]:5436/postgres'})
+            alias2 = Member(0, 'test.3', 28, {'conn_url': 'postgres://replicator:[email protected]:5436/postgres'})
+            cluster.members.extend([alias1, alias2])
+            self.s.sync_replication_slots(cluster, False)
+            self.assertEqual(errorlog_mock.call_count, 5)
+            ca = errorlog_mock.call_args_list[0][0][1]
+            self.assertTrue("test-3" in ca, "non matching {0}".format(ca))
+            self.assertTrue("test.3" in ca, "non matching {0}".format(ca))
+            with patch.object(Postgresql, 'major_version', PropertyMock(return_value=90618)):
+                self.s.sync_replication_slots(cluster, False)
+
+    def test_process_permanent_slots(self):
+        config = ClusterConfig(1, {'slots': {'ls': {'database': 'a', 'plugin': 'b'}},
+                                   'ignore_slots': [{'name': 'blabla'}]}, 1)
+        cluster = Cluster(True, config, self.leader, 0, [self.me, self.other, self.leadermem], None, None, None, None)
+
+        self.s.sync_replication_slots(cluster, False)
+        with patch.object(Postgresql, '_query') as mock_query:
+            self.p.reset_cluster_info_state(None)
+            mock_query.return_value.fetchone.return_value = (
+                1, 0, 0, 0, 0, 0, 0, 0, 0,
+                [{"slot_name": "ls", "type": "logical", "datoid": 5, "plugin": "b",
+                  "confirmed_flush_lsn": 12345, "catalog_xmin": 105}])
+            self.assertEqual(self.p.slots(), {'ls': 12345})
+
+            self.p.reset_cluster_info_state(None)
+            mock_query.return_value.fetchone.return_value = (
+                1, 0, 0, 0, 0, 0, 0, 0, 0,
+                [{"slot_name": "ls", "type": "logical", "datoid": 6, "plugin": "b",
+                  "confirmed_flush_lsn": 12345, "catalog_xmin": 105}])
+            self.assertEqual(self.p.slots(), {})
+
+    @patch.object(Postgresql, 'is_leader', Mock(return_value=False))
+    def test__ensure_logical_slots_replica(self):
+        self.p.set_role('replica')
+        config = ClusterConfig(1, {'slots': {'ls': {'database': 'a', 'plugin': 'b'}}}, 1)
+        cluster = Cluster(True, config, self.leader, 0,
+                          [self.me, self.other, self.leadermem], None, None, None, {'ls': 12346})
+        self.assertEqual(self.s.sync_replication_slots(cluster, False), [])
+        self.s._schedule_load_slots = False
+        with patch.object(MockCursor, 'execute', Mock(side_effect=psycopg2.OperationalError)):
+            self.assertEqual(self.s.sync_replication_slots(cluster, False), [])
+        cluster.slots['ls'] = 'a'
+        self.assertEqual(self.s.sync_replication_slots(cluster, False), [])
+        with patch.object(MockCursor, 'rowcount', PropertyMock(return_value=1), create=True):
+            self.assertEqual(self.s.sync_replication_slots(cluster, False), ['ls'])
+
+    @patch.object(MockCursor, 'execute', Mock(side_effect=psycopg2.OperationalError))
+    def test_copy_logical_slots(self):
+        self.s.copy_logical_slots(self.leader, ['foo'])
+
+    @patch.object(Postgresql, 'stop', Mock(return_value=True))
+    @patch.object(Postgresql, 'start', Mock(return_value=True))
+    @patch.object(Postgresql, 'is_leader', Mock(return_value=False))
+    def test_check_logical_slots_readiness(self):
+        self.s.copy_logical_slots(self.leader, ['ls'])
+        config = ClusterConfig(1, {'slots': {'ls': {'database': 'a', 'plugin': 'b'}}}, 1)
+        cluster = Cluster(True, config, self.leader, 0,
+                          [self.me, self.other, self.leadermem], None, None, None, {'ls': 12345})
+        self.assertEqual(self.s.sync_replication_slots(cluster, False), [])
+        with patch.object(MockCursor, 'rowcount', PropertyMock(return_value=1), create=True):
+            self.s.check_logical_slots_readiness(cluster, False, None)
+
+    @patch.object(Postgresql, 'stop', Mock(return_value=True))
+    @patch.object(Postgresql, 'start', Mock(return_value=True))
+    @patch.object(Postgresql, 'is_leader', Mock(return_value=False))
+    def test_on_promote(self):
+        self.s.copy_logical_slots(self.leader, ['ls'])
+        self.s.on_promote()
+
+    @unittest.skipIf(os.name == 'nt', "Windows not supported")
+    @patch('os.open', Mock())
+    @patch('os.close', Mock())
+    @patch('os.fsync', Mock(side_effect=OSError))
+    def test_fsync_dir(self):
+        self.assertRaises(OSError, fsync_dir, 'foo')

tests/test_zookeeper.py

@@ -30,7 +30,9 @@ class MockKazooClient(Mock):
     def get(self, path, watch=None):
         if not isinstance(path, six.string_types):
             raise TypeError("Invalid type for 'path' (string expected)")
-        if path == '/no_node':
+        if path == '/broken/status':
+            return (b'{', ZnodeStat(0, 0, 0, 0, 0, 0, 0, -1, 0, 0, 0))
+        elif path in ('/no_node', '/legacy/status'):
             raise NoNodeError
         elif '/members/' in path:
             return (
@@ -45,6 +47,8 @@ class MockKazooClient(Mock):
             return (b'foo', ZnodeStat(0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0))
         elif path.endswith('/initialize'):
             return (b'foo', ZnodeStat(0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0))
+        elif path.endswith('/status'):
+            return (b'{"optime":500,"slots":{"ls":1234567}}', ZnodeStat(0, 0, 0, 0, 0, 0, 0, -1, 0, 0, 0))
         return (b'', ZnodeStat(0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0))
 
     @staticmethod
@@ -147,6 +151,10 @@ class TestZooKeeper(unittest.TestCase):
     def test__inner_load_cluster(self):
         self.zk._base_path = self.zk._base_path.replace('test', 'bla')
         self.zk._inner_load_cluster()
+        self.zk._base_path = self.zk._base_path = '/broken'
+        self.zk._inner_load_cluster()
+        self.zk._base_path = self.zk._base_path = '/legacy'
+        self.zk._inner_load_cluster()
         self.zk._base_path = self.zk._base_path = '/no_node'
         self.zk._inner_load_cluster()
 
@@ -156,11 +164,11 @@ class TestZooKeeper(unittest.TestCase):
         self.assertIsInstance(cluster.leader, Leader)
         self.zk.touch_member({'foo': 'foo'})
         self.zk._name = 'bar'
-        self.zk.optime_watcher(None)
+        self.zk.status_watcher(None)
         with patch.object(ZooKeeper, 'get_node', Mock(side_effect=Exception)):
             self.zk.get_cluster()
         cluster = self.zk.get_cluster()
-        self.assertEqual(cluster.last_leader_operation, 500)
+        self.assertEqual(cluster.last_lsn, 500)
 
     def test_delete_leader(self):
         self.assertTrue(self.zk.delete_leader())
@@ -203,10 +211,10 @@ class TestZooKeeper(unittest.TestCase):
             self.zk.take_leader()
 
     def test_update_leader(self):
-        self.assertTrue(self.zk.update_leader(None))
+        self.assertTrue(self.zk.update_leader(12345))
 
     def test_write_leader_optime(self):
-        self.zk.last_leader_operation = '0'
+        self.zk.last_lsn = '0'
         self.zk.write_leader_optime('1')
         with patch.object(MockKazooClient, 'create_async', Mock()):
             self.zk.write_leader_optime('1')
@@ -221,7 +229,7 @@ class TestZooKeeper(unittest.TestCase):
     def test_watch(self):
         self.zk.watch(None, 0)
         self.zk.event.isSet = Mock(return_value=True)
-        self.zk._fetch_optime = False
+        self.zk._fetch_status = False
         self.zk.watch(None, 0)
 
     def test__kazoo_connect(self):