733 lines
21 KiB
Go
733 lines
21 KiB
Go
package migrate
|
|
|
|
import (
|
|
"compress/gzip"
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/arescom/docker-migrate/internal/dkr"
|
|
"github.com/arescom/docker-migrate/internal/job"
|
|
"github.com/arescom/docker-migrate/internal/spec"
|
|
"github.com/arescom/docker-migrate/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])
|
|
}
|