From d35bc0d993beb024838308f4fb7a9571483a59a5 Mon Sep 17 00:00:00 2001 From: Claudiu Belu Date: Mon, 19 Oct 2026 10:04:08 +0000 Subject: [PATCH 1/2] integration: Update test providers for incremental transfers deploy_replica_disks() allocated a new, empty loop device on every execution, ignoring the volumes_info from the previous one. This dropped the disks' "replica_state", turning every transfer execution into a full sync, and leaked the previously allocated loop devices. Disks allocated by a previous execution are now reused, like the real import providers do. The replicator only loads its state file at startup, and it is already running when replicate_disks() is called: deploy_replica_source_resources() starts it without any state. It was only restarted with the replica state on reused pool minions, so incremental executions on fresh source minions were always full syncs. The replicator is now always restarted with the replica state for incremental executions. --- coriolis/tests/integration/test_provider/exp.py | 13 +++++++------ coriolis/tests/integration/test_provider/imp.py | 12 +++++++++++- 2 files changed, 18 insertions(+), 7 deletions(-) diff --git a/coriolis/tests/integration/test_provider/exp.py b/coriolis/tests/integration/test_provider/exp.py index 047a0504..319afd36 100644 --- a/coriolis/tests/integration/test_provider/exp.py +++ b/coriolis/tests/integration/test_provider/exp.py @@ -350,7 +350,6 @@ def replicate_disks( repl_state = _extract_repl_state(volumes_info) if incremental else None disk_mappings = source_resources.get("disk_mappings") - reused_minion = disk_mappings is None if disk_mappings is None: # Minion pool case: "source_resources" only carries the pool minion's # container_id. the block devices were never attached to the container, @@ -369,11 +368,13 @@ def replicate_disks( source_conn_info, self._event_manager(), volumes_info, repl_state ) replicator.init_replicator() - if reused_minion and incremental: - # The pool minion's replicator process persists across executions and it - # only computes whole-disk checksums once, at process startup. Without - # restarting it, "verify_disk_integrity" would keep comparing against that - # stale, pre-sync checksum on every subsequent execution against the minion. + if incremental: + # The replicator process is already running, either started by + # deploy_replica_source_resources() or persisting on a pool minion across + # executions, and it only loads the state file and computes whole-disk + # checksums at startup. Without restarting it, the previous replica state + # would be ignored (resulting in full syncs), and "verify_disk_integrity" + # would compare against a stale, pre-sync checksum on a pool minion. replicator.update_state(repl_state or [], restart=True) replicator.wait_for_chunks() diff --git a/coriolis/tests/integration/test_provider/imp.py b/coriolis/tests/integration/test_provider/imp.py index fa45f3b3..6d8d9d35 100644 --- a/coriolis/tests/integration/test_provider/imp.py +++ b/coriolis/tests/integration/test_provider/imp.py @@ -160,11 +160,21 @@ def deploy_replica_disks( export_info, volumes_info, ): - """Allocate disks and return volumes_info.""" + """Allocate disks and return volumes_info. + + Disks allocated by a previous execution are reused, along with their + "replica_state", so that subsequent executions are incremental. + """ src_disks = export_info.get("devices", {}).get("disks", []) + existing_vols = {vol["disk_id"]: vol for vol in volumes_info or []} result = [] for disk in src_disks: + existing_vol = existing_vols.get(disk["id"]) + if existing_vol and existing_vol.get("volume_dev"): + result.append(existing_vol) + continue + owner = disk.get("owner") if owner and owner != instance_name: # Shared disk owned by another instance of a clustered From 8b30fc37593c9dca0d74a626a024362666179436 Mon Sep 17 00:00:00 2001 From: Claudiu Belu Date: Fri, 18 Sep 2026 09:24:07 +0000 Subject: [PATCH 2/2] integration: Verify incremental transfers for any provider pair Disk integrity verification is now enabled for both executions of test_incremental_replica_transfer. The execution if the source and destination checksums differ, so the transferred content is verified for any provider pair, not only with the core test import provider. The test also asserts that the incremental execution wrote less data than the full one, which was not checked before: every execution was a full sync with the core test providers, and the test still passed. Test providers can now implement exec_test_vm, which runs a shell command inside the test instance, and mutate_test_instance_data is built on top of it. The same hook can then be used by other tests that need to act on the guest. Also adds ssh_utils, so providers whose test instances are reachable over SSH only have to supply the instance's address. The core test export provider's instances were raw loop devices, so the only way to change their data was a hard-coded write at an offset. Instances are now formatted with a tiny ext4 filesystem holding only the host's static busybox, and exec_test_vm chroots into it. This keeps the transfer size small, compared to writing a full OS image. Document the new busybox requirement in the integration tests README, and install it in the integration tests workflow. --- .github/workflows/integration-tests.yml | 4 + coriolis/tests/integration/README.md | 3 +- coriolis/tests/integration/base.py | 41 +++++++++ .../tests/integration/osmorphing_utils.py | 56 ++++++++++++ .../tests/integration/provider_test_base.py | 39 ++++++++ coriolis/tests/integration/ssh_utils.py | 59 ++++++++++++ .../tests/integration/test_provider/exp.py | 23 ++++- .../integration/transfers/test_transfer.py | 91 ++++++++++--------- 8 files changed, 270 insertions(+), 46 deletions(-) create mode 100644 coriolis/tests/integration/ssh_utils.py diff --git a/.github/workflows/integration-tests.yml b/.github/workflows/integration-tests.yml index c957be32..96e2e91c 100644 --- a/.github/workflows/integration-tests.yml +++ b/.github/workflows/integration-tests.yml @@ -31,6 +31,10 @@ jobs: run: | sudo apt install tox + - name: Install busybox for the test instances' root filesystem + shell: bash + run: sudo apt-get install -y busybox-static + - name: Build write_data resource binary shell: bash run: | diff --git a/coriolis/tests/integration/README.md b/coriolis/tests/integration/README.md index 6ec090f5..31efcace 100644 --- a/coriolis/tests/integration/README.md +++ b/coriolis/tests/integration/README.md @@ -40,10 +40,11 @@ devices. | `dd`, `sync`, `cmp` | Test-pattern writes and device comparison | | `docker` | MariaDB database container; data-minion container image | | `ssh-keygen` | Generates the ephemeral SSH key pair used by the test provider | +| `busybox` (statically linked) | Minimal root filesystem of the core test provider's instances, which `exec_test_vm` chroots into | On Ubuntu / Debian: ```bash -sudo apt-get install util-linux coreutils +sudo apt-get install util-linux coreutils busybox-static ``` ### Docker image - data-minion diff --git a/coriolis/tests/integration/base.py b/coriolis/tests/integration/base.py index 47e38975..424378a3 100644 --- a/coriolis/tests/integration/base.py +++ b/coriolis/tests/integration/base.py @@ -26,6 +26,7 @@ from coriolis import constants, context, exception from coriolis.db import api as db_api +from coriolis.providers import backup_writers from coriolis.providers import factory as providers_factory from coriolis.tests import test_base from coriolis.tests.integration import harness @@ -414,6 +415,17 @@ def _dst_device(self): return None + @property + def _transfer_disk_ids(self): + """The "disk_id" of every disk in the transfer's volumes_info.""" + ctxt = self._get_db_context() + + transfer = db_api.get_transfer(ctxt, self._transfer.id, include_task_info=True) + info = transfer.get("info", {}).get(self._instance_name, {}) + return [ + vol["disk_id"] for vol in info.get("volumes_info", []) if vol.get("disk_id") + ] + def _cleanup_provider_dst_devices(self, transfer_id): """Remove any devices the provider allocated for this transfer.""" ctxt = self._get_db_context() @@ -444,6 +456,35 @@ def _execute_and_wait(self, transfer_id, timeout=600): ) self.assertExecutionCompleted(execution.id, timeout=timeout) + def _execute_and_spy_on_writer(self, transfer_id): + """Execute *transfer_id*, recording what the HTTP backup writer did. + + Returns a ``(checksummed_disk_ids, bytes_written)`` tuple. + """ + checksummed_disks = [] + written = [] + writer_cls = backup_writers.HTTPBackupWriterImpl + original_get_disk_checksum = writer_cls.get_disk_checksum + original_write = writer_cls.write + + def _record_get_disk_checksum(writer, *args, **kwargs): + checksummed_disks.append(writer._disk_id) + return original_get_disk_checksum(writer, *args, **kwargs) + + def _record_write(writer, data, encoding=None, uncompressed_size=None): + written.append(uncompressed_size or len(data)) + return original_write(writer, data, encoding, uncompressed_size) + + with ( + mock.patch.object( + writer_cls, "get_disk_checksum", _record_get_disk_checksum + ), + mock.patch.object(writer_cls, "write", _record_write), + ): + self._execute_and_wait(transfer_id) + + return checksummed_disks, sum(written) + def _start_executions(self, transfer_ids): """Start one execution per transfer id without waiting on any.""" return [ diff --git a/coriolis/tests/integration/osmorphing_utils.py b/coriolis/tests/integration/osmorphing_utils.py index 2f80e4c4..b98bad67 100644 --- a/coriolis/tests/integration/osmorphing_utils.py +++ b/coriolis/tests/integration/osmorphing_utils.py @@ -10,6 +10,8 @@ import contextlib import json import os +import shutil +import stat import subprocess import tempfile from typing import Iterator @@ -70,6 +72,60 @@ def write_os_image_to_disk(device_path, container_image): _run(["docker", "rm", "-f", container_id], check=False) +def write_busybox_rootfs_to_device(device_path): + """Write a minimal, chroot-able busybox rootfs to *device_path*. + + Much smaller than a full OS image: a tiny ext4 filesystem holding only the + statically linked ``busybox`` of the host, and its applet links. + """ + busybox = shutil.which("busybox") + _run( + [ + "mkfs.ext4", + "-F", + "-q", + "-m", + "0", + "-N", + "1024", + "-O", + "^has_journal", + device_path, + ] + ) + + with mounted(device_path) as mount_point: + bin_dir = os.path.join(mount_point, "bin") + os.makedirs(bin_dir) + shutil.copy(busybox, os.path.join(bin_dir, "busybox")) + + # Relative links, as "busybox --install -s" would point them at the + # host's mount point, which does not exist inside the chroot. + applets = _run([busybox, "--list"]).stdout.decode().split() + for applet in applets: + if applet != "busybox": + os.symlink("busybox", os.path.join(bin_dir, applet)) + + for name in ("dev", "proc", "tmp", "etc"): + os.makedirs(os.path.join(mount_point, name)) + + for name, minor in (("null", 3), ("zero", 5), ("urandom", 9)): + os.mknod( + os.path.join(mount_point, "dev", name), + 0o666 | stat.S_IFCHR, + os.makedev(1, minor), + ) + + +def chroot_exec(device_path, command): + """Run the shell *command* chrooted in the filesystem of *device_path*. + + Returns the ``subprocess.CompletedProcess``, without raising on failure. + """ + with mounted(device_path) as mount_point: + return _run(["chroot", mount_point, "/bin/sh", "-c", command], check=False) + + def _fixup_luks_inner_os(mapper_path, luks_uuid): """Patch the OS image inside a LUKS mapper to work with OS morphing. diff --git a/coriolis/tests/integration/provider_test_base.py b/coriolis/tests/integration/provider_test_base.py index b0235c68..7b93b444 100644 --- a/coriolis/tests/integration/provider_test_base.py +++ b/coriolis/tests/integration/provider_test_base.py @@ -19,9 +19,13 @@ from oslo_log import log as logging from coriolis import context as coriolis_context +from coriolis import exception LOG = logging.getLogger(__name__) +# Writes random data to a file in the current (home) directory, and flushes it. +_MUTATE_DATA_COMMAND = "dd if=/dev/urandom of=coriolis-marker bs=512 count=8 && sync" + class BaseTestExportProvider(abc.ABC): """Base test provider to be used by export providers. @@ -107,6 +111,41 @@ def get_test_instance_device(self, instance_name: str): """ return None + def can_exec_test_vm(self) -> bool: + """Whether ``exec_test_vm`` is implemented. + + The default is False; tests that need to run commands inside the source + instance (e.g.: to change its data) are skipped when this is False. + """ + return False + + def exec_test_vm(self, instance_name: str, command: str): + """Run the shell *command* inside *instance_name*. + + Returns an ``ssh_utils.ExecResult`` (``returncode``, ``stdout``, ``stderr``). + Only called when ``can_exec_test_vm()`` is True. Providers are expected to + SSH into the instance, or to otherwise enter its filesystem if it is not + reachable (e.g.: chroot for the core test provider). + """ + raise NotImplementedError( + "%s does not support executing commands in test instances" + % type(self).__name__ + ) + + def mutate_test_instance_data(self, instance_name: str): + """Change some data on *instance_name*'s primary disk. + + Called between two executions of the same transfer to force a detectable + difference for an incremental sync to pick up. Only called when + ``can_exec_test_vm()`` is True. + """ + result = self.exec_test_vm(instance_name, _MUTATE_DATA_COMMAND) + if result.returncode != 0: + raise exception.CoriolisException( + "Failed to mutate the data of instance '%s' (exit code %s): %s" + % (instance_name, result.returncode, result.stderr) + ) + class BaseTestImportProvider(abc.ABC): def initialize(self, connection_info: dict, source_config: dict): diff --git a/coriolis/tests/integration/ssh_utils.py b/coriolis/tests/integration/ssh_utils.py new file mode 100644 index 00000000..1e1b6494 --- /dev/null +++ b/coriolis/tests/integration/ssh_utils.py @@ -0,0 +1,59 @@ +# Copyright 2026 Cloudbase Solutions Srl +# All Rights Reserved. + +""" +SSH helpers for test providers which need to run commands inside test VMs. +""" + +import collections + +import paramiko +from oslo_log import log as logging + +from coriolis import utils + +LOG = logging.getLogger(__name__) + +# Default guest login, matching the well-known Cirros test image. Can be overridden +# through ``source.guest_ssh`` in providers.yaml. +DEFAULT_GUEST_SSH = {"username": "cirros", "password": "gocubsgo"} + +ExecResult = collections.namedtuple("ExecResult", ["returncode", "stdout", "stderr"]) + + +def get_guest_ssh_config(source_config: dict) -> dict: + """Return the guest SSH login: the defaults, updated by ``guest_ssh``. + + Recognized keys are ``username``, ``password``, ``pkey_path`` and ``port``. + """ + return {**DEFAULT_GUEST_SSH, **source_config.get("guest_ssh", {})} + + +def exec_ssh_command( + ip: str, command: str, guest_ssh: dict, connect_attempts: int = 10 +) -> ExecResult: + """Run *command* on *ip* over SSH, waiting for the guest's SSH port to open. + + *guest_ssh* is a dict as returned by ``get_guest_ssh_config``. + """ + port = guest_ssh.get("port", 22) + utils.wait_for_port_connectivity(ip, port) + + ssh = paramiko.SSHClient() + ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) + try: + utils.retry_on_error(max_attempts=connect_attempts, sleep_seconds=5)( + ssh.connect + )( + hostname=ip, + port=port, + username=guest_ssh["username"], + password=guest_ssh.get("password"), + key_filename=guest_ssh.get("pkey_path"), + timeout=10, + ) + _, stdout, stderr = ssh.exec_command(command) + returncode = stdout.channel.recv_exit_status() + return ExecResult(returncode, stdout.read().decode(), stderr.read().decode()) + finally: + ssh.close() diff --git a/coriolis/tests/integration/test_provider/exp.py b/coriolis/tests/integration/test_provider/exp.py index 319afd36..ebd33560 100644 --- a/coriolis/tests/integration/test_provider/exp.py +++ b/coriolis/tests/integration/test_provider/exp.py @@ -11,6 +11,7 @@ import csv import io import os +import shutil import unittest import uuid @@ -29,7 +30,7 @@ BaseSourceMinionPoolProvider, BaseUpdateSourceReplicaProvider, ) -from coriolis.tests.integration import provider_test_base +from coriolis.tests.integration import osmorphing_utils, provider_test_base, ssh_utils from coriolis.tests.integration import utils as test_utils from coriolis.tests.integration.test_provider import common @@ -103,6 +104,12 @@ def teardown(self, connection_info: dict): ) def check_prerequisites(self): + if not shutil.which("busybox"): + raise unittest.SkipTest( + "A statically linked 'busybox' binary is required on the host " + "(e.g.: the busybox-static package)" + ) + if not test_utils.container_image_exists(test_utils.DATA_MINION_IMAGE): raise unittest.SkipTest( "Docker image '%s' not found; build it with: " @@ -113,7 +120,7 @@ def check_prerequisites(self): def get_test_instance(self, size_mb: int): device = test_utils.create_loop_device(size_mb * 1024 * 1024) - test_utils.write_test_pattern(device, 8192) + osmorphing_utils.write_busybox_rootfs_to_device(device) instance_name = "%s-%s" % (os.path.basename(device), uuid.uuid4().hex[:8]) self._test_devices[instance_name] = device @@ -130,6 +137,18 @@ def delete_test_instance(self, instance_name: str): def get_test_instance_device(self, instance_name: str): return self._test_devices.get(instance_name) + def can_exec_test_vm(self): + return True + + def exec_test_vm(self, instance_name: str, command: str): + """Run *command* chrooted in the instance's busybox rootfs.""" + result = osmorphing_utils.chroot_exec( + self._test_devices[instance_name], command + ) + return ssh_utils.ExecResult( + result.returncode, result.stdout.decode(), result.stderr.decode() + ) + def _event_manager(self): return events.EventManager(self._event_handler) diff --git a/coriolis/tests/integration/transfers/test_transfer.py b/coriolis/tests/integration/transfers/test_transfer.py index 1eb1dfa2..4a133194 100644 --- a/coriolis/tests/integration/transfers/test_transfer.py +++ b/coriolis/tests/integration/transfers/test_transfer.py @@ -63,45 +63,27 @@ def test_transfer(self): def test_incremental_replica_transfer(self): """Full transfer followed by incremental after source modification. - - Write a known byte pattern to the source loop device. - Create source / destination endpoints and a Replica transfer via the Coriolis REST API (using coriolisclient). - - Execute the transfer and wait for it to complete. - - Overwrite a single chunk on the source device. - Update the transfer to enable disk integrity verification. + - Execute the transfer and wait for it to complete. + - Mutate some data on the source instance's primary disk. - Execute a second transfer run (incremental=True). - - Assert that the destination disk checksum was computed. + - Assert that the disk checksums were verified on both runs, and that + the second run wrote less data than the first one. - The content is verified only if the test import provider is being used. + The execution fails if the source and destination checksums differ, so the + content is verified for any provider pair. The devices are also compared + directly if the test import provider is being used. Requires a source provider + that implements ``can_exec_test_vm()``, and which supports disk checksumming. """ - if not self._harness.uses_core_test_export_provider(): + if not self._exp_provider.can_exec_test_vm(): self.skipTest( - "Incremental transfer verification requires direct access " - "to the source device" - ) - - # First run: full transfer - self._execute_and_wait(self._transfer.id) - - if self._harness.uses_core_test_import_provider(): - self.assertTrue( - test_utils.devices_match(self._src_device, self._dst_device), - "Devices do not match after initial full transfer", + "%s does not support incremental transfer verification" + % type(self._exp_provider).__name__ ) - # Mutate source: write a different pattern at the second chunk - test_utils.write_bytes_at_offset( - self._src_device, - offset=4096, - data=b"\xff\xfe\xfd\xfc" * 1024, - ) - if self._harness.uses_core_test_import_provider(): - self.assertFalse( - test_utils.devices_match(self._src_device, self._dst_device), - "Devices should differ after mutating the source", - ) - - # Enable disk integrity verification for the incremental run. + # Enable disk integrity verification for both runs. # Note that the HTTP backup writer is the only one implementing # get_disk_checksum(). execution = self._client.transfers.update( @@ -117,25 +99,48 @@ def test_incremental_replica_transfer(self): ) self.assertExecutionCompleted(execution.id) - checksummed_disks = [] - original = backup_writers.HTTPBackupWriterImpl.get_disk_checksum + # First run: full transfer + checksummed_disks, full_bytes = self._execute_and_spy_on_writer( + self._transfer.id + ) + + self.assertEqual( + self._transfer_disk_ids, + checksummed_disks, + "The checksum was not computed after the initial full transfer", + ) + if self._harness.uses_core_test_import_provider(): + self.assertTrue( + test_utils.devices_match(self._src_device, self._dst_device), + "Devices do not match after initial full transfer", + ) - def _recording_get_disk_checksum(writer, *args, **kwargs): - checksummed_disks.append(writer._disk_id) - return original(writer, *args, **kwargs) + # Mutate source. + self._exp_provider.mutate_test_instance_data(self._instance_name) + + if self._harness.uses_core_test_import_provider(): + self.assertFalse( + test_utils.devices_match(self._src_device, self._dst_device), + "Devices should differ after mutating the source", + ) # Second run: incremental - with mock.patch.object( - backup_writers.HTTPBackupWriterImpl, - "get_disk_checksum", - _recording_get_disk_checksum, - ): - self._execute_and_wait(self._transfer.id) + checksummed_disks, incremental_bytes = self._execute_and_spy_on_writer( + self._transfer.id + ) self.assertEqual( - [os.path.basename(self._src_device)], + self._transfer_disk_ids, checksummed_disks, - "The checksum was not computed for the transferred disk", + "The checksum was not computed after the incremental transfer", + ) + self.assertGreater( + incremental_bytes, 0, "The incremental transfer wrote no data" + ) + self.assertLess( + incremental_bytes, + full_bytes, + "The incremental transfer wrote as much data as the full one", ) if self._harness.uses_core_test_import_provider():