daemon/containerd/migration/migration.go

package migration

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"strings"
	"sync"
	"time"

	"github.com/containerd/containerd/v2/core/content"
	"github.com/containerd/containerd/v2/core/images"
	"github.com/containerd/containerd/v2/core/leases"
	"github.com/containerd/containerd/v2/core/mount"
	"github.com/containerd/containerd/v2/core/snapshots"
	"github.com/containerd/containerd/v2/pkg/archive/compression"
	"github.com/containerd/continuity/fs"
	cerrdefs "github.com/containerd/errdefs"
	"github.com/containerd/log"
	"github.com/moby/moby/v2/daemon/internal/image"
	"github.com/moby/moby/v2/daemon/internal/layer"
	refstore "github.com/moby/moby/v2/daemon/internal/refstore"
	"github.com/opencontainers/go-digest"
	"github.com/opencontainers/image-spec/identity"
	"github.com/opencontainers/image-spec/specs-go"
	ocispec "github.com/opencontainers/image-spec/specs-go/v1"
	"golang.org/x/sync/errgroup"
)

type LayerMigrator struct {
	layers  layer.Store
	refs    refstore.Store
	dis     image.Store
	leases  leases.Manager
	content content.Store
	cis     images.Store
}

type Config struct {
	ImageCount       int
	LayerStore       layer.Store
	ReferenceStore   refstore.Store
	DockerImageStore image.Store
	Leases           leases.Manager
	Content          content.Store
	ImageStore       images.Store
}

func NewLayerMigrator(config Config) *LayerMigrator {
	return &LayerMigrator{
		layers:  config.LayerStore,
		refs:    config.ReferenceStore,
		dis:     config.DockerImageStore,
		leases:  config.Leases,
		content: config.Content,
		cis:     config.ImageStore,
	}
}

