diff --git a/patroni/dcs.py b/patroni/dcs.py index 25e44cd1..48d49cf5 100644 --- a/patroni/dcs.py +++ b/patroni/dcs.py @@ -240,8 +240,11 @@ class AbstractDCS: overwriting the key if necessary.""" @abc.abstractmethod - def initialize(self): + def initialize(self, create_new=True, sysid=None): """Race for cluster initialization. + + :param create_new: False if the key should already exist (in the case we are setting the system_id) + :param sysid: PostgreSQL cluster system identifier, if specified, is written to the key :returns: `!True` if key has been created successfully. this method should create atomically initialize key and return `!True` diff --git a/patroni/etcd.py b/patroni/etcd.py index 79379700..c20d73ab 100644 --- a/patroni/etcd.py +++ b/patroni/etcd.py @@ -177,7 +177,8 @@ class Etcd(AbstractDCS): nodes = {os.path.relpath(node.key, result.key): node for node in result.leaves} # get initialize flag - initialize = bool(nodes.get(self._INITIALIZE, False)) + initialize = nodes.get(self._INITIALIZE, None) + initialize = initialize and initialize.value # get last leader operation last_leader_operation = nodes.get(self._LEADER_OPTIME, None) @@ -235,8 +236,8 @@ class Etcd(AbstractDCS): return self.retry(self.client.test_and_set, self.leader_path, self._name, self._name, self.ttl) @catch_etcd_errors - def initialize(self): - return self.retry(self.client.write, self.initialize_path, self._name, prevExist=False) + def initialize(self, create_new=True, sysid=None): + return self.retry(self.client.write, self.initialize_path, sysid or "", prevExist=(not create_new)) @catch_etcd_errors def delete_leader(self): diff --git a/patroni/ha.py b/patroni/ha.py index 0019253d..82921a74 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -73,9 +73,10 @@ class Ha: self._async_executor.run_async(self.copy_backup_from_leader, args=(self.cluster.leader, )) return 'trying to bootstrap from leader' elif not self.cluster.initialize: # no initialize key - if self.dcs.initialize(): # race for initialization + if self.dcs.initialize(create_new=True): # race for initialization try: self.state_handler.bootstrap() + self.dcs.initialize(create_new=False, sysid=self.state_handler.sysid) except: # initdb or start failed # remove initialization key and give a chance to other members logger.info("removing initialize key after failed attempt to initialize the cluster") @@ -350,6 +351,11 @@ class Ha: else: return self._async_executor.scheduled_action + ' in progress' + def sysid_valid(self, sysid): + # sysid does tv_sec << 32, where tv_sec is the number of seconds sine 1970, + # so even 1 << 32 would have 10 digits. + return str(sysid) and len(str(sysid)) >= 10 and str(sysid).isdigit() + def _run_cycle(self): try: self.load_cluster_from_dcs() @@ -357,8 +363,8 @@ class Ha: self.touch_member() # cluster has leader key but not initialize key - if not self.cluster.is_unlocked() and not self.cluster.initialize: - self.dcs.initialize() # fix it + if not self.cluster.is_unlocked() and not self.sysid_valid(self.cluster.initialize) and self.has_lock(): + self.dcs.initialize(create_new=(self.cluster.initialize is None), sysid=self.state_handler.sysid) if self._async_executor.busy: return self.handle_long_action_in_progress() @@ -372,8 +378,12 @@ class Ha: if self.state_handler.data_directory_empty(): return self.bootstrap() # new node # "bootstrap", but data directory is not empty - elif not self.cluster.initialize and self.cluster.is_unlocked(): - self.dcs.initialize() + elif not self.sysid_valid(self.cluster.initialize) and self.cluster.is_unlocked(): + self.dcs.initialize(create_new=(self.cluster.initialize is None), sysid=self.state_handler.sysid) + else: + # check if we are allowed to join + if self.sysid_valid(self.cluster.initialize) and self.cluster.initialize != self.state_handler.sysid: + return "system ID mismatch, node {0} belongs to a different cluster".format(self.state_handler.name) # try to start dead postgres if not self.state_handler.is_healthy(): diff --git a/patroni/postgresql.py b/patroni/postgresql.py index fc7956e9..07f62fdf 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -69,6 +69,7 @@ class Postgresql: self._connection = None self._cursor_holder = None self._need_rewind = False + self._sysid = None self.replication_slots = [] # list of already existing replication slots self.retry = Retry(max_tries=-1, deadline=5, max_delay=1, retry_exceptions=PostgresConnectionException) @@ -105,6 +106,13 @@ class Postgresql: data.get('Data page checksum version', '0') != '0' return False + @property + def sysid(self): + if not self._sysid: + data = self.controldata() + self._sysid = data and data.get('Database system identifier', None) + return self._sysid + def require_rewind(self): self._need_rewind = True diff --git a/patroni/zookeeper.py b/patroni/zookeeper.py index 6f8ab981..d3e4fd57 100644 --- a/patroni/zookeeper.py +++ b/patroni/zookeeper.py @@ -139,7 +139,7 @@ class ZooKeeper(AbstractDCS): self.fetch_cluster = True # get initialize flag - initialize = self._INITIALIZE in nodes + initialize = self.get_node(self._INITIALIZE)[0] if self._INITIALIZE in nodes else None # get list of members members = self.load_members() if self._MEMBERS[:-1] in nodes else [] @@ -203,8 +203,9 @@ class ZooKeeper(AbstractDCS): logging.exception('set_failover_value') return False - def initialize(self): - return self._create(self.initialize_path, self._name, makepath=True) + def initialize(self, create_new=True, sysid=None): + return self._create(self.initialize_path, sysid if sysid else "", makepath=True) if create_new \ + else self.client.retry(self.client.set, self.initialize_path, sysid.encode("utf-8") if sysid else "") def touch_member(self, data, ttl=None): cluster = self.cluster