Support separate execution contexts for spawned nodes

src/backup.py

@@ -35,7 +35,8 @@ def __init__(self,
                  base_dir=None,
                  username=None,
                  xlog_method=XLogMethod.fetch,
-                 options=None):
+                 options=None,
+                 os_ops=None):
         """
         Create a new backup.
 
@@ -44,13 +45,18 @@ def __init__(self,
             base_dir: where should we store it?
             username: database user name.
             xlog_method: none | fetch | stream (see docs)
+            os_ops: execution context where pg_basebackup is run and the
+                backup is stored. If omitted, use the source node's os_ops.
+                The source node must be reachable from this context using its
+                configured host and port.
         """
         assert node.os_ops is not None
         assert isinstance(node.os_ops, OsOperations)
+        assert os_ops is None or isinstance(os_ops, OsOperations)
 
         if not options:
             options = []
-        self.os_ops = node.os_ops
+        self.os_ops = os_ops if os_ops is not None else node.os_ops
         if not node.status():
             raise BackupException('Node must be running')
 
@@ -63,7 +69,9 @@ def __init__(self,
                 raise BackupException(msg)
 
         # Set default arguments
-        username = username or self.os_ops.get_user()
+        # The username belongs to the source PostgreSQL instance, not to the
+        # environment where pg_basebackup happens to run.
+        username = username or node.os_ops.get_user()
         base_dir = base_dir or self.os_ops.mkdtemp(prefix=TMP_BACKUP)
 
         # public
@@ -139,13 +147,24 @@ def _prepare_dir(self, destroy):
         # Return path to new node
         return dest_base_dir
 
-    def spawn_primary(self, name=None, destroy=True):
+    def spawn_primary(
+        self,
+        name=None,
+        destroy=True,
+        target_host=None,
+        target_port=None,
+    ):
         """
         Create a primary node from a backup.
 
         Args:
             name: primary's application name.
             destroy: should we convert this backup into a node?
+            target_host: host/address for the new node. If omitted, preserve
+                the source host when the backup uses the source os_ops,
+                otherwise use the backup os_ops host.
+            target_port: explicit port for the new node. If omitted, reserve a
+                port in the backup's execution context.
 
         Returns:
             New instance of :class:`.PostgresNode`.
@@ -157,33 +176,61 @@ def spawn_primary(self, name=None, destroy=True):
         # Build a new PostgresNode
         assert self.original_node is not None
 
-        node = self.original_node.clone_with_new_name_and_base_dir(name=name, base_dir=base_dir)
+        node = self.original_node.clone_with_new_name_and_base_dir(
+            name=name,
+            base_dir=base_dir,
+            target_os_ops=self.os_ops,
+            target_host=target_host,
+            target_port=target_port,
+        )
 
         assert node is not None
         assert type(node) is self.original_node.__class__
 
         with clean_on_error(node) as node:
-            # Set a new port
+            # The backup contains the source node's connection settings.
+            # Override settings that belong to the spawned node.
             node.append_conf(filename=PG_CONF_FILE, line='\n')
-            node.append_conf(filename=PG_CONF_FILE, port=node.port)
+            node.append_conf(
+                filename=PG_CONF_FILE,
+                listen_addresses=node.host,
+                port=node.port,
+            )
 
             return node
 
-    def spawn_replica(self, name=None, destroy=True, slot=None):
+    def spawn_replica(
+        self,
+        name=None,
+        destroy=True,
+        slot=None,
+        target_host=None,
+        target_port=None,
+    ):
         """
         Create a replica of the original node from a backup.
 
         Args:
             name: replica's application name.
             slot: create a replication slot with the specified name.
             destroy: should we convert this backup into a node?
+            target_host: host/address for the replica. If omitted, preserve
+                the source host when the backup uses the source os_ops,
+                otherwise use the backup os_ops host.
+            target_port: explicit port for the replica. If omitted, reserve a
+                port in the backup's execution context.
 
         Returns:
             New instance of :class:`.PostgresNode`.
         """
 
         # Build a new PostgresNode
-        node = self.spawn_primary(name=name, destroy=destroy)
+        node = self.spawn_primary(
+            name=name,
+            destroy=destroy,
+            target_host=target_host,
+            target_port=target_port,
+        )
         assert node is not None
 
         try:

src/node.py

@@ -310,29 +310,58 @@ def _get_port_manager(os_ops: OsOperations) -> PortManager:
         # TODO: Throw the exception "Please define a port manager." ?
         return PortManager__Generic2(os_ops)
 
-    def clone_with_new_name_and_base_dir(self, name: str, base_dir: str):
+    def clone_with_new_name_and_base_dir(
+        self,
+        name: str,
+        base_dir: str,
+        target_os_ops: typing.Optional[OsOperations] = None,
+        target_host: typing.Optional[str] = None,
+        target_port: typing.Optional[int] = None,
+    ):
         assert name is None or type(name) is str
         assert base_dir is None or type(base_dir) is str
+        assert target_os_ops is None or isinstance(target_os_ops, OsOperations)
+        assert target_host is None or type(target_host) is str
+        assert target_port is None or type(target_port) is int
 
         assert __class__ == PostgresNode
 
-        if self._port_manager is None:
-            raise InvalidOperationException("PostgresNode without PortManager can't be cloned.")
+        if target_os_ops is None:
+            target_os_ops = self._os_ops
+
+        same_os_ops = target_os_ops is self._os_ops
+
+        if target_port is None and same_os_ops:
+            if self._port_manager is None:
+                raise InvalidOperationException(
+                    "PostgresNode without PortManager can't be cloned."
+                )
+
+            target_port_manager = self._port_manager
+        else:
+            target_port_manager = None
+
+        # Paths and default host names belong to an OsOperations context.
+        # Reusing them for another context is only safe when it is the same
+        # OsOperations object.  Otherwise let PostgresNode resolve them using
+        # the target environment.
+        target_bin_dir = self._bin_dir if same_os_ops else None
+        if target_host is None and same_os_ops:
+            target_host = self._host
 
-        assert self._port_manager is not None
-        assert isinstance(self._port_manager, PortManager)
         assert self._os_ops is not None
         assert isinstance(self._os_ops, OsOperations)
         assert type(self._host) is str
 
         node = PostgresNode(
             name=name,
             base_dir=base_dir,
-            bin_dir=self._bin_dir,
+            port=target_port,
+            bin_dir=target_bin_dir,
             prefix=self._prefix,
-            os_ops=self._os_ops,
-            port_manager=self._port_manager,
-            host=self._host,
+            os_ops=target_os_ops,
+            port_manager=target_port_manager,
+            host=target_host,
         )
 
         return node
@@ -530,9 +559,6 @@ def source_walsender(self):
 
         assert type(self.master) is PostgresNode
 
-        # master should be on the same host
-        assert self.master.host == self._host
-
         with self.master.connect() as con:
             for row in con.execute(sql, self.name):
                 for child in self.master.auxiliary_processes:
@@ -2014,6 +2040,10 @@ def backup(
             username: database user name.
             xlog_method: a method for collecting the logs ('fetch' | 'stream').
             base_dir: the base directory for data files and logs
+            os_ops: execution context where pg_basebackup is run and the
+                backup is stored. If omitted, use this node's os_ops. This
+                node must be reachable from that context using its configured
+                host and port.
 
         Returns:
             A smart object of type NodeBackup.
@@ -2028,6 +2058,9 @@ def replicate(
         self,
         name=None,
         slot=None,
+        target_os_ops=None,
+        target_host=None,
+        target_port=None,
         **kwargs,
     ):
         """
@@ -2036,17 +2069,37 @@ def replicate(
         Args:
             name: replica's application name.
             slot: create a replication slot with the specified name.
+            target_os_ops: OS operations object for the replica. If omitted,
+                use this node's os_ops. pg_basebackup is executed in this
+                context and writes the backup directly there. The primary must
+                be reachable from this context using its configured host and
+                port.
+            target_host: host/address for the replica. If omitted, preserve
+                the source host in the same execution context, otherwise use
+                the host from target_os_ops.
+            target_port: explicit port for the replica. If omitted, reserve a
+                port in the target execution context.
             username: database user name.
             xlog_method: a method for collecting the logs ('fetch' | 'stream').
             base_dir: the base directory for data files and logs
         """
 
-        # transform backup into a replica
-        with utils.clean_on_error(self.backup(**kwargs)) as backup:
+        backup_os_ops = (
+            target_os_ops if target_os_ops is not None else self._os_ops
+        )
+
+        # Create the backup in the same execution context where the replica
+        # will live.  This avoids copying a backup across OsOperations
+        # instances and also makes _prepare_dir() operate in the right context.
+        with utils.clean_on_error(
+            self.backup(os_ops=backup_os_ops, **kwargs)
+        ) as backup:
             return backup.spawn_replica(
                 name=name,
                 destroy=True,
                 slot=slot,
+                target_host=target_host,
+                target_port=target_port,
             )
 
     def set_synchronous_standbys(self, standbys):