// MigrateTocontainerd migrates containers from overlay2 to overlayfs or vfs to native
func (lm *LayerMigrator) MigrateTocontainerd(ctx context.Context, snKey string, sn snapshots.Snapshotter) error {
	if sn == nil {
		return fmt.Errorf("no snapshotter to migrate to: %w", cerrdefs.ErrNotImplemented)
	}

	switch driver := lm.layers.DriverName(); driver {
	case "overlay2":
	case "vfs":
	default:
		return fmt.Errorf("%q not supported for migration: %w", driver, cerrdefs.ErrNotImplemented)
	}

	var (
		// Zstd makes migration 10x faster
		// TODO: make configurable
		layerMediaType   = ocispec.MediaTypeImageLayerZstd
		layerCompression = compression.Zstd
	)

	l, err := lm.leases.Create(ctx, leases.WithRandomID(), leases.WithExpiration(24*time.Hour))
	if err != nil {
		return err
	}
	defer func() {
		lm.leases.Delete(ctx, l)
	}()
	ctx = leases.WithLease(ctx, l.ID)

	for imgID, img := range lm.dis.Heads() {
		diffids := img.RootFS.DiffIDs
		if len(diffids) == 0 {
			continue
		}
		var (
			parent   string
			manifest = ocispec.Manifest{
				MediaType: ocispec.MediaTypeImageManifest,
				Versioned: specs.Versioned{
					SchemaVersion: 2,
				},
				Layers: make([]ocispec.Descriptor, len(diffids)),
			}
			ml        sync.Mutex
			eg, egctx = errgroup.WithContext(ctx)
		)
		for i := range diffids {
			chainID := identity.ChainID(diffids[:i+1])
			l, err := lm.layers.Get(chainID)
			if err != nil {
				return fmt.Errorf("failed to get layer [%d] %q: %w", i, chainID, err)
			}
			layerIndex := i
			eg.Go(func() error {
				ctx := egctx
				t1 := time.Now()
				ts, err := l.TarStream()
				if err != nil {
					return err
				}

				desc := ocispec.Descriptor{
					MediaType: layerMediaType,
				}

				cw, err := lm.content.Writer(ctx,
					content.WithRef(fmt.Sprintf("ingest-%s", chainID)),
					content.WithDescriptor(desc))
				if err != nil {
					return fmt.Errorf("failed to get content writer: %w", err)
				}

				dgstr := digest.Canonical.Digester()
				cs, _ := compression.CompressStream(io.MultiWriter(cw, dgstr.Hash()), layerCompression)
				_, err = io.Copy(cs, ts)
				if err != nil {
					return fmt.Errorf("failed to copy to compressed stream: %w", err)
				}
				cs.Close()

				status, err := cw.Status()
				if err != nil {
					return err
				}

				desc.Size = status.Offset
				desc.Digest = dgstr.Digest()

				if err := cw.Commit(ctx, desc.Size, desc.Digest); err != nil && !cerrdefs.IsAlreadyExists(err) {
					return err
				}

				log.G(ctx).WithFields(log.Fields{
					"t":      time.Since(t1),
					"size":   desc.Size,
					"digest": desc.Digest,
				}).Debug("Converted layer to content tar")

				ml.Lock()
				manifest.Layers[layerIndex] = desc
				ml.Unlock()
				return nil
			})

			metadata, err := l.Metadata()
			if err != nil {
				return err
			}
			src, ok := metadata["UpperDir"]
			if !ok {
				src, ok = metadata["SourceDir"]
				if !ok {
					log.G(ctx).WithField("metadata", metadata).WithField("driver", lm.layers.DriverName()).Debug("no source directory metadata")
					return fmt.Errorf("graphdriver not supported: %w", cerrdefs.ErrNotImplemented)
				}
			}
			log.G(ctx).WithField("metadata", metadata).Debugf("migrating %s from %s", chainID, src)

			active := fmt.Sprintf("migration-%s", chainID)

			key := chainID.String()

			snapshotLabels := map[string]string{
				"containerd.io/snapshot.ref": key,
			}
			mounts, err := sn.Prepare(ctx, active, parent, snapshots.WithLabels(snapshotLabels))
			parent = key
			if err != nil {
				if cerrdefs.IsAlreadyExists(err) {
					continue
				}
				return err
			}

			dst, err := extractSource(mounts)
			if err != nil {
				return err
			}

			t1 := time.Now()
			if err := fs.CopyDir(dst, src); err != nil {
				return err
			}
			log.G(ctx).WithFields(log.Fields{
				"t":   time.Since(t1),
				"key": key,
			}).Debug("Copied layer to snapshot")

			if err := sn.Commit(ctx, key, active); err != nil && !cerrdefs.IsAlreadyExists(err) {
				return err
			}
		}

		configBytes := img.RawJSON()
		digest.FromBytes(configBytes)
		manifest.Config = ocispec.Descriptor{
			MediaType: ocispec.MediaTypeImageConfig,
			Digest:    digest.FromBytes(configBytes),
			Size:      int64(len(configBytes)),
		}

		configLabels := map[string]string{
			fmt.Sprintf("containerd.io/gc.ref.snapshot.%s", snKey): parent,
		}
		if err = content.WriteBlob(ctx, lm.content, "config"+manifest.Config.Digest.String(), bytes.NewReader(configBytes), manifest.Config, content.WithLabels(configLabels)); err != nil && !cerrdefs.IsAlreadyExists(err) {
			return err
		}

		if err := eg.Wait(); err != nil {
			return err
		}

		manifestBytes, err := json.MarshalIndent(manifest, "", "   ")
		if err != nil {
			return err
		}

		manifestDesc := ocispec.Descriptor{
			MediaType: manifest.MediaType,
			Digest:    digest.FromBytes(manifestBytes),
			Size:      int64(len(manifestBytes)),
		}

		manifestLabels := map[string]string{
			"containerd.io/gc.ref.content.config": manifest.Config.Digest.String(),
		}
		for i := range manifest.Layers {
			manifestLabels[fmt.Sprintf("containerd.io/gc.ref.content.%d", i)] = manifest.Layers[i].Digest.String()
		}

		if err = content.WriteBlob(ctx, lm.content, "manifest"+manifestDesc.Digest.String(), bytes.NewReader(manifestBytes), manifestDesc, content.WithLabels(manifestLabels)); err != nil && !cerrdefs.IsAlreadyExists(err) {
			return err
		}

		childrenHandler := images.ChildrenHandler(lm.content)
		childrenHandler = images.SetChildrenMappedLabels(lm.content, childrenHandler, nil)
		if err = images.Walk(ctx, childrenHandler, manifestDesc); err != nil {
			return err
		}

		var added bool
		for _, named := range lm.refs.References(digest.Digest(imgID)) {
			img := images.Image{
				Name:   named.String(),
				Target: manifestDesc,
				// TODO: Any labels?
			}
			img, err = lm.cis.Create(ctx, img)
			if err != nil && !cerrdefs.IsAlreadyExists(err) {
				return err
			} else if err != nil {
				log.G(ctx).Infof("Tag already exists: %s", named)
				continue
			}

			log.G(ctx).Infof("Migrated image %s to %s", img.Name, img.Target.Digest)
			added = true
		}

		if !added {
			img := images.Image{
				Name:   "moby-dangling@" + manifestDesc.Digest.String(),
				Target: manifestDesc,
				// TODO: Any labels?
			}
			img, err = lm.cis.Create(ctx, img)
			if err != nil && !cerrdefs.IsAlreadyExists(err) {
				return err
			} else if err == nil {
				log.G(ctx).Infof("Migrated image %s to %s", img.Name, img.Target.Digest)
			}
		}
	}

	return nil
}

func extractSource(mounts []mount.Mount) (string, error) {
	if len(mounts) != 1 {
		return "", fmt.Errorf("cannot support snapshotters with multiple mount sources: %w", cerrdefs.ErrNotImplemented)
	}
	switch mounts[0].Type {
	case "bind":
		return mounts[0].Source, nil
	case "overlay":
		for _, option := range mounts[0].Options {
			if strings.HasPrefix(option, "upperdir=") {
				return option[9:], nil
			}
		}
	default:
		return "", fmt.Errorf("mount type %q not supported: %w", mounts[0].Type, cerrdefs.ErrNotImplemented)
	}

	return "", fmt.Errorf("mount is missing upper option: %w", cerrdefs.ErrNotImplemented)
}