package migrate import ( "compress/gzip" "context" "errors" "fmt" "io" "strings" "sync" "time" "github.com/arescom/dockmv/internal/dkr" "github.com/arescom/dockmv/internal/job" "github.com/arescom/dockmv/internal/spec" "github.com/arescom/dockmv/internal/sshx" ) // SSHRunner migrates containers straight from the local daemon to a target // host over one SSH connection. Nothing is written to disk on either side: // tar streams go from the source daemon into a remote `docker cp`. type SSHRunner struct { Src *dkr.Client Dst *sshx.RemoteDocker Containers []spec.Container Volumes []spec.Volume Networks []spec.Network Plan spec.Plan } // Run executes the whole plan, reporting into j. func (r *SSHRunner) Run(ctx context.Context, j *job.Job) error { opts := r.Plan.Options if opts.Parallelism < 1 { opts.Parallelism = 1 } pre, err := r.Dst.Preflight(ctx) if err != nil { return fmt.Errorf("target preflight: %w", err) } for _, p := range pre.Problems { j.Logf(job.LevelWarn, "", "target: %s", p) } if pre.ServerVersion == "" { return errors.New("target host cannot run docker; see the warnings above") } j.Logf(job.LevelInfo, "", "target docker %s (%s/%s), free space on %s: %s", pre.ServerVersion, pre.OS, pre.Arch, pre.DockerRoot, humanBytes(pre.DiskFreeBytes)) compress := opts.Compress && pre.HasGzip if opts.Compress && !pre.HasGzip { j.Logf(job.LevelWarn, "", "gzip missing on target; sending data uncompressed") } if opts.DryRun { j.Logf(job.LevelInfo, "", "dry run: no command below is executed on the target") } byID := map[string]*spec.Container{} for i := range r.Containers { byID[r.Containers[i].ID] = &r.Containers[i] } // Prepare everything up front so a bad selection fails before any change. var prepared []*Prepared for _, sel := range r.Plan.Items { if !sel.Include { continue } p, err := Prepare(byID[sel.ContainerID], sel, r.Volumes, r.Networks) if err != nil { return fmt.Errorf("container %s: %w", sel.ContainerID, err) } prepared = append(prepared, p) } if len(prepared) == 0 { return errors.New("nothing selected to migrate") } // Networks are shared between containers, so they are created once, before // the per-container work fans out. if err := r.ensureNetworks(ctx, j, prepared, opts); err != nil { return err } sem := make(chan struct{}, opts.Parallelism) var wg sync.WaitGroup var mu sync.Mutex var failures int for _, p := range prepared { p := p item := j.AddItem(p.Source.ID, p.Source.Name) for _, n := range p.Source.Warnings { j.AddItemWarning(item, "%s: %s", p.Source.Name, n) } for _, n := range p.Notes { j.AddItemWarning(item, "%s: %s", p.Source.Name, n) } wg.Add(1) go func() { defer wg.Done() select { case sem <- struct{}{}: case <-ctx.Done(): j.SetItemState(item, job.StateCanceled, nil) return } defer func() { <-sem }() err := r.migrateOne(ctx, j, item, p, opts, compress) switch { case err == nil: j.SetItemState(item, job.StateSucceeded, nil) case errors.Is(err, errSkipped): j.SetItemState(item, job.StateSkipped, err) case errors.Is(err, context.Canceled): j.SetItemState(item, job.StateCanceled, err) default: j.SetItemState(item, job.StateFailed, err) j.Logf(job.LevelError, item.ID, "%s: %v", p.Source.Name, err) mu.Lock() failures++ mu.Unlock() } }() } wg.Wait() if ctx.Err() != nil { return context.Canceled } if failures > 0 { return fmt.Errorf("%d of %d containers failed to migrate", failures, len(prepared)) } return nil } var errSkipped = errors.New("skipped") func (r *SSHRunner) migrateOne(ctx context.Context, j *job.Job, item *job.Item, p *Prepared, opts spec.Options, compress bool) error { name := p.ContainerName() // 1. Name conflict on the target. stepConflict := j.AddStep(item, "conflict", "check target for an existing "+name, 0) j.StartStep(stepConflict) newName, action, err := r.resolveConflict(ctx, j, name, opts) if err != nil { j.FinishStep(stepConflict, err) return err } if action == "skip" { j.SkipStep(stepConflict, "container already exists on target") return fmt.Errorf("%w: %s already exists on the target", errSkipped, name) } if newName != name { j.Logf(job.LevelWarn, item.ID, "%s already exists on target; creating %s instead", name, newName) p.Render.NameOverride = newName name = newName } j.FinishStep(stepConflict, nil) // 2. Image. if err := r.ensureImage(ctx, j, item, p, opts, compress); err != nil { return err } // 3. Named volumes. if err := r.ensureVolumes(ctx, j, item, p, opts); err != nil { return err } // 4. Create the container, stopped. Creating it before the data copy is // what makes volumes and bind directories exist with the right identity. stepCreate := j.AddStep(item, "create", "create container "+name, 0) j.StartStep(stepCreate) createArgs := p.Target.CreateArgs(p.Render) j.Logf(job.LevelCmd, item.ID, "target: %s", r.Dst.Cmd(createArgs...)) if !opts.DryRun { if _, err := r.Dst.Run(ctx, createArgs...); err != nil { j.FinishStep(stepCreate, err) return fmt.Errorf("create container on target: %w", err) } } j.FinishStep(stepCreate, nil) for _, args := range p.Target.NetworkConnectArgs(p.Render) { st := j.AddStep(item, "netconnect", "attach "+args[len(args)-2], 0) j.StartStep(st) j.Logf(job.LevelCmd, item.ID, "target: %s", r.Dst.Cmd(args...)) if !opts.DryRun { if _, err := r.Dst.Run(ctx, args...); err != nil { j.FinishStep(st, err) return fmt.Errorf("attach extra network: %w", err) } } j.FinishStep(st, nil) } // 5. Data. The source container is stopped first when asked, so the files // are not changing underneath the copy. if len(p.Transfers) > 0 { restore, err := r.quiesceSource(ctx, j, item, p, opts) if err != nil { return err } copyErr := r.transferAll(ctx, j, item, p, opts, compress, name) if restore != nil { restore() } if copyErr != nil { return copyErr } } else if p.Selection.StopSourceAfter && !opts.DryRun { if err := r.Src.Stop(ctx, p.Source.ID, 30*time.Second); err != nil { j.AddItemWarning(item, "could not stop source container: %v", err) } } // 6. Start. if p.Selection.StartAfter { st := j.AddStep(item, "start", "start "+name+" on target", 0) j.StartStep(st) j.Logf(job.LevelCmd, item.ID, "target: %s", r.Dst.Cmd("start", name)) if !opts.DryRun { if _, err := r.Dst.Run(ctx, "start", name); err != nil { j.FinishStep(st, err) return fmt.Errorf("start container on target: %w", err) } } j.FinishStep(st, nil) } // 7. Verify. if opts.VerifyAfter && !opts.DryRun { st := j.AddStep(item, "verify", "verify "+name+" on target", 0) j.StartStep(st) err := r.verify(ctx, j, item, p, name) j.FinishStep(st, err) if err != nil { return err } } return nil } // quiesceSource stops the source container when the selection asks for a // consistent copy, and returns the function that puts it back the way the // operator wants it afterwards. func (r *SSHRunner) quiesceSource(ctx context.Context, j *job.Job, item *job.Item, p *Prepared, opts spec.Options) (func(), error) { if opts.DryRun { return nil, nil } wasRunning := p.Source.State == "running" if !p.Selection.StopSourceDuringCopy { if wasRunning { j.AddItemWarning(item, "copying %s while it is running; data written during the copy may be inconsistent", p.Source.Name) } return func() { if p.Selection.StopSourceAfter && wasRunning { if err := r.Src.Stop(ctx, p.Source.ID, 30*time.Second); err != nil { j.AddItemWarning(item, "could not stop source container: %v", err) } } }, nil } if wasRunning { st := j.AddStep(item, "quiesce", "stop source "+p.Source.Name, 0) j.StartStep(st) err := r.Src.Stop(ctx, p.Source.ID, 30*time.Second) j.FinishStep(st, err) if err != nil { return nil, fmt.Errorf("stop source container: %w", err) } } return func() { if !wasRunning || p.Selection.StopSourceAfter { return } if err := r.Src.Start(context.WithoutCancel(ctx), p.Source.ID); err != nil { j.AddItemWarning(item, "could not restart source container: %v", err) } else { j.Logf(job.LevelInfo, item.ID, "source container %s restarted", p.Source.Name) } }, nil } func (r *SSHRunner) transferAll(ctx context.Context, j *job.Job, item *job.Item, p *Prepared, opts spec.Options, compress bool, targetName string) error { // Ask the target what it actually created, so read-only mounts can be // seeded through a staging container that mounts the same volume writable. var resolved map[string]targetMount if !opts.DryRun && anyReadOnlyTransfer(p.Transfers) { var err error resolved, err = r.inspectTargetMounts(ctx, targetName) if err != nil { return fmt.Errorf("inspect target mounts: %w", err) } } for i, t := range p.Transfers { st := j.AddStep(item, fmt.Sprintf("data-%d", i), t.Label, t.SizeBytes) j.StartStep(st) if opts.DryRun { j.Logf(job.LevelCmd, item.ID, "target: %s (fed with a tar stream of %s from %s)", r.Dst.Cmd("cp", "-a", "-", targetName+":"+t.RestoreInto), t.SourcePath, p.Source.Name) j.SkipStep(st, "dry run") continue } err := r.transferOne(ctx, j, item, st, p, t, i, opts, compress, targetName, resolved) j.FinishStep(st, err) if err != nil { return err } } return nil } func (r *SSHRunner) transferOne( ctx context.Context, j *job.Job, item *job.Item, st *job.Step, p *Prepared, t Transfer, index int, opts spec.Options, compress bool, targetName string, resolved map[string]targetMount, ) error { // Pick where the archive is extracted. A writable mount is filled through // the real container; a read-only one goes through a staging container that // mounts the same volume or host path writable. cpTarget := targetName extractInto := t.RestoreInto var cleanup func() if t.ReadOnly { tm, ok := resolved[t.Destination] if !ok { return fmt.Errorf("target does not report a mount at %s", t.Destination) } stageName := StagingName(targetName, index) mountAt, into := StagingPaths(t.Destination) var source string switch { case tm.Name != "": source = tm.Name case tm.Source != "": source = tm.Source default: return fmt.Errorf("cannot resolve the storage behind read-only mount %s", t.Destination) } _, _ = r.Dst.Run(ctx, "rm", "-f", stageName) args := []string{"create", "--name", stageName, "--volume", source + ":" + mountAt, p.Target.Image} j.Logf(job.LevelCmd, item.ID, "target: %s", r.Dst.Cmd(args...)) if _, err := r.Dst.Run(ctx, args...); err != nil { return fmt.Errorf("create staging container for read-only mount %s: %w", t.Destination, err) } cleanup = func() { _, _ = r.Dst.Run(context.WithoutCancel(ctx), "rm", "-f", stageName) } cpTarget, extractInto = stageName, into j.Logf(job.LevelInfo, item.ID, "seeding read-only mount %s through staging container %s", t.Destination, stageName) } if cleanup != nil { defer cleanup() } src, err := r.Src.CopyOut(ctx, p.Source.ID, t.SourcePath) if err != nil { return err } defer src.Close() counted := job.NewCountingReader(src, j, st) body := io.Reader(counted) if compress { pr, pw := io.Pipe() go func() { gz, gerr := gzip.NewWriterLevel(pw, gzipLevel(p, opts.CompressLevel)) if gerr != nil { pw.CloseWithError(gerr) return } _, cerr := io.Copy(gz, counted) if closeErr := gz.Close(); cerr == nil { cerr = closeErr } pw.CloseWithError(cerr) }() body = pr } cpArgs := []string{"cp", "-a", "-", cpTarget + ":" + extractInto} j.Logf(job.LevelCmd, item.ID, "target: %s", r.Dst.Cmd(cpArgs...)) if err := r.Dst.Feed(ctx, body, compress, cpArgs...); err != nil { return fmt.Errorf("copy %s: %w", t.Label, err) } counted.Flush() return nil } type targetMount struct { Kind string Name string Source string } // inspectTargetMounts reads back the mounts the target actually created, // which is the only way to learn the generated name of an anonymous volume. func (r *SSHRunner) inspectTargetMounts(ctx context.Context, name string) (map[string]targetMount, error) { const format = `{{range .Mounts}}{{.Type}}` + "\t" + `{{.Name}}` + "\t" + `{{.Source}}` + "\t" + `{{.Destination}}{{"\n"}}{{end}}` out, err := r.Dst.Run(ctx, "inspect", "--format", format, name) if err != nil { return nil, err } res := map[string]targetMount{} for _, line := range strings.Split(out, "\n") { if strings.TrimSpace(line) == "" { continue } f := strings.Split(strings.TrimRight(line, "\r"), "\t") if len(f) != 4 { continue } res[f[3]] = targetMount{Kind: f[0], Name: f[1], Source: f[2]} } return res, nil } func (r *SSHRunner) ensureImage(ctx context.Context, j *job.Job, item *job.Item, p *Prepared, opts spec.Options, compress bool) error { ref := p.Target.Image sel := p.Selection if !sel.MigrateImage || sel.ImageMode == spec.ImageSkip { st := j.AddStep(item, "image", "check image "+ref+" on target", 0) j.StartStep(st) if opts.DryRun { j.SkipStep(st, "dry run") return nil } ok, err := r.Dst.Exists(ctx, "image", ref) j.FinishStep(st, err) if err != nil { return err } if !ok { return fmt.Errorf("image %s is not on the target and image migration is disabled", ref) } return nil } mode := sel.ImageMode if mode == "" { mode = spec.ImageAuto } if mode == spec.ImageAuto && !opts.DryRun { if ok, err := r.Dst.Exists(ctx, "image", ref); err == nil && ok { st := j.AddStep(item, "image", "image "+ref+" already on target", 0) j.StartStep(st) j.SkipStep(st, "already present") return nil } } tryPull := mode == spec.ImagePull || (mode == spec.ImageAuto && looksPullable(ref) && len(r.Src.ImageRepoDigests(ctx, ref)) > 0) if tryPull { st := j.AddStep(item, "image", "pull "+ref+" on target", 0) j.StartStep(st) j.Logf(job.LevelCmd, item.ID, "target: %s", r.Dst.Cmd("pull", ref)) if opts.DryRun { j.SkipStep(st, "dry run") return nil } _, err := r.Dst.Run(ctx, "pull", ref) if err == nil { j.FinishStep(st, nil) return nil } if mode == spec.ImagePull { j.FinishStep(st, err) return fmt.Errorf("pull image on target: %w", err) } j.SkipStep(st, "pull failed, falling back to streaming the image") j.Logf(job.LevelWarn, item.ID, "pull of %s failed on target (%v); streaming layers instead", ref, err) } size := r.Src.ImageSizeBytes(ctx, ref) st := j.AddStep(item, "image", "stream image "+ref, size) j.StartStep(st) j.Logf(job.LevelCmd, item.ID, "target: %s", r.Dst.Cmd("load")) if opts.DryRun { j.SkipStep(st, "dry run") return nil } src, err := r.Src.SaveImage(ctx, ref) if err != nil { j.FinishStep(st, err) return err } defer src.Close() counted := job.NewCountingReader(src, j, st) body := io.Reader(counted) if compress { pr, pw := io.Pipe() go func() { gz, gerr := gzip.NewWriterLevel(pw, gzipLevel(p, opts.CompressLevel)) if gerr != nil { pw.CloseWithError(gerr) return } _, cerr := io.Copy(gz, counted) if closeErr := gz.Close(); cerr == nil { cerr = closeErr } pw.CloseWithError(cerr) }() body = pr } err = r.Dst.Feed(ctx, body, compress, "load") counted.Flush() j.FinishStep(st, err) if err != nil { return fmt.Errorf("load image on target: %w", err) } return nil } func (r *SSHRunner) ensureVolumes(ctx context.Context, j *job.Job, item *job.Item, p *Prepared, opts spec.Options) error { for _, v := range p.Volumes { st := j.AddStep(item, "volume-"+v.Name, "ensure volume "+v.Name, 0) j.StartStep(st) if !opts.DryRun { exists, err := r.Dst.Exists(ctx, "volume", v.Name) if err != nil { j.FinishStep(st, err) return err } if exists { switch opts.Conflict { case spec.ConflictReplace: j.Logf(job.LevelWarn, item.ID, "removing existing volume %s on target", v.Name) if _, err := r.Dst.Run(ctx, "volume", "rm", "-f", v.Name); err != nil { j.FinishStep(st, err) return fmt.Errorf("remove existing volume %s: %w", v.Name, err) } default: // Reusing an existing volume is the safe default: the copy // below writes into it without destroying anything else. j.SkipStep(st, "volume already exists on target and is reused") j.AddItemWarning(item, "volume %s already exists on the target; its current contents will be merged with the copied data", v.Name) continue } } } args := v.CreateArgs() j.Logf(job.LevelCmd, item.ID, "target: %s", r.Dst.Cmd(args...)) if !opts.DryRun { if _, err := r.Dst.Run(ctx, args...); err != nil { j.FinishStep(st, err) return fmt.Errorf("create volume %s: %w", v.Name, err) } } j.FinishStep(st, nil) } return nil } func (r *SSHRunner) ensureNetworks(ctx context.Context, j *job.Job, prepared []*Prepared, opts spec.Options) error { seen := map[string]bool{} for _, p := range prepared { for _, n := range p.Networks { if seen[n.Name] || isBuiltin(n.Name) { continue } seen[n.Name] = true if !opts.DryRun { exists, err := r.Dst.Exists(ctx, "network", n.Name) if err != nil { return err } if exists { j.Logf(job.LevelInfo, "", "network %s already exists on target; reusing it", n.Name) continue } } args := n.CreateArgs() j.Logf(job.LevelCmd, "", "target: %s", r.Dst.Cmd(args...)) if opts.DryRun { continue } if _, err := r.Dst.Run(ctx, args...); err != nil { return fmt.Errorf("create network %s: %w", n.Name, err) } j.Logf(job.LevelInfo, "", "created network %s on target", n.Name) } } return nil } // resolveConflict decides what to do about an existing container on the target // and returns the name to use. func (r *SSHRunner) resolveConflict(ctx context.Context, j *job.Job, name string, opts spec.Options) (string, string, error) { exists, err := r.Dst.Exists(ctx, "container", name) if err != nil { return "", "", err } if !exists { return name, "create", nil } switch opts.Conflict { case spec.ConflictSkip: return name, "skip", nil case spec.ConflictReplace: j.Logf(job.LevelWarn, "", "removing existing container %s on target", name) if opts.DryRun { return name, "create", nil } if _, err := r.Dst.Run(ctx, "rm", "-f", name); err != nil { return "", "", fmt.Errorf("remove existing container %s: %w", name, err) } return name, "create", nil case spec.ConflictRename: suffix := opts.RenameSuffix if suffix == "" { suffix = "-migrated" } candidate := name + suffix for i := 2; ; i++ { ok, err := r.Dst.Exists(ctx, "container", candidate) if err != nil { return "", "", err } if !ok { return candidate, "create", nil } candidate = fmt.Sprintf("%s%s-%d", name, suffix, i) if i > 50 { return "", "", fmt.Errorf("could not find a free name based on %s", name) } } default: return "", "", fmt.Errorf("container %s already exists on the target", name) } } func (r *SSHRunner) verify(ctx context.Context, j *job.Job, item *job.Item, p *Prepared, name string) error { const format = `{{.State.Status}}` + "\t" + `{{.Config.Image}}` + "\t" + `{{len .Mounts}}` out, err := r.Dst.Run(ctx, "inspect", "--format", format, name) if err != nil { return fmt.Errorf("container %s is not inspectable on the target: %w", name, err) } f := strings.Split(strings.TrimSpace(out), "\t") if len(f) != 3 { return fmt.Errorf("unexpected inspect output for %s", name) } status, image, mountCount := f[0], f[1], f[2] wantMounts := 0 for _, m := range p.Target.Mounts { if !p.Render.DropMounts[m.Destination] { wantMounts++ } } if fmt.Sprint(wantMounts) != mountCount { j.AddItemWarning(item, "target container %s reports %s mounts, expected %d", name, mountCount, wantMounts) } if image != p.Target.Image { j.AddItemWarning(item, "target container %s runs image %s, expected %s", name, image, p.Target.Image) } if p.Selection.StartAfter && status != "running" { logs, _ := r.Dst.Run(ctx, "logs", "--tail", "20", name) if s := strings.TrimSpace(logs); s != "" { j.Logf(job.LevelError, item.ID, "last logs from %s:\n%s", name, s) } return fmt.Errorf("container %s did not stay running on the target (status %s)", name, status) } j.Logf(job.LevelInfo, item.ID, "verified %s on target: status=%s image=%s mounts=%s", name, status, image, mountCount) return nil } func anyReadOnlyTransfer(ts []Transfer) bool { for _, t := range ts { if t.ReadOnly { return true } } return false } func gzipLevel(_ *Prepared, level int) int { if level < gzip.BestSpeed || level > gzip.BestCompression { return gzip.BestSpeed } return level } // looksPullable reports whether a reference is a name a registry could serve, // as opposed to a bare image id or a locally built, never-pushed tag. func looksPullable(ref string) bool { if ref == "" || strings.HasPrefix(ref, "sha256:") { return false } if len(ref) == 64 && !strings.ContainsAny(ref, ":/.-_") { return false } return true } func isBuiltin(name string) bool { switch name { case "bridge", "host", "none": return true } return false } func humanBytes(n int64) string { if n <= 0 { return "unknown" } const unit = 1024 if n < unit { return fmt.Sprintf("%d B", n) } div, exp := int64(unit), 0 for v := n / unit; v >= unit; v /= unit { div *= unit exp++ } return fmt.Sprintf("%.1f %ciB", float64(n)/float64(div), "KMGTPE"[exp]) }