diff --git a/.github/workflows/integration-tests.yml b/.github/workflows/integration-tests.yml index c957be325..96e2e91ce 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/providers/replicator.py b/coriolis/providers/replicator.py index 7beca5643..625c4a8b2 100644 --- a/coriolis/providers/replicator.py +++ b/coriolis/providers/replicator.py @@ -1065,8 +1065,10 @@ def replicate_disks( "No new chunks to replicate for disk \"%s\" (%s)" % (volume['disk_id'], devName) ) - self._repl_state = curr_state - return self._repl_state + # NOTE: the remaining disks must still be processed, as they may + # have changed even if this one did not. + dst_vol["replica_state"] = state_for_vol + continue size = self._get_size_from_chunks(chunks) diff --git a/coriolis/tests/integration/README.md b/coriolis/tests/integration/README.md index 6ec090f5c..31efcace3 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 7ba36b648..395367200 100644 --- a/coriolis/tests/integration/base.py +++ b/coriolis/tests/integration/base.py @@ -27,6 +27,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 @@ -523,6 +524,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() @@ -553,6 +565,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 2f80e4c4c..b98bad671 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 b0235c682..7b93b4444 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 000000000..1e1b6494d --- /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 047a0504d..ebd33560d 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) @@ -350,7 +369,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 +387,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 b6a9121e9..14ff40997 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 diff --git a/coriolis/tests/integration/transfers/test_transfer.py b/coriolis/tests/integration/transfers/test_transfer.py index 1feed4cab..0ba4d77be 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", + ) + + # Mutate source. + self._exp_provider.mutate_test_instance_data(self._instance_name) - def _recording_get_disk_checksum(writer, *args, **kwargs): - checksummed_disks.append(writer._disk_id) - return original(writer, *args, **kwargs) + 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(): @@ -492,3 +497,126 @@ def _spy_get_ssh_tunnel(client_self): test_utils.devices_match(self._src_device, self._dst_device), "Devices do not match after transfer via SSH tunnel", ) + + +class MultiDiskTransferIntegrationTest(base.ReplicaIntegrationTestBase): + """Replica transfer tests for an instance with more than one disk.""" + + @classmethod + def setUpClass(cls): + h = integration_harness._IntegrationHarness.get() + if not h.uses_core_test_export_provider(): + raise unittest.SkipTest( + "Multi-disk instances require the core test export provider" + ) + super().setUpClass() + + def setUp(self): + super().setUp() + + self._root_device = self._src_device + self._data_device = test_utils.create_loop_device( + self._SRC_DEVICE_SIZE_MB * 1024 * 1024 + ) + self.addCleanup(test_utils.remove_loop_device, self._data_device) + test_utils.write_test_pattern(self._data_device, 4096) + + self._multi_disk_instance = "%s-%s-multidisk" % ( + os.path.basename(self._root_device), + uuid.uuid4().hex[:8], + ) + # NOTE: the HTTP backup writer is the only one implementing + # get_disk_checksum(). + self._multi_disk_transfer = self._create_transfer( + self._src_endpoint.id, + self._dst_endpoint.id, + instances=[self._multi_disk_instance], + source_environment={ + **self._exp_env_options, + "instance_block_devices": { + self._multi_disk_instance: [ + self._root_device, + self._data_device, + ], + }, + "verify_disk_integrity": True, + }, + destination_environment={ + **self._imp_env_options, + "data_transfer_mechanism": ( + backup_writers.DATA_TRANSFER_MECHANISM_HTTPS + ), + }, + ) + self.addCleanup( + self._cleanup_provider_dst_devices, self._multi_disk_transfer.id + ) + + def _dst_devices_by_disk_id(self): + ctxt = self._get_db_context() + transfer = db_api.get_transfer( + ctxt, self._multi_disk_transfer.id, include_task_info=True + ) + info = transfer.get("info", {}).get(self._multi_disk_instance, {}) + return { + vol["disk_id"]: vol["volume_dev"] + for vol in info.get("volumes_info", []) + if vol.get("volume_dev") + } + + def test_incremental_transfer_with_unchanged_first_disk(self): + """Incremental transfer in which only the second disk changed. + + Disks are replicated in order, so an unchanged first disk (e.g. a + mostly idle root disk) must not prevent the following disks from + being replicated. + """ + root_disk_id = os.path.basename(self._root_device) + data_disk_id = os.path.basename(self._data_device) + + # First run: full transfer + checksummed_disks, _ = self._execute_and_spy_on_writer( + self._multi_disk_transfer.id + ) + + self.assertEqual( + [root_disk_id, data_disk_id], + checksummed_disks, + "Both disks should have been replicated in order by the initial " + "full transfer", + ) + + # Mutate only the second disk. + test_utils.write_bytes_at_offset( + self._data_device, offset=4096, data=b"\xff\xfe\xfd\xfc" * 1024 + ) + + # Second run: incremental + checksummed_disks, incremental_bytes = self._execute_and_spy_on_writer( + self._multi_disk_transfer.id + ) + + self.assertIn( + data_disk_id, + checksummed_disks, + "The changed data disk was not replicated by the incremental transfer", + ) + self.assertGreater( + incremental_bytes, 0, "The incremental transfer wrote no data" + ) + + if not self._harness.uses_core_test_import_provider(): + # "volume_dev" is only a host-readable device path with the in-tree + # test provider. + return + + dst_devices = self._dst_devices_by_disk_id() + self.assertTrue( + test_utils.devices_match(self._root_device, dst_devices[root_disk_id]), + "Root disk destination does not match its source", + ) + self.assertTrue( + test_utils.devices_match(self._data_device, dst_devices[data_disk_id]), + "Data disk destination does not match its source after the " + "incremental transfer", + ) diff --git a/coriolis/tests/providers/test_replicator.py b/coriolis/tests/providers/test_replicator.py index 2b5ddbced..ae11d5c64 100644 --- a/coriolis/tests/providers/test_replicator.py +++ b/coriolis/tests/providers/test_replicator.py @@ -1593,6 +1593,33 @@ def test_replicate_disks_no_chunks(self): self.backup_writer.open.assert_not_called() self.assertEqual(result, self.replicator._repl_state) + @mock.patch.object(replicator_module, 'Client') + def test_replicate_disks_unchanged_disk_does_not_skip_others(self, mock_Client): + self.replicator._cli = mock_Client.return_value + changes = {"sdb": [], "sdc": [{'length': 100, 'offset': 0}]} + self.replicator._cli.get_changes.side_effect = lambda dev: changes[dev] + self.replicator._cli.get_status.return_value = [ + {"device-name": "sdb"}, + {"device-name": "sdc"}, + ] + self.replicator._volumes_info = [{"disk_id": "disk1"}, {"disk_id": "disk2"}] + source_volumes_info = [ + {"disk_id": "disk1", "disk_path": "/dev/sdb"}, + {"disk_id": "disk2", "disk_path": "/dev/sdc"}, + ] + self.replicator._repl_state = ['non-empty'] + + self.replicator.replicate_disks(source_volumes_info, self.backup_writer) + + self.backup_writer.open.assert_called_once_with("", "disk2") + self.replicator._cli.download_chunk.assert_called_once_with( + "sdc", {'length': 100, 'offset': 0} + ) + self.assertEqual( + [{"device-name": "sdb"}, {"device-name": "sdc"}], + [vol["replica_state"] for vol in self.replicator._volumes_info], + ) + @mock.patch('builtins.open', new_callable=mock.mock_open) @mock.patch.object(replicator_module, 'Client') def test__download_full_disk(self, mock_Client, mock_open):