gitops/patroni
1
0
Fork 0
mirror of https://github.com/outbackdingo/patroni.git synced 2026-09-01 09:09:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

Citus integration (#2504)

Browse Source 
Citus cluster (coordinator and workers) will be stored in DCS as a fleet of Patroni logically grouped together:
```
/service/batman/
/service/batman/0/
/service/batman/0/initialize
/service/batman/0/leader
/service/batman/0/members/
/service/batman/0/members/m1
/service/batman/0/members/m2
/service/batman/
/service/batman/1/
/service/batman/1/initialize
/service/batman/1/leader
/service/batman/1/members/
/service/batman/1/members/m1
/service/batman/1/members/m2
...
```

Where 0 is a Citus group for coordinator and 1, 2, etc are worker groups.

Such hierarchy allows reading the entire Citus cluster with a single call to DCS (except Zookeeper).

The get_cluster() method will be reading the entire Citus cluster on the coordinator because it needs to discover workers. For the worker cluster it will be reading the subtree of its own group.

Besides that we introduce a new method  get_citus_coordinator(). It will be used only by worker clusters.

Since there is no hierarchical structures on K8s we will use the citus group suffix on all objects that Patroni creates.
E.g.
```
batman-0-leader  # the leader config map for the coordinator
batman-0-config  # the config map holding initialize, config, and history "keys"
...
batman-1-leader  # the leader config map for worker group 1
batman-1-config
...
```

Citus integration is enabled from patroni.yaml:
```yaml
citus:
  database: citus
  group: 0  # 0 is for coordinator, 1, 2, etc are for workers
```

If enabled, Patroni will create the database, citus extension in it, and INSERTs INTO `pg_dist_authinfo` information required for Citus nodes to communicate between each other, i.e. 'password', 'sslcert', 'sslkey' for superuser if they are defined in the Patroni configuration file.

When the new Citus coordinator/worker is bootstrapped, Patroni adds `synchronous_mode: on` to the `bootstrap.dcs` section.

Besides that, Patroni takes over management of some Postgres GUCs:
- `shared_preload_libraries` - Patroni ensures that the "citus" is added to the first place
- `max_prepared_transactions` - if not set or set to 0, Patroni changes the value to `max_connections*2`
- wal_level - automatically set to logical. It is used by Citus to move/split shards. Under the hood Citus is creating/removing replication slots and they are automatically added by Patroni to the `ignore_slots` configuration to avoid accidental removal.

The coordinator primary actively discovers worker primary nodes and registers/updates them in the `pg_dist_node` table using
citus_add_node() and citus_update_node() functions.

Patroni running on the coordinator provides the new REST API endpoint: `POST /citus`. It is used by workers to facilitate controlled switchovers and restarts of worker primaries.
When the worker primary needs to shut down Postgres because of restart or switchover, it calls the `POST /citus` endpoint on the coordinator and the Patroni on the coordinator starts a transaction and calls `citus_update_node(nodeid, 'host-demoted', port)` in order to pause client connections that work with the given worker.
Once the new leader is elected or postgres started back, they perform another call to the `POST/citus` endpoint, that does another `citus_update_node()` call with actual hostname and port and commits a transaction. After transaction is committed, coordinator reestablishes connections to the worker node and client connections are unblocked.
If clients don't run long transaction the operation finishes without client visible errors, but only a short latency spike.

All operations on the `pg_dist_node` are serialized by Patroni on the coordinator. It allows to have more control and ROLLBACK transaction in progress if its lifetime exceeding a certain threshold and there are other worker nodes should be updated.
This commit is contained in:
Alexander Kukushkin
2023-01-24 16:14:58 +01:00
committed by GitHub
parent 3161f31088
commit 4872ac51e0
54 changed files with 3611 additions and 624 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/__init__.py
+6 -1
View File
@@ -128,6 +128,10 @@ class MockCursor(object):
b'1\t0/40159C0\tno recovery target specified\n\n'
b'2\t0/402DD98\tno recovery target specified\n\n'
b'3\t0/403DD98\tno recovery target specified\n')]
elif sql.startswith('SELECT pg_catalog.citus_add_node'):
self.results = [(2,)]
elif sql.startswith('SELECT nodeid, groupid'):
self.results = [(1, 0, 'host1', 5432, 'primary'), (2, 1, 'host2', 5432, 'primary')]
else:
self.results = [(None, None, None, None, None, None, None, None, None, None)]
@@ -205,7 +209,8 @@ class PostgresInit(unittest.TestCase):
'pg_hba': ['host all all 0.0.0.0/0 md5'],
'pg_ident': ['krb realm postgres'],
'callbacks': {'on_start': 'true', 'on_stop': 'true', 'on_reload': 'true',
'on_restart': 'true', 'on_role_change': 'true'}})
'on_restart': 'true', 'on_role_change': 'true'},
'citus': {'group': 0, 'database': 'citus'}})
class BaseTestPostgresql(PostgresInit):
tests/test_api.py
+8
View File
@@ -34,6 +34,7 @@ class MockPostgresql(object):
lsn_name = 'lsn'
POSTMASTER_START_TIME = 'pg_catalog.pg_postmaster_start_time()'
TL_LSN = 'CASE WHEN pg_catalog.pg_is_in_recovery()'
citus_handler = Mock()
@staticmethod
def connection():
@@ -573,6 +574,13 @@ class TestRestApiHandler(unittest.TestCase):
MockRestApiServer(RestApiHandler, post + '14\n\n{"leader":"1"}')
MockRestApiServer(RestApiHandler, post + '37\n\n{"candidate":"2","scheduled_at": "1"}')
@patch.object(MockPatroni, 'dcs', Mock())
@patch.object(MockHa, 'is_leader', Mock(return_value=True))
def test_do_POST_citus(self):
post = 'POST /citus HTTP/1.0' + self._authorization + '\nContent-Length: '
MockRestApiServer(RestApiHandler, post + '0\n\n')
MockRestApiServer(RestApiHandler, post + '14\n\n{"leader":"1"}')
class TestRestApiServer(unittest.TestCase):
tests/test_citus.py
+159
View File
@@ -0,0 +1,159 @@
from mock import Mock, patch
from patroni.postgresql.citus import CitusHandler
from . import BaseTestPostgresql, MockCursor, psycopg_connect, SleepException
from .test_ha import get_cluster_initialized_with_leader
@patch('patroni.postgresql.citus.Thread', Mock())
@patch('patroni.psycopg.connect', psycopg_connect)
class TestCitus(BaseTestPostgresql):
def setUp(self):
super(TestCitus, self).setUp()
self.c = self.p.citus_handler
self.c.set_conn_kwargs({'host': 'localhost', 'dbname': 'postgres'})
self.cluster = get_cluster_initialized_with_leader()
self.cluster.workers[1] = self.cluster
@patch('time.time', Mock(side_effect=[100, 130, 160, 190, 220, 250, 280]))
@patch('patroni.postgresql.citus.logger.exception', Mock(side_effect=SleepException))
@patch('patroni.postgresql.citus.logger.warning')
@patch('patroni.postgresql.citus.PgDistNode.wait', Mock())
@patch.object(CitusHandler, 'is_alive', Mock(return_value=True))
def test_run(self, mock_logger_warning):
# `before_demote` or `before_promote` REST API calls starting a
# transaction. We want to make sure that it finishes during
# certain timeout. In case if it is not, we want to roll it back
# in order to not block other workers that want to update
# `pg_dist_node`.
self.c._condition.wait = Mock(side_effect=[Mock(), Mock(), Mock(), SleepException])
self.c.handle_event(self.cluster, {'type': 'before_demote', 'group': 1,
'leader': 'leader', 'timeout': 30, 'cooldown': 10})
self.c.add_task('after_promote', 2, 'postgres://host3:5432/postgres')
self.assertRaises(SleepException, self.c.run)
mock_logger_warning.assert_called_once()
self.assertTrue(mock_logger_warning.call_args[0][0].startswith('Rolling back transaction'))
self.assertTrue(repr(mock_logger_warning.call_args[0][1]).startswith('PgDistNode'))
@patch.object(CitusHandler, 'is_alive', Mock(return_value=False))
@patch.object(CitusHandler, 'start', Mock())
def test_sync_pg_dist_node(self):
with patch.object(CitusHandler, 'is_enabled', Mock(return_value=False)):
self.c.sync_pg_dist_node(self.cluster)
self.c.sync_pg_dist_node(self.cluster)
def test_handle_event(self):
self.c.handle_event(self.cluster, {})
with patch.object(CitusHandler, 'is_alive', Mock(return_value=True)):
self.c.handle_event(self.cluster, {'type': 'after_promote', 'group': 2,
'leader': 'leader', 'timeout': 30, 'cooldown': 10})
def test_add_task(self):
with patch('patroni.postgresql.citus.logger.error') as mock_logger,\
patch('patroni.postgresql.citus.urlparse', Mock(side_effect=Exception)):
self.c.add_task('', 1, None)
mock_logger.assert_called_once()
with patch('patroni.postgresql.citus.logger.debug') as mock_logger:
self.c.add_task('before_demote', 1, 'postgres://host:5432/postgres', 30)
mock_logger.assert_called_once()
self.assertTrue(mock_logger.call_args[0][0].startswith('Adding the new task:'))
with patch('patroni.postgresql.citus.logger.debug') as mock_logger:
self.c.add_task('before_promote', 1, 'postgres://host:5432/postgres', 30)
mock_logger.assert_called_once()
self.assertTrue(mock_logger.call_args[0][0].startswith('Overriding existing task:'))
# add_task called from sync_pg_dist_node should not override already scheduled or in flight task
self.assertIsNotNone(self.c.add_task('after_promote', 1, 'postgres://host:5432/postgres', 30))
self.assertIsNone(self.c.add_task('after_promote', 1, 'postgres://host:5432/postgres'))
self.c._in_flight = self.c._tasks.pop()
self.assertIsNone(self.c.add_task('after_promote', 1, 'postgres://host:5432/postgres'))
# If there is no transaction in progress and cached pg_dist_node matching desired state task should not be added
self.c._schedule_load_pg_dist_node = False
self.c._pg_dist_node[self.c._in_flight.group] = self.c._in_flight
self.c._in_flight = None
self.assertIsNone(self.c.add_task('after_promote', 1, 'postgres://host:5432/postgres'))
def test_pick_task(self):
self.c.add_task('after_promote', 1, 'postgres://host2:5432/postgres')
with patch.object(CitusHandler, 'process_task') as mock_process_task:
self.c.process_tasks()
# process_task() shouln't be called because pick_task double checks with _pg_dist_node
mock_process_task.assert_not_called()
def test_process_task(self):
self.c.add_task('after_promote', 0, 'postgres://host2:5432/postgres')
task = self.c.add_task('before_promote', 1, 'postgres://host4:5432/postgres', 30)
self.c.process_tasks()
self.assertTrue(task._event.is_set())
# the after_promote should result only in COMMIT
task = self.c.add_task('after_promote', 1, 'postgres://host4:5432/postgres', 30)
with patch.object(CitusHandler, 'query') as mock_query:
self.c.process_tasks()
mock_query.assert_called_once()
self.assertEqual(mock_query.call_args[0][0], 'COMMIT')
def test_process_tasks(self):
self.c.add_task('after_promote', 0, 'postgres://host2:5432/postgres')
self.c.process_tasks()
self.c.add_task('after_promote', 0, 'postgres://host3:5432/postgres')
with patch('patroni.postgresql.citus.logger.error') as mock_logger,\
patch.object(CitusHandler, 'query', Mock(side_effect=Exception)):
self.c.process_tasks()
mock_logger.assert_called_once()
self.assertTrue(mock_logger.call_args[0][0].startswith('Exception when working with pg_dist_node: '))
def test_on_demote(self):
self.c.on_demote()
@patch('patroni.postgresql.citus.logger.error')
@patch.object(MockCursor, 'execute', Mock(side_effect=Exception))
def test_load_pg_dist_node(self, mock_logger):
# load_pg_dist_node() triggers, query fails and exception is property handled
self.c.process_tasks()
self.assertTrue(self.c._schedule_load_pg_dist_node)
mock_logger.assert_called_once()
self.assertTrue(mock_logger.call_args[0][0].startswith('Exception when executing query'))
self.assertTrue(mock_logger.call_args[0][1].startswith('SELECT nodeid, groupid, '))
def test_wait(self):
task = self.c.add_task('before_demote', 1, 'postgres://host:5432/postgres', 30)
task._event.wait = Mock()
task.wait()
def test_adjust_postgres_gucs(self):
parameters = {'max_connections': 101,
'max_prepared_transactions': 0,
'shared_preload_libraries': 'foo , citus, bar '}
self.c.adjust_postgres_gucs(parameters)
self.assertEqual(parameters['max_prepared_transactions'], 202)
self.assertEqual(parameters['shared_preload_libraries'], 'citus,foo,bar')
self.assertEqual(parameters['wal_level'], 'logical')
@patch.object(CitusHandler, 'is_enabled', Mock(return_value=False))
def test_bootstrap(self):
self.c.bootstrap()
def test_ignore_replication_slot(self):
self.assertFalse(self.c.ignore_replication_slot({'name': 'foo', 'type': 'physical',
'database': 'bar', 'plugin': 'wal2json'}))
self.assertFalse(self.c.ignore_replication_slot({'name': 'foo', 'type': 'logical',
'database': 'bar', 'plugin': 'wal2json'}))
self.assertFalse(self.c.ignore_replication_slot({'name': 'foo', 'type': 'logical',
'database': 'bar', 'plugin': 'pgoutput'}))
self.assertFalse(self.c.ignore_replication_slot({'name': 'foo', 'type': 'logical',
'database': 'citus', 'plugin': 'pgoutput'}))
self.assertTrue(self.c.ignore_replication_slot({'name': 'citus_shard_move_slot_1_2_3',
'type': 'logical', 'database': 'citus', 'plugin': 'pgoutput'}))
self.assertFalse(self.c.ignore_replication_slot({'name': 'citus_shard_move_slot_1_2_3',
'type': 'logical', 'database': 'citus', 'plugin': 'citus'}))
self.assertFalse(self.c.ignore_replication_slot({'name': 'citus_shard_split_slot_1_2_3',
'type': 'logical', 'database': 'citus', 'plugin': 'pgoutput'}))
self.assertTrue(self.c.ignore_replication_slot({'name': 'citus_shard_split_slot_1_2_3',
'type': 'logical', 'database': 'citus', 'plugin': 'citus'}))
tests/test_config.py
+3
View File
@@ -31,6 +31,9 @@ class TestConfig(unittest.TestCase):
'PATRONI_LOGLEVEL': 'ERROR',
'PATRONI_LOG_LOGGERS': 'patroni.postmaster: WARNING, urllib3: DEBUG',
'PATRONI_LOG_FILE_NUM': '5',
'PATRONI_CITUS_DATABASE': 'citus',
'PATRONI_CITUS_GROUP': '0',
'PATRONI_CITUS_HOST': '0',
'PATRONI_RESTAPI_USERNAME': 'username',
'PATRONI_RESTAPI_PASSWORD': 'password',
'PATRONI_RESTAPI_LISTEN': '0.0.0.0:8008',
tests/test_consul.py
+13 -5
View File
@@ -18,6 +18,8 @@ def kv_get(self, key, **kwargs):
good_cls = ('6429',
[{'CreateIndex': 1334, 'Flags': 0, 'Key': key + 'failover', 'LockIndex': 0,
'ModifyIndex': 1334, 'Value': b''},
{'CreateIndex': 1334, 'Flags': 0, 'Key': key + '1/initialize', 'LockIndex': 0,
'ModifyIndex': 1334, 'Value': b'postgresql0'},
{'CreateIndex': 1334, 'Flags': 0, 'Key': key + 'initialize', 'LockIndex': 0,
'ModifyIndex': 1334, 'Value': b'postgresql0'},
{'CreateIndex': 2621, 'Flags': 0, 'Key': key + 'leader', 'LockIndex': 1,
@@ -94,7 +96,7 @@ class TestConsul(unittest.TestCase):
'verify': 'on', 'cert': 'bar', 'cacert': 'buz', 'register_service': True})
self.c = Consul({'ttl': 30, 'scope': 'test', 'name': 'postgresql1', 'host': 'localhost:1', 'retry_timeout': 10,
'register_service': True, 'service_check_tls_server_name': True})
self.c._base_path = '/service/good'
self.c._base_path = 'service/good'
self.c.get_cluster()
@patch('time.sleep', Mock(side_effect=SleepException))
@@ -115,16 +117,22 @@ class TestConsul(unittest.TestCase):
@patch.object(consul.Consul.KV, 'delete', Mock())
def test_get_cluster(self):
self.c._base_path = '/service/test'
self.c._base_path = 'service/test'
self.assertIsInstance(self.c.get_cluster(), Cluster)
self.assertIsInstance(self.c.get_cluster(), Cluster)
self.c._base_path = '/service/fail'
self.c._base_path = 'service/fail'
self.assertRaises(ConsulError, self.c.get_cluster)
self.c._base_path = '/service/broken'
self.c._base_path = 'service/broken'
self.assertIsInstance(self.c.get_cluster(), Cluster)
self.c._base_path = '/service/legacy'
self.c._base_path = 'service/legacy'
self.assertIsInstance(self.c.get_cluster(), Cluster)
def test__get_citus_cluster(self):
self.c._citus_group = '0'
cluster = self.c.get_cluster()
self.assertIsInstance(cluster, Cluster)
self.assertIsInstance(cluster.workers[1], Cluster)
@patch.object(consul.Consul.KV, 'delete', Mock(side_effect=[ConsulException, True, True, True]))
@patch.object(consul.Consul.KV, 'put', Mock(side_effect=[True, ConsulException, InvalidSession]))
def test_touch_member(self):
tests/test_ctl.py
+61 -49
View File
@@ -21,7 +21,8 @@ from .test_ha import get_cluster_initialized_without_leader, get_cluster_initial
@patch('patroni.ctl.load_config', Mock(return_value={
'scope': 'alpha', 'restapi': {'listen': '::', 'certfile': 'a'}, 'etcd': {'host': 'localhost:2379'},
'scope': 'alpha', 'restapi': {'listen': '::', 'certfile': 'a'},
'etcd': {'host': 'localhost:2379'}, 'citus': {'database': 'citus', 'group': 0},
'postgresql': {'data_dir': '.', 'pgpass': './pgpass', 'parameters': {}, 'retry_timeout': 5}}))
class TestCtl(unittest.TestCase):
@@ -30,7 +31,8 @@ class TestCtl(unittest.TestCase):
with patch.object(AbstractEtcdClientWithFailover, 'machines') as mock_machines:
mock_machines.__get__ = Mock(return_value=['http://remotehost:2379'])
self.runner = CliRunner()
self.e = get_dcs({'etcd': {'ttl': 30, 'host': 'ok:2379', 'retry_timeout': 10}}, 'foo')
self.e = get_dcs({'etcd': {'ttl': 30, 'host': 'ok:2379', 'retry_timeout': 10},
'citus': {'group': 0}}, 'foo', None)
@patch('patroni.ctl.logging.debug')
def test_load_config(self, mock_logger_debug):
@@ -56,14 +58,14 @@ class TestCtl(unittest.TestCase):
@patch('patroni.psycopg.connect', psycopg_connect)
def test_get_cursor(self):
self.assertIsNone(get_cursor(get_cluster_initialized_without_leader(), {}, role='master'))
self.assertIsNone(get_cursor({}, get_cluster_initialized_without_leader(), None, {}, role='master'))
self.assertIsNotNone(get_cursor(get_cluster_initialized_with_leader(), {}, role='master'))
self.assertIsNotNone(get_cursor({}, get_cluster_initialized_with_leader(), None, {}, role='master'))
# MockCursor returns pg_is_in_recovery as false
self.assertIsNone(get_cursor(get_cluster_initialized_with_leader(), {}, role='replica'))
self.assertIsNone(get_cursor({}, get_cluster_initialized_with_leader(), None, {}, role='replica'))
self.assertIsNotNone(get_cursor(get_cluster_initialized_with_leader(), {'dbname': 'foo'}, role='any'))
self.assertIsNotNone(get_cursor({}, get_cluster_initialized_with_leader(), None, {'dbname': 'foo'}, role='any'))
def test_parse_dcs(self):
assert parse_dcs(None) is None
@@ -80,7 +82,7 @@ class TestCtl(unittest.TestCase):
cluster = get_cluster_initialized_with_leader(Failover(1, 'foo', 'bar', scheduled_at))
del cluster.members[1].data['conn_url']
for fmt in ('pretty', 'json', 'yaml', 'tsv', 'topology'):
self.assertIsNone(output_members(cluster, name='abc', fmt=fmt))
self.assertIsNone(output_members({}, cluster, name='abc', fmt=fmt))
@patch('patroni.ctl.get_dcs')
@patch.object(PoolManager, 'request', Mock(return_value=MockResponse()))
@@ -88,74 +90,79 @@ class TestCtl(unittest.TestCase):
mock_get_dcs.return_value = self.e
mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader
mock_get_dcs.return_value.set_failover_value = Mock()
result = self.runner.invoke(ctl, ['switchover', 'dummy'], input='leader\nother\n\ny')
result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='leader\nother\n\ny')
assert 'leader' in result.output
result = self.runner.invoke(ctl, ['switchover', 'dummy'], input='leader\nother\n2300-01-01T12:23:00\ny')
result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'],
input='leader\nother\n2300-01-01T12:23:00\ny')
assert result.exit_code == 0
with patch('patroni.dcs.Cluster.is_paused', Mock(return_value=True)):
result = self.runner.invoke(ctl, ['switchover', 'dummy', '--force', '--scheduled', '2015-01-01T12:00:00'])
result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0',
'--force', '--scheduled', '2015-01-01T12:00:00'])
assert result.exit_code == 1
# Aborting switchover, as we answer NO to the confirmation
result = self.runner.invoke(ctl, ['switchover', 'dummy'], input='leader\nother\n\nN')
result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='leader\nother\n\nN')
assert result.exit_code == 1
# Aborting scheduled switchover, as we answer NO to the confirmation
result = self.runner.invoke(ctl, ['switchover', 'dummy', '--scheduled', '2015-01-01T12:00:00+01:00'],
input='leader\nother\n\nN')
result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0',
'--scheduled', '2015-01-01T12:00:00+01:00'], input='leader\nother\n\nN')
assert result.exit_code == 1
# Target and source are equal
result = self.runner.invoke(ctl, ['switchover', 'dummy'], input='leader\nleader\n\ny')
result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='leader\nleader\n\ny')
assert result.exit_code == 1
# Reality is not part of this cluster
result = self.runner.invoke(ctl, ['switchover', 'dummy'], input='leader\nReality\n\ny')
result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='leader\nReality\n\ny')
assert result.exit_code == 1
result = self.runner.invoke(ctl, ['switchover', 'dummy', '--force'])
result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0', '--force'])
assert 'Member' in result.output
result = self.runner.invoke(ctl, ['switchover', 'dummy', '--force', '--scheduled', '2015-01-01T12:00:00+01:00'])
result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0',
'--force', '--scheduled', '2015-01-01T12:00:00+01:00'])
assert result.exit_code == 0
# Invalid timestamp
result = self.runner.invoke(ctl, ['switchover', 'dummy', '--force', '--scheduled', 'invalid'])
result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0', '--force', '--scheduled', 'invalid'])
assert result.exit_code != 0
# Invalid timestamp
result = self.runner.invoke(ctl, ['switchover', 'dummy', '--force', '--scheduled', '2115-02-30T12:00:00+01:00'])
result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0',
'--force', '--scheduled', '2115-02-30T12:00:00+01:00'])
assert result.exit_code != 0
# Specifying wrong leader
result = self.runner.invoke(ctl, ['switchover', 'dummy'], input='dummy')
result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='dummy')
assert result.exit_code == 1
with patch.object(PoolManager, 'request', Mock(side_effect=Exception)):
# Non-responding patroni
result = self.runner.invoke(ctl, ['switchover', 'dummy'], input='leader\nother\n2300-01-01T12:23:00\ny')
result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'],
input='leader\nother\n2300-01-01T12:23:00\ny')
assert 'falling back to DCS' in result.output
with patch.object(PoolManager, 'request') as mocked:
mocked.return_value.status = 500
result = self.runner.invoke(ctl, ['switchover', 'dummy'], input='leader\nother\n\ny')
result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='leader\nother\n\ny')
assert 'Switchover failed' in result.output
mocked.return_value.status = 501
mocked.return_value.data = b'Server does not support this operation'
result = self.runner.invoke(ctl, ['switchover', 'dummy'], input='leader\nother\n\ny')
result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='leader\nother\n\ny')
assert 'Switchover failed' in result.output
# No members available
mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_only_leader
result = self.runner.invoke(ctl, ['switchover', 'dummy'], input='leader\nother\n\ny')
result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='leader\nother\n\ny')
assert result.exit_code == 1
# No master available
mock_get_dcs.return_value.get_cluster = get_cluster_initialized_without_leader
result = self.runner.invoke(ctl, ['switchover', 'dummy'], input='leader\nother\n\ny')
result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='leader\nother\n\ny')
assert result.exit_code == 1
@patch('patroni.ctl.get_dcs')
@@ -164,12 +171,14 @@ class TestCtl(unittest.TestCase):
mock_get_dcs.return_value = self.e
mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader
mock_get_dcs.return_value.set_failover_value = Mock()
result = self.runner.invoke(ctl, ['failover', 'dummy'], input='\n')
result = self.runner.invoke(ctl, ['failover', 'dummy', '--force'], input='\n')
assert 'For Citus clusters the --group must me specified' in result.output
result = self.runner.invoke(ctl, ['failover', 'dummy'], input='0\n')
assert 'Failover could be performed only to a specific candidate' in result.output
@patch('patroni.dcs.dcs_modules', Mock(return_value=['patroni.dcs.dummy', 'patroni.dcs.etcd']))
def test_get_dcs(self):
self.assertRaises(PatroniCtlException, get_dcs, {'dummy': {}}, 'dummy')
self.assertRaises(PatroniCtlException, get_dcs, {'dummy': {}}, 'dummy', 0)
@patch('patroni.psycopg.connect', psycopg_connect)
@patch('patroni.ctl.query_member', Mock(return_value=([['mock column']], None)))
@@ -189,7 +198,7 @@ class TestCtl(unittest.TestCase):
result = self.runner.invoke(ctl, ['query', 'alpha', '--file', 'dummy', '--command', 'dummy'])
assert result.exit_code == 1
result = self.runner.invoke(ctl, ['query', 'alpha', '--file', 'dummy'])
result = self.runner.invoke(ctl, ['query', 'alpha', '--member', 'abc', '--file', 'dummy'])
assert result.exit_code == 0
os.remove('dummy')
@@ -207,21 +216,21 @@ class TestCtl(unittest.TestCase):
def test_query_member(self):
with patch('patroni.ctl.get_cursor', Mock(return_value=MockConnect().cursor())):
rows = query_member(None, None, None, 'master', 'SELECT pg_catalog.pg_is_in_recovery()', {})
rows = query_member({}, None, None, None, None, 'master', 'SELECT pg_catalog.pg_is_in_recovery()', {})
self.assertTrue('False' in str(rows))
with patch.object(MockCursor, 'execute', Mock(side_effect=OperationalError('bla'))):
rows = query_member(None, None, None, 'replica', 'SELECT pg_catalog.pg_is_in_recovery()', {})
rows = query_member({}, None, None, None, None, 'replica', 'SELECT pg_catalog.pg_is_in_recovery()', {})
with patch('patroni.ctl.get_cursor', Mock(return_value=None)):
rows = query_member(None, None, None, None, 'SELECT pg_catalog.pg_is_in_recovery()', {})
rows = query_member({}, None, None, None, None, None, 'SELECT pg_catalog.pg_is_in_recovery()', {})
self.assertTrue('No connection to' in str(rows))
rows = query_member(None, None, None, 'replica', 'SELECT pg_catalog.pg_is_in_recovery()', {})
rows = query_member({}, None, None, None, 'foo', 'replica', 'SELECT pg_catalog.pg_is_in_recovery()', {})
self.assertTrue('No connection to' in str(rows))
with patch('patroni.ctl.get_cursor', Mock(side_effect=OperationalError('bla'))):
rows = query_member(None, None, None, 'replica', 'SELECT pg_catalog.pg_is_in_recovery()', {})
rows = query_member({}, None, None, None, None, 'replica', 'SELECT pg_catalog.pg_is_in_recovery()', {})
@patch('patroni.ctl.get_dcs')
def test_dsn(self, mock_get_dcs):
@@ -334,21 +343,23 @@ class TestCtl(unittest.TestCase):
@patch('patroni.ctl.get_dcs')
def test_remove(self, mock_get_dcs):
mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader
result = self.runner.invoke(ctl, ['-k', 'remove', 'alpha'], input='alpha\nslave')
result = self.runner.invoke(ctl, ['remove', 'dummy'], input='\n')
assert 'For Citus clusters the --group must me specified' in result.output
result = self.runner.invoke(ctl, ['-k', 'remove', 'alpha', '--group', '0'], input='alpha\nslave')
assert 'Please confirm' in result.output
assert 'You are about to remove all' in result.output
# Not typing an exact confirmation
assert result.exit_code == 1
# master specified does not match master of cluster
result = self.runner.invoke(ctl, ['remove', 'alpha'], input='alpha\nYes I am aware\nslave')
result = self.runner.invoke(ctl, ['remove', 'alpha', '--group', '0'], input='alpha\nYes I am aware\nslave')
assert result.exit_code == 1
# cluster specified on cmdline does not match verification prompt
result = self.runner.invoke(ctl, ['remove', 'alpha'], input='beta\nleader')
result = self.runner.invoke(ctl, ['remove', 'alpha', '--group', '0'], input='beta\nleader')
assert result.exit_code == 1
result = self.runner.invoke(ctl, ['remove', 'alpha'], input='alpha\nYes I am aware\nleader')
result = self.runner.invoke(ctl, ['remove', 'alpha', '--group', '0'], input='alpha\nYes I am aware\nleader')
assert result.exit_code == 0
def test_ctl(self):
@@ -358,23 +369,24 @@ class TestCtl(unittest.TestCase):
assert 'Usage:' in result.output
def test_get_any_member(self):
self.assertIsNone(get_any_member(get_cluster_initialized_without_leader(), role='master'))
self.assertIsNone(get_any_member({}, get_cluster_initialized_without_leader(), None, role='master'))
m = get_any_member(get_cluster_initialized_with_leader(), role='master')
m = get_any_member({}, get_cluster_initialized_with_leader(), None, role='master')
self.assertEqual(m.name, 'leader')
def test_get_all_members(self):
self.assertEqual(list(get_all_members(get_cluster_initialized_without_leader(), role='master')), [])
self.assertEqual(list(get_all_members({}, get_cluster_initialized_without_leader(), None, role='master')), [])
r = list(get_all_members(get_cluster_initialized_with_leader(), role='master'))
r = list(get_all_members({}, get_cluster_initialized_with_leader(), None, role='master'))
self.assertEqual(len(r), 1)
self.assertEqual(r[0].name, 'leader')
r = list(get_all_members(get_cluster_initialized_with_leader(), role='replica'))
r = list(get_all_members({}, get_cluster_initialized_with_leader(), None, role='replica'))
self.assertEqual(len(r), 1)
self.assertEqual(r[0].name, 'other')
self.assertEqual(len(list(get_all_members(get_cluster_initialized_without_leader(), role='replica'))), 2)
self.assertEqual(len(list(get_all_members({}, get_cluster_initialized_without_leader(),
None, role='replica'))), 2)
@patch('patroni.ctl.get_dcs')
def test_members(self, mock_get_dcs):
@@ -438,16 +450,16 @@ class TestCtl(unittest.TestCase):
cluster.members.append(cascade_member_wrong_tags)
mock_get_dcs.return_value.get_cluster = Mock(return_value=cluster)
result = self.runner.invoke(ctl, ['topology', 'dummy'])
assert '+\n| leader | 127.0.0.1:5435 | Leader |' in result.output
assert '|\n| + other | 127.0.0.1:5436 | Replica |' in result.output
assert '|\n| + cascade | 127.0.0.1:5437 | Replica |' in result.output
assert '|\n| + wrong_cascade | 127.0.0.1:5438 | Replica |' in result.output
assert '+\n| 0 | leader | 127.0.0.1:5435 | Leader |' in result.output
assert '|\n| 0 | + other | 127.0.0.1:5436 | Replica |' in result.output
assert '|\n| 0 | + cascade | 127.0.0.1:5437 | Replica |' in result.output
assert '|\n| 0 | + wrong_cascade | 127.0.0.1:5438 | Replica |' in result.output
cluster = get_cluster_initialized_without_leader()
mock_get_dcs.return_value.get_cluster = Mock(return_value=cluster)
result = self.runner.invoke(ctl, ['topology', 'dummy'])
assert '+\n| + leader | 127.0.0.1:5435 | Replica |' in result.output
assert '|\n| + other | 127.0.0.1:5436 | Replica |' in result.output
assert '+\n| 0 | + leader | 127.0.0.1:5435 | Replica |' in result.output
assert '|\n| 0 | + other | 127.0.0.1:5436 | Replica |' in result.output
@patch('patroni.ctl.get_dcs')
@patch.object(PoolManager, 'request', Mock(return_value=MockResponse()))
tests/test_etcd.py
+10
View File
@@ -40,6 +40,10 @@ def etcd_read(self, key, **kwargs):
raise etcd.EtcdKeyNotFound
response = {"action": "get", "node": {"key": "/service/batman5", "dir": True, "nodes": [
{"key": "/service/batman5/1", "dir": True, "nodes": [
{"key": "/service/batman5/1/initialize", "value": "2164261704",
"modifiedIndex": 20729, "createdIndex": 20729}],
"modifiedIndex": 20437, "createdIndex": 20437},
{"key": "/service/batman5/config", "value": '{"synchronous_mode": 0, "failsafe_mode": true}',
"modifiedIndex": 1582, "createdIndex": 1582},
{"key": "/service/batman5/failover", "value": "",
@@ -266,6 +270,12 @@ class TestEtcd(unittest.TestCase):
self.etcd._base_path = '/service/noleader'
self.assertRaises(EtcdError, self.etcd.get_cluster)
def test__get_citus_cluster(self):
self.etcd._citus_group = '0'
cluster = self.etcd.get_cluster()
self.assertIsInstance(cluster, Cluster)
self.assertIsInstance(cluster.workers[1], Cluster)
def test_touch_member(self):
self.assertFalse(self.etcd.touch_member('', ''))
tests/test_etcd3.py
+8
View File
@@ -30,6 +30,8 @@ def mock_urlopen(self, method, url, **kwargs):
ret.content = json.dumps({
"header": {"revision": "1"},
"kvs": [
{"key": base64_encode('/patroni/test/1/initialize'),
"value": base64_encode('12345'), "mod_revision": '1'},
{"key": base64_encode('/patroni/test/leader'),
"value": base64_encode('foo'), "lease": "bla", "mod_revision": '1'},
{"key": base64_encode('/patroni/test/members/foo'),
@@ -207,6 +209,12 @@ class TestEtcd3(BaseTestEtcd3):
mock_urlopen.side_effect = SleepException()
self.assertRaises(Etcd3Error, self.etcd3.get_cluster)
def test__get_citus_cluster(self):
self.etcd3._citus_group = '0'
cluster = self.etcd3.get_cluster()
self.assertIsInstance(cluster, Cluster)
self.assertIsInstance(cluster.workers[1], Cluster)
def test_touch_member(self):
self.etcd3.touch_member({})
self.etcd3._lease = 'bla'
tests/test_ha.py
+61 -4
View File
@@ -13,6 +13,7 @@ from patroni.postgresql import Postgresql
from patroni.postgresql.bootstrap import Bootstrap
from patroni.postgresql.cancellable import CancellableSubprocess
from patroni.postgresql.config import ConfigHandler
from patroni.postgresql.postmaster import PostmasterProcess
from patroni.postgresql.rewind import Rewind
from patroni.postgresql.slots import SlotsHandler
from patroni.utils import tzutc
@@ -51,7 +52,8 @@ def get_cluster_bootstrapping_without_leader(cluster_config=None):
def get_cluster_initialized_without_leader(leader=False, failover=None, sync=None, cluster_config=None, failsafe=False):
m1 = Member(0, 'leader', 28, {'conn_url': 'postgres://replicator:[email protected]:5435/postgres',
'api_url': 'http://127.0.0.1:8008/patroni', 'xlog_location': 4})
'api_url': 'http://127.0.0.1:8008/patroni', 'xlog_location': 4,
'role': 'master', 'state': 'running'})
leader = Leader(0, 0, m1 if leader else Member(0, '', 28, {}))
m2 = Member(0, 'other', 28, {'conn_url': 'postgres://replicator:[email protected]:5436/postgres',
'api_url': 'http://127.0.0.1:8011/patroni',
@@ -188,6 +190,7 @@ def run_async(self, func, args=()):
@patch('patroni.async_executor.AsyncExecutor.busy', PropertyMock(return_value=False))
@patch('patroni.async_executor.AsyncExecutor.run_async', run_async)
@patch('patroni.postgresql.rewind.Thread', Mock())
@patch('patroni.postgresql.citus.CitusHandler.start', Mock())
@patch('subprocess.call', Mock(return_value=0))
@patch('time.sleep', Mock())
class TestHa(PostgresInit):
@@ -204,7 +207,8 @@ class TestHa(PostgresInit):
self.p.postmaster_start_time = MagicMock(return_value=str(postmaster_start_time))
self.p.can_create_replica_without_replication_connection = MagicMock(return_value=False)
self.e = get_dcs({'etcd': {'ttl': 30, 'host': 'ok:2379', 'scope': 'test',
'name': 'foo', 'retry_timeout': 10}})
'name': 'foo', 'retry_timeout': 10},
'citus': {'database': 'citus', 'group': None}})
self.ha = Ha(MockPatroni(self.p, self.e))
self.ha.old_cluster = self.e.get_cluster()
self.ha.cluster = get_cluster_initialized_without_leader()
@@ -225,6 +229,9 @@ class TestHa(PostgresInit):
self.p.timeline_wal_position = Mock(return_value=(0, 1, 1))
self.p.set_role('standby_leader')
self.ha.touch_member()
self.p.set_role('master')
self.ha.dcs.touch_member = true
self.ha.touch_member()
def test_is_leader(self):
self.assertFalse(self.ha.is_leader())
@@ -420,6 +427,12 @@ class TestHa(PostgresInit):
self.ha.has_lock = true
self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock')
def test_coordinator_leader_with_lock(self):
self.ha.cluster = get_cluster_initialized_with_leader()
self.ha.cluster.is_unlocked = false
self.ha.has_lock = true
self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock')
@patch.object(Postgresql, '_wait_for_connection_close', Mock())
def test_demote_because_not_having_lock(self):
self.ha.cluster.is_unlocked = false
@@ -553,6 +566,8 @@ class TestHa(PostgresInit):
self.assertEqual(self.ha.bootstrap(), 'failed to acquire initialize lock')
@patch('patroni.psycopg.connect', psycopg_connect)
@patch('patroni.postgresql.citus.connect', psycopg_connect)
@patch('patroni.postgresql.citus.quote_ident', Mock())
@patch.object(Postgresql, 'connection', Mock(return_value=None))
def test_bootstrap_initialized_new_cluster(self):
self.ha.cluster = get_cluster_not_initialized_without_leader()
@@ -572,16 +587,20 @@ class TestHa(PostgresInit):
self.assertRaises(PatroniFatalException, self.ha.post_bootstrap)
@patch('patroni.psycopg.connect', psycopg_connect)
@patch('patroni.postgresql.citus.connect', psycopg_connect)
@patch('patroni.postgresql.citus.quote_ident', Mock())
@patch.object(Postgresql, 'connection', Mock(return_value=None))
def test_bootstrap_release_initialize_key_on_watchdog_failure(self):
self.ha.cluster = get_cluster_not_initialized_without_leader()
self.e.initialize = true
self.ha.bootstrap()
self.p.is_running.return_value = MockPostmaster()
self.p.is_leader = true
with patch.object(Watchdog, 'activate', Mock(return_value=False)):
with patch.object(Watchdog, 'activate', Mock(return_value=False)),\
patch('patroni.ha.logger.error') as mock_logger:
self.assertEqual(self.ha.post_bootstrap(), 'running post_bootstrap')
self.assertRaises(PatroniFatalException, self.ha.post_bootstrap)
self.assertTrue(mock_logger.call_args[0][0].startswith('Cancelling bootstrap because'
' watchdog activation failed'))
@patch('patroni.psycopg.connect', psycopg_connect)
def test_reinitialize(self):
@@ -608,6 +627,20 @@ class TestHa(PostgresInit):
with patch.object(self.ha, "restart_matches", return_value=False):
self.assertEqual(self.ha.restart({'foo': 'bar'}), (False, "restart conditions are not satisfied"))
@patch('time.sleep', Mock())
@patch.object(ConfigHandler, 'replace_pg_hba', Mock())
@patch.object(ConfigHandler, 'replace_pg_ident', Mock())
@patch.object(PostmasterProcess, 'start', Mock(return_value=MockPostmaster()))
@patch('patroni.postgresql.citus.CitusHandler.is_coordinator', Mock(return_value=False))
def test_worker_restart(self):
self.ha.has_lock = true
self.ha.patroni.request = Mock()
self.p.is_running = Mock(side_effect=[Mock(), False])
self.assertEqual(self.ha.restart({}), (True, 'restarted successfully'))
self.ha.patroni.request.assert_called()
self.assertEqual(self.ha.patroni.request.call_args_list[0][0][3]['type'], 'before_demote')
self.assertEqual(self.ha.patroni.request.call_args_list[1][0][3]['type'], 'after_promote')
@patch('os.kill', Mock())
def test_restart_in_progress(self):
with patch('patroni.async_executor.AsyncExecutor.busy', PropertyMock(return_value=True)):
@@ -631,6 +664,7 @@ class TestHa(PostgresInit):
self.ha.is_paused = true
self.assertEqual(self.ha.run_cycle(), 'PAUSE: restart in progress')
@patch('patroni.postgresql.citus.CitusHandler.is_coordinator', Mock(return_value=False))
def test_manual_failover_from_leader(self):
self.ha.fetch_node_status = get_node_status()
self.ha.has_lock = true
@@ -1266,6 +1300,16 @@ class TestHa(PostgresInit):
self.ha.is_failover_possible = true
self.ha.shutdown()
@patch('patroni.postgresql.citus.CitusHandler.is_coordinator', Mock(return_value=False))
def test_shutdown_citus_worker(self):
self.ha.is_leader = true
self.p.is_running = Mock(side_effect=[Mock(), False])
self.ha.patroni.request = Mock()
self.ha.shutdown()
self.ha.patroni.request.assert_called()
self.assertEqual(self.ha.patroni.request.call_args[0][2], 'citus')
self.assertEqual(self.ha.patroni.request.call_args[0][3]['type'], 'before_demote')
@patch('time.sleep', Mock())
def test_leader_with_not_accessible_data_directory(self):
self.ha.cluster = get_cluster_initialized_with_leader()
@@ -1366,3 +1410,16 @@ class TestHa(PostgresInit):
self.ha.dcs.attempt_to_acquire_leader = Mock(side_effect=[DCSError('foo'), Exception])
self.assertRaises(DCSError, self.ha.acquire_lock)
self.assertFalse(self.ha.acquire_lock())
@patch('patroni.postgresql.citus.CitusHandler.is_coordinator', Mock(return_value=False))
def test_notify_citus_coordinator(self):
self.ha.patroni.request = Mock()
self.ha.notify_citus_coordinator('before_demote')
self.ha.patroni.request.assert_called_once()
self.assertEqual(self.ha.patroni.request.call_args[1]['timeout'], 30)
self.ha.patroni.request = Mock(side_effect=Exception)
with patch('patroni.ha.logger.warning') as mock_logger:
self.ha.notify_citus_coordinator('before_promote')
self.assertEqual(self.ha.patroni.request.call_args[1]['timeout'], 2)
mock_logger.assert_called()
self.assertTrue(mock_logger.call_args[0][0].startswith('Request to Citus coordinator'))
tests/test_kubernetes.py
+29 -3
View File
@@ -7,7 +7,7 @@ import time
import unittest
from mock import Mock, PropertyMock, mock_open, patch
from patroni.dcs.kubernetes import k8s_client, k8s_config, K8sConfig, K8sConnectionFailed,\
from patroni.dcs.kubernetes import Cluster, k8s_client, k8s_config, K8sConfig, K8sConnectionFailed,\
K8sException, K8sObject, Kubernetes, KubernetesError, KubernetesRetriableException,\
Retry, RetryFailedError, SERVICE_HOST_ENV_NAME, SERVICE_PORT_ENV_NAME
from six.moves import builtins
@@ -26,6 +26,15 @@ def mock_list_namespaced_config_map(*args, **kwargs):
items.append(k8s_client.V1ConfigMap(metadata=k8s_client.V1ObjectMeta(**metadata)))
metadata.update({'name': 'test-sync', 'annotations': {'leader': 'p-0'}})
items.append(k8s_client.V1ConfigMap(metadata=k8s_client.V1ObjectMeta(**metadata)))
metadata.update({'name': 'test-0-leader', 'labels': {Kubernetes._CITUS_LABEL: '0'},
'annotations': {'optime': '1234x', 'leader': 'p-0', 'ttl': '30s', 'slots': '{', 'failsafe': '{'}})
items.append(k8s_client.V1ConfigMap(metadata=k8s_client.V1ObjectMeta(**metadata)))
metadata.update({'name': 'test-0-config', 'labels': {Kubernetes._CITUS_LABEL: '0'},
'annotations': {'initialize': '123', 'config': '{}'}})
items.append(k8s_client.V1ConfigMap(metadata=k8s_client.V1ObjectMeta(**metadata)))
metadata.update({'name': 'test-1-leader', 'labels': {Kubernetes._CITUS_LABEL: '1'},
'annotations': {'leader': 'p-3', 'ttl': '30s'}})
items.append(k8s_client.V1ConfigMap(metadata=k8s_client.V1ObjectMeta(**metadata)))
metadata = k8s_client.V1ObjectMeta(resource_version='1')
return k8s_client.V1ConfigMapList(metadata=metadata, items=items, kind='ConfigMapList')
@@ -48,7 +57,8 @@ def mock_list_namespaced_endpoints(*args, **kwargs):
def mock_list_namespaced_pod(*args, **kwargs):
metadata = k8s_client.V1ObjectMeta(resource_version='1', name='p-0', annotations={'status': '{}'},
metadata = k8s_client.V1ObjectMeta(resource_version='1', labels={'f': 'b', Kubernetes._CITUS_LABEL: '1'},
name='p-0', annotations={'status': '{}'},
uid='964dfeae-e79b-4476-8a5a-1920b5c2a69d')
status = k8s_client.V1PodStatus(pod_ip='10.0.0.0')
spec = k8s_client.V1PodSpec(hostname='p-0', node_name='kind-control-plane', containers=[])
@@ -213,9 +223,10 @@ class BaseTestKubernetes(unittest.TestCase):
@patch.object(k8s_client.CoreV1Api, 'list_namespaced_config_map', mock_list_namespaced_config_map, create=True)
def setUp(self, config=None):
config = config or {}
config.update(ttl=30, scope='test', name='p-0', loop_wait=10,
config.update(ttl=30, scope='test', name='p-0', loop_wait=10, group=0,
retry_timeout=10, labels={'f': 'b'}, bypass_api_service=True)
self.k = Kubernetes(config)
self.k._citus_group = None
self.assertRaises(AttributeError, self.k._pods._build_cache)
self.k._pods._is_ready = True
self.assertRaises(TypeError, self.k._kinds._build_cache)
@@ -239,6 +250,20 @@ class TestKubernetesConfigMaps(BaseTestKubernetes):
with patch.object(Kubernetes, '_wait_caches', Mock(side_effect=Exception)):
self.assertRaises(KubernetesError, self.k.get_cluster)
def test__get_citus_cluster(self):
self.k._citus_group = '0'
cluster = self.k.get_cluster()
self.assertIsInstance(cluster, Cluster)
self.assertIsInstance(cluster.workers[1], Cluster)
@patch('patroni.dcs.kubernetes.logger.error')
def test_get_citus_coordinator(self, mock_logger):
self.assertIsInstance(self.k.get_citus_coordinator(), Cluster)
with patch.object(Kubernetes, '_cluster_loader', Mock(side_effect=Exception)):
self.assertIsNone(self.k.get_citus_coordinator())
mock_logger.assert_called()
self.assertTrue(mock_logger.call_args[0][0].startswith('Failed to load Citus coordinator'))
def test_attempt_to_acquire_leader(self):
with patch.object(k8s_client.CoreV1Api, 'patch_namespaced_config_map', create=True) as mock_patch:
mock_patch.side_effect = K8sException
@@ -373,6 +398,7 @@ class TestCacheBuilder(BaseTestKubernetes):
@patch.object(k8s_client.CoreV1Api, 'list_namespaced_config_map', mock_list_namespaced_config_map, create=True)
@patch('patroni.dcs.kubernetes.ObjectCache._watch')
def test__build_cache(self, mock_response):
self.k._citus_group = '0'
mock_response.return_value.read_chunked.return_value = [json.dumps(
{'type': 'MODIFIED', 'object': {'metadata': {
'name': self.k.config_path, 'resourceVersion': '2', 'annotations': {self.k._CONFIG: 'foo'}}}}
tests/test_raft.py
+14 -4
View File
@@ -4,7 +4,8 @@ import tempfile
import time
from mock import Mock, PropertyMock, patch
from patroni.dcs.raft import DynMemberSyncObj, KVStoreTTL, Raft, RaftError, SyncObjUtility, TCPTransport, _TCPTransport
from patroni.dcs.raft import Cluster, DynMemberSyncObj, KVStoreTTL,\
Raft, RaftError, SyncObjUtility, TCPTransport, _TCPTransport
from pysyncobj import SyncObjConf, FAIL_REASON
@@ -128,7 +129,8 @@ class TestRaft(unittest.TestCase):
def test_raft(self):
raft = Raft({'ttl': 30, 'scope': 'test', 'name': 'pg', 'self_addr': '127.0.0.1:1234',
'retry_timeout': 10, 'data_dir': self._TMP})
'retry_timeout': 10, 'data_dir': self._TMP,
'database': 'citus', 'group': 0})
raft.reload_config({'retry_timeout': 20, 'ttl': 60, 'loop_wait': 10})
self.assertTrue(raft._sync_obj.set(raft.members_path + 'legacy', '{"version":"2.0.0"}'))
self.assertTrue(raft.touch_member(''))
@@ -136,20 +138,28 @@ class TestRaft(unittest.TestCase):
self.assertTrue(raft.cancel_initialization())
self.assertTrue(raft.set_config_value('{}'))
self.assertTrue(raft.write_sync_state('foo', 'bar'))
raft._citus_group = '1'
self.assertTrue(raft.manual_failover('foo', 'bar'))
raft.get_cluster()
raft._citus_group = '0'
cluster = raft.get_cluster()
self.assertIsInstance(cluster, Cluster)
self.assertIsInstance(cluster.workers[1], Cluster)
self.assertTrue(raft._sync_obj.set(raft.status_path, '{"optime":1234567,"slots":{"ls":12345}}'))
raft.get_cluster()
self.assertTrue(raft.update_leader('1', failsafe={'foo': 'bat'}))
self.assertTrue(raft._sync_obj.set(raft.failsafe_path, '{"foo"}'))
self.assertTrue(raft._sync_obj.set(raft.status_path, '{'))
raft.get_cluster()
raft.get_citus_coordinator()
self.assertTrue(raft.delete_sync_state())
self.assertTrue(raft.delete_leader())
self.assertTrue(raft.set_history_value(''))
self.assertTrue(raft.delete_cluster())
raft._citus_group = '1'
self.assertTrue(raft.delete_cluster())
raft._citus_group = None
raft.get_cluster()
self.assertTrue(raft.take_leader())
raft.get_cluster()
raft.watch(None, 0.001)
raft._sync_obj.destroy()
tests/test_zookeeper.py
+24 -6
View File
@@ -62,7 +62,7 @@ class MockKazooClient(Mock):
if path.startswith('/no_node'):
raise NoNodeError
elif path in ['/service/bla/', '/service/test/']:
return ['initialize', 'leader', 'members', 'optime', 'failover', 'sync', 'failsafe']
return ['initialize', 'leader', 'members', 'optime', 'failover', 'sync', 'failsafe', '0', '1']
return ['foo', 'bar', 'buzz']
def create(self, path, value=b"", acl=None, ephemeral=False, sequence=False, makepath=False):
@@ -155,6 +155,11 @@ class TestZooKeeper(unittest.TestCase):
def test_session_listener(self):
self.zk.session_listener(KazooState.SUSPENDED)
def test_members_watcher(self):
self.zk._fetch_cluster = False
self.zk.members_watcher(None)
self.assertTrue(self.zk._fetch_cluster)
def test_reload_config(self):
self.zk.reload_config({'ttl': 20, 'retry_timeout': 10, 'loop_wait': 10})
self.zk.reload_config({'ttl': 20, 'retry_timeout': 10, 'loop_wait': 5})
@@ -165,15 +170,15 @@ class TestZooKeeper(unittest.TestCase):
def test_get_children(self):
self.assertListEqual(self.zk.get_children('/no_node'), [])
def test__inner_load_cluster(self):
def test__cluster_loader(self):
self.zk._base_path = self.zk._base_path.replace('test', 'bla')
self.zk._inner_load_cluster()
self.zk._cluster_loader(self.zk.client_path(''))
self.zk._base_path = self.zk._base_path = '/broken'
self.zk._inner_load_cluster()
self.zk._cluster_loader(self.zk.client_path(''))
self.zk._base_path = self.zk._base_path = '/legacy'
self.zk._inner_load_cluster()
self.zk._cluster_loader(self.zk.client_path(''))
self.zk._base_path = self.zk._base_path = '/no_node'
self.zk._inner_load_cluster()
self.zk._cluster_loader(self.zk.client_path(''))
def test_get_cluster(self):
cluster = self.zk.get_cluster(True)
@@ -188,6 +193,19 @@ class TestZooKeeper(unittest.TestCase):
cluster = self.zk.get_cluster()
self.assertEqual(cluster.last_lsn, 500)
def test__get_citus_cluster(self):
self.zk._citus_group = '0'
for _ in range(0, 2):
cluster = self.zk.get_cluster()
self.assertIsInstance(cluster, Cluster)
self.assertIsInstance(cluster.workers[1], Cluster)
@patch('patroni.dcs.zookeeper.logger.error')
@patch.object(ZooKeeper, '_cluster_loader', Mock(side_effect=Exception))
def test_get_citus_coordinator(self, mock_logger):
self.assertIsNone(self.zk.get_citus_coordinator())
mock_logger.assert_called_once()
def test_delete_leader(self):
self.assertTrue(self.zk.delete_leader())