mirror of
https://github.com/outbackdingo/patroni.git
synced 2026-08-25 14:53:37 +00:00
Multi Sync Standby Support (#1594)
The new parameter `synchronous_node_count` is used by Patroni to manage number of synchronous standby databases. It is set to 1 by default. It has no effect when synchronous_mode is set to off. When enabled, Patroni manages precise number of synchronous standby databases based on parameter synchronous_node_count and adjusts the state in DCS & synchronous_standby_names as members join and leave. This functionality can be further extended to support Priority (FIRST n) based synchronous replication & Quorum (ANY n) based synchronous replication in future.
This commit is contained in:
ksarabu1
@@ -59,11 +59,14 @@ Synchronous mode can be switched on and off via Patroni REST interface. See :ref
|
||||
|
||||
Note: Because of the way synchronous replication is implemented in PostgreSQL it is still possible to lose transactions even when using ``synchronous_mode_strict``. If the PostgreSQL backend is cancelled while waiting to acknowledge replication (as a result of packet cancellation due to client timeout or backend failure) transaction changes become visible for other backends. Such changes are not yet replicated and may be lost in case of standby promotion.
|
||||
|
||||
Synchronous Replication Factor
|
||||
------------------------------
|
||||
The parameter ``synchronous_node_count`` is used by Patroni to manage number of synchronous standby databases. It is set to 1 by default. It has no effect when ``synchronous_mode`` is set to off. When enabled, Patroni manages precise number of synchronous standby databases based on parameter ``synchronous_node_count`` and adjusts the state in DCS & synchronous_standby_names as members join and leave.
|
||||
|
||||
Synchronous mode implementation
|
||||
-------------------------------
|
||||
|
||||
When in synchronous mode Patroni maintains synchronization state in the DCS, containing the latest primary and current synchronous standby. This state is updated with strict ordering constraints to ensure the following invariants:
|
||||
When in synchronous mode Patroni maintains synchronization state in the DCS, containing the latest primary and current synchronous standby databases. This state is updated with strict ordering constraints to ensure the following invariants:
|
||||
|
||||
- A node must be marked as the latest leader whenever it can accept write transactions. Patroni crashing or PostgreSQL not shutting down can cause violations of this invariant.
|
||||
|
||||
@@ -71,9 +74,9 @@ When in synchronous mode Patroni maintains synchronization state in the DCS, con
|
||||
|
||||
- A node that is not the leader or current synchronous standby is not allowed to promote itself automatically.
|
||||
|
||||
Patroni will only ever assign one standby to ``synchronous_standby_names`` because with multiple candidates it is not possible to know which node was acting as synchronous during the failure.
|
||||
Patroni will only assign one or more synchronous standby nodes based on ``synchronous_node_count`` parameter to ``synchronous_standby_names``.
|
||||
|
||||
On each HA loop iteration Patroni re-evaluates synchronous standby choice. If the current synchronous standby is connected and has not requested its synchronous status to be removed it remains picked. Otherwise the cluster member available for sync that is furthest ahead in replication is picked.
|
||||
On each HA loop iteration Patroni re-evaluates synchronous standby nodes choice. If the current list of synchronous standby nodes are connected and has not requested its synchronous status to be removed it remains picked. Otherwise the cluster member available for sync that is furthest ahead in replication is picked.
|
||||
|
||||
|
||||
.. [1] The data is still there, but recovering it requires a manual recovery effort by data recovery specialists. When Patroni is allowed to rewind with ``use_pg_rewind`` the forked timeline will be automatically erased to rejoin the failed primary with the cluster.
|
||||
|
||||
@@ -28,6 +28,27 @@ Feature: basic replication
|
||||
When I issue a GET request to http://127.0.0.1:8009/async
|
||||
Then I receive a response code 200
|
||||
|
||||
Scenario: check multi sync replication
|
||||
Given I issue a PATCH request to http://127.0.0.1:8008/config with {"synchronous_node_count": 2}
|
||||
Then I receive a response code 200
|
||||
And I sleep for 10 seconds
|
||||
Then "sync" key in DCS has sync_standby=postgres1,postgres2 after 5 seconds
|
||||
When I issue a GET request to http://127.0.0.1:8010/sync
|
||||
Then I receive a response code 200
|
||||
When I issue a GET request to http://127.0.0.1:8009/sync
|
||||
Then I receive a response code 200
|
||||
When I issue a PATCH request to http://127.0.0.1:8008/config with {"synchronous_node_count": 1}
|
||||
Then I receive a response code 200
|
||||
And I shut down postgres1
|
||||
And I sleep for 10 seconds
|
||||
Then "sync" key in DCS has sync_standby=postgres2 after 10 seconds
|
||||
When I start postgres1
|
||||
And "members/postgres1" key in DCS has state=running after 10 seconds
|
||||
When I issue a GET request to http://127.0.0.1:8010/sync
|
||||
Then I receive a response code 200
|
||||
When I issue a GET request to http://127.0.0.1:8009/async
|
||||
Then I receive a response code 200
|
||||
|
||||
Scenario: check the basic failover in synchronous mode
|
||||
Given I run patronictl.py pause batman
|
||||
Then I receive a response returncode 0
|
||||
|
||||
@@ -21,7 +21,7 @@ def write_label(context, content, name):
|
||||
context.pctl.write_label(name, content)
|
||||
|
||||
|
||||
@step('"{name}" key in DCS has {key:w}={value:w} after {time_limit:d} seconds')
|
||||
@step('"{name}" key in DCS has {key:w}={value} after {time_limit:d} seconds')
|
||||
def check_member(context, name, key, value, time_limit):
|
||||
time_limit *= context.timeout_multiplier
|
||||
max_time = time.time() + int(time_limit)
|
||||
|
||||
+3
-3
@@ -131,7 +131,7 @@ class RestApiHandler(BaseHTTPRequestHandler):
|
||||
status_code = 200 if response.get('state') == 'running' else 503
|
||||
elif cluster: # dcs is available
|
||||
is_synchronous = cluster.is_synchronous_mode() and cluster.sync \
|
||||
and cluster.sync.sync_standby == patroni.postgresql.name
|
||||
and patroni.postgresql.name in cluster.sync.members
|
||||
if path in ('/sync', '/synchronous') and is_synchronous:
|
||||
status_code = replica_status_code
|
||||
elif path in ('/async', '/asynchronous') and not is_synchronous:
|
||||
@@ -367,13 +367,13 @@ class RestApiHandler(BaseHTTPRequestHandler):
|
||||
if leader and (not cluster.leader or cluster.leader.name != leader):
|
||||
return 'leader name does not match'
|
||||
if candidate:
|
||||
if action == 'switchover' and cluster.is_synchronous_mode() and cluster.sync.sync_standby != candidate:
|
||||
if action == 'switchover' and cluster.is_synchronous_mode() and candidate not in cluster.sync.members:
|
||||
return 'candidate name does not match with sync_standby'
|
||||
members = [m for m in cluster.members if m.name == candidate]
|
||||
if not members:
|
||||
return 'candidate does not exists'
|
||||
elif cluster.is_synchronous_mode():
|
||||
members = [m for m in cluster.members if m.name == cluster.sync.sync_standby]
|
||||
members = [m for m in cluster.members if m.name in cluster.sync.members]
|
||||
if not members:
|
||||
return action + ' is not possible: can not find sync_standby'
|
||||
else:
|
||||
|
||||
@@ -64,6 +64,7 @@ class Config(object):
|
||||
'master_stop_timeout': 0,
|
||||
'synchronous_mode': False,
|
||||
'synchronous_mode_strict': False,
|
||||
'synchronous_node_count': 1,
|
||||
'standby_cluster': {
|
||||
'create_replica_methods': '',
|
||||
'host': '',
|
||||
@@ -386,6 +387,7 @@ class Config(object):
|
||||
'retry_timeout',
|
||||
'synchronous_mode',
|
||||
'synchronous_mode_strict',
|
||||
'synchronous_node_count',
|
||||
)
|
||||
|
||||
pg_config.update({p: config[p] for p in updated_fields if p in config})
|
||||
|
||||
+14
-5
@@ -351,7 +351,7 @@ class SyncState(namedtuple('SyncState', 'index,leader,sync_standby')):
|
||||
|
||||
:param index: modification index of a synchronization key in a Configuration Store
|
||||
:param leader: reference to member that was leader
|
||||
:param sync_standby: standby that was last synchronized to leader
|
||||
:param sync_standby: synchronous standby list (comma delimited) which are last synchronized to leader
|
||||
"""
|
||||
|
||||
@staticmethod
|
||||
@@ -383,15 +383,22 @@ class SyncState(namedtuple('SyncState', 'index,leader,sync_standby')):
|
||||
data = {}
|
||||
return SyncState(index, data.get('leader'), data.get('sync_standby'))
|
||||
|
||||
@property
|
||||
def members(self):
|
||||
""" Returns sync_standby in list """
|
||||
return self.sync_standby and self.sync_standby.split(',') or []
|
||||
|
||||
def matches(self, name):
|
||||
"""
|
||||
Returns if a node name matches one of the nodes in the sync state
|
||||
|
||||
>>> s = SyncState(1, 'foo', 'bar')
|
||||
>>> s = SyncState(1, 'foo', 'bar,zoo')
|
||||
>>> s.matches('foo')
|
||||
True
|
||||
>>> s.matches('bar')
|
||||
True
|
||||
>>> s.matches('zoo')
|
||||
True
|
||||
>>> s.matches('baz')
|
||||
False
|
||||
>>> s.matches(None)
|
||||
@@ -399,7 +406,7 @@ class SyncState(namedtuple('SyncState', 'index,leader,sync_standby')):
|
||||
>>> SyncState(1, None, None).matches('foo')
|
||||
False
|
||||
"""
|
||||
return name is not None and name in (self.leader, self.sync_standby)
|
||||
return name is not None and name in [self.leader] + self.members
|
||||
|
||||
|
||||
class TimelineHistory(namedtuple('TimelineHistory', 'index,value,lines')):
|
||||
@@ -787,8 +794,10 @@ class AbstractDCS(object):
|
||||
|
||||
@staticmethod
|
||||
def sync_state(leader, sync_standby):
|
||||
"""Build sync_state dict"""
|
||||
return {'leader': leader, 'sync_standby': sync_standby}
|
||||
"""Build sync_state dict
|
||||
sync_standby dictionary key being kept for backward compatibility
|
||||
"""
|
||||
return {'leader': leader, 'sync_standby': sync_standby and ','.join(sorted(sync_standby)) or None}
|
||||
|
||||
def write_sync_state(self, leader, sync_standby, index=None):
|
||||
sync_value = self.sync_state(leader, sync_standby)
|
||||
|
||||
@@ -165,7 +165,7 @@ class ZooKeeper(AbstractDCS):
|
||||
def load_members(self, sync_standby):
|
||||
members = []
|
||||
for member in self.get_children(self.members_path, self.cluster_watcher):
|
||||
watch = member == sync_standby and self.cluster_watcher or None
|
||||
watch = member in sync_standby and self.cluster_watcher or None
|
||||
data = self.get_node(self.members_path + member, watch)
|
||||
if data is not None:
|
||||
members.append(self.member(member, *data))
|
||||
@@ -194,7 +194,7 @@ class ZooKeeper(AbstractDCS):
|
||||
sync = SyncState.from_node(sync and sync[1].version, sync and sync[0])
|
||||
|
||||
# get list of members
|
||||
sync_standby = sync.leader == self._name and sync.sync_standby or None
|
||||
sync_standby = sync.leader == self._name and sync.members or []
|
||||
members = self.load_members(sync_standby) if self._MEMBERS[:-1] in nodes else []
|
||||
|
||||
# get leader
|
||||
|
||||
+25
-19
@@ -447,28 +447,34 @@ class Ha(object):
|
||||
promoting standbys that were guaranteed to be replicating synchronously.
|
||||
"""
|
||||
if self.is_synchronous_mode():
|
||||
current = self.cluster.sync.leader and self.cluster.sync.sync_standby
|
||||
picked, allow_promote = self.state_handler.pick_synchronous_standby(self.cluster)
|
||||
if picked != current:
|
||||
# We need to revoke privilege from current before replacing it in the config
|
||||
if current:
|
||||
logger.info("Removing synchronous privilege from %s", current)
|
||||
if not self.dcs.write_sync_state(self.state_handler.name, None, index=self.cluster.sync.index):
|
||||
sync_node_count = self.patroni.config['synchronous_node_count']
|
||||
current = self.cluster.sync.leader and self.cluster.sync.members or []
|
||||
picked, allow_promote = self.state_handler.pick_synchronous_standby(self.cluster, sync_node_count)
|
||||
if set(picked) != set(current):
|
||||
# update synchronous standby list in dcs temporarily to point to common nodes in current and picked
|
||||
sync_common = list(set(current).intersection(set(allow_promote)))
|
||||
if set(sync_common) != set(current):
|
||||
logger.info("Updating synchronous privilege temporarily from %s to %s", current, sync_common)
|
||||
if not self.dcs.write_sync_state(self.state_handler.name,
|
||||
sync_common or None,
|
||||
index=self.cluster.sync.index):
|
||||
logger.info('Synchronous replication key updated by someone else.')
|
||||
return
|
||||
|
||||
if self.is_synchronous_mode_strict() and picked is None:
|
||||
picked = '*'
|
||||
# Update db param and wait for x secs
|
||||
if self.is_synchronous_mode_strict() and not picked:
|
||||
picked = ['*']
|
||||
logger.warning("No standbys available!")
|
||||
|
||||
logger.info("Assigning synchronous standby status to %s", picked)
|
||||
self.state_handler.config.set_synchronous_standby(picked)
|
||||
|
||||
if picked and picked != '*' and not allow_promote:
|
||||
if picked and picked[0] != '*' and set(allow_promote) != set(picked) and not allow_promote:
|
||||
# Wait for PostgreSQL to enable synchronous mode and see if we can immediately set sync_standby
|
||||
time.sleep(2)
|
||||
picked, allow_promote = self.state_handler.pick_synchronous_standby(self.cluster)
|
||||
if allow_promote:
|
||||
_, allow_promote = self.state_handler.pick_synchronous_standby(self.cluster,
|
||||
sync_node_count)
|
||||
if allow_promote and set(allow_promote) != set(sync_common):
|
||||
try:
|
||||
cluster = self.dcs.get_cluster()
|
||||
except DCSError:
|
||||
@@ -476,18 +482,18 @@ class Ha(object):
|
||||
if cluster.sync.leader and cluster.sync.leader != self.state_handler.name:
|
||||
logger.info("Synchronous replication key updated by someone else")
|
||||
return
|
||||
if not self.dcs.write_sync_state(self.state_handler.name, picked, index=cluster.sync.index):
|
||||
if not self.dcs.write_sync_state(self.state_handler.name, allow_promote, index=cluster.sync.index):
|
||||
logger.info("Synchronous replication key updated by someone else")
|
||||
return
|
||||
logger.info("Synchronous standby status assigned to %s", picked)
|
||||
logger.info("Synchronous standby status assigned to %s", allow_promote)
|
||||
else:
|
||||
if self.cluster.sync.leader and self.dcs.delete_sync_state(index=self.cluster.sync.index):
|
||||
logger.info("Disabled synchronous replication")
|
||||
self.state_handler.config.set_synchronous_standby(None)
|
||||
self.state_handler.config.set_synchronous_standby([])
|
||||
|
||||
def is_sync_standby(self, cluster):
|
||||
return cluster.leader and cluster.sync.leader == cluster.leader.name \
|
||||
and cluster.sync.sync_standby == self.state_handler.name
|
||||
and self.state_handler.name in cluster.sync.members
|
||||
|
||||
def while_not_sync_standby(self, func):
|
||||
"""Runs specified action while trying to make sure that the node is not assigned synchronous standby status.
|
||||
@@ -576,7 +582,7 @@ class Ha(object):
|
||||
# Somebody else updated sync state, it may be due to us losing the lock. To be safe, postpone
|
||||
# promotion until next cycle. TODO: trigger immediate retry of run_cycle
|
||||
return 'Postponing promotion because synchronous replication state was updated by somebody else'
|
||||
self.state_handler.config.set_synchronous_standby('*' if self.is_synchronous_mode_strict() else None)
|
||||
self.state_handler.config.set_synchronous_standby(['*'] if self.is_synchronous_mode_strict() else [])
|
||||
if self.state_handler.role != 'master':
|
||||
self.set_leader_access_is_restricted(self.cluster.has_permanent_logical_slots(self.state_handler.name))
|
||||
|
||||
@@ -811,7 +817,7 @@ class Ha(object):
|
||||
self._async_executor.try_run_async('starting after demotion', self.state_handler.follow, (node_to_follow,))
|
||||
else:
|
||||
if self.is_synchronous_mode():
|
||||
self.state_handler.config.set_synchronous_standby(None)
|
||||
self.state_handler.config.set_synchronous_standby([])
|
||||
if self._rewind.rewind_or_reinitialize_needed_and_possible(leader):
|
||||
return False # do not start postgres, but run pg_rewind on the next iteration
|
||||
self.state_handler.follow(node_to_follow)
|
||||
@@ -872,7 +878,7 @@ class Ha(object):
|
||||
else:
|
||||
if self.is_synchronous_mode():
|
||||
if failover.candidate and not self.cluster.sync.matches(failover.candidate):
|
||||
logger.warning('Failover candidate=%s does not match with sync_standby=%s',
|
||||
logger.warning('Failover candidate=%s does not match with sync_standbys=%s',
|
||||
failover.candidate, self.cluster.sync.sync_standby)
|
||||
members = []
|
||||
else:
|
||||
|
||||
@@ -894,38 +894,46 @@ class Postgresql(object):
|
||||
logger.exception('Could not remove data directory %s', self._data_dir)
|
||||
self.move_data_directory()
|
||||
|
||||
def pick_synchronous_standby(self, cluster):
|
||||
def _get_synchronous_commit_param(self):
|
||||
return self.query("SHOW synchronous_commit").fetchone()[0]
|
||||
|
||||
def pick_synchronous_standby(self, cluster, sync_node_count=1):
|
||||
"""Finds the best candidate to be the synchronous standby.
|
||||
|
||||
Current synchronous standby is always preferred, unless it has disconnected or does not want to be a
|
||||
synchronous standby any longer.
|
||||
|
||||
:returns tuple of candidate name or None, and bool showing if the member is the active synchronous standby.
|
||||
:returns tuple of candidates list and synchronous standby list.
|
||||
"""
|
||||
current = cluster.sync.sync_standby
|
||||
current = current.lower() if current else current
|
||||
if self._major_version < 90600:
|
||||
sync_node_count = 1
|
||||
members = {m.name.lower(): m for m in cluster.members}
|
||||
candidates = []
|
||||
# Pick candidates based on who has flushed WAL farthest.
|
||||
# TODO: for synchronous_commit = remote_write we actually want to order on write_location
|
||||
sync_nodes = []
|
||||
# Pick candidates based on who has higher replay/remote_write/flush lsn.
|
||||
sync_commit_par = self._get_synchronous_commit_param()
|
||||
sort_col = {'remote_apply': 'replay', 'remote_write': 'write'}.get(sync_commit_par, 'flush')
|
||||
# pg_stat_replication.sync_state has 4 possible states - async, potential, quorum, sync.
|
||||
# Sort clause "ORDER BY sync_state DESC" is to get the result in required order and to keep
|
||||
# the result consistent in case if a synchronous standby member is slowed down OR async node
|
||||
# receiving changes faster than the sync member (very rare but possible). Such cases would
|
||||
# trigger sync standby member swapping frequently and the sort on sync_state desc should
|
||||
# help in keeping the query result consistent.
|
||||
for app_name, state, sync_state in self.query(
|
||||
"SELECT pg_catalog.lower(application_name), state, sync_state"
|
||||
" FROM pg_catalog.pg_stat_replication"
|
||||
" ORDER BY flush_{0} DESC".format(self.lsn_name)):
|
||||
" WHERE state = 'streaming'"
|
||||
" ORDER BY sync_state DESC, {0}_{1} DESC".format(sort_col, self.lsn_name)):
|
||||
member = members.get(app_name)
|
||||
if state != 'streaming' or not member or member.tags.get('nosync', False):
|
||||
if not member or member.tags.get('nosync', False):
|
||||
continue
|
||||
candidates.append(member.name)
|
||||
if sync_state == 'sync':
|
||||
return member.name, True
|
||||
if sync_state == 'potential' and app_name == current:
|
||||
# Prefer current even if not the best one any more to avoid indecisivness and spurious swaps.
|
||||
return cluster.sync.sync_standby, False
|
||||
if sync_state in ('async', 'potential'):
|
||||
candidates.append(member.name)
|
||||
sync_nodes.append(member.name)
|
||||
if len(candidates) >= sync_node_count:
|
||||
break
|
||||
|
||||
if candidates:
|
||||
return candidates[0], False
|
||||
return None, False
|
||||
return candidates, sync_nodes
|
||||
|
||||
def schedule_sanity_checks_after_pause(self):
|
||||
"""
|
||||
|
||||
@@ -1007,16 +1007,20 @@ class ConfigHandler(object):
|
||||
else:
|
||||
logger.info('No PostgreSQL configuration items changed, nothing to reload.')
|
||||
|
||||
def set_synchronous_standby(self, name):
|
||||
def set_synchronous_standby(self, sync_members):
|
||||
"""Sets a node to be synchronous standby and if changed does a reload for PostgreSQL."""
|
||||
if name and name != '*':
|
||||
name = quote_ident(name)
|
||||
if name != self._synchronous_standby_names:
|
||||
if name is None:
|
||||
if sync_members and sync_members != ['*']:
|
||||
sync_members = [quote_ident(x) for x in sync_members]
|
||||
if self._postgresql.major_version >= 90600 and len(sync_members) > 1:
|
||||
sync_param = '{0} ({1})'.format(len(sync_members), ','.join(sync_members))
|
||||
else:
|
||||
sync_param = next(iter(sync_members), None)
|
||||
if sync_param != self._synchronous_standby_names:
|
||||
if sync_param is None:
|
||||
self._server_parameters.pop('synchronous_standby_names', None)
|
||||
else:
|
||||
self._server_parameters['synchronous_standby_names'] = name
|
||||
self._synchronous_standby_names = name
|
||||
self._server_parameters['synchronous_standby_names'] = sync_param
|
||||
self._synchronous_standby_names = sync_param
|
||||
if self._postgresql.state == 'running':
|
||||
self.write_postgresql_conf()
|
||||
self._postgresql.reload()
|
||||
|
||||
+1
-1
@@ -411,7 +411,7 @@ def cluster_as_json(cluster):
|
||||
if m.name == leader_name:
|
||||
config = cluster.config.data if cluster.config and cluster.config.modify_index else {}
|
||||
role = 'standby_leader' if is_standby_cluster(config.get('standby_cluster')) else 'leader'
|
||||
elif m.name == cluster.sync.sync_standby:
|
||||
elif m.name in cluster.sync.members:
|
||||
role = 'sync_standby'
|
||||
else:
|
||||
role = 'replica'
|
||||
|
||||
+1
-1
@@ -174,7 +174,7 @@ class TestRestApiHandler(unittest.TestCase):
|
||||
with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={'state': 'running'})):
|
||||
MockRestApiServer(RestApiHandler, 'GET /health')
|
||||
MockRestApiServer(RestApiHandler, 'GET /master')
|
||||
MockPatroni.dcs.cluster.sync.sync_standby = MockPostgresql.name
|
||||
MockPatroni.dcs.cluster.sync.members = [MockPostgresql.name]
|
||||
MockPatroni.dcs.cluster.is_synchronous_mode = Mock(return_value=True)
|
||||
with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={'role': 'replica'})):
|
||||
MockRestApiServer(RestApiHandler, 'GET /synchronous')
|
||||
|
||||
+18
-10
@@ -3,7 +3,7 @@ import etcd
|
||||
import os
|
||||
import sys
|
||||
|
||||
from mock import Mock, MagicMock, PropertyMock, patch, mock_open
|
||||
from mock import call, Mock, MagicMock, PropertyMock, patch, mock_open
|
||||
from patroni.config import Config
|
||||
from patroni.dcs import Cluster, ClusterConfig, Failover, Leader, Member, get_dcs, SyncState, TimelineHistory
|
||||
from patroni.dcs.etcd import AbstractEtcdClientWithFailover
|
||||
@@ -856,7 +856,7 @@ class TestHa(PostgresInit):
|
||||
with patch.object(self.ha.dcs, 'delete_sync_state') as mock_delete_sync:
|
||||
self.ha.run_cycle()
|
||||
mock_delete_sync.assert_called_once()
|
||||
mock_set_sync.assert_called_once_with(None)
|
||||
mock_set_sync.assert_called_once_with([])
|
||||
|
||||
mock_set_sync.reset_mock()
|
||||
# Test sync key not touched when not there
|
||||
@@ -864,14 +864,14 @@ class TestHa(PostgresInit):
|
||||
with patch.object(self.ha.dcs, 'delete_sync_state') as mock_delete_sync:
|
||||
self.ha.run_cycle()
|
||||
mock_delete_sync.assert_not_called()
|
||||
mock_set_sync.assert_called_once_with(None)
|
||||
mock_set_sync.assert_called_once_with([])
|
||||
|
||||
mock_set_sync.reset_mock()
|
||||
|
||||
self.ha.is_synchronous_mode = true
|
||||
|
||||
# Test sync standby not touched when picking the same node
|
||||
self.p.pick_synchronous_standby = Mock(return_value=('other', True))
|
||||
self.p.pick_synchronous_standby = Mock(return_value=(['other'], ['other']))
|
||||
self.ha.cluster = get_cluster_initialized_with_leader(sync=('leader', 'other'))
|
||||
self.ha.run_cycle()
|
||||
mock_set_sync.assert_not_called()
|
||||
@@ -879,10 +879,18 @@ class TestHa(PostgresInit):
|
||||
mock_set_sync.reset_mock()
|
||||
|
||||
# Test sync standby is replaced when switching standbys
|
||||
self.p.pick_synchronous_standby = Mock(return_value=('other2', False))
|
||||
self.p.pick_synchronous_standby = Mock(return_value=(['other2'], []))
|
||||
self.ha.dcs.write_sync_state = Mock(return_value=True)
|
||||
self.ha.run_cycle()
|
||||
mock_set_sync.assert_called_once_with('other2')
|
||||
mock_set_sync.assert_called_once_with(['other2'])
|
||||
|
||||
# Test sync standby is replaced when new standby is joined
|
||||
self.p.pick_synchronous_standby = Mock(return_value=(['other2', 'other3'], ['other2']))
|
||||
self.ha.dcs.write_sync_state = Mock(return_value=True)
|
||||
self.ha.run_cycle()
|
||||
# mock_set_sync.assert_called_once_with(['other2'])
|
||||
calls = [call(['other2']), call(['other2', 'other3'])]
|
||||
mock_set_sync.assert_has_calls(calls)
|
||||
|
||||
mock_set_sync.reset_mock()
|
||||
# Test sync standby is not disabled when updating dcs fails
|
||||
@@ -895,7 +903,7 @@ class TestHa(PostgresInit):
|
||||
self.ha.dcs.write_sync_state = Mock(return_value=True)
|
||||
self.ha.dcs.get_cluster = Mock(return_value=get_cluster_initialized_with_leader(sync=('leader', 'other')))
|
||||
# self.ha.cluster = get_cluster_initialized_with_leader(sync=('leader', 'other'))
|
||||
self.p.pick_synchronous_standby = Mock(return_value=('other2', True))
|
||||
self.p.pick_synchronous_standby = Mock(return_value=(['other2'], ['other2']))
|
||||
self.ha.run_cycle()
|
||||
self.ha.dcs.get_cluster.assert_called_once()
|
||||
self.assertEqual(self.ha.dcs.write_sync_state.call_count, 2)
|
||||
@@ -918,9 +926,9 @@ class TestHa(PostgresInit):
|
||||
# Test sync set to '*' when synchronous_mode_strict is enabled
|
||||
mock_set_sync.reset_mock()
|
||||
self.ha.is_synchronous_mode_strict = true
|
||||
self.p.pick_synchronous_standby = Mock(return_value=(None, False))
|
||||
self.p.pick_synchronous_standby = Mock(return_value=([], []))
|
||||
self.ha.run_cycle()
|
||||
mock_set_sync.assert_called_once_with('*')
|
||||
mock_set_sync.assert_called_once_with(['*'])
|
||||
|
||||
def test_sync_replication_become_master(self):
|
||||
self.ha.is_synchronous_mode = true
|
||||
@@ -935,7 +943,7 @@ class TestHa(PostgresInit):
|
||||
|
||||
# When we just became master nobody is sync
|
||||
self.assertEqual(self.ha.enforce_master_role('msg', 'promote msg'), 'promote msg')
|
||||
mock_set_sync.assert_called_once_with(None)
|
||||
mock_set_sync.assert_called_once_with([])
|
||||
mock_write_sync.assert_called_once_with('leader', None, index=0)
|
||||
|
||||
mock_set_sync.reset_mock()
|
||||
|
||||
+33
-26
@@ -595,36 +595,43 @@ 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)
|
||||
mock_cursor = Mock()
|
||||
mock_cursor.fetchone.return_value = ('remote_apply',)
|
||||
|
||||
with patch.object(Postgresql, "query", return_value=[
|
||||
(self.leadermem.name, 'streaming', 'sync'),
|
||||
(self.me.name, 'streaming', 'async'),
|
||||
(self.other.name, 'streaming', 'async'),
|
||||
with patch.object(Postgresql, "query", side_effect=[
|
||||
mock_cursor,
|
||||
[(self.leadermem.name, 'streaming', 'sync'),
|
||||
(self.me.name, 'streaming', 'async'),
|
||||
(self.other.name, 'streaming', 'async')]
|
||||
]):
|
||||
self.assertEqual(self.p.pick_synchronous_standby(cluster), (self.leadermem.name, True))
|
||||
self.assertEqual(self.p.pick_synchronous_standby(cluster), ([self.leadermem.name], [self.leadermem.name]))
|
||||
|
||||
with patch.object(Postgresql, "query", return_value=[
|
||||
(self.me.name, 'streaming', 'async'),
|
||||
(self.leadermem.name, 'streaming', 'potential'),
|
||||
(self.other.name, 'streaming', 'async'),
|
||||
with patch.object(Postgresql, "query", side_effect=[
|
||||
mock_cursor,
|
||||
[(self.leadermem.name, 'streaming', 'potential'),
|
||||
(self.me.name, 'streaming', 'async'),
|
||||
(self.other.name, 'streaming', 'async')]
|
||||
]):
|
||||
self.assertEqual(self.p.pick_synchronous_standby(cluster), (self.leadermem.name, False))
|
||||
self.assertEqual(self.p.pick_synchronous_standby(cluster), ([self.leadermem.name], []))
|
||||
|
||||
with patch.object(Postgresql, "query", return_value=[
|
||||
(self.me.name, 'streaming', 'async'),
|
||||
(self.other.name, 'streaming', 'async'),
|
||||
with patch.object(Postgresql, "query", side_effect=[
|
||||
mock_cursor,
|
||||
[(self.me.name, 'streaming', 'async'),
|
||||
(self.other.name, 'streaming', 'async')]
|
||||
]):
|
||||
self.assertEqual(self.p.pick_synchronous_standby(cluster), (self.me.name, False))
|
||||
self.assertEqual(self.p.pick_synchronous_standby(cluster), ([self.me.name], []))
|
||||
|
||||
with patch.object(Postgresql, "query", return_value=[
|
||||
('missing', 'streaming', 'sync'),
|
||||
(self.me.name, 'streaming', 'async'),
|
||||
(self.other.name, 'streaming', 'async'),
|
||||
with patch.object(Postgresql, "query", side_effect=[
|
||||
mock_cursor,
|
||||
[('missing', 'streaming', 'sync'),
|
||||
(self.me.name, 'streaming', 'async'),
|
||||
(self.other.name, 'streaming', 'async')]
|
||||
]):
|
||||
self.assertEqual(self.p.pick_synchronous_standby(cluster), (self.me.name, False))
|
||||
self.assertEqual(self.p.pick_synchronous_standby(cluster), ([self.me.name], []))
|
||||
|
||||
with patch.object(Postgresql, "query", return_value=[]):
|
||||
self.assertEqual(self.p.pick_synchronous_standby(cluster), (None, False))
|
||||
with patch.object(Postgresql, "query", side_effect=[mock_cursor, []]):
|
||||
self.p._major_version = 90400
|
||||
self.assertEqual(self.p.pick_synchronous_standby(cluster), ([], []))
|
||||
|
||||
def test_set_sync_standby(self):
|
||||
def value_in_conf():
|
||||
@@ -634,21 +641,21 @@ class TestPostgresql(BaseTestPostgresql):
|
||||
return line.strip()
|
||||
|
||||
mock_reload = self.p.reload = Mock()
|
||||
self.p.config.set_synchronous_standby('n1')
|
||||
self.p.config.set_synchronous_standby(['n1'])
|
||||
self.assertEqual(value_in_conf(), "synchronous_standby_names = 'n1'")
|
||||
mock_reload.assert_called()
|
||||
|
||||
mock_reload.reset_mock()
|
||||
self.p.config.set_synchronous_standby('n1')
|
||||
self.p.config.set_synchronous_standby(['n1'])
|
||||
mock_reload.assert_not_called()
|
||||
self.assertEqual(value_in_conf(), "synchronous_standby_names = 'n1'")
|
||||
|
||||
self.p.config.set_synchronous_standby('n2')
|
||||
self.p.config.set_synchronous_standby(['n1', 'n2'])
|
||||
mock_reload.assert_called()
|
||||
self.assertEqual(value_in_conf(), "synchronous_standby_names = 'n2'")
|
||||
self.assertEqual(value_in_conf(), "synchronous_standby_names = '2 (n1,n2)'")
|
||||
|
||||
mock_reload.reset_mock()
|
||||
self.p.config.set_synchronous_standby(None)
|
||||
self.p.config.set_synchronous_standby([])
|
||||
mock_reload.assert_called()
|
||||
self.assertEqual(value_in_conf(), None)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user