mirror of
https://github.com/outbackdingo/patroni.git
synced 2026-08-31 16:49:46 +00:00
Alexander KukushkinandGitHub
c7173aadd7
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
265 lines
12 KiB
Python
265 lines
12 KiB
Python
import etcd
|
|
import json
|
|
import unittest
|
|
import urllib3
|
|
|
|
from mock import Mock, patch
|
|
from patroni.dcs.etcd3 import PatroniEtcd3Client, Cluster, Etcd3, Etcd3Error, Etcd3ClientError, RetryFailedError,\
|
|
InvalidAuthToken, Unavailable, Unknown, UnsupportedEtcdVersion, UserEmpty, AuthFailed, base64_encode
|
|
from threading import Thread
|
|
|
|
from . import SleepException, MockResponse
|
|
|
|
|
|
def mock_urlopen(self, method, url, **kwargs):
|
|
ret = MockResponse()
|
|
if method == 'GET' and url.endswith('/version'):
|
|
ret.content = '{"etcdserver": "3.3.13", "etcdcluster": "3.3.0"}'
|
|
elif method != 'POST':
|
|
raise Exception('Unexpected request method: {0} {1} {2}'.format(method, url, kwargs))
|
|
elif url.endswith('/cluster/member/list'):
|
|
ret.content = '{"members":[{"clientURLs":["http://localhost:2379", "http://localhost:4001"]}]}'
|
|
elif url.endswith('/auth/authenticate'):
|
|
ret.content = '{"token":"authtoken"}'
|
|
elif url.endswith('/lease/grant'):
|
|
ret.content = '{"ID": "123"}'
|
|
elif url.endswith('/lease/keepalive'):
|
|
ret.content = '{"result":{"TTL":30}}'
|
|
elif url.endswith('/kv/range'):
|
|
ret.content = json.dumps({
|
|
"header": {"revision": "1"},
|
|
"kvs": [
|
|
{"key": base64_encode('/patroni/test/leader'),
|
|
"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'}
|
|
]
|
|
})
|
|
elif url.endswith('/watch'):
|
|
key = base64_encode('/patroni/test/config')
|
|
ret.read_chunked = Mock(return_value=[json.dumps({
|
|
'result': {'events': [
|
|
{'kv': {'key': key, 'value': base64_encode('bar'), 'mod_revision': '2'}},
|
|
{'kv': {'key': key, 'value': base64_encode('buzz'), 'mod_revision': '3'}},
|
|
{'type': 'DELETE', 'kv': {'key': key, 'mod_revision': '4'}},
|
|
{'kv': {'key': base64_encode('/patroni/test/optime/leader'),
|
|
'value': base64_encode('1234567'), 'mod_revision': '5'}},
|
|
]}
|
|
})[:-1].encode('utf-8'), b'}{"error":{"grpc_code":14,"message":"","http_code":503}}'])
|
|
elif url.endswith('/kv/put') or url.endswith('/kv/txn'):
|
|
ret.status_code = 400
|
|
ret.content = '{"code":5,"error":"etcdserver: requested lease not found"}'
|
|
elif not url.endswith('/kv/deleterange'):
|
|
raise Exception('Unexpected url: {0} {1} {2}'.format(method, url, kwargs))
|
|
return ret
|
|
|
|
|
|
class BaseTestEtcd3(unittest.TestCase):
|
|
|
|
@patch.object(Thread, 'start', Mock())
|
|
@patch.object(urllib3.PoolManager, 'urlopen', mock_urlopen)
|
|
def setUp(self):
|
|
self.etcd3 = Etcd3({'namespace': '/patroni/', 'ttl': 30, 'retry_timeout': 10,
|
|
'host': 'localhost:2378', 'scope': 'test', 'name': 'foo',
|
|
'username': 'etcduser', 'password': 'etcdpassword'})
|
|
self.client = self.etcd3._client
|
|
self.kv_cache = self.client._kv_cache
|
|
|
|
|
|
class TestKVCache(BaseTestEtcd3):
|
|
|
|
def test__do_watch(self):
|
|
self.client.watchprefix = Mock(return_value=False)
|
|
self.assertRaises(AttributeError, self.kv_cache._do_watch, '1')
|
|
|
|
@patch('time.sleep', Mock(side_effect=SleepException))
|
|
@patch('patroni.dcs.etcd3.KVCache._build_cache', Mock(side_effect=Exception))
|
|
def test_run(self):
|
|
self.assertRaises(SleepException, self.kv_cache.run)
|
|
|
|
@patch.object(urllib3.PoolManager, 'urlopen', mock_urlopen)
|
|
def test_kill_stream(self):
|
|
self.assertRaises(Unavailable, self.kv_cache._do_watch, '1')
|
|
self.kv_cache.kill_stream()
|
|
with patch.object(MockResponse, 'connection', create=True) as mock_conn:
|
|
self.kv_cache.kill_stream()
|
|
mock_conn.sock.close.side_effect = Exception
|
|
self.kv_cache.kill_stream()
|
|
|
|
|
|
class TestPatroniEtcd3Client(BaseTestEtcd3):
|
|
|
|
@patch('patroni.dcs.etcd3.Etcd3Client.authenticate', Mock(side_effect=AuthFailed))
|
|
def test__init__(self):
|
|
self.assertRaises(SystemExit, self.setUp)
|
|
|
|
@patch.object(urllib3.PoolManager, 'urlopen')
|
|
def test_call_rpc(self, mock_urlopen):
|
|
request = {'key': base64_encode('/patroni/test/leader')}
|
|
mock_urlopen.return_value = MockResponse()
|
|
mock_urlopen.return_value.content = '{"succeeded":true,"header":{"revision":"1"}}'
|
|
self.client.call_rpc('/kv/txn', {'success': [{'request_delete_range': request}]})
|
|
self.client.call_rpc('/kv/put', request)
|
|
self.client.call_rpc('/kv/deleterange', request)
|
|
|
|
@patch('time.time', Mock(side_effect=[1, 10.9, 100]))
|
|
def test__wait_cache(self):
|
|
with self.kv_cache.condition:
|
|
self.assertRaises(RetryFailedError, self.client._wait_cache, 10)
|
|
|
|
@patch.object(urllib3.PoolManager, 'urlopen')
|
|
def test__restart_watcher(self, mock_urlopen):
|
|
mock_urlopen.return_value = MockResponse()
|
|
mock_urlopen.return_value.status_code = 400
|
|
mock_urlopen.return_value.content = '{"code":9,"error":"etcdserver: authentication is not enabled"}'
|
|
self.client.authenticate()
|
|
|
|
@patch.object(urllib3.PoolManager, 'urlopen')
|
|
def test__handle_auth_errors(self, mock_urlopen):
|
|
mock_urlopen.return_value = MockResponse()
|
|
mock_urlopen.return_value.content = '{"code":3,"error":"etcdserver: user name is empty"}'
|
|
mock_urlopen.return_value.status_code = 403
|
|
self.client._cluster_version = (3, 1, 5)
|
|
self.assertRaises(UnsupportedEtcdVersion, self.client.deleteprefix, 'foo')
|
|
self.client._cluster_version = (3, 3, 13)
|
|
self.assertRaises(UserEmpty, self.client.deleteprefix, 'foo')
|
|
mock_urlopen.return_value.content = '{"code":16,"error":"etcdserver: invalid auth token"}'
|
|
self.assertRaises(InvalidAuthToken, self.client.deleteprefix, 'foo')
|
|
with patch.object(PatroniEtcd3Client, 'authenticate', Mock(return_value=True)):
|
|
self.assertRaises(InvalidAuthToken, self.client.deleteprefix, 'foo')
|
|
self.client.username = None
|
|
self.assertRaises(InvalidAuthToken, self.client.deleteprefix, 'foo')
|
|
|
|
def test__handle_server_response(self):
|
|
response = MockResponse()
|
|
response.content = '{"code":0,"error":"'
|
|
self.assertRaises(etcd.EtcdException, self.client._handle_server_response, response)
|
|
response.status_code = 400
|
|
self.assertRaises(Unknown, self.client._handle_server_response, response)
|
|
response.content = '{"error":{"grpc_code":0,"message":"","http_code":400}}'
|
|
try:
|
|
self.client._handle_server_response(response)
|
|
except Unknown as e:
|
|
self.assertEqual(e.as_dict(), {'code': 2, 'codeText': 'OK', 'error': u'', 'status': 400})
|
|
|
|
@patch.object(urllib3.PoolManager, 'urlopen')
|
|
def test__ensure_version_prefix(self, mock_urlopen):
|
|
self.client.version_prefix = None
|
|
mock_urlopen.return_value = MockResponse()
|
|
mock_urlopen.return_value.content = '{"etcdserver": "3.0.3", "etcdcluster": "3.0.0"}'
|
|
self.assertRaises(UnsupportedEtcdVersion, self.client._ensure_version_prefix, '')
|
|
mock_urlopen.return_value.content = '{"etcdserver": "3.0.4", "etcdcluster": "3.0.0"}'
|
|
self.client._ensure_version_prefix('')
|
|
self.assertEqual(self.client.version_prefix, '/v3alpha')
|
|
mock_urlopen.return_value.content = '{"etcdserver": "3.4.4", "etcdcluster": "3.4.0"}'
|
|
self.client._ensure_version_prefix('')
|
|
self.assertEqual(self.client.version_prefix, '/v3')
|
|
|
|
|
|
@patch.object(urllib3.PoolManager, 'urlopen', mock_urlopen)
|
|
class TestEtcd3(BaseTestEtcd3):
|
|
|
|
@patch.object(Thread, 'start', Mock())
|
|
@patch.object(urllib3.PoolManager, 'urlopen', mock_urlopen)
|
|
def setUp(self):
|
|
super(TestEtcd3, self).setUp()
|
|
self.assertRaises(AttributeError, self.kv_cache._build_cache)
|
|
self.kv_cache._is_ready = True
|
|
self.etcd3.get_cluster()
|
|
|
|
def test_get_cluster(self):
|
|
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()
|
|
self.assertRaises(Etcd3Error, self.etcd3.get_cluster)
|
|
|
|
def test_touch_member(self):
|
|
self.etcd3.touch_member({})
|
|
self.etcd3._lease = 'bla'
|
|
self.etcd3.touch_member({})
|
|
with patch.object(PatroniEtcd3Client, 'lease_grant', Mock(side_effect=Etcd3ClientError)):
|
|
self.etcd3.touch_member({})
|
|
|
|
def test__update_leader(self):
|
|
self.etcd3._lease = None
|
|
self.etcd3.update_leader('123')
|
|
self.etcd3.update_leader('124')
|
|
|
|
def test_attempt_to_acquire_leader(self):
|
|
self.etcd3._lease = None
|
|
self.assertFalse(self.etcd3.attempt_to_acquire_leader())
|
|
|
|
def test_set_ttl(self):
|
|
self.etcd3.set_ttl(20)
|
|
|
|
@patch.object(PatroniEtcd3Client, 'lease_keepalive', Mock(return_value=False))
|
|
def test_refresh_lease(self):
|
|
self.etcd3._last_lease_refresh = 0
|
|
self.etcd3.refresh_lease()
|
|
|
|
@patch('time.sleep', Mock(side_effect=SleepException))
|
|
@patch.object(PatroniEtcd3Client, 'lease_keepalive', Mock(return_value=False))
|
|
@patch.object(PatroniEtcd3Client, 'lease_grant', Mock(side_effect=Etcd3ClientError))
|
|
def test_create_lease(self):
|
|
self.etcd3._lease = None
|
|
self.etcd3._last_lease_refresh = 0
|
|
self.assertRaises(SleepException, self.etcd3.create_lease)
|
|
|
|
def test_set_failover_value(self):
|
|
self.etcd3.set_failover_value('', 1)
|
|
|
|
def test_set_config_value(self):
|
|
self.etcd3.set_config_value('')
|
|
|
|
def test_initialize(self):
|
|
self.etcd3.initialize()
|
|
|
|
def test_cancel_initialization(self):
|
|
self.etcd3.cancel_initialization()
|
|
|
|
def test_delete_leader(self):
|
|
self.etcd3.delete_leader()
|
|
|
|
def test_delete_cluster(self):
|
|
self.etcd3.delete_cluster()
|
|
|
|
def test_set_history_value(self):
|
|
self.etcd3.set_history_value('')
|
|
|
|
def test_set_sync_state_value(self):
|
|
self.etcd3.set_sync_state_value('')
|
|
|
|
def test_delete_sync_state(self):
|
|
self.etcd3.delete_sync_state()
|
|
|
|
def test_watch(self):
|
|
self.etcd3.set_ttl(10)
|
|
self.etcd3.watch(None, 0)
|
|
self.etcd3.watch(None, 0)
|
|
|
|
def test_set_socket_options(self):
|
|
with patch('socket.SIO_KEEPALIVE_VALS', 1, create=True):
|
|
self.etcd3.set_socket_options(Mock(), None)
|