Add --all-tags flag to crane cp (#1682)

This will copy every tag in the src repo to dst.

This adds a --no-clobber flag that crane cp will respect that avoids
overwriting an existing tag (at the time of command invocation).

cmd/crane/cmd/copy.go

@@ -15,20 +15,36 @@
 package cmd
 
 import (
+	"runtime"
+
 	"github.com/google/go-containerregistry/pkg/crane"
 	"github.com/spf13/cobra"
 )
 
 // NewCmdCopy creates a new cobra.Command for the copy subcommand.
 func NewCmdCopy(options *[]crane.Option) *cobra.Command {
-	return &cobra.Command{
+	allTags := false
+	noclobber := false
+	jobs := runtime.GOMAXPROCS(0)
+	cmd := &cobra.Command{
 		Use:     "copy SRC DST",
 		Aliases: []string{"cp"},
 		Short:   "Efficiently copy a remote image from src to dst while retaining the digest value",
 		Args:    cobra.ExactArgs(2),
 		RunE: func(_ *cobra.Command, args []string) error {
+			opts := append(*options, crane.WithJobs(jobs), crane.WithNoClobber(noclobber))
 			src, dst := args[0], args[1]
-			return crane.Copy(src, dst, *options...)
+			if allTags {
+				return crane.CopyRepository(src, dst, opts...)
+			}
+
+			return crane.Copy(src, dst, opts...)
 		},
 	}
+
+	cmd.Flags().BoolVarP(&allTags, "all-tags", "a", false, "(Optional) if true, copy all tags from SRC to DST")
+	cmd.Flags().BoolVarP(&noclobber, "no-clobber", "n", false, "(Optional) if true, avoid overwriting existing tags in DST")
+	cmd.Flags().IntVarP(&jobs, "jobs", "j", 0, "(Optional) The maximum number of concurrent copies, defaults to GOMAXPROCS")
+
+	return cmd
 }

pkg/crane/copy.go

@@ -15,11 +15,15 @@
 package crane
 
 import (
+	"errors"
 	"fmt"
+	"net/http"
 
 	"github.com/google/go-containerregistry/pkg/logs"
 	"github.com/google/go-containerregistry/pkg/name"
 	"github.com/google/go-containerregistry/pkg/v1/remote"
+	"github.com/google/go-containerregistry/pkg/v1/remote/transport"
+	"golang.org/x/sync/errgroup"
 )
 
 // Copy copies a remote image or index from src to dst.
@@ -35,12 +39,31 @@ func Copy(src, dst string, opt ...Option) error {
 		return fmt.Errorf("parsing reference for %q: %w", dst, err)
 	}
 
-	pusher, err := remote.NewPusher(o.Remote...)
+	puller, err := remote.NewPuller(o.Remote...)
 	if err != nil {
 		return err
 	}
 
-	puller, err := remote.NewPuller(o.Remote...)
+	if tag, ok := dstRef.(name.Tag); ok {
+		if o.noclobber {
+			logs.Progress.Printf("Checking existing tag %v", tag)
+			head, err := puller.Head(o.ctx, tag)
+			var terr *transport.Error
+			if errors.As(err, &terr) {
+				if terr.StatusCode != http.StatusNotFound && terr.StatusCode != http.StatusForbidden {
+					return err
+				}
+			} else if err != nil {
+				return err
+			}
+
+			if head != nil {
+				return fmt.Errorf("refusing to clobber existing tag %s@%s", tag, head.Digest)
+			}
+		}
+	}
+
+	pusher, err := remote.NewPusher(o.Remote...)
 	if err != nil {
 		return err
 	}
@@ -62,3 +85,97 @@ func Copy(src, dst string, opt ...Option) error {
 	}
 	return pusher.Push(o.ctx, dstRef, img)
 }
+
+// CopyRepository copies every tag from src to dst.
+func CopyRepository(src, dst string, opt ...Option) error {
+	o := makeOptions(opt...)
+
+	srcRepo, err := name.NewRepository(src, o.Name...)
+	if err != nil {
+		return err
+	}
+
+	dstRepo, err := name.NewRepository(dst, o.Name...)
+	if err != nil {
+		return fmt.Errorf("parsing reference for %q: %w", dst, err)
+	}
+
+	puller, err := remote.NewPuller(o.Remote...)
+	if err != nil {
+		return err
+	}
+
+	ignoredTags := map[string]struct{}{}
+	if o.noclobber {
+		// TODO: It would be good to propagate noclobber down into remote so we can use Etags.
+		have, err := puller.List(o.ctx, dstRepo)
+		if err != nil {
+			var terr *transport.Error
+			if errors.As(err, &terr) {
+				// Some registries create repository on first push, so listing tags will fail.
+				// If we see 404 or 403, assume we failed because the repository hasn't been created yet.
+				if !(terr.StatusCode == http.StatusNotFound || terr.StatusCode == http.StatusForbidden) {
+					return err
+				}
+			} else {
+				return err
+			}
+		}
+		for _, tag := range have {
+			ignoredTags[tag] = struct{}{}
+		}
+	}
+
+	pusher, err := remote.NewPusher(o.Remote...)
+	if err != nil {
+		return err
+	}
+
+	lister, err := puller.Lister(o.ctx, srcRepo)
+	if err != nil {
+		return err
+	}
+
+	g, ctx := errgroup.WithContext(o.ctx)
+	g.SetLimit(o.jobs)
+
+	for lister.HasNext() {
+		tags, err := lister.Next(ctx)
+		if err != nil {
+			return err
+		}
+
+		for _, tag := range tags.Tags {
+			tag := tag
+
+			if o.noclobber {
+				if _, ok := ignoredTags[tag]; ok {
+					logs.Progress.Printf("Skipping %s due to no-clobber", tag)
+					continue
+				}
+			}
+
+			g.Go(func() error {
+				srcTag, err := name.ParseReference(src+":"+tag, o.Name...)
+				if err != nil {
+					return fmt.Errorf("failed to parse tag: %w", err)
+				}
+				dstTag, err := name.ParseReference(dst+":"+tag, o.Name...)
+				if err != nil {
+					return fmt.Errorf("failed to parse tag: %w", err)
+				}
+
+				logs.Progress.Printf("Fetching %s", srcTag)
+				desc, err := puller.Get(ctx, srcTag)
+				if err != nil {
+					return err
+				}
+
+				logs.Progress.Printf("Pushing %s", dstTag)
+				return pusher.Push(ctx, dstTag, desc)
+			})
+		}
+	}
+
+	return g.Wait()
+}

pkg/crane/options.go

@@ -32,8 +32,11 @@ type Options struct {
 	Platform *v1.Platform
 	Keychain authn.Keychain
 
+	auth      authn.Authenticator
 	transport http.RoundTripper
 	insecure  bool
+	jobs      int
+	noclobber bool
 	ctx       context.Context
 }
 
@@ -51,6 +54,7 @@ func makeOptions(opts ...Option) Options {
 			remote.WithAuthFromKeychain(authn.DefaultKeychain),
 		},
 		Keychain: authn.DefaultKeychain,
+		jobs:     4,
 		ctx:      context.Background(),
 	}
 
@@ -124,6 +128,7 @@ func WithAuth(auth authn.Authenticator) Option {
 	return func(o *Options) {
 		// Replace the default keychain at position 0.
 		o.Remote[0] = remote.WithAuth(auth)
+		o.auth = auth
 	}
 }
 
@@ -150,3 +155,22 @@ func WithContext(ctx context.Context) Option {
 		o.Remote = append(o.Remote, remote.WithContext(ctx))
 	}
 }
+
+// WithJobs sets the number of concurrent jobs to run.
+//
+// The default number of jobs is GOMAXPROCS.
+func WithJobs(jobs int) Option {
+	return func(o *Options) {
+		if jobs > 0 {
+			o.jobs = jobs
+		}
+		o.Remote = append(o.Remote, remote.WithJobs(o.jobs))
+	}
+}
+
+// WithNoClobber modifies behavior to avoid overwriting existing tags, if possible.
+func WithNoClobber(noclobber bool) Option {
+	return func(o *Options) {
+		o.noclobber = noclobber
+	}
+}