Store the cluster sysid in the initialize flag.

Make sure that the new PostgreSQL node will only
join the cluster if its sysid matches the one
stored in DCS.

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`

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):

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():

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
 

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