diff --git a/src/backup.py b/src/backup.py index 75ac1af2..522dfb4a 100644 --- a/src/backup.py +++ b/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,19 +176,37 @@ 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. @@ -177,13 +214,23 @@ def spawn_replica(self, name=None, destroy=True, slot=None): 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: diff --git a/src/node.py b/src/node.py index c680bf96..5458e830 100644 --- a/src/node.py +++ b/src/node.py @@ -310,17 +310,45 @@ 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 @@ -328,11 +356,12 @@ def clone_with_new_name_and_base_dir(self, name: str, base_dir: 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): diff --git a/tests/test_testgres_common.py b/tests/test_testgres_common.py index 6931c090..37fd2147 100644 --- a/tests/test_testgres_common.py +++ b/tests/test_testgres_common.py @@ -17,6 +17,7 @@ from src.node import PostgresNodeLogReader from src.node import PostgresNodeUtils from src.node import ProcessProxy +from src.node import PortManager__Generic2 from src.utils import PgVer from src.utils import get_pg_version2 from src.utils import file_tail @@ -63,6 +64,7 @@ import psutil import threading import testgres.postgres_configuration as testgres_pgconf +from testgres.operations.local_ops import LocalOperations from packaging.version import Version from concurrent.futures import ThreadPoolExecutor @@ -238,6 +240,69 @@ def test_node_constructor__host(self): assert node2._port != node._port assert node2._os_ops is node._os_ops assert node2._port_manager is node._port_manager + + assert node2.name == C_NODE2_NAME + assert node2.base_dir == nodedir2 + return + + def test_node_clone__target_context(self, node_svc: PostgresNodeService): + assert isinstance(node_svc, PostgresNodeService) + + with __class__.helper__get_node(node_svc) as node: + target_port = node_svc.port_manager.reserve_port() + try: + nodedir = node_svc.os_ops.mkdtemp(prefix="node-target--") + + with node.clone_with_new_name_and_base_dir( + name="target", + base_dir=nodedir, + target_os_ops=node_svc.os_ops, + target_host="abc", + target_port=target_port, + ) as node2: + assert node2 is not node + assert node2.os_ops is node_svc.os_ops + assert node2.name == "target" + assert node2.host == "abc" + assert node2.port == target_port + assert node2.port_manager is None + finally: + node_svc.port_manager.release_port(target_port) + return + + def test_node_clone__different_os_ops( + self, + node_svc: PostgresNodeService, + ): + assert isinstance(node_svc, PostgresNodeService) + + source_svc = node_svc + target_os_ops = LocalOperations() + + assert target_os_ops is not source_svc.os_ops + + unique_id = uuid.uuid4().hex + tmpdir = target_os_ops.get_tempdir() + + with __class__.helper__get_node(source_svc) as node: + nodedir = target_os_ops.build_path( + tmpdir, + "node-target--" + unique_id, + ) + + with node.clone_with_new_name_and_base_dir( + name="target", + base_dir=nodedir, + target_os_ops=target_os_ops, + ) as node2: + assert node2 is not node + assert node2.os_ops is target_os_ops + assert node2.name == "target" + assert node2.base_dir == nodedir + assert node2.host == target_os_ops.host + assert node2.port != node.port + assert node2.port_manager is not node.port_manager + assert type(node2.port_manager) is PortManager__Generic2 return def test_node_repr(self, node_svc: PostgresNodeService): @@ -1914,6 +1979,43 @@ def test_backup_multiple(self, node_svc: PostgresNodeService): backup.spawn_primary('node2', destroy=False) as node2: assert (node1.base_dir != node2.base_dir) + @pytest.mark.parametrize( + ("source_svc", "target_svc"), + [ + pytest.param( + PostgresNodeServices.sm_remote, + PostgresNodeServices.sm_local, + id="remote_to_local", + ), + ], + ) + def test_backup__different_os_ops( + self, + source_svc: PostgresNodeService, + target_svc: PostgresNodeService, + ): + assert source_svc.os_ops is not target_svc.os_ops + + with __class__.helper__get_node(source_svc) as source: + source.init(allow_streaming=True).start() + + with source.backup(os_ops=target_svc.os_ops) as backup: + assert backup.os_ops is target_svc.os_ops + + with backup.spawn_primary(destroy=False) as target: + assert target is not source + assert type(target) is PostgresNode + assert target.os_ops is target_svc.os_ops + assert target.host == target_svc.os_ops.host + assert target.base_dir != backup.base_dir + assert target.os_ops.isdir(target.data_dir) + assert target.port_manager is not None + assert target.port_manager is not source.port_manager + + target.slow_start() + target.stop() + return + def test_backup_exhaust(self, node_svc: PostgresNodeService): assert isinstance(node_svc, PostgresNodeService) with __class__.helper__get_node(node_svc) as node: @@ -2099,6 +2201,91 @@ def test_replicate(self, node_svc: PostgresNodeService): res = node.execute('select * from test') assert (res == []) + def test_replicate__target_context(self, node_svc: PostgresNodeService): + assert isinstance(node_svc, PostgresNodeService) + + with __class__.helper__get_node(node_svc) as master: + master.init(allow_streaming=True).start() + + target_port = node_svc.port_manager.reserve_port() + assert target_port != master.port + try: + # For local operations use an equivalent host spelling to + # exercise source_walsender with different host strings. A + # remote host must keep its configured address. + if isinstance(node_svc.os_ops, LocalOperations): + target_host = "localhost" if master.host != "localhost" else "127.0.0.1" + else: + target_host = master.host + + with master.replicate( + target_os_ops=node_svc.os_ops, + target_host=target_host, + target_port=target_port, + ) as replica: + assert type(replica) is PostgresNode + assert replica.os_ops is node_svc.os_ops + assert replica.host == target_host + assert replica.port == target_port + assert replica.port_manager is None + + replica.start() + + master.execute('create table test_target_context (val int)', commit=True) + master.execute('insert into test_target_context values (42)', commit=True) + + replica.catchup() + + res = replica.execute('select val from test_target_context') + assert res == [(42, )] + + assert replica.source_walsender is not None + finally: + node_svc.port_manager.release_port(target_port) + return + + @pytest.mark.parametrize( + ("source_svc", "target_svc"), + [ + pytest.param( + PostgresNodeServices.sm_remote, + PostgresNodeServices.sm_local, + id="remote_to_local", + ), + ], + ) + def test_replicate__different_os_ops( + self, + source_svc: PostgresNodeService, + target_svc: PostgresNodeService, + ): + assert source_svc.os_ops is not target_svc.os_ops + + with __class__.helper__get_node(source_svc) as master: + master.init(allow_streaming=True).start() + + with master.replicate( + target_os_ops=target_svc.os_ops, + ).start() as replica: + assert replica.os_ops is target_svc.os_ops + assert replica.host == target_svc.os_ops.host + + master.execute( + 'create table test_cross_context (val int)', + commit=True, + ) + master.execute( + 'insert into test_cross_context values (42)', + commit=True, + ) + + replica.catchup() + + res = replica.execute('select val from test_cross_context') + assert res == [(42, )] + + assert replica.source_walsender is not None + def test_synchronous_replication(self, node_svc: PostgresNodeService): assert isinstance(node_svc, PostgresNodeService)