From e7393d0fc58f07c582915562ded41c67bdbed3a1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Pawe=C5=82=20Gronowski?= Date: Tue, 7 Feb 2023 14:02:20 +0100 Subject: [PATCH 1/2] images/push: Accept reference MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Push the reference parsing from repo and tag names into the api and pass a reference object to the ImageService. Signed-off-by: Paweł Gronowski --- api/server/router/image/backend.go | 2 +- api/server/router/image/image_routes.go | 20 +++++++++++++++++++- daemon/containerd/image_push.go | 3 ++- daemon/image_service.go | 4 ++-- daemon/images/image_push.go | 16 ++-------------- 5 files changed, 26 insertions(+), 19 deletions(-) diff --git a/api/server/router/image/backend.go b/api/server/router/image/backend.go index 302b48d68e..f7cacba90f 100644 --- a/api/server/router/image/backend.go +++ b/api/server/router/image/backend.go @@ -38,7 +38,7 @@ type importExportBackend interface { type registryBackend interface { PullImage(ctx context.Context, image, tag string, platform *specs.Platform, metaHeaders map[string][]string, authConfig *registry.AuthConfig, outStream io.Writer) error - PushImage(ctx context.Context, image, tag string, metaHeaders map[string][]string, authConfig *registry.AuthConfig, outStream io.Writer) error + PushImage(ctx context.Context, ref reference.Named, metaHeaders map[string][]string, authConfig *registry.AuthConfig, outStream io.Writer) error } type Searcher interface { diff --git a/api/server/router/image/image_routes.go b/api/server/router/image/image_routes.go index d891b8158e..e39e59e3bb 100644 --- a/api/server/router/image/image_routes.go +++ b/api/server/router/image/image_routes.go @@ -154,7 +154,25 @@ func (ir *imageRouter) postImagesPush(ctx context.Context, w http.ResponseWriter img := vars["name"] tag := r.Form.Get("tag") - if err := ir.backend.PushImage(ctx, img, tag, metaHeaders, authConfig, output); err != nil { + + var ref reference.Named + + // Tag is empty only in case ImagePushOptions.All is true. + if tag != "" { + r, err := httputils.RepoTagReference(img, tag) + if err != nil { + return errdefs.InvalidParameter(err) + } + ref = r + } else { + r, err := reference.ParseNormalizedNamed(img) + if err != nil { + return errdefs.InvalidParameter(err) + } + ref = r + } + + if err := ir.backend.PushImage(ctx, ref, metaHeaders, authConfig, output); err != nil { if !output.Flushed() { return err } diff --git a/daemon/containerd/image_push.go b/daemon/containerd/image_push.go index 4a406d4b8b..90f6c2eeb0 100644 --- a/daemon/containerd/image_push.go +++ b/daemon/containerd/image_push.go @@ -5,11 +5,12 @@ import ( "errors" "io" + "github.com/docker/distribution/reference" "github.com/docker/docker/api/types/registry" "github.com/docker/docker/errdefs" ) // PushImage initiates a push operation on the repository named localName. -func (i *ImageService) PushImage(ctx context.Context, image, tag string, metaHeaders map[string][]string, authConfig *registry.AuthConfig, outStream io.Writer) error { +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")) } diff --git a/daemon/image_service.go b/daemon/image_service.go index 78ac874f4e..02f1f7c6dc 100644 --- a/daemon/image_service.go +++ b/daemon/image_service.go @@ -25,8 +25,8 @@ import ( type ImageService interface { // Images - PullImage(ctx context.Context, image, tag string, platform *v1.Platform, metaHeaders map[string][]string, authConfig *registry.AuthConfig, outStream io.Writer) error - PushImage(ctx context.Context, image, tag string, metaHeaders map[string][]string, authConfig *registry.AuthConfig, outStream io.Writer) error + PullImage(ctx context.Context, name, tag string, platform *v1.Platform, metaHeaders map[string][]string, authConfig *registry.AuthConfig, outStream io.Writer) error + PushImage(ctx context.Context, ref reference.Named, metaHeaders map[string][]string, authConfig *registry.AuthConfig, outStream io.Writer) error CreateImage(config []byte, parent string) (builder.Image, error) ImageDelete(ctx context.Context, imageRef string, force, prune bool) ([]types.ImageDeleteResponseItem, error) ExportImage(ctx context.Context, names []string, outStream io.Writer) error diff --git a/daemon/images/image_push.go b/daemon/images/image_push.go index c461133c30..88ba18293a 100644 --- a/daemon/images/image_push.go +++ b/daemon/images/image_push.go @@ -14,20 +14,8 @@ import ( ) // PushImage initiates a push operation on the repository named localName. -func (i *ImageService) PushImage(ctx context.Context, image, tag string, metaHeaders map[string][]string, authConfig *registry.AuthConfig, outStream io.Writer) error { +func (i *ImageService) PushImage(ctx context.Context, ref reference.Named, metaHeaders map[string][]string, authConfig *registry.AuthConfig, outStream io.Writer) error { start := time.Now() - 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 - } - } - // Include a buffer so that slow client connections don't affect // transfer performance. progressChan := make(chan progress.Progress, 100) @@ -57,7 +45,7 @@ func (i *ImageService) PushImage(ctx context.Context, image, tag string, metaHea UploadManager: i.uploadManager, } - err = distribution.Push(ctx, ref, imagePushConfig) + err := distribution.Push(ctx, ref, imagePushConfig) close(progressChan) <-writesDone imageActions.WithValues("push").UpdateSince(start) From a75354c04560d0e2b7a8469dae7743033d77ed8a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Pawe=C5=82=20Gronowski?= Date: Fri, 8 Jul 2022 15:12:35 +0200 Subject: [PATCH 2/2] c8d: Implement push MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- daemon/containerd/image_push.go | 248 +++++++++++++++++++++++++++++++- daemon/containerd/progress.go | 74 ++++++++++ daemon/containerd/store.go | 84 +++++++++++ 3 files changed, 402 insertions(+), 4 deletions(-) create mode 100644 daemon/containerd/store.go diff --git a/daemon/containerd/image_push.go b/daemon/containerd/image_push.go index 90f6c2eeb0..ab35cfc075 100644 --- a/daemon/containerd/image_push.go +++ b/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) } diff --git a/daemon/containerd/progress.go b/daemon/containerd/progress.go index 1e75d63f87..eb1d0f08bc 100644 --- a/daemon/containerd/progress.go +++ b/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 +} diff --git a/daemon/containerd/store.go b/daemon/containerd/store.go new file mode 100644 index 0000000000..41810c13db --- /dev/null +++ b/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...) +}