c8d: Implement push

This implements `docker push` under containerd image store.  When
pushing manifest lists that reference a content which is not present in
the local content store, it will attempt to perform the cross-repo mount
the content if possible.

Considering this scenario:

```bash
$ docker pull docker.io/library/busybox
```
This will download manifest list and only host platform-specific
manifest and blobs.

Note, tagging to a different repository (but still the same registry) and pushing:
```bash
$ docker tag docker.io/library/busybox docker.io/private-repo/mybusybox
$ docker push docker.io/private-repo/mybusybox
```

will result in error, because the neither we nor the target repository
doesn't have the manifests that the busybox manifest list references
(because manifests can't be cross-repo mounted).

If for some reason the manifests and configs for all other platforms
would be present in the content store, but only layer blobs were
missing, then the push would work, because the blobs can be cross-repo
mounted (only if we push to the same registry).

Signed-off-by: Paweł Gronowski <pawel.gronowski@docker.com>

daemon/containerd/image_push.go

@@ -2,15 +2,255 @@ package containerd
 
 import (
 	"context"
-	"errors"
+	"fmt"
 	"io"
+	"strings"
 
+	"github.com/containerd/containerd/content"
+	cerrdefs "github.com/containerd/containerd/errdefs"
+	"github.com/containerd/containerd/images"
+	containerdimages "github.com/containerd/containerd/images"
+	"github.com/containerd/containerd/platforms"
+	"github.com/containerd/containerd/remotes"
+	"github.com/containerd/containerd/remotes/docker"
 	"github.com/docker/distribution/reference"
 	"github.com/docker/docker/api/types/registry"
 	"github.com/docker/docker/errdefs"
+	"github.com/docker/docker/pkg/streamformatter"
+	"github.com/opencontainers/go-digest"
+	ocispec "github.com/opencontainers/image-spec/specs-go/v1"
+	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
+	"golang.org/x/sync/semaphore"
 )
 
-// PushImage initiates a push operation on the repository named localName.
-func (i *ImageService) PushImage(ctx context.Context, ref reference.NamedTagged, metaHeaders map[string][]string, authConfig *registry.AuthConfig, outStream io.Writer) (outerr error) {
-	return errdefs.NotImplemented(errors.New("not implemented"))
+// PushImage initiates a push operation of the image pointed to by targetRef.
+// Image manifest (or index) is pushed as is, which will probably fail if you
+// don't have all content referenced by the index.
+// Cross-repo mounts will be attempted for non-existing blobs.
+//
+// It will also add distribution source labels to the pushed content
+// pointing to the new target repository. This will allow subsequent pushes
+// to perform cross-repo mounts of the shared content when pushing to a different
+// repository on the same registry.
+func (i *ImageService) PushImage(ctx context.Context, targetRef reference.Named, metaHeaders map[string][]string, authConfig *registry.AuthConfig, outStream io.Writer) error {
+	if _, tagged := targetRef.(reference.Tagged); !tagged {
+		if _, digested := targetRef.(reference.Digested); !digested {
+			return errdefs.NotImplemented(errors.New("push all tags is not implemented"))
+		}
+	}
+
+	leasedCtx, release, err := i.client.WithLease(ctx)
+	if err != nil {
+		return err
+	}
+	defer release(leasedCtx)
+	out := streamformatter.NewJSONProgressOutput(outStream, false)
+
+	img, err := i.client.ImageService().Get(ctx, targetRef.String())
+	if err != nil {
+		return errdefs.NotFound(err)
+	}
+
+	target := img.Target
+	store := i.client.ContentStore()
+
+	resolver, tracker := i.newResolverFromAuthConfig(authConfig)
+	pushProgress := pushProgress{Tracker: tracker}
+	jobs := newJobs()
+	finishProgress := jobs.showProgress(ctx, out, combinedProgress([]progressUpdater{
+		&pushProgress,
+		pullProgress{ShowExists: false, Store: store},
+	}))
+	defer finishProgress()
+
+	var limiter *semaphore.Weighted = nil // TODO: Respect max concurrent downloads/uploads
+
+	mountableBlobs, err := i.findMissingMountable(ctx, store, jobs, target, targetRef, limiter)
+	if err != nil {
+		return err
+	}
+	for dgst := range mountableBlobs {
+		pushProgress.addMountable(dgst)
+	}
+
+	// Create a store which fakes the local existence of possibly mountable blobs.
+	// Otherwise they can't be pushed at all.
+	realStore := store
+	wrapped := wrapWithFakeMountableBlobs(store, mountableBlobs)
+	store = wrapped
+
+	pusher, err := resolver.Pusher(ctx, targetRef.String())
+	if err != nil {
+		return err
+	}
+
+	addChildrenToJobs := containerdimages.HandlerFunc(
+		func(ctx context.Context, desc ocispec.Descriptor) ([]ocispec.Descriptor, error) {
+			children, err := containerdimages.Children(ctx, store, desc)
+			if err != nil {
+				return nil, err
+			}
+			for _, c := range children {
+				jobs.Add(c)
+			}
+
+			jobs.Add(desc)
+
+			return nil, nil
+		},
+	)
+
+	appendSource, err := docker.AppendDistributionSourceLabel(realStore, targetRef.String())
+	if err != nil {
+		// This shouldn't happen at this point because the reference would have to be invalid
+		// and if it was, then it would error out earlier.
+		return errdefs.Unknown(errors.Wrap(err, "failed to create an handler that appends distribution source label to pushed content"))
+	}
+
+	handlerWrapper := func(h images.Handler) images.Handler {
+		return containerdimages.Handlers(addChildrenToJobs, h, appendSource)
+	}
+
+	err = remotes.PushContent(ctx, pusher, target, store, limiter, platforms.All, handlerWrapper)
+	if err != nil {
+		if containerdimages.IsIndexType(target.MediaType) {
+			if cerrdefs.IsNotFound(err) {
+				err = errdefs.NotFound(fmt.Errorf(
+					"missing content: %w\n"+
+						"Note: You're trying to push a manifest list/index which "+
+						"references multiple platform specific manifests, but not all of them are available locally "+
+						"or available to the remote repository.\n"+
+						"Make sure you have all the referenced content and try again.",
+					err))
+			}
+		}
+	}
+	return err
+}
+
+// findMissingMountable will walk the target descriptor recursively and return
+// missing contents with their distribution source which could potentially
+// be cross-repo mounted.
+func (i *ImageService) findMissingMountable(ctx context.Context, store content.Store, jobs *jobs,
+	target ocispec.Descriptor, targetRef reference.Named, limiter *semaphore.Weighted,
+) (map[digest.Digest]distributionSource, error) {
+	mountableBlobs := map[digest.Digest]distributionSource{}
+	sources, err := getDigestSources(ctx, store, target.Digest)
+
+	if err != nil {
+		if !errdefs.IsNotFound(err) {
+			return nil, err
+		}
+		logrus.WithField("target", target).Debug("distribution source label not found")
+		return mountableBlobs, nil
+	}
+
+	handler := func(ctx context.Context, desc ocispec.Descriptor) ([]ocispec.Descriptor, error) {
+		_, err := store.Info(ctx, desc.Digest)
+		if err != nil {
+			if !cerrdefs.IsNotFound(err) {
+				return nil, errdefs.System(errors.Wrapf(err, "failed to get metadata of content %s", desc.Digest.String()))
+			}
+
+			for _, source := range sources {
+				if canBeMounted(desc.MediaType, targetRef, i.registryService.IsInsecureRegistry, source) {
+					mountableBlobs[desc.Digest] = source
+					jobs.Add(desc)
+					break
+				}
+			}
+			return nil, nil
+		}
+
+		return containerdimages.Children(ctx, store, desc)
+	}
+
+	err = containerdimages.Dispatch(ctx, containerdimages.HandlerFunc(handler), limiter, target)
+	if err != nil {
+		return nil, err
+	}
+
+	return mountableBlobs, nil
+}
+
+func getDigestSources(ctx context.Context, store content.Manager, digest digest.Digest) ([]distributionSource, error) {
+	info, err := store.Info(ctx, digest)
+	if err != nil {
+		if cerrdefs.IsNotFound(err) {
+			return nil, errdefs.NotFound(err)
+		}
+		return nil, errdefs.System(err)
+	}
+
+	sources := extractDistributionSources(info.Labels)
+	if sources == nil {
+		return nil, errdefs.NotFound(fmt.Errorf("label %q is not attached to %s", labelDistributionSource, digest.String()))
+	}
+
+	return sources, nil
+}
+
+// TODO(vvoland): Remove and use containerd const in containerd 1.7+
+// https://github.com/containerd/containerd/pull/8224
+const labelDistributionSource = "containerd.io/distribution.source."
+
+func extractDistributionSources(labels map[string]string) []distributionSource {
+	var sources []distributionSource
+
+	// Check if this blob has a distributionSource label
+	// if yes, read it as source
+	for k, v := range labels {
+		registry := strings.TrimPrefix(k, labelDistributionSource)
+		if registry != k {
+			ref, err := reference.ParseNamed(registry + "/" + v)
+			if err != nil {
+				continue
+			}
+
+			sources = append(sources, distributionSource{
+				registryRef: ref,
+			})
+		}
+	}
+
+	return sources
+}
+
+type distributionSource struct {
+	registryRef reference.Named
+}
+
+// ToAnnotation returns key and value
+func (source distributionSource) ToAnnotation() (string, string) {
+	domain := reference.Domain(source.registryRef)
+	v := reference.Path(source.registryRef)
+	return labelDistributionSource + domain, v
+}
+
+func (source distributionSource) GetReference(dgst digest.Digest) (reference.Named, error) {
+	return reference.WithDigest(source.registryRef, dgst)
+}
+
+// canBeMounted returns if the content with given media type can be cross-repo
+// mounted when pushing it to a remote reference ref.
+func canBeMounted(mediaType string, targetRef reference.Named, isInsecureFunc func(string) bool, source distributionSource) bool {
+	if containerdimages.IsManifestType(mediaType) {
+		return false
+	}
+	if containerdimages.IsIndexType(mediaType) {
+		return false
+	}
+
+	registry := reference.Domain(targetRef)
+
+	// Cross-repo mount doesn't seem to work with insecure registries.
+	isInsecure := isInsecureFunc(registry)
+	if isInsecure {
+		return false
+	}
+
+	// If the source registry is the same as the one we are pushing to
+	// then the cross-repo mount will work.
+	return registry == reference.Domain(source.registryRef)
 }

daemon/containerd/progress.go

@@ -9,6 +9,7 @@ import (
 	"github.com/containerd/containerd/content"
 	cerrdefs "github.com/containerd/containerd/errdefs"
 	"github.com/containerd/containerd/remotes"
+	"github.com/containerd/containerd/remotes/docker"
 	"github.com/docker/docker/pkg/progress"
 	"github.com/docker/docker/pkg/stringid"
 	"github.com/opencontainers/go-digest"
@@ -158,3 +159,76 @@ func (p pullProgress) UpdateProgress(ctx context.Context, ongoing *jobs, out pro
 	}
 	return nil
 }
+
+type pushProgress struct {
+	Tracker   docker.StatusTracker
+	mountable map[digest.Digest]struct{}
+	mutex     sync.Mutex
+}
+
+func (p *pushProgress) addMountable(dgst digest.Digest) {
+	p.mutex.Lock()
+	defer p.mutex.Unlock()
+
+	if p.mountable == nil {
+		p.mountable = map[digest.Digest]struct{}{}
+	}
+	p.mountable[dgst] = struct{}{}
+}
+
+func (p *pushProgress) isMountable(dgst digest.Digest) bool {
+	p.mutex.Lock()
+	defer p.mutex.Unlock()
+
+	if p.mountable == nil {
+		return false
+	}
+	_, has := p.mountable[dgst]
+	return has
+}
+
+func (p *pushProgress) UpdateProgress(ctx context.Context, ongoing *jobs, out progress.Output, start time.Time) error {
+	for _, j := range ongoing.Jobs() {
+		key := remotes.MakeRefKey(ctx, j)
+		id := stringid.TruncateID(j.Digest.Encoded())
+
+		status, err := p.Tracker.GetStatus(key)
+		if err != nil {
+			if cerrdefs.IsNotFound(err) {
+				progress.Update(out, id, "Waiting")
+				continue
+			}
+		}
+
+		if status.Committed && status.Offset >= status.Total {
+			if p.isMountable(j.Digest) {
+				progress.Update(out, id, "Mounted")
+			} else {
+				progress.Update(out, id, "Pushed")
+			}
+			ongoing.Remove(j)
+			continue
+		}
+
+		out.WriteProgress(progress.Progress{
+			ID:      id,
+			Action:  "Pushing",
+			Current: status.Offset,
+			Total:   status.Total,
+		})
+	}
+
+	return nil
+}
+
+type combinedProgress []progressUpdater
+
+func (combined combinedProgress) UpdateProgress(ctx context.Context, ongoing *jobs, out progress.Output, start time.Time) error {
+	for _, p := range combined {
+		err := p.UpdateProgress(ctx, ongoing, out, start)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}

daemon/containerd/store.go

@@ -0,0 +1,84 @@
+package containerd
+
+import (
+	"context"
+
+	"github.com/containerd/containerd/content"
+	cerrdefs "github.com/containerd/containerd/errdefs"
+	"github.com/docker/distribution/reference"
+	"github.com/opencontainers/go-digest"
+	ocispec "github.com/opencontainers/image-spec/specs-go/v1"
+)
+
+// fakeStoreWithSources fakes the existence of the specified content.
+// Only existence is faked - Info function will include the distribution source label
+// which makes it possible to perform cross-repo mount.
+// ReaderAt will still fail with ErrNotFound.
+type fakeStoreWithSources struct {
+	s       content.Store
+	sources map[digest.Digest]distributionSource
+}
+
+// wrapWithFakeMountableBlobs wraps the provided content store.
+func wrapWithFakeMountableBlobs(s content.Store, sources map[digest.Digest]distributionSource) fakeStoreWithSources {
+	return fakeStoreWithSources{
+		s:       s,
+		sources: sources,
+	}
+}
+
+func (p fakeStoreWithSources) Delete(ctx context.Context, dgst digest.Digest) error {
+	return p.s.Delete(ctx, dgst)
+}
+
+func (p fakeStoreWithSources) Info(ctx context.Context, dgst digest.Digest) (content.Info, error) {
+	info, err := p.s.Info(ctx, dgst)
+	if err != nil {
+		if !cerrdefs.IsNotFound(err) {
+			return info, err
+		}
+		source, ok := p.sources[dgst]
+		if !ok {
+			return info, err
+		}
+
+		key := labelDistributionSource + reference.Domain(source.registryRef)
+		value := reference.Path(source.registryRef)
+		return content.Info{
+			Digest: dgst,
+			Labels: map[string]string{
+				key: value,
+			},
+		}, nil
+	}
+
+	return info, nil
+}
+
+func (p fakeStoreWithSources) Update(ctx context.Context, info content.Info, fieldpaths ...string) (content.Info, error) {
+	return p.s.Update(ctx, info, fieldpaths...)
+}
+
+func (p fakeStoreWithSources) Walk(ctx context.Context, fn content.WalkFunc, filters ...string) error {
+	return p.s.Walk(ctx, fn, filters...)
+}
+
+func (p fakeStoreWithSources) ReaderAt(ctx context.Context, desc ocispec.Descriptor) (content.ReaderAt, error) {
+	return p.s.ReaderAt(ctx, desc)
+}
+
+func (p fakeStoreWithSources) Abort(ctx context.Context, ref string) error {
+	return p.s.Abort(ctx, ref)
+}
+
+func (p fakeStoreWithSources) ListStatuses(ctx context.Context, filters ...string) ([]content.Status, error) {
+	return p.s.ListStatuses(ctx, filters...)
+}
+
+func (p fakeStoreWithSources) Status(ctx context.Context, ref string) (content.Status, error) {
+	return p.s.Status(ctx, ref)
+}
+
+func (p fakeStoreWithSources) Writer(ctx context.Context, opts ...content.WriterOpt) (content.Writer, error) {
+	return p.s.Writer(ctx, opts...)
+}