From 412c51ddf1cca682a49cf1c4d11a46fbb446d644 Mon Sep 17 00:00:00 2001 From: Mark Pekala Date: Mon, 10 Jul 2023 22:43:57 -0700 Subject: [PATCH] Prevent splitbrain from duplicate names in configuration (#2724) When starting check if node with the same is registered in DCS and try to query it's REST API. If REST API is accessible exit with the error. Close #1804 --- features/basic_replication.feature | 4 ++++ features/environment.py | 12 +++++++--- features/steps/basic_replication.py | 23 +++++++++++++++++++ patroni/__main__.py | 22 ++++++++++++++++++- tests/test_patroni.py | 34 +++++++++++++++++++++++++++++ 5 files changed, 91 insertions(+), 4 deletions(-) diff --git a/features/basic_replication.feature b/features/basic_replication.feature index 4b8b2686..4a0eda68 100644 --- a/features/basic_replication.feature +++ b/features/basic_replication.feature @@ -83,3 +83,7 @@ Feature: basic replication Then postgres0 role is the secondary after 20 seconds When I add the table buz to postgres1 Then table buz is present on postgres0 after 20 seconds + + Scenario: check graceful rejection when two nodes have the same name + Given I start duplicate postgres0 on port 8011 + Then there is a "Can't start; there is already a node named 'postgres0' running" CRITICAL in the dup-postgres0 patroni log diff --git a/features/environment.py b/features/environment.py index a4d22015..f7690ad2 100644 --- a/features/environment.py +++ b/features/environment.py @@ -52,10 +52,9 @@ class AbstractController(abc.ABC): self._log = open(os.path.join(self._output_dir, self._name + '.log'), 'a') self._handle = self._start() - assert self._has_started(), "Process {0} is not running after being started".format(self._name) - max_wait_limit *= self._context.timeout_multiplier for _ in range(max_wait_limit): + assert self._has_started(), "Process {0} is not running after being started".format(self._name) if self._is_accessible(): break time.sleep(1) @@ -344,6 +343,13 @@ class PatroniController(AbstractController): '--datadir=' + os.path.join(self._work_directory, dest), '--dbname=' + self.backup_source]) + def read_patroni_log(self, level): + try: + with open(str(os.path.join(self._output_dir or '', self._name + ".log"))) as f: + return [line for line in f.readlines() if line[24:24 + len(level)] == level] + except IOError: + return [] + class ProcessHang(object): @@ -827,7 +833,7 @@ class PatroniPoolController(object): def __getattr__(self, func): if func not in ['stop', 'query', 'write_label', 'read_label', 'check_role_has_changed_to', - 'add_tag_to_config', 'get_watchdog', 'patroni_hang', 'backup']: + 'add_tag_to_config', 'get_watchdog', 'patroni_hang', 'backup', 'read_patroni_log']: raise AttributeError("PatroniPoolController instance has no attribute '{0}'".format(func)) def wrapper(name, *args, **kwargs): diff --git a/features/steps/basic_replication.py b/features/steps/basic_replication.py index 9b51a859..6499a03f 100644 --- a/features/steps/basic_replication.py +++ b/features/steps/basic_replication.py @@ -9,6 +9,22 @@ def start_patroni(context, name): return context.pctl.start(name) +@step('I start duplicate {name:w} on port {port:d}') +def start_duplicate_patroni(context, name, port): + config = { + "name": name, + "restapi": { + "listen": "127.0.0.1:{0}".format(port) + } + } + try: + context.pctl.start('dup-' + name, custom_config=config) + assert False, "Process was expected to fail" + except AssertionError as e: + assert 'is not running after being started' in str(e),\ + "No error was raised by duplicate start of {0} ".format(name) + + @step('I shut down {name:w}') def stop_patroni(context, name): return context.pctl.stop(name, timeout=60) @@ -90,3 +106,10 @@ def replication_works(context, primary, replica, time_limit): When I add the table test_{0} to {1} Then table test_{0} is present on {2} after {3} seconds """.format(int(time()), primary, replica, time_limit)) + + +@then('there is a "{message}" {level:w} in the {node} patroni log') +def check_patroni_log(context, message, level, node): + messsages_of_level = context.pctl.read_patroni_log(node, level) + assert any(message in line for line in messsages_of_level),\ + "There was no {0} {1} in the {2} patroni log".format(message, level, node) diff --git a/patroni/__main__.py b/patroni/__main__.py index 2a669d3f..b365bfe2 100644 --- a/patroni/__main__.py +++ b/patroni/__main__.py @@ -30,12 +30,15 @@ class Patroni(AbstractPatroniDaemon): self.version = __version__ self.dcs = get_dcs(self.config) + self.request = PatroniRequest(self.config, True) + + self.ensure_unique_name() + self.watchdog = Watchdog(self.config) self.load_dynamic_configuration() self.postgresql = Postgresql(self.config['postgresql']) self.api = RestApiServer(self, self.config['restapi']) - self.request = PatroniRequest(self.config, True) self.ha = Ha(self) self.tags = self.get_tags() @@ -60,6 +63,23 @@ class Patroni(AbstractPatroniDaemon): logger.warning('Can not get cluster from dcs') time.sleep(5) + def ensure_unique_name(self) -> None: + """A helper method to prevent splitbrain from operator naming error.""" + from patroni.dcs import Member + + cluster = self.dcs.get_cluster() + if not cluster: + return + member = cluster.get_member(self.config['name'], False) + if not isinstance(member, Member): + return + try: + _ = self.request(member, endpoint="/liveness") + logger.fatal("Can't start; there is already a node named '%s' running", self.config['name']) + sys.exit(1) + except Exception: + return + def get_tags(self) -> Dict[str, Any]: return {tag: value for tag, value in self.config.get('tags', {}).items() if tag not in ('clonefrom', 'nofailover', 'noloadbalance', 'nosync') or value} diff --git a/tests/test_patroni.py b/tests/test_patroni.py index d5c76c8b..e5785231 100644 --- a/tests/test_patroni.py +++ b/tests/test_patroni.py @@ -10,6 +10,7 @@ from http.server import HTTPServer from mock import Mock, PropertyMock, patch from patroni.api import RestApiServer from patroni.async_executor import AsyncExecutor +from patroni.dcs import Cluster, Member from patroni.dcs.etcd import AbstractEtcdClientWithFailover from patroni.exceptions import DCSError from patroni.postgresql import Postgresql @@ -202,3 +203,36 @@ class TestPatroni(unittest.TestCase): self.assertRaises(SystemExit, check_psycopg) with patch('builtins.__import__', mock_import): self.assertRaises(SystemExit, check_psycopg) + + def test_ensure_unique_name(self): + # None/empty cluster implies unique name + with patch('patroni.dcs.AbstractDCS.get_cluster', Mock(return_value=None)): + self.assertIsNone(self.p.ensure_unique_name()) + empty_cluster = Cluster.empty() + with patch('patroni.dcs.AbstractDCS.get_cluster', Mock(return_value=empty_cluster)): + self.assertIsNone(self.p.ensure_unique_name()) + without_members = empty_cluster._asdict() + del without_members['members'] + + # Cluster with members with different names implies unique name + okay_cluster = Cluster( + members=[Member(version=1, name="distinct", session=1, data={})], + **without_members + ) + with patch('patroni.dcs.AbstractDCS.get_cluster', Mock(return_value=okay_cluster)): + self.assertIsNone(self.p.ensure_unique_name()) + + # Cluster with a member with the same name that is running + bad_cluster = Cluster( + members=[Member(version=1, name="postgresql0", session=1, data={ + "api_url": "https://127.0.0.1:8008", + })], + **without_members + ) + with patch('patroni.dcs.AbstractDCS.get_cluster', Mock(return_value=bad_cluster)): + # If the api of the running node cannot be reached, this implies unique name + with patch.object(self.p, 'request', Mock(side_effect=ConnectionError)): + self.assertIsNone(self.p.ensure_unique_name()) + # Only if the api of the running node is reachable do we throw an error + with patch.object(self.p, 'request', Mock()): + self.assertRaises(SystemExit, self.p.ensure_unique_name)