containerdstore: Initial push support

daemon/containerdstore/prune.go

@@ -0,0 +1,98 @@
+package containerdstore
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/containerd/containerd/content"
+	containerdimages "github.com/containerd/containerd/images"
+	"github.com/containerd/containerd/platforms"
+	"github.com/docker/docker/api/types"
+	"github.com/docker/docker/api/types/filters"
+	"github.com/opencontainers/go-digest"
+	ocispec "github.com/opencontainers/image-spec/specs-go/v1"
+	"github.com/pkg/errors"
+)
+
+// TODO: handle pruneFilters
+func (cs *containerdStore) ImagesPrune(ctx context.Context, pruneFilters filters.Args) (*types.ImagesPruneReport, error) {
+	is := cs.client.ImageService()
+	store := cs.client.ContentStore()
+
+	images, err := is.List(ctx)
+	if err != nil {
+		return nil, errors.Wrapf(err, "Failed to list images")
+	}
+
+	platform := platforms.DefaultStrict()
+	report := types.ImagesPruneReport{}
+	toDelete := map[digest.Digest]uint64{}
+	errs := []error{}
+
+	for _, img := range images {
+		err := getContentDigestsWithSizes(ctx, img, store, platform, toDelete)
+		if err != nil {
+			errs = append(errs, err)
+			continue
+		}
+
+		err = is.Delete(ctx, img.Name, containerdimages.SynchronousDelete())
+		if err != nil {
+			errs = append(errs, err)
+			continue
+		}
+
+		report.ImagesDeleted = append(report.ImagesDeleted,
+			types.ImageDeleteResponseItem{
+				Untagged: img.Name,
+			},
+		)
+	}
+
+	for digest, size := range toDelete {
+		err := store.Delete(ctx, digest)
+		if err != nil {
+			errs = append(errs, errors.Wrapf(err, "Failed to delete %s", digest.String()))
+		}
+		report.SpaceReclaimed += size
+		report.ImagesDeleted = append(report.ImagesDeleted,
+			types.ImageDeleteResponseItem{
+				Deleted: digest.String(),
+			},
+		)
+	}
+
+	if len(errs) > 1 {
+		return &report, MultipleErrors{all: errs}
+	} else if len(errs) == 1 {
+		return &report, errs[0]
+	}
+
+	return &report, nil
+}
+
+func getContentDigestsWithSizes(ctx context.Context, img containerdimages.Image, store content.Store, platform platforms.MatchComparer, toDelete map[digest.Digest]uint64) error {
+	return containerdimages.Walk(ctx, containerdimages.Handlers(containerdimages.HandlerFunc(func(ctx context.Context, desc ocispec.Descriptor) ([]ocispec.Descriptor, error) {
+		if desc.Size < 0 {
+			return nil, fmt.Errorf("invalid size %v in %v (%v)", desc.Size, desc.Digest, desc.MediaType)
+		}
+		toDelete[desc.Digest] = uint64(desc.Size)
+		return nil, nil
+	}), containerdimages.LimitManifests(containerdimages.FilterPlatforms(containerdimages.ChildrenHandler(store), platform), platform, 1)), img.Target)
+}
+
+type MultipleErrors struct {
+	all []error
+}
+
+func (m MultipleErrors) Error() string {
+	errString := ""
+	for _, err := range m.all {
+		if len(m.all) > 0 {
+			errString += "\n"
+		}
+		errString += err.Error()
+	}
+
+	return errString
+}

daemon/containerdstore/service.go

