Merge pull request #2200 from jessvalarezo/multiarch-pulls

allow content to be pulled for specific platform(s), all platforms

client.go

@@ -43,7 +43,6 @@ import (
 	"github.com/containerd/containerd/events"
 	"github.com/containerd/containerd/images"
 	"github.com/containerd/containerd/namespaces"
-	"github.com/containerd/containerd/platforms"
 	"github.com/containerd/containerd/plugin"
 	"github.com/containerd/containerd/remotes"
 	"github.com/containerd/containerd/remotes/docker"
@@ -236,6 +235,10 @@ type RemoteContext struct {
 	// If no resolver is provided, defaults to Docker registry resolver.
 	Resolver remotes.Resolver
 
+	// Platforms defines which platforms to handle when doing the image operation.
+	// If this field is empty, content for all platforms will be pulled.
+	Platforms []string
+
 	// Unpack is done after an image is pulled to extract into a snapshotter.
 	// If an image is not unpacked on pull, it can be unpacked any time
 	// afterwards. Unpacking is required to run an image.
@@ -287,6 +290,7 @@ func (c *Client) Pull(ctx context.Context, ref string, opts ...RemoteOpt) (Image
 	if err != nil {
 		return nil, errors.Wrapf(err, "failed to resolve reference %q", ref)
 	}
+
 	fetcher, err := pullCtx.Resolver.Fetcher(ctx, name)
 	if err != nil {
 		return nil, errors.Wrapf(err, "failed to get fetcher for %q", name)
@@ -304,8 +308,8 @@ func (c *Client) Pull(ctx context.Context, ref string, opts ...RemoteOpt) (Image
 		childrenHandler := images.ChildrenHandler(store)
 		// Set any children labels for that content
 		childrenHandler = images.SetChildrenLabels(store, childrenHandler)
-		// Filter the childen by the platform
+		// Filter childen by platforms
-		childrenHandler = images.FilterPlatform(platforms.Default(), childrenHandler)
+		childrenHandler = images.FilterPlatforms(childrenHandler, pullCtx.Platforms...)
 
 		handler = images.Handlers(append(pullCtx.BaseHandlers,
 			remotes.FetchHandler(store, fetcher),
@@ -371,7 +375,7 @@ func (c *Client) Push(ctx context.Context, ref string, desc ocispec.Descriptor,
 		return err
 	}
 
-	return remotes.PushContent(ctx, pusher, desc, c.ContentStore(), pushCtx.BaseHandlers...)
+	return remotes.PushContent(ctx, pusher, desc, c.ContentStore(), pushCtx.Platforms, pushCtx.BaseHandlers...)
 }
 
 // GetImage returns an existing image

client_opts.go

@@ -64,6 +64,21 @@ func WithServices(opts ...ServicesOpt) ClientOpt {
 // RemoteOpt allows the caller to set distribution options for a remote
 type RemoteOpt func(*Client, *RemoteContext) error
 
+// WithPlatform allows the caller to specify a platform to retrieve
+// content for
+func WithPlatform(platform string) RemoteOpt {
+	return func(_ *Client, c *RemoteContext) error {
+		for _, p := range c.Platforms {
+			if p == platform {
+				return nil
+			}
+		}
+
+		c.Platforms = append(c.Platforms, platform)
+		return nil
+	}
+}
+
 // WithPullUnpack is used to unpack an image after pull. This
 // uses the snapshotter, content store, and diff service
 // configured for the client.

client_test.go

@@ -30,8 +30,10 @@ import (
 
 	"google.golang.org/grpc/grpclog"
 
+	"github.com/containerd/containerd/images"
 	"github.com/containerd/containerd/log"
 	"github.com/containerd/containerd/namespaces"
+	"github.com/containerd/containerd/platforms"
 	"github.com/containerd/containerd/sys"
 	"github.com/containerd/containerd/testutil"
 	"github.com/sirupsen/logrus"
@@ -116,7 +118,7 @@ func TestMain(m *testing.M) {
 	}).Info("running tests against containerd")
 
 	// pull a seed image
-	if _, err = client.Pull(ctx, testImage, WithPullUnpack); err != nil {
+	if _, err = client.Pull(ctx, testImage, WithPullUnpack, WithPlatform(platforms.Default())); err != nil {
 		ctrd.Stop()
 		ctrd.Wait()
 		fmt.Fprintf(os.Stderr, "%s: %s\n", err, buf.String())
@@ -187,12 +189,115 @@ func TestImagePull(t *testing.T) {
 
 	ctx, cancel := testContext()
 	defer cancel()
-	_, err = client.Pull(ctx, testImage)
+	_, err = client.Pull(ctx, testImage, WithPlatform(platforms.Default()))
 	if err != nil {
 		t.Fatal(err)
 	}
 }
 
+func TestImagePullAllPlatforms(t *testing.T) {
+	client, err := newClient(t, address)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer client.Close()
+	ctx, cancel := testContext()
+	defer cancel()
+
+	cs := client.ContentStore()
+	img, err := client.Pull(ctx, testImage)
+	if err != nil {
+		t.Fatal(err)
+	}
+	index := img.Target()
+	manifests, err := images.Children(ctx, cs, index)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, manifest := range manifests {
+		children, err := images.Children(ctx, cs, manifest)
+		if err != nil {
+			t.Fatal("Th")
+		}
+		// check if childless data type has blob in content store
+		for _, desc := range children {
+			ra, err := cs.ReaderAt(ctx, desc.Digest)
+			if err != nil {
+				t.Fatal(err)
+			}
+			ra.Close()
+		}
+	}
+}
+
+func TestImagePullSomePlatforms(t *testing.T) {
+	client, err := newClient(t, address)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer client.Close()
+	ctx, cancel := testContext()
+	defer cancel()
+
+	cs := client.ContentStore()
+	platformList := []string{"linux/arm64/v8", "linux/386"}
+	m := make(map[string]platforms.Matcher)
+	var opts []RemoteOpt
+
+	for _, platform := range platformList {
+		p, err := platforms.Parse(platform)
+		if err != nil {
+			t.Fatal(err)
+		}
+		m[platform] = platforms.NewMatcher(p)
+		opts = append(opts, WithPlatform(platform))
+	}
+
+	img, err := client.Pull(ctx, "docker.io/library/busybox:latest", opts...)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	index := img.Target()
+	manifests, err := images.Children(ctx, cs, index)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	count := 0
+	for _, manifest := range manifests {
+		children, err := images.Children(ctx, cs, manifest)
+		found := false
+		for _, matcher := range m {
+			if matcher.Match(*manifest.Platform) {
+				count++
+				found = true
+			}
+		}
+
+		if found {
+			if len(children) == 0 {
+				t.Fatal("manifest should have pulled children content")
+			}
+
+			// check if childless data type has blob in content store
+			for _, desc := range children {
+				ra, err := cs.ReaderAt(ctx, desc.Digest)
+				if err != nil {
+					t.Fatal(err)
+				}
+				ra.Close()
+			}
+		} else if !found && err == nil {
+			t.Fatal("manifest should not have pulled children content")
+		}
+	}
+
+	if count != len(platformList) {
+		t.Fatal("expected a different number of pulled manifests")
+	}
+}
+
 func TestClientReconnect(t *testing.T) {
 	t.Parallel()
 

cmd/ctr/commands/content/fetch.go

@@ -101,12 +101,20 @@ func Fetch(ref string, cliContext *cli.Context) (containerd.Image, error) {
 
 	log.G(pctx).WithField("image", ref).Debug("fetching")
 	labels := commands.LabelArgs(cliContext.StringSlice("label"))
-	img, err := client.Pull(pctx, ref,
+	opts := []containerd.RemoteOpt{
 		containerd.WithPullLabels(labels),
 		containerd.WithResolver(resolver),
 		containerd.WithImageHandler(h),
 		containerd.WithSchema1Conversion,
-	)
+	}
+
+	if !cliContext.Bool("all-platforms") {
+		for _, platform := range cliContext.StringSlice("platform") {
+			opts = append(opts, containerd.WithPlatform(platform))
+		}
+	}
+
+	img, err := client.Pull(pctx, ref, opts...)
 	stopProgress()
 	if err != nil {
 		return nil, err

cmd/ctr/commands/images/pull.go

@@ -22,6 +22,7 @@ import (
 	"github.com/containerd/containerd/cmd/ctr/commands"
 	"github.com/containerd/containerd/cmd/ctr/commands/content"
 	"github.com/containerd/containerd/log"
+	"github.com/containerd/containerd/platforms"
 	"github.com/urfave/cli"
 )
 
@@ -38,7 +39,17 @@ command. As part of this process, we do the following:
 2. Prepare the snapshot filesystem with the pulled resources.
 3. Register metadata for the image.
 `,
-	Flags: append(commands.RegistryFlags, append(commands.SnapshotterFlags, commands.LabelFlag)...),
+	Flags: append(append(commands.RegistryFlags, append(commands.SnapshotterFlags, commands.LabelFlag)...),
+		cli.StringSliceFlag{
+			Name:  "platform",
+			Usage: "Pull content from a specific platform",
+			Value: &cli.StringSlice{platforms.Default()},
+		},
+		cli.BoolFlag{
+			Name:  "all-platforms",
+			Usage: "pull content from all platforms",
+		},
+	),
 	Action: func(context *cli.Context) error {
 		var (
 			ref = context.Args().First()

images/handlers.go

@@ -182,9 +182,9 @@ func SetChildrenLabels(manager content.Manager, f HandlerFunc) HandlerFunc {
 	}
 }
 
-// FilterPlatform is a handler wrapper which limits the descriptors returned
+// FilterPlatforms is a handler wrapper which limits the descriptors returned
-// by a handler to a single platform.
+// by a handler to the specified platforms.
-func FilterPlatform(platform string, f HandlerFunc) HandlerFunc {
+func FilterPlatforms(f HandlerFunc, platformList ...string) HandlerFunc {
 	return func(ctx context.Context, desc ocispec.Descriptor) ([]ocispec.Descriptor, error) {
 		children, err := f(ctx, desc)
 		if err != nil {
@@ -192,32 +192,25 @@ func FilterPlatform(platform string, f HandlerFunc) HandlerFunc {
 		}
 
 		var descs []ocispec.Descriptor
-		if platform != "" && isMultiPlatform(desc.MediaType) {
-			p, err := platforms.Parse(platform)
-			if err != nil {
-				return nil, err
-			}
-			matcher := platforms.NewMatcher(p)
 
-			for _, d := range children {
+		if len(platformList) == 0 {
-				if d.Platform == nil || matcher.Match(*d.Platform) {
+			descs = children
-					descs = append(descs, d)
+		} else {
+			for _, platform := range platformList {
+				p, err := platforms.Parse(platform)
+				if err != nil {
+					return nil, err
+				}
+				matcher := platforms.NewMatcher(p)
+
+				for _, d := range children {
+					if d.Platform == nil || matcher.Match(*d.Platform) {
+						descs = append(descs, d)
+					}
 				}
 			}
-		} else {
-			descs = children
 		}
 
 		return descs, nil
 	}
-
-}
-
-func isMultiPlatform(mediaType string) bool {
-	switch mediaType {
-	case MediaTypeDockerSchema2ManifestList, ocispec.MediaTypeImageIndex:
-		return true
-	default:
-		return false
-	}
 }

images/image.go

@@ -118,7 +118,7 @@ func (image *Image) Size(ctx context.Context, provider content.Provider, platfor
 		}
 		size += desc.Size
 		return nil, nil
-	}), FilterPlatform(platform, ChildrenHandler(provider))), image.Target)
+	}), FilterPlatforms(ChildrenHandler(provider), platform)), image.Target)
 }
 
 // Manifest resolves a manifest from the image for the given platform.

images/oci/exporter.go

@@ -58,7 +58,7 @@ func (oe *V1Exporter) Export(ctx context.Context, store content.Provider, desc o
 	}
 
 	handlers := images.Handlers(
-		images.FilterPlatform(platforms.Default(), images.ChildrenHandler(store)),
+		images.FilterPlatforms(images.ChildrenHandler(store), platforms.Default()),
 		images.HandlerFunc(exportHandler),
 	)
 

remotes/handlers.go

@@ -29,7 +29,6 @@ import (
 	"github.com/containerd/containerd/errdefs"
 	"github.com/containerd/containerd/images"
 	"github.com/containerd/containerd/log"
-	"github.com/containerd/containerd/platforms"
 	ocispec "github.com/opencontainers/image-spec/specs-go/v1"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
@@ -182,7 +181,7 @@ func push(ctx context.Context, provider content.Provider, pusher Pusher, desc oc
 //
 // Base handlers can be provided which will be called before any push specific
 // handlers.
-func PushContent(ctx context.Context, pusher Pusher, desc ocispec.Descriptor, provider content.Provider, baseHandlers ...images.Handler) error {
+func PushContent(ctx context.Context, pusher Pusher, desc ocispec.Descriptor, provider content.Provider, platforms []string, baseHandlers ...images.Handler) error {
 	var m sync.Mutex
 	manifestStack := []ocispec.Descriptor{}
 
@@ -202,7 +201,7 @@ func PushContent(ctx context.Context, pusher Pusher, desc ocispec.Descriptor, pr
 	pushHandler := PushHandler(pusher, provider)
 
 	handlers := append(baseHandlers,
-		images.FilterPlatform(platforms.Default(), images.ChildrenHandler(provider)),
+		images.FilterPlatforms(images.ChildrenHandler(provider), platforms...),
 		filterHandler,
 		pushHandler,
 	)