use crate::error::{MigrationError, Result};
use crate::helper_image::HELPER_OBJECT_PREFIX;
use crate::model::{
ContainerMount, ContainerPlan, ContainerSpec, MigrationPlan, NetworkModeSpec, PortPublish,
SourceConfig,
};
use crate::progress::{MigrationProgress, MigrationStage};
use crate::runner::{CreateNetworkOptions, DockerCliRunner};
use std::collections::{BTreeSet, HashMap};
#[derive(Debug, Clone, Copy, Default)]
pub struct MigrationExecutorOptions {
pub confirm_replace: bool,
pub confirm_stop_source_containers: bool,
pub start_containers: bool,
}
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct MigrationOutcome {
pub warnings: Vec<String>,
}
#[derive(Debug, Clone)]
pub struct MigrationExecutor {
target: DockerCliRunner,
}
impl MigrationExecutor {
#[must_use]
pub fn new(target: DockerCliRunner) -> Self {
Self { target }
}
pub async fn execute<F>(
&self,
source: SourceConfig,
plan: &MigrationPlan,
options: MigrationExecutorOptions,
mut progress: F,
) -> Result<MigrationOutcome>
where
F: FnMut(MigrationProgress),
{
if !plan.unsupported_resources.is_empty() {
return Err(MigrationError::Blocked(format!(
"unsupported resources: {}",
plan.unsupported_resources.join(", ")
)));
}
if !plan.replacements.is_empty() && !options.confirm_replace {
return Err(MigrationError::Blocked(
"replace confirmation is required".to_string(),
));
}
if !plan.blockers.is_empty() && !options.confirm_stop_source_containers {
return Err(MigrationError::Blocked(
"stopping source containers is required".to_string(),
));
}
let source_runner = DockerCliRunner::new(source.socket_path)?;
stop_source_blockers(&source_runner, plan, &mut progress).await?;
remove_target_conflicts(&self.target, plan, &mut progress).await?;
source_runner.ensure_helper_image().await?;
self.target.ensure_helper_image().await?;
let image_rewrites =
import_images(&source_runner, &self.target, plan, &mut progress).await?;
import_volumes(&source_runner, &self.target, plan, &mut progress).await?;
recreate_networks(&self.target, plan, &mut progress).await?;
let (created, mut warnings) =
recreate_containers(&self.target, plan, &image_rewrites, &mut progress).await;
if options.start_containers {
warnings.extend(start_containers(&self.target, plan, &created, &mut progress).await);
}
Ok(MigrationOutcome { warnings })
}
}
async fn stop_source_blockers<F>(
source: &DockerCliRunner,
plan: &MigrationPlan,
progress: &mut F,
) -> Result<()>
where
F: FnMut(MigrationProgress),
{
let mut containers = plan
.blockers
.iter()
.flat_map(|blocker| blocker.containers.clone())
.collect::<Vec<_>>();
containers.sort();
containers.dedup();
let total = u32::try_from(containers.len()).unwrap_or(0);
for (index, container) in containers.into_iter().enumerate() {
progress(MigrationProgress {
stage: MigrationStage::StopSourceContainers,
detail: format!("stopping source container '{container}'"),
resource_type: Some("container".to_string()),
resource_name: Some(container.clone()),
current: Some(u32::try_from(index + 1).unwrap_or(total)),
total: Some(total),
});
source.stop_container(&container).await?;
}
Ok(())
}
async fn remove_target_conflicts<F>(
target: &DockerCliRunner,
plan: &MigrationPlan,
progress: &mut F,
) -> Result<()>
where
F: FnMut(MigrationProgress),
{
for container in &plan.replacements.containers {
progress(MigrationProgress {
stage: MigrationStage::Cleanup,
detail: format!("removing target container '{container}'"),
resource_type: Some("container".to_string()),
resource_name: Some(container.clone()),
current: None,
total: None,
});
target.remove_container(container).await?;
}
for network in &plan.replacements.networks {
progress(MigrationProgress {
stage: MigrationStage::Cleanup,
detail: format!("removing target network '{network}'"),
resource_type: Some("network".to_string()),
resource_name: Some(network.clone()),
current: None,
total: None,
});
target.remove_network(network).await?;
}
for volume in &plan.replacements.volumes {
progress(MigrationProgress {
stage: MigrationStage::Cleanup,
detail: format!("removing target volume '{volume}'"),
resource_type: Some("volume".to_string()),
resource_name: Some(volume.clone()),
current: None,
total: None,
});
target.remove_volume(volume).await?;
}
Ok(())
}
async fn import_images<F>(
source: &DockerCliRunner,
target: &DockerCliRunner,
plan: &MigrationPlan,
progress: &mut F,
) -> Result<HashMap<String, String>>
where
F: FnMut(MigrationProgress),
{
let mut rewrites = HashMap::new();
let total = u32::try_from(plan.images.len()).unwrap_or(0);
for (index, image) in plan.images.iter().enumerate() {
progress(MigrationProgress {
stage: MigrationStage::ImportImages,
detail: format!("importing image '{}'", image.primary_reference()),
resource_type: Some("image".to_string()),
resource_name: Some(image.primary_reference().to_string()),
current: Some(u32::try_from(index + 1).unwrap_or(total)),
total: Some(total),
});
let transfer = source
.pipe_save_into(target, &image.export_references)
.await?;
if !image.repo_tags.is_empty() {
continue;
}
let assigned = transfer.loaded_image_id.ok_or_else(|| {
MigrationError::Docker(format!(
"docker load reported no image ID for untagged image '{}'",
image.image_id
))
})?;
rewrites.insert(image.primary_reference().to_string(), assigned);
}
Ok(rewrites)
}
async fn import_volumes<F>(
source: &DockerCliRunner,
target: &DockerCliRunner,
plan: &MigrationPlan,
progress: &mut F,
) -> Result<()>
where
F: FnMut(MigrationProgress),
{
let total = u32::try_from(plan.volumes.len()).unwrap_or(0);
for (index, volume) in plan.volumes.iter().enumerate() {
progress(MigrationProgress {
stage: MigrationStage::ImportVolumes,
detail: format!("importing volume '{}'", volume.name),
resource_type: Some("volume".to_string()),
resource_name: Some(volume.name.clone()),
current: Some(u32::try_from(index + 1).unwrap_or(total)),
total: Some(total),
});
target
.create_volume(
&volume.name,
&volume
.labels
.iter()
.map(|(key, value)| (key.clone(), value.clone()))
.collect::<Vec<_>>(),
&volume
.options
.iter()
.map(|(key, value)| (key.clone(), value.clone()))
.collect::<Vec<_>>(),
)
.await?;
let source_helper_name =
format!("{HELPER_OBJECT_PREFIX}src-{}", sanitize_name(&volume.name));
let target_helper_name =
format!("{HELPER_OBJECT_PREFIX}dst-{}", sanitize_name(&volume.name));
source.remove_stale_helper(&source_helper_name).await?;
target.remove_stale_helper(&target_helper_name).await?;
let source_helper = source
.create_helper_container(&source_helper_name, &volume.name)
.await?;
let target_helper = target
.create_helper_container(&target_helper_name, &volume.name)
.await?;
let archive_result = source.copy_from_container(&source_helper, "/volume").await;
let copy_result = match archive_result {
Ok(archive) => {
target
.copy_to_container(archive.path(), &target_helper, "/")
.await
}
Err(err) => Err(err),
};
let cleanup_source = source.remove_container(&source_helper).await;
let cleanup_target = target.remove_container(&target_helper).await;
copy_result?;
cleanup_source?;
cleanup_target?;
}
Ok(())
}
async fn recreate_networks<F>(
target: &DockerCliRunner,
plan: &MigrationPlan,
progress: &mut F,
) -> Result<()>
where
F: FnMut(MigrationProgress),
{
let total = u32::try_from(plan.networks.len()).unwrap_or(0);
for (index, network) in plan.networks.iter().enumerate() {
progress(MigrationProgress {
stage: MigrationStage::RecreateNetworks,
detail: format!("recreating network '{}'", network.name),
resource_type: Some("network".to_string()),
resource_name: Some(network.name.clone()),
current: Some(u32::try_from(index + 1).unwrap_or(total)),
total: Some(total),
});
let ipam = network
.ipam
.iter()
.map(|entry| {
(
entry.subnet.clone(),
entry.gateway.clone(),
entry.ip_range.clone(),
)
})
.collect::<Vec<_>>();
let create_options = CreateNetworkOptions {
internal: network.internal,
enable_ipv6: network.enable_ipv6,
attachable: network.attachable,
labels: network
.labels
.iter()
.map(|(key, value)| (key.clone(), value.clone()))
.collect(),
options: network
.options
.iter()
.map(|(key, value)| (key.clone(), value.clone()))
.collect(),
ipam,
};
target
.create_network(&network.name, &create_options)
.await?;
}
Ok(())
}
async fn recreate_containers<F>(
target: &DockerCliRunner,
plan: &MigrationPlan,
image_rewrites: &HashMap<String, String>,
progress: &mut F,
) -> (BTreeSet<String>, Vec<String>)
where
F: FnMut(MigrationProgress),
{
let mut created = BTreeSet::new();
let mut warnings = Vec::new();
let total = u32::try_from(plan.containers.len()).unwrap_or(0);
for (index, container) in plan.containers.iter().enumerate() {
let current = Some(u32::try_from(index + 1).unwrap_or(total));
let create_args = build_create_args(container, resolve_image(container, image_rewrites));
let detail = match target.create_container(create_args).await {
Ok(container_id) => {
created.insert(container.name.clone());
match attach_extra_networks(target, container, &container_id).await {
Ok(()) => format!("recreated container '{}'", container.name),
Err(error) => {
let warning = format!(
"container '{}' was migrated but could not join every network: {error}",
container.name
);
warnings.push(warning.clone());
warning
}
}
}
Err(error) => {
let warning = format!(
"container '{}' could not be recreated: {error}",
container.name
);
warnings.push(warning.clone());
warning
}
};
progress(MigrationProgress {
stage: MigrationStage::RecreateContainers,
detail,
resource_type: Some("container".to_string()),
resource_name: Some(container.name.clone()),
current,
total: Some(total),
});
}
(created, warnings)
}
async fn attach_extra_networks(
target: &DockerCliRunner,
container: &ContainerPlan,
container_id: &str,
) -> Result<()> {
if container.spec.network_mode.forbids_extra_networks() {
return Ok(());
}
for attachment in &container.extra_networks {
target
.connect_network(&attachment.network, container_id, &attachment.aliases)
.await?;
}
Ok(())
}
async fn start_containers<F>(
target: &DockerCliRunner,
plan: &MigrationPlan,
created: &BTreeSet<String>,
progress: &mut F,
) -> Vec<String>
where
F: FnMut(MigrationProgress),
{
let running = containers_to_start(plan, created);
let mut warnings = Vec::new();
let total = u32::try_from(running.len()).unwrap_or(0);
for (index, container) in running.into_iter().enumerate() {
let current = Some(u32::try_from(index + 1).unwrap_or(total));
let detail = match target.start_container(&container.name).await {
Ok(()) => format!("started container '{}'", container.name),
Err(error) => {
let warning = format!(
"container '{}' was migrated but did not start: {error}",
container.name
);
warnings.push(warning.clone());
warning
}
};
progress(MigrationProgress {
stage: MigrationStage::StartContainers,
detail,
resource_type: Some("container".to_string()),
resource_name: Some(container.name.clone()),
current,
total: Some(total),
});
}
warnings
}
fn containers_to_start<'a>(
plan: &'a MigrationPlan,
created: &BTreeSet<String>,
) -> Vec<&'a ContainerPlan> {
plan.containers
.iter()
.filter(|container| container.was_running && created.contains(&container.name))
.collect()
}
fn resolve_image<'a>(plan: &'a ContainerPlan, rewrites: &'a HashMap<String, String>) -> &'a str {
rewrites
.get(&plan.image_reference)
.map_or(plan.image_reference.as_str(), String::as_str)
}
fn build_create_args(plan: &ContainerPlan, image_reference: &str) -> Vec<String> {
let mut args = vec!["--name".to_string(), plan.name.clone()];
append_container_spec_args(&mut args, &plan.spec);
args.push(image_reference.to_string());
args.extend(final_command(&plan.spec));
args
}
fn append_container_spec_args(args: &mut Vec<String>, spec: &ContainerSpec) {
if let Some(hostname) = &spec.hostname {
args.push("--hostname".to_string());
args.push(hostname.clone());
}
if let Some(domainname) = &spec.domainname {
args.push("--domainname".to_string());
args.push(domainname.clone());
}
if let Some(user) = &spec.user {
args.push("--user".to_string());
args.push(user.clone());
}
for env in &spec.env {
args.push("--env".to_string());
args.push(env.clone());
}
for (key, value) in &spec.labels {
args.push("--label".to_string());
args.push(format!("{key}={value}"));
}
for port in &spec.exposed_ports {
args.push("--expose".to_string());
args.push(port.clone());
}
if spec.tty {
args.push("--tty".to_string());
}
if spec.open_stdin {
args.push("--interactive".to_string());
}
if let Some(working_dir) = &spec.working_dir {
args.push("--workdir".to_string());
args.push(working_dir.clone());
}
if let Some(entrypoint) = spec.entrypoint.first() {
args.push("--entrypoint".to_string());
args.push(entrypoint.clone());
}
for mount in &spec.mounts {
match mount {
ContainerMount::Volume { source, target, rw } => {
args.push("--mount".to_string());
args.push(format!(
"type=volume,src={source},dst={target}{}",
if *rw { "" } else { ",readonly" }
));
}
ContainerMount::Bind { source, target, rw } => {
args.push("--mount".to_string());
args.push(format!(
"type=bind,src={source},dst={target}{}",
if *rw { "" } else { ",readonly" }
));
}
ContainerMount::Tmpfs { target, options } => {
args.push("--tmpfs".to_string());
args.push(if let Some(options) = options {
format!("{target}:{options}")
} else {
target.clone()
});
}
}
}
for publish in &spec.publishes {
args.push("--publish".to_string());
args.push(format_publish(publish));
}
if let Some(restart_policy) = &spec.restart_policy {
args.push("--restart".to_string());
let mut value = restart_policy.name.clone();
if let Some(count) = restart_policy.maximum_retry_count {
if restart_policy.name == "on-failure" {
value.push(':');
value.push_str(&count.to_string());
}
}
args.push(value);
}
if spec.privileged {
args.push("--privileged".to_string());
}
if spec.read_only_rootfs {
args.push("--read-only".to_string());
}
for host in &spec.extra_hosts {
args.push("--add-host".to_string());
args.push(host.clone());
}
if spec.auto_remove {
args.push("--rm".to_string());
}
if let Some(memory) = spec.memory {
args.push("--memory".to_string());
args.push(memory.to_string());
}
if let Some(nano_cpus) = spec.nano_cpus {
args.push("--cpus".to_string());
args.push(format_cpus(nano_cpus));
}
for capability in &spec.cap_add {
args.push("--cap-add".to_string());
args.push(capability.clone());
}
append_network_args(args, &spec.network_mode);
}
fn append_network_args(args: &mut Vec<String>, mode: &NetworkModeSpec) {
let network = match mode {
NetworkModeSpec::Default => return,
NetworkModeSpec::Host => "host",
NetworkModeSpec::None => "none",
NetworkModeSpec::Named(attachment) => {
args.push("--network".to_string());
args.push(attachment.network.clone());
for alias in &attachment.aliases {
args.push("--network-alias".to_string());
args.push(alias.clone());
}
return;
}
};
args.push("--network".to_string());
args.push(network.to_string());
}
fn format_cpus(nano_cpus: i64) -> String {
const NANOS_PER_CPU: i64 = 1_000_000_000;
let whole = nano_cpus / NANOS_PER_CPU;
let fraction = nano_cpus % NANOS_PER_CPU;
if fraction == 0 {
return whole.to_string();
}
let fraction = format!("{:09}", fraction.abs());
format!("{whole}.{}", fraction.trim_end_matches('0'))
}
fn final_command(spec: &ContainerSpec) -> Vec<String> {
let mut command = Vec::new();
if spec.entrypoint.len() > 1 {
command.extend(spec.entrypoint.iter().skip(1).cloned());
}
command.extend(spec.cmd.clone());
command
}
fn format_publish(publish: &PortPublish) -> String {
let mut out = String::new();
if let Some(host_ip) = &publish.host_ip {
if !host_ip.is_empty() {
out.push_str(host_ip);
out.push(':');
}
}
if let Some(host_port) = &publish.host_port {
if !host_port.is_empty() {
out.push_str(host_port);
out.push(':');
}
}
out.push_str(&publish.container_port);
out
}
fn sanitize_name(name: &str) -> String {
name.chars()
.map(|ch| if ch.is_ascii_alphanumeric() { ch } else { '-' })
.collect()
}
#[cfg(test)]
mod tests {
use super::*;
use crate::model::{
ContainerNetworkAttachment, ReplacementSummary, RestartPolicySpec, SourceInfo, SourceKind,
};
fn args_for(spec: &ContainerSpec) -> Vec<String> {
let mut args = Vec::new();
append_container_spec_args(&mut args, spec);
args
}
fn contains_pair(args: &[String], flag: &str, value: &str) -> bool {
args.windows(2).any(|window| window == [flag, value])
}
#[test]
fn final_command_merges_entrypoint_tail_and_cmd() {
let spec = ContainerSpec {
entrypoint: vec!["/bin/sh".into(), "-c".into()],
cmd: vec!["echo hi".into()],
..ContainerSpec::default()
};
assert_eq!(
final_command(&spec),
vec!["-c".to_string(), "echo hi".to_string()]
);
}
#[test]
fn publish_format_handles_host_ip_and_port() {
let publish = PortPublish {
container_port: "5432/tcp".into(),
host_ip: Some("127.0.0.1".into()),
host_port: Some("15432".into()),
};
assert_eq!(format_publish(&publish), "127.0.0.1:15432:5432/tcp");
}
#[test]
fn restart_policy_on_failure_keeps_retry_count() {
let spec = ContainerSpec {
restart_policy: Some(RestartPolicySpec {
name: "on-failure".into(),
maximum_retry_count: Some(5),
}),
..ContainerSpec::default()
};
assert!(contains_pair(&args_for(&spec), "--restart", "on-failure:5"));
}
#[test]
fn default_network_mode_emits_no_network_flag() {
let args = args_for(&ContainerSpec::default());
assert!(!args.iter().any(|arg| arg == "--network"));
}
#[test]
fn host_and_none_network_modes_are_emitted_verbatim() {
for (mode, expected) in [
(NetworkModeSpec::Host, "host"),
(NetworkModeSpec::None, "none"),
] {
let spec = ContainerSpec {
network_mode: mode,
..ContainerSpec::default()
};
assert!(contains_pair(&args_for(&spec), "--network", expected));
}
}
#[test]
fn host_mode_keeps_hostname_and_published_ports() {
let spec = ContainerSpec {
network_mode: NetworkModeSpec::Host,
hostname: Some("api".into()),
publishes: vec![PortPublish {
container_port: "80/tcp".into(),
host_ip: None,
host_port: Some("8080".into()),
}],
..ContainerSpec::default()
};
let args = args_for(&spec);
assert!(contains_pair(&args, "--hostname", "api"));
assert!(contains_pair(&args, "--publish", "8080:80/tcp"));
}
#[test]
fn named_network_carries_its_aliases() {
let spec = ContainerSpec {
network_mode: NetworkModeSpec::Named(ContainerNetworkAttachment {
network: "usernet".into(),
aliases: vec!["api".into()],
}),
..ContainerSpec::default()
};
let args = args_for(&spec);
assert!(contains_pair(&args, "--network", "usernet"));
assert!(contains_pair(&args, "--network-alias", "api"));
}
#[test]
fn resource_limits_are_emitted_when_set() {
let spec = ContainerSpec {
memory: Some(536_870_912),
nano_cpus: Some(1_500_000_000),
cap_add: vec!["NET_ADMIN".into(), "SYS_PTRACE".into()],
..ContainerSpec::default()
};
let args = args_for(&spec);
assert!(contains_pair(&args, "--memory", "536870912"));
assert!(contains_pair(&args, "--cpus", "1.5"));
assert!(contains_pair(&args, "--cap-add", "NET_ADMIN"));
assert!(contains_pair(&args, "--cap-add", "SYS_PTRACE"));
}
#[test]
fn cpu_quotas_round_trip_exactly() {
assert_eq!(format_cpus(1_234_567_890), "1.23456789");
assert_eq!(format_cpus(100_000), "0.0001");
assert_eq!(format_cpus(1_500_000_000), "1.5");
assert_eq!(format_cpus(2_000_000_000), "2");
assert_eq!(format_cpus(1), "0.000000001");
}
#[test]
fn unset_resource_limits_emit_no_flags() {
let args = args_for(&ContainerSpec::default());
for flag in ["--memory", "--cpus", "--cap-add"] {
assert!(
!args.iter().any(|arg| arg == flag),
"{flag} should be absent"
);
}
}
fn container_on(image_reference: &str) -> ContainerPlan {
ContainerPlan {
name: "demo".into(),
id: "cid".into(),
image_reference: image_reference.into(),
spec: ContainerSpec::default(),
extra_networks: Vec::new(),
replace_existing: false,
was_running: false,
created: String::new(),
}
}
#[test]
fn an_untagged_image_reference_is_rewritten_to_the_assigned_id() {
let rewrites = HashMap::from([("sha256:source".to_string(), "sha256:target".to_string())]);
let container = container_on("sha256:source");
assert_eq!(resolve_image(&container, &rewrites), "sha256:target");
assert_eq!(
build_create_args(&container, resolve_image(&container, &rewrites)).last(),
Some(&"sha256:target".to_string())
);
}
#[test]
fn tagged_image_references_pass_through_untouched() {
let rewrites = HashMap::from([("sha256:source".to_string(), "sha256:target".to_string())]);
let container = container_on("myapp:dev");
assert_eq!(resolve_image(&container, &rewrites), "myapp:dev");
}
#[test]
fn only_host_mode_forbids_extra_networks() {
assert!(NetworkModeSpec::Host.forbids_extra_networks());
assert!(!NetworkModeSpec::None.forbids_extra_networks());
assert!(!NetworkModeSpec::Default.forbids_extra_networks());
}
fn named_container(name: &str, was_running: bool) -> ContainerPlan {
ContainerPlan {
name: name.into(),
was_running,
..container_on("alpine:3.21")
}
}
fn plan_of(containers: Vec<ContainerPlan>) -> MigrationPlan {
MigrationPlan {
source: SourceInfo {
kind: SourceKind::OrbStack,
socket_path: std::path::PathBuf::new(),
daemon_name: String::new(),
server_version: String::new(),
operating_system: String::new(),
architecture: String::new(),
},
helper_image: String::new(),
images: Vec::new(),
volumes: Vec::new(),
networks: Vec::new(),
containers,
unsupported_resources: Vec::new(),
warnings: Vec::new(),
replacements: ReplacementSummary::default(),
blockers: Vec::new(),
}
}
#[test]
fn a_container_that_failed_to_recreate_is_not_started() {
let plan = plan_of(vec![
named_container("survived", true),
named_container("failed-create", true),
named_container("was-stopped", false),
]);
let created = BTreeSet::from(["survived".to_string(), "was-stopped".to_string()]);
let names: Vec<_> = containers_to_start(&plan, &created)
.iter()
.map(|container| container.name.as_str())
.collect();
assert_eq!(
names,
vec!["survived"],
"only a container that was both running on the source and recreated here"
);
}
}