@@ -15,6 +15,7 @@ import (
 	"github.com/containerd/containerd/images/archive"
 	"github.com/containerd/containerd/log"
 	"github.com/containerd/containerd/remotes"
+	"github.com/containerd/containerd/remotes/docker"
 	containertypes "github.com/docker/docker/api/types/container"
 	"github.com/docker/docker/pkg/progress"
 	"github.com/docker/docker/pkg/streamformatter"
@@ -376,10 +377,6 @@ func (cs *containerdStore) ImageHistory(ctx context.Context, name string) ([]*im
 	panic("not implemented")
 }
 
-func (cs *containerdStore) ImagesPrune(ctx context.Context, pruneFilters filters.Args) (*types.ImagesPruneReport, error) {
-	panic("not implemented")
-}
-
 func (cs *containerdStore) ImportImage(ctx context.Context, src string, repository string, platform *ocispec.Platform, tag string, msg string, inConfig io.ReadCloser, outStream io.Writer, changes []string) error {
 	panic("not implemented")
 }
@@ -396,7 +393,82 @@ func (cs *containerdStore) LookupImage(ctx context.Context, name string) (*types
 }
 
 func (cs *containerdStore) PushImage(ctx context.Context, image, tag string, metaHeaders map[string][]string, authConfig *types.AuthConfig, outStream io.Writer) error {
-	panic("not implemented")
+	ref, err := reference.ParseNormalizedNamed(image)
+	if err != nil {
+		return err
+	}
+	if tag != "" {
+		// Push by digest is not supported, so only tags are supported.
+		ref, err = reference.WithTag(ref, tag)
+		if err != nil {
+			return err
+		}
+	}
+
+	is := cs.client.ImageService()
+
+	img, err := is.Get(ctx, ref.String())
+	if err != nil {
+		return errors.Wrap(err, "Failed to get image")
+	}
+
+	platformMatcher := platforms.DefaultStrict()
+
+	target := img.Target
+
+	// If the image is an index/manifest list, then push only default platform.
+	// We don't want to push other platforms, because pull only fetches the default platform
+	// and manifest list includes other platforms, which we or the remote repository may not have.
+	if containerdimages.IsIndexType(img.Target.MediaType) {
+		children, err := containerdimages.Children(ctx, cs.client.ContentStore(), img.Target)
+		if err != nil {
+			return err
+		}
+
+		for _, child := range children {
+			if child.Platform == nil {
+				continue
+			}
+
+			if platformMatcher.Match(*child.Platform) {
+				target = child
+			}
+		}
+	}
+
+	imageHandler := containerdimages.HandlerFunc(func(ctx context.Context, desc ocispec.Descriptor) (subdescs []ocispec.Descriptor, err error) {
+		logrus.WithField("desc", desc).Debug("Push, handle digest")
+		return nil, nil
+	})
+	imageHandler = remotes.SkipNonDistributableBlobs(imageHandler)
+
+	resolver := newResolverFromAuthConfig(authConfig)
+
+	log.G(ctx).WithField("desc", target).WithField("ref", ref.String()).Info("Pushing desc to remote ref")
+	return cs.client.Push(ctx, ref.String(), target,
+		containerd.WithResolver(resolver),
+		containerd.WithPlatformMatcher(platformMatcher),
+		containerd.WithImageHandler(imageHandler),
+	)
+}
+
+func newResolverFromAuthConfig(authConfig *types.AuthConfig) remotes.Resolver {
+	opts := []docker.RegistryOpt{}
+
+	if authConfig != nil {
+		authorizer := docker.NewDockerAuthorizer(docker.WithAuthCreds(func(s string) (string, string, error) {
+			if authConfig.IdentityToken != "" {
+				return "", authConfig.IdentityToken, nil
+			}
+			return authConfig.Username, authConfig.Password, nil
+		}))
+
+		opts = append(opts, docker.WithAuthorizer(authorizer))
+	}
+
+	return docker.NewResolver(docker.ResolverOptions{
+		Hosts: docker.ConfigureDefaultRegistries(opts...),
+	})
 }
 
 func (cs *containerdStore) SearchRegistryForImages(ctx context.Context, searchFilters filters.Args, term string, limit int, authConfig *types.AuthConfig, metaHeaders map[string][]string) (*registrytypes.SearchResults, error) {