integration: Adds BaseDestinationMinionPoolProvider to TestImportProvider

Adds BaseDestinationMinionPoolProvider to TestImportProvider and implements
all abstract methods. The test provider now advertises the
PROVIDER_TYPE_DESTINATION_MINION_POOL capability.

TestImportProvider.create_minion() starts a coriolis-data-minion
container, waits for SSH, deploys the HTTP backup writer, and returns
real connection / writer info.

start_minion / shutdown_minion / delete_minion map to
docker start / stop / remove.

attach_volumes_to_minion hotplugs block devices into the container's mount
namespace via nsenter + mknod.

healthcheck_minion opens an SSH connection to verify liveness.

coriolis/tests/integration/providers/test_provider/exp.py

@@ -187,7 +187,7 @@ class TestExportProvider(
         pkey_path = connection_info["pkey_path"]
 
         container_name = "coriolis-replicator-%s" % uuid.uuid4().hex[:8]
-        container_id = test_utils.start_container(
+        container_id = test_utils.run_container(
             test_utils.DATA_MINION_IMAGE,
             container_name,
             is_systemd=True,
@@ -218,7 +218,7 @@ class TestExportProvider(
                 },
             }
         except Exception:
-            test_utils.stop_container(container_id)
+            test_utils.remove_container(container_id)
             raise
 
     def delete_replica_source_resources(
@@ -226,7 +226,7 @@ class TestExportProvider(
             migr_resources_dict):
         container_id = (migr_resources_dict or {}).get("container_id")
         if container_id:
-            test_utils.stop_container(container_id)
+            test_utils.remove_container(container_id)
 
     def replicate_disks(
             self, ctxt, connection_info, source_environment, instance_name,

coriolis/tests/integration/providers/test_provider/imp.py

@@ -16,6 +16,7 @@ from oslo_log import log as logging
 import paramiko
 
 from coriolis.providers import backup_writers
+from coriolis.providers.base import BaseDestinationMinionPoolProvider
 from coriolis.providers.base import BaseEndpointDestinationOptionsProvider
 from coriolis.providers.base import BaseEndpointNetworksProvider
 from coriolis.providers.base import BaseEndpointProvider
@@ -24,6 +25,7 @@ from coriolis.providers.base import BaseReplicaImportProvider
 from coriolis.providers.base import BaseReplicaImportValidationProvider
 from coriolis.providers.base import BaseUpdateDestinationReplicaProvider
 from coriolis.tests.integration import utils as test_utils
+from coriolis import utils as coriolis_utils
 
 LOG = logging.getLogger(__name__)
 
@@ -38,7 +40,8 @@ class TestImportProvider(
         BaseEndpointStorageProvider,
         BaseUpdateDestinationReplicaProvider,
         BaseReplicaImportProvider,
-        BaseReplicaImportValidationProvider):
+        BaseReplicaImportValidationProvider,
+        BaseDestinationMinionPoolProvider):
     """Destination-side provider backed by a local `scsi_debug` block device.
 
     ``connection_info`` (the destination endpoint's connection info) has the
@@ -144,16 +147,29 @@ class TestImportProvider(
 
     def deploy_replica_target_resources(
             self, ctxt, connection_info, target_environment, volumes_info):
+        result = self._create_minion(
+            "coriolis-writer", connection_info, volumes_info)
+
+        return {
+            "volumes_info": volumes_info,
+            "connection_info": result["backup_writer_connection_info"],
+            "migr_resources": {"container_id": result["container_id"]},
+        }
+
+    def _create_minion(
+            self, name_prefix, connection_info, volumes_info,
+            device_cgroup_rules=None):
         pkey_path = connection_info["pkey_path"]
         dest_devices = [vol["volume_dev"] for vol in volumes_info]
-        container_name = "coriolis-writer-%s" % uuid.uuid4().hex[:8]
+        container_name = "%s-%s" % (name_prefix, uuid.uuid4().hex[:8])
 
-        container_id = test_utils.start_container(
+        container_id = test_utils.run_container(
             test_utils.DATA_MINION_IMAGE,
             container_name,
             is_systemd=True,
             ssh_key=f"{pkey_path}.pub",
             devices=dest_devices,
+            device_cgroup_rules=device_cgroup_rules,
         )
 
         try:
@@ -172,15 +188,15 @@ class TestImportProvider(
             writer_conn_details = bootstrapper.setup_writer()
 
             return {
-                "volumes_info": volumes_info,
-                "connection_info": {
+                "container_id": container_id,
+                "ssh_connection_info": ssh_conn_info,
+                "backup_writer_connection_info": {
                     "backend": "http_backup_writer",
                     "connection_details": writer_conn_details,
                 },
-                "migr_resources": {"container_id": container_id},
             }
         except Exception:
-            test_utils.stop_container(container_id)
+            test_utils.remove_container(container_id)
             raise
 
     def delete_replica_target_resources(
@@ -188,7 +204,7 @@ class TestImportProvider(
             migr_resources_dict):
         container_id = (migr_resources_dict or {}).get("container_id")
         if container_id:
-            test_utils.stop_container(container_id)
+            test_utils.remove_container(container_id)
 
     def delete_replica_disks(
             self, ctxt, connection_info, target_environment, volumes_info):
@@ -265,3 +281,119 @@ class TestImportProvider(
     def validate_replica_deployment_input(
             self, ctxt, connection_info, target_environment, export_info):
         return {}
+
+    # BaseDestinationMinionPoolProvider
+
+    def get_minion_pool_environment_schema(self):
+        return self.get_target_environment_schema()
+
+    def get_minion_pool_options(
+            self, ctxt, connection_info, env=None, option_names=None):
+        return self.get_target_environment_options(
+            ctxt, connection_info, env, option_names)
+
+    def validate_minion_compatibility_for_transfer(
+            self, ctxt, connection_info, export_info, environment_options,
+            minion_properties):
+        pass
+
+    def validate_minion_pool_environment_options(
+            self, ctxt, connection_info, environment_options):
+        pass
+
+    def set_up_pool_shared_resources(
+            self, ctxt, connection_info, environment_options, pool_identifier):
+        return {}
+
+    def tear_down_pool_shared_resources(
+            self, ctxt, connection_info, environment_options,
+            pool_shared_resources):
+        pass
+
+    def create_minion(
+            self, ctxt, connection_info, environment_options, pool_identifier,
+            pool_os_type, pool_shared_resources, new_minion_identifier):
+        # Devices are hotplugged after container creation via mknod / nsenter.
+        # We must pre-authorize all block devices through the
+        # --device-cgroup-rule option, otherwise any device added will be
+        # inaccessible ("operation not permitted" error on open).
+        result = self._create_minion(
+            "coriolis-pool-minion", connection_info, [],
+            device_cgroup_rules=["b *:* rwm"])
+
+        backup_writer_conn_info = result["backup_writer_connection_info"]
+        return {
+            "connection_info": result["ssh_connection_info"],
+            "backup_writer_connection_info": backup_writer_conn_info,
+            "minion_provider_properties": {
+                "container_id": result["container_id"],
+            },
+        }
+
+    def delete_minion(self, ctxt, connection_info, minion_properties):
+        container_id = (minion_properties or {}).get("container_id")
+        if container_id:
+            test_utils.remove_container(container_id)
+
+    def shutdown_minion(self, ctxt, connection_info, minion_properties):
+        container_id = (minion_properties or {}).get("container_id")
+        if container_id:
+            test_utils.stop_container(container_id)
+
+    def start_minion(self, ctxt, connection_info, minion_properties):
+        container_id = (minion_properties or {}).get("container_id")
+        if container_id:
+            test_utils.start_container(container_id)
+
+    def attach_volumes_to_minion(
+            self, ctxt, connection_info, minion_properties,
+            minion_connection_info, volumes_info):
+        container_id = minion_properties["container_id"]
+        for vol in volumes_info:
+            device_path = vol["volume_dev"]
+            test_utils.hotplug_device_to_container(container_id, device_path)
+
+        return {
+            "minion_properties": minion_properties,
+            "volumes_info": volumes_info,
+        }
+
+    def detach_volumes_from_minion(
+            self, ctxt, connection_info, minion_properties,
+            minion_connection_info, volumes_info):
+        container_id = (minion_properties or {}).get("container_id")
+        if not container_id:
+            return
+
+        for vol in (volumes_info or []):
+            dev_path = vol.get("volume_dev")
+            if not dev_path:
+                continue
+
+            test_utils.unplug_device_from_container(container_id, dev_path)
+
+        return {
+            "minion_properties": minion_properties,
+            "volumes_info": volumes_info,
+        }
+
+    def healthcheck_minion(
+            self, ctxt, connection_info, minion_properties,
+            minion_connection_info):
+        ip = minion_connection_info.get("ip")
+        port = minion_connection_info.get("port", 22)
+        username = minion_connection_info.get("username", "root")
+        pkey = minion_connection_info.get("pkey")
+
+        client = coriolis_utils.connect_ssh(ip, port, username, pkey=pkey)
+        client.close()
+
+    def validate_osmorphing_minion_compatibility_for_transfer(
+            self, ctxt, connection_info, export_info, environment_options,
+            minion_properties):
+        pass
+
+    def get_additional_os_morphing_info(
+            self, ctxt, connection_info, target_environment,
+            instance_deployment_info):
+        return {}

coriolis/tests/integration/utils.py

@@ -182,7 +182,12 @@ def wait_for_ssh(host, port, username, pkey_path, timeout=30):
 # Docker utils
 
 
-def _start_container(image, name, extra_args=None):
+def start_container(container_id):
+    """Start a stopped Docker container."""
+    _run(["docker", "start", container_id], check=False)
+
+
+def _run_container(image, name, extra_args=None):
     cmd = ["docker", "run", "--detach", "--name", name]
     if extra_args:
         cmd.extend(extra_args)
@@ -191,9 +196,9 @@ def _start_container(image, name, extra_args=None):
     return result.stdout.decode().strip()
 
 
-def start_container(
+def run_container(
     image, name, is_systemd=False, ssh_key=None, volumes=None, devices=None,
-    extra_args=None,
+    device_cgroup_rules=None, extra_args=None,
 ):
     """Start a detached Docker container and return its container ID.
 
@@ -204,11 +209,15 @@ def start_container(
     :param ssh_key: SSH key to add as a volume to the authorized_keys.
     :param volumes: List of volumes to attach to the container.
     :param devices: List of devices to attach to the container.
+    :param device_cgroup_rules: List of device cgroup rules (e.g.
+      ``["b *:* rwm"]``). This is needed for device hotplug after container
+      creation.
     :param extra_args: Optional list of extra ``docker run`` arguments.
     :returns: container ID string (stripped).
     """
     volumes = volumes or []
     devices = devices or []
+    device_cgroup_rules = device_cgroup_rules or []
     extra_args = extra_args or []
     sec_opts = []
     caps = []
@@ -228,21 +237,29 @@ def start_container(
     for device in devices:
         extra_args += ["--device", f"{device}:{device}"]
 
+    for rule in device_cgroup_rules:
+        extra_args += ["--device-cgroup-rule", rule]
+
     for cap in caps:
         extra_args += ["--cap-add", cap]
 
     for sec_opt in sec_opts:
         extra_args += ["--security-opt", sec_opt]
 
-    return _start_container(image, name, extra_args)
+    return _run_container(image, name, extra_args)
 
 
 def stop_container(container_id):
+    """Stop a Docker container."""
+    _run(["docker", "stop", "--time", "5", container_id], check=False)
+
+
+def remove_container(container_id):
     """Stop and remove a Docker container, ignoring errors.
 
     :param container_id: container ID or name to stop / remove.
     """
-    _run(["docker", "stop", "--time", "5", container_id], check=False)
+    stop_container(container_id)
     _run(["docker", "rm", "--force", container_id], check=False)
 
 
@@ -257,3 +274,32 @@ def get_container_ip(container_id):
          "{{.NetworkSettings.IPAddress}}",
          container_id])
     return result.stdout.decode().strip()
+
+
+def _get_container_pid(container_id):
+    """Return the host PID of the init process of *container_id*."""
+    result = _run(
+        ["docker", "inspect", "--format", "{{.State.Pid}}", container_id])
+    return int(result.stdout.decode().strip())
+
+
+def hotplug_device_to_container(container_id, device_path):
+    """Create a device node for *device_path* in *container_id*'s namespace."""
+    pid = _get_container_pid(container_id)
+    stat_result = os.stat(device_path)
+    major = os.major(stat_result.st_rdev)
+    minor = os.minor(stat_result.st_rdev)
+
+    _run([
+        "nsenter", "--target", str(pid), "--mount", "--",
+        "mknod", device_path, "b", str(major), str(minor),
+    ])
+
+
+def unplug_device_from_container(container_id, device_path):
+    """Remove a device node from *container_id*'s mount namespace."""
+    pid = _get_container_pid(container_id)
+    _run([
+        "nsenter", "--target", str(pid), "--mount", "--",
+        "rm", "-f", device_path,
+    ], check=False)