Support separate execution contexts for spawned nodes

.github/workflows/ci.yml

@@ -289,6 +289,76 @@ jobs:
         name: testgres--test_logs--${{ env.RUN_CFG__NOW }}-${{ env.BASE_SIGN }}-id${{ github.run_id }}
         path: "${{ env.RUN_CFG__LOGS_DIR }}/"
 
+  test-pglab:
+    runs-on: ubuntu-24.04
+    needs: build-check
+    name: "test: pglab | ubuntu-24.04 | pg16"
+
+    steps:
+    - name: Checkout
+      uses: actions/checkout@v7
+
+    - name: Set up Python 3.12
+      uses: actions/setup-python@v7
+      with:
+        python-version: "3.12"
+        cache: 'pip'
+
+    - name: Install PostgreSQL and PgLab system dependencies
+      run: |
+        sudo apt-get update
+        sudo apt-get install -y --no-install-recommends \
+          postgresql-16 \
+          postgresql-client-16 \
+          libpq-dev \
+          iproute2 \
+          iputils-ping \
+          iptables \
+          util-linux
+
+    - name: Install Python test dependencies
+      run: |
+        python -m pip install --upgrade pip
+        python -m pip install -r tests/requirements.txt
+
+    - name: PgLab preflight
+      run: |
+        set -eux
+        "$(pg_config --bindir)/postgres" --version
+        pg_config --bindir
+        findmnt -n -o SOURCE,FSTYPE -T /var/tmp
+        mount | grep cgroup || true
+        cat /sys/fs/cgroup/cgroup.controllers
+        cat /sys/fs/cgroup/cgroup.subtree_control || true
+
+    - name: Run privileged PgLab tests
+      run: |
+        set -eux
+        mkdir -p "${{ github.workspace }}/logs-pglab"
+        PYTHON="$(command -v python)"
+        PG_BIN="$(pg_config --bindir)"
+        sudo env \
+          PATH="$PATH" \
+          TESTGRES_PGLAB_TESTS=1 \
+          PG_BIN="$PG_BIN" \
+          PGLAB_OS_USER="$USER" \
+          PGLAB_DB_USER=postgres \
+          PGLAB_TEST_ROOT=/var/tmp \
+          TEST_CFG__LOG_DIR="${{ github.workspace }}/logs-pglab" \
+          "$PYTHON" -m pytest -vv -s \
+            tests/test_pglab_compat.py \
+            tests/test_pglab.py \
+            tests/test_pglab_faults.py \
+            tests/test_pglab_netem.py \
+            tests/test_pglab_netem_effects.py
+
+    - name: Upload PgLab logs
+      uses: actions/upload-artifact@v7
+      if: always()
+      with:
+        name: testgres--pglab-logs--id${{ github.run_id }}
+        path: logs-pglab/
+
   test-macos-14:
     runs-on: macos-14
     needs: build-check

README.md

@@ -120,6 +120,34 @@ with testgres.get_new_node('master') as master:
         print(replica.execute('postgres', 'select 1'))
 ```
 
+### Linux fault injection with PgLab
+
+`testgres.pglab` builds Linux-isolated PostgreSQL nodes on top of the normal
+Testgres lifecycle. Each PgLab node gets its own network namespace and cgroup,
+while Testgres still owns `initdb`, `pg_ctl`, `pg_basebackup`, replication and
+SQL access.
+
+This is also a concrete use case for the target execution-context arguments of
+`PostgresNode.replicate()`: a replica must be created with its own
+`OsOperations`, host and port, and `pg_basebackup` must run in that target
+context.
+
+```python
+from testgres.pglab import PgLab
+
+with PgLab(pg_bin='/path/to/postgres/bin', os_user='postgres') as lab:
+    primary = lab.create_primary('p1', '10.203.0.10').init().start()
+    replica = lab.create_replica('p1', 'r1', '10.203.0.11').start()
+    replica.catchup()
+
+    lab.partition('p1', 'r1')
+    lab.heal_partition('p1', 'r1')
+```
+
+PgLab is Linux-only and currently requires root privileges because it creates
+network namespaces, veth devices and cgroups. Its privileged integration suite
+is run separately in CI.
+
 ### Benchmarks
 
 Use `pgbench` through `testgres` to run quick benchmarks:

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