Refactor image management to use transfer service

cmd/nerdctl/image/image_import.go

@@ -45,6 +45,7 @@ func ImportCommand() *cobra.Command {
 
 	cmd.Flags().StringP("message", "m", "", "Set commit message for imported image")
 	cmd.Flags().String("platform", "", "Set platform for imported image (e.g., linux/amd64)")
+	cmd.Flags().Bool("local", false, "Import content locally rather than using transfer service")
 	return cmd
 }
 
@@ -61,6 +62,10 @@ func importOptions(cmd *cobra.Command, args []string) (types.ImageImportOptions,
 	if err != nil {
 		return types.ImageImportOptions{}, err
 	}
+	useLocal, err := cmd.Flags().GetBool("local")
+	if err != nil {
+		return types.ImageImportOptions{}, err
+	}
 	var reference string
 	if len(args) > 1 {
 		reference = args[1]
@@ -97,6 +102,7 @@ func importOptions(cmd *cobra.Command, args []string) (types.ImageImportOptions,
 		Reference: reference,
 		Message:   message,
 		Platform:  platform,
+		Local:     useLocal,
 	}, nil
 }
 

cmd/nerdctl/image/image_pull.go

@@ -68,6 +68,7 @@ func PullCommand() *cobra.Command {
 	cmd.Flags().BoolP("quiet", "q", false, "Suppress verbose output")
 
 	cmd.Flags().String("ipfs-address", "", "multiaddr of IPFS API (default uses $IPFS_PATH env variable if defined or local directory ~/.ipfs)")
+	cmd.Flags().Bool("local", false, "Fetch content from local client rather than using transfer service")
 
 	return cmd
 }
@@ -108,6 +109,10 @@ func processPullCommandFlags(cmd *cobra.Command) (types.ImagePullOptions, error)
 	if err != nil {
 		return types.ImagePullOptions{}, err
 	}
+	useLocal, err := cmd.Flags().GetBool("local")
+	if err != nil {
+		return types.ImagePullOptions{}, err
+	}
 
 	sociIndexDigest, err := cmd.Flags().GetString("soci-index-digest")
 	if err != nil {
@@ -132,6 +137,7 @@ func processPullCommandFlags(cmd *cobra.Command) (types.ImagePullOptions, error)
 		Stdout:                 cmd.OutOrStdout(),
 		Stderr:                 cmd.OutOrStderr(),
 		ProgressOutputToStdout: true,
+		Local:                  useLocal,
 	}, nil
 }
 

cmd/nerdctl/image/image_push.go

@@ -68,6 +68,7 @@ func PushCommand() *cobra.Command {
 	cmd.Flags().BoolP("quiet", "q", false, "Suppress verbose output")
 
 	cmd.Flags().Bool(allowNonDistFlag, false, "Allow pushing images with non-distributable blobs")
+	cmd.Flags().Bool("local", false, "Push content from local client rather than using transfer service")
 
 	return cmd
 }
@@ -105,6 +106,10 @@ func pushOptions(cmd *cobra.Command) (types.ImagePushOptions, error) {
 	if err != nil {
 		return types.ImagePushOptions{}, err
 	}
+	useLocal, err := cmd.Flags().GetBool("local")
+	if err != nil {
+		return types.ImagePushOptions{}, err
+	}
 	signOptions, err := signOptions(cmd)
 	if err != nil {
 		return types.ImagePushOptions{}, err
@@ -124,6 +129,7 @@ func pushOptions(cmd *cobra.Command) (types.ImagePushOptions, error) {
 		IpfsAddress:                    ipfsAddress,
 		Quiet:                          quiet,
 		AllowNondistributableArtifacts: allowNonDist,
+		Local:                          useLocal,
 		Stdout:                         cmd.OutOrStdout(),
 	}, nil
 }

cmd/nerdctl/image/image_save.go

@@ -50,6 +50,8 @@ func SaveCommand() *cobra.Command {
 	cmd.Flags().Bool("all-platforms", false, "Export content for all platforms")
 	// #endregion
 
+	cmd.Flags().Bool("local", false, "Export images locally rather than using transfer service")
+
 	return cmd
 }
 
@@ -67,11 +69,16 @@ func saveOptions(cmd *cobra.Command) (types.ImageSaveOptions, error) {
 	if err != nil {
 		return types.ImageSaveOptions{}, err
 	}
+	useLocal, err := cmd.Flags().GetBool("local")
+	if err != nil {
+		return types.ImageSaveOptions{}, err
+	}
 
 	return types.ImageSaveOptions{
 		GOptions:     globalOptions,
 		AllPlatforms: allPlatforms,
 		Platform:     platform,
+		Local:        useLocal,
 	}, err
 }
 

cmd/nerdctl/image/image_tag.go

@@ -36,6 +36,7 @@ func TagCommand() *cobra.Command {
 		SilenceUsage:      true,
 		SilenceErrors:     true,
 	}
+	cmd.Flags().Bool("local", false, "Run tag locally rather than through transfer API")
 	return cmd
 }
 
@@ -45,10 +46,16 @@ func tagAction(cmd *cobra.Command, args []string) error {
 		return err
 	}
 
+	useLocal, err := cmd.Flags().GetBool("local")
+	if err != nil {
+		return err
+	}
+
 	options := types.ImageTagOptions{
 		GOptions: globalOptions,
 		Source:   args[0],
 		Target:   args[1],
+		Local:    useLocal,
 	}
 
 	client, ctx, cancel, err := clientutil.NewClient(cmd.Context(), options.GOptions.Namespace, options.GOptions.Address)

pkg/api/types/image_types.go

@@ -202,6 +202,8 @@ type ImagePushOptions struct {
 	Quiet bool
 	// AllowNondistributableArtifacts allow pushing non-distributable artifacts
 	AllowNondistributableArtifacts bool
+	// Local controls whether nerdctl should rely on containerd's transfer service for pushes.
+	Local bool
 }
 
 // RemoteSnapshotterFlags are used for pulling with remote snapshotters
@@ -232,6 +234,8 @@ type ImagePullOptions struct {
 	IPFSAddress string
 	// Flags to pass into remote snapshotters
 	RFlags RemoteSnapshotterFlags
+	// Local controls whether to rely on containerd's transfer service for registry pulls.
+	Local bool
 }
 
 // ImageTagOptions specifies options for `nerdctl (image) tag`.
@@ -242,6 +246,8 @@ type ImageTagOptions struct {
 	Source string
 	// Target is the image to be created.
 	Target string
+	// Local controls whether to use containerd's transfer service for tagging.
+	Local bool
 }
 
 // ImageRemoveOptions specifies options for `nerdctl rmi` and `nerdctl image rm`.
@@ -276,6 +282,8 @@ type ImageSaveOptions struct {
 	AllPlatforms bool
 	// Export content for a specific platform
 	Platform []string
+	// Local controls whether touses transfer service for exporting images (default: true, --local sets to false)
+	Local bool
 }
 
 // ImageSignOptions contains options for signing an image. It contains options from

pkg/api/types/import_types.go

@@ -28,4 +28,6 @@ type ImageImportOptions struct {
 	Reference string
 	Message   string
 	Platform  string
+	// Local controls whether nerdctl should rely on containerd's transfer service for imports.
+	Local bool
 }

pkg/cmd/image/import.go

@@ -25,6 +25,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
+	"os"
 	"time"
 
 	"github.com/opencontainers/go-digest"
@@ -36,15 +37,74 @@ import (
 	"github.com/containerd/containerd/v2/core/content"
 	"github.com/containerd/containerd/v2/core/images"
 	"github.com/containerd/containerd/v2/core/leases"
+	"github.com/containerd/containerd/v2/core/transfer"
+	tarchive "github.com/containerd/containerd/v2/core/transfer/archive"
+	transferimage "github.com/containerd/containerd/v2/core/transfer/image"
 	"github.com/containerd/containerd/v2/pkg/archive/compression"
 	"github.com/containerd/errdefs"
 	"github.com/containerd/platforms"
 
 	"github.com/containerd/nerdctl/v2/pkg/api/types"
 	"github.com/containerd/nerdctl/v2/pkg/referenceutil"
+	"github.com/containerd/nerdctl/v2/pkg/transferutil"
 )
 
 func Import(ctx context.Context, client *containerd.Client, options types.ImageImportOptions) (string, error) {
+	if !options.Local {
+		return importWithTransfer(ctx, client, options)
+	}
+	return importWithLegacy(ctx, client, options)
+}
+
+func importWithTransfer(ctx context.Context, client *containerd.Client, options types.ImageImportOptions) (string, error) {
+	var opts []transferimage.StoreOpt
+
+	prefix := options.Reference
+	var overwrite bool
+	if prefix == "" {
+		prefix = fmt.Sprintf("import-%s", time.Now().Format("2006-01-02"))
+		overwrite = true
+	}
+
+	opts = append(opts, transferimage.WithNamedPrefix(prefix, overwrite))
+
+	platUnpack := platforms.DefaultSpec()
+	if options.Platform != "" {
+		p, err := platforms.Parse(options.Platform)
+		if err != nil {
+			return "", err
+		}
+		platUnpack = p
+		opts = append(opts, transferimage.WithPlatforms(platUnpack))
+	}
+
+	snapshotter := options.GOptions.Snapshotter
+	opts = append(opts, transferimage.WithUnpack(platUnpack, snapshotter))
+
+	is := transferimage.NewStore("", opts...)
+
+	var r io.ReadCloser
+	if rc, ok := options.Stdin.(io.ReadCloser); ok {
+		r = rc
+	} else {
+		r = io.NopCloser(options.Stdin)
+	}
+
+	var iopts []tarchive.ImportOpt
+	iis := tarchive.NewImageImportStream(r, "", iopts...)
+
+	pf, done := transferutil.ProgressHandler(ctx, os.Stderr)
+	defer done()
+
+	err := client.Transfer(ctx, iis, is, transfer.WithProgress(pf))
+	if err != nil {
+		return "", err
+	}
+
+	return prefix, nil
+}
+
+func importWithLegacy(ctx context.Context, client *containerd.Client, options types.ImageImportOptions) (string, error) {
 	img, err := importRootfs(ctx, client, options.GOptions.Snapshotter, options)
 	if err != nil {
 		return "", err

pkg/cmd/image/push.go

@@ -37,12 +37,14 @@ import (
 	dockerconfig "github.com/containerd/containerd/v2/core/remotes/docker/config"
 	"github.com/containerd/containerd/v2/pkg/reference"
 	"github.com/containerd/log"
+	"github.com/containerd/platforms"
 	"github.com/containerd/stargz-snapshotter/estargz"
 	"github.com/containerd/stargz-snapshotter/estargz/zstdchunked"
 	estargzconvert "github.com/containerd/stargz-snapshotter/nativeconverter/estargz"
 
 	"github.com/containerd/nerdctl/v2/pkg/api/types"
 	"github.com/containerd/nerdctl/v2/pkg/errutil"
+	"github.com/containerd/nerdctl/v2/pkg/imgutil"
 	nerdconverter "github.com/containerd/nerdctl/v2/pkg/imgutil/converter"
 	"github.com/containerd/nerdctl/v2/pkg/imgutil/dockerconfigresolver"
 	"github.com/containerd/nerdctl/v2/pkg/imgutil/push"
@@ -110,7 +112,6 @@ func Push(ctx context.Context, client *containerd.Client, rawRef string, options
 		return err
 	}
 	ref := parsedReference.String()
-	refDomain := parsedReference.Domain
 
 	platMC, err := platformutil.NewMatchComparer(options.AllPlatforms, options.Platforms)
 	if err != nil {
@@ -147,11 +148,53 @@ func Push(ctx context.Context, client *containerd.Client, rawRef string, options
 		log.G(ctx).Infof("pushing as an eStargz image (%s, %s)", esgzImg.Target.MediaType, esgzImg.Target.Digest)
 	}
 
-	// In order to push images where most layers are the same but the
-	// repository name is different, it is necessary to refresh the
-	// PushTracker. Otherwise, the MANIFEST_BLOB_UNKNOWN error will occur due
-	// to the registry not creating the corresponding layer link file,
-	// resulting in the failure of the entire image push.
+	useTransfer := !options.Local
+	if useTransfer && options.AllowNondistributableArtifacts {
+		log.G(ctx).Warn("transfer service currently cannot honor --allow-nondistributable-artifacts, falling back to legacy push path")
+		useTransfer = false
+	}
+
+	if useTransfer {
+		if err := imgutil.PushWithTransfer(ctx, client, parsedReference, pushRef, ref, options); err != nil {
+			return err
+		}
+	} else {
+		if err := pushWithLegacy(ctx, client, parsedReference, pushRef, ref, platMC, options); err != nil {
+			return err
+		}
+	}
+
+	img, err := client.ImageService().Get(ctx, pushRef)
+	if err != nil {
+		return err
+	}
+	refSpec, err := reference.Parse(pushRef)
+	if err != nil {
+		return err
+	}
+	signRef := fmt.Sprintf("%s@%s", refSpec.String(), img.Target.Digest.String())
+	if err = signutil.Sign(signRef,
+		options.GOptions.Experimental,
+		options.SignOptions); err != nil {
+		return err
+	}
+	if options.GOptions.Snapshotter == "soci" {
+		if err = snapshotterutil.CreateSociIndexV1(ref, options.GOptions, options.AllPlatforms, options.Platforms, options.SociOptions); err != nil {
+			return err
+		}
+		if err = snapshotterutil.PushSoci(ref, options.GOptions, options.AllPlatforms, options.Platforms); err != nil {
+			return err
+		}
+	}
+	if options.Quiet {
+		fmt.Fprintln(options.Stdout, ref)
+	}
+	return nil
+}
+
+func pushWithLegacy(ctx context.Context, client *containerd.Client, parsedReference *referenceutil.ImageReference, pushRef, ref string, platMC platforms.MatchComparer, options types.ImagePushOptions) error {
+	refDomain := parsedReference.Domain
+
 	pushTracker := docker.NewInMemoryTracker()
 
 	pushFunc := func(r remotes.Resolver) error {
@@ -177,7 +220,6 @@ func Push(ctx context.Context, client *containerd.Client, rawRef string, options
 
 	resolver := docker.NewResolver(resolverOpts)
 	if err = pushFunc(resolver); err != nil {
-		// In some circumstance (e.g. people just use 80 port to support pure http), the error will contain message like "dial tcp <port>: connection refused"
 		if !errors.Is(err, http.ErrSchemeMismatch) && !errutil.IsErrConnectionRefused(err) {
 			return err
 		}
@@ -194,32 +236,6 @@ func Push(ctx context.Context, client *containerd.Client, rawRef string, options
 		log.G(ctx).Info("Hint: you may want to try --insecure-registry to allow plain HTTP (if you are in a trusted network)")
 		return err
 	}
-
-	img, err := client.ImageService().Get(ctx, pushRef)
-	if err != nil {
-		return err
-	}
-	refSpec, err := reference.Parse(pushRef)
-	if err != nil {
-		return err
-	}
-	signRef := fmt.Sprintf("%s@%s", refSpec.String(), img.Target.Digest.String())
-	if err = signutil.Sign(signRef,
-		options.GOptions.Experimental,
-		options.SignOptions); err != nil {
-		return err
-	}
-	if options.GOptions.Snapshotter == "soci" {
-		if err = snapshotterutil.CreateSociIndexV1(ref, options.GOptions, options.AllPlatforms, options.Platforms, options.SociOptions); err != nil {
-			return err
-		}
-		if err = snapshotterutil.PushSoci(ref, options.GOptions, options.AllPlatforms, options.Platforms); err != nil {
-			return err
-		}
-	}
-	if options.Quiet {
-		fmt.Fprintln(options.Stdout, ref)
-	}
 	return nil
 }