refactor: allow custom handlers for processing push/pull status updates

cmd/nerdctl/container_create.go

@@ -24,6 +24,7 @@ import (
 	"github.com/containerd/nerdctl/pkg/clientutil"
 	"github.com/containerd/nerdctl/pkg/cmd/container"
 	"github.com/containerd/nerdctl/pkg/containerutil"
+	"github.com/containerd/nerdctl/pkg/imgutil/jobs"
 	"github.com/spf13/cobra"
 )
 
@@ -396,11 +397,10 @@ func processContainerCreateOptions(cmd *cobra.Command) (opt types.ContainerCreat
 		return
 	}
 	opt.ImagePullOpt = types.ImagePullOptions{
-		GOptions:      opt.GOptions,
-		VerifyOptions: imageVerifyOpt,
-		IPFSAddress:   opt.IPFSAddress,
-		Stdout:        opt.Stdout,
-		Stderr:        opt.Stderr,
+		GOptions:        opt.GOptions,
+		VerifyOptions:   imageVerifyOpt,
+		IPFSAddress:     opt.IPFSAddress,
+		ProgressHandler: jobs.PrintProgress(opt.Stderr),
 	}
 	// #endregion
 

cmd/nerdctl/image_pull.go

@@ -17,9 +17,12 @@
 package main
 
 import (
+	"fmt"
+
 	"github.com/containerd/nerdctl/pkg/api/types"
 	"github.com/containerd/nerdctl/pkg/clientutil"
 	"github.com/containerd/nerdctl/pkg/cmd/image"
+	"github.com/containerd/nerdctl/pkg/imgutil/jobs"
 	"github.com/spf13/cobra"
 )
 
@@ -94,15 +97,14 @@ func processPullCommandFlags(cmd *cobra.Command) (types.ImagePullOptions, error)
 		return types.ImagePullOptions{}, err
 	}
 	return types.ImagePullOptions{
-		GOptions:      globalOptions,
-		VerifyOptions: verifyOptions,
-		AllPlatforms:  allPlatforms,
-		Platform:      platform,
-		Unpack:        unpackStr,
-		Quiet:         quiet,
-		IPFSAddress:   ipfsAddressStr,
-		Stdout:        cmd.OutOrStdout(),
-		Stderr:        cmd.OutOrStderr(),
+		GOptions:        globalOptions,
+		VerifyOptions:   verifyOptions,
+		AllPlatforms:    allPlatforms,
+		Platform:        platform,
+		Unpack:          unpackStr,
+		Quiet:           quiet,
+		IPFSAddress:     ipfsAddressStr,
+		ProgressHandler: jobs.PrintProgress(cmd.OutOrStderr()),
 	}, nil
 }
 
@@ -118,5 +120,9 @@ func pullAction(cmd *cobra.Command, args []string) error {
 	}
 	defer cancel()
 
-	return image.Pull(ctx, client, args[0], options)
+	ref, err := image.Pull(ctx, client, args[0], options)
+	if options.Quiet {
+		fmt.Fprintln(cmd.OutOrStderr(), ref)
+	}
+	return err
 }

cmd/nerdctl/image_push.go

@@ -17,9 +17,12 @@
 package main
 
 import (
+	"fmt"
+
 	"github.com/containerd/nerdctl/pkg/api/types"
 	"github.com/containerd/nerdctl/pkg/clientutil"
 	"github.com/containerd/nerdctl/pkg/cmd/image"
+	"github.com/containerd/nerdctl/pkg/imgutil/jobs"
 	"github.com/spf13/cobra"
 )
 
@@ -111,7 +114,7 @@ func processImagePushOptions(cmd *cobra.Command) (types.ImagePushOptions, error)
 		IpfsAddress:                    ipfsAddress,
 		Quiet:                          quiet,
 		AllowNondistributableArtifacts: allowNonDist,
-		Stdout:                         cmd.OutOrStdout(),
+		ProgressHandler:                jobs.PrintProgress(cmd.OutOrStdout()),
 	}, nil
 }
 
@@ -128,7 +131,11 @@ func pushAction(cmd *cobra.Command, args []string) error {
 	}
 	defer cancel()
 
-	return image.Push(ctx, client, rawRef, options)
+	ref, err := image.Push(ctx, client, rawRef, options)
+	if options.Quiet {
+		fmt.Fprintln(cmd.OutOrStdout(), ref)
+	}
+	return err
 }
 
 func pushShellComplete(cmd *cobra.Command, args []string, toComplete string) ([]string, cobra.ShellCompDirective) {

pkg/api/types/image_types.go

@@ -16,7 +16,11 @@
 
 package types
 
-import "io"
+import (
+	"io"
+
+	"github.com/containerd/nerdctl/pkg/imgutil/jobs"
+)
 
 // ImageListOptions specifies options for `nerdctl image list`.
 type ImageListOptions struct {
@@ -149,7 +153,6 @@ type ImageInspectOptions struct {
 
 // ImagePushOptions specifies options for `nerdctl (image) push`.
 type ImagePushOptions struct {
-	Stdout      io.Writer
 	GOptions    GlobalCommandOptions
 	SignOptions ImageSignOptions
 	// Platforms convert content for a specific platform
@@ -165,14 +168,14 @@ type ImagePushOptions struct {
 	IpfsAddress string
 	// Suppress verbose output
 	Quiet bool
+	// If non-nil, the Push job will pass upload statuses to the handler in small intervals
+	ProgressHandler jobs.StatusHandler
 	// AllowNondistributableArtifacts allow pushing non-distributable artifacts
 	AllowNondistributableArtifacts bool
 }
 
 // ImagePullOptions specifies options for `nerdctl (image) pull`.
 type ImagePullOptions struct {
-	Stdout        io.Writer
-	Stderr        io.Writer
 	GOptions      GlobalCommandOptions
 	VerifyOptions ImageVerifyOptions
 	// Unpack the image for the current single platform (auto/true/false)
@@ -183,6 +186,8 @@ type ImagePullOptions struct {
 	AllPlatforms bool
 	// Suppress verbose output
 	Quiet bool
+	// If non-nil, the Pull job will pass download statuses to the handler in small intervals
+	ProgressHandler jobs.StatusHandler
 	// multiaddr of IPFS API (default uses $IPFS_PATH env variable if defined or local directory ~/.ipfs)
 	IPFSAddress string
 }

pkg/cmd/compose/compose.go

@@ -31,6 +31,7 @@ import (
 	"github.com/containerd/nerdctl/pkg/composer"
 	"github.com/containerd/nerdctl/pkg/composer/serviceparser"
 	"github.com/containerd/nerdctl/pkg/imgutil"
+	"github.com/containerd/nerdctl/pkg/imgutil/jobs"
 	"github.com/containerd/nerdctl/pkg/ipfs"
 	"github.com/containerd/nerdctl/pkg/netutil"
 	"github.com/containerd/nerdctl/pkg/referenceutil"
@@ -110,6 +111,18 @@ func New(client *containerd.Client, globalOptions types.GlobalCommandOptions, op
 			ocispecPlatforms = []ocispec.Platform{parsed} // no append
 		}
 
+		pullCfg := imgutil.PullConfig{
+			Ref:             imageName,
+			Platforms:       ocispecPlatforms,
+			Snapshotter:     globalOptions.Snapshotter,
+			Insecure:        globalOptions.InsecureRegistry,
+			HostsDir:        globalOptions.HostsDir,
+			Mode:            pullMode,
+			Unpack:          nil,
+			Quiet:           quiet,
+			ProgressHandler: jobs.PrintProgress(stderr),
+		}
+
 		// IPFS reference
 		if scheme, ref, err := referenceutil.ParseIPFSRefWithScheme(imageName); err == nil {
 			var ipfsPath string
@@ -124,8 +137,7 @@ func New(client *containerd.Client, globalOptions types.GlobalCommandOptions, op
 				}
 				ipfsPath = dir
 			}
-			_, err = ipfs.EnsureImage(ctx, client, stdout, stderr, globalOptions.Snapshotter, scheme, ref,
-				pullMode, ocispecPlatforms, nil, quiet, ipfsPath)
+			_, err = ipfs.EnsureImage(ctx, client, scheme, ref, ipfsPath, pullCfg)
 			return err
 		}
 
@@ -135,8 +147,7 @@ func New(client *containerd.Client, globalOptions types.GlobalCommandOptions, op
 			return err
 		}
 
-		_, err = imgutil.EnsureImage(ctx, client, stdout, stderr, globalOptions.Snapshotter, ref,
-			pullMode, globalOptions.InsecureRegistry, globalOptions.HostsDir, ocispecPlatforms, nil, quiet)
+		_, err = imgutil.EnsureImage(ctx, client, ref, pullCfg)
 		return err
 	}
 

pkg/cmd/image/pull.go

@@ -34,29 +34,41 @@ import (
 )
 
 // Pull pulls an image specified by `rawRef`.
-func Pull(ctx context.Context, client *containerd.Client, rawRef string, options types.ImagePullOptions) error {
+func Pull(ctx context.Context, client *containerd.Client, rawRef string, options types.ImagePullOptions) (imageRef string, err error) {
 	ocispecPlatforms, err := platformutil.NewOCISpecPlatformSlice(options.AllPlatforms, options.Platform)
 	if err != nil {
-		return err
+		return "", err
 	}
 
 	unpack, err := strutil.ParseBoolOrAuto(options.Unpack)
 	if err != nil {
-		return err
+		return "", err
 	}
 
-	_, err = EnsureImage(ctx, client, rawRef, ocispecPlatforms, "always", unpack, options.Quiet, options)
+	img, err := EnsureImage(ctx, client, rawRef, ocispecPlatforms, "always", unpack, options.Quiet, options)
 	if err != nil {
-		return err
+		return "", err
 	}
 
-	return nil
+	return img.Ref, nil
 }
 
 // EnsureImage pulls an image either from ipfs or from registry.
 func EnsureImage(ctx context.Context, client *containerd.Client, rawRef string, ocispecPlatforms []v1.Platform, pull string, unpack *bool, quiet bool, options types.ImagePullOptions) (*imgutil.EnsuredImage, error) {
 	var ensured *imgutil.EnsuredImage
 
+	pullCfg := imgutil.PullConfig{
+		Ref:             rawRef,
+		Platforms:       ocispecPlatforms,
+		Snapshotter:     options.GOptions.Snapshotter,
+		Insecure:        options.GOptions.InsecureRegistry,
+		HostsDir:        options.GOptions.HostsDir,
+		Mode:            pull,
+		Unpack:          unpack,
+		Quiet:           quiet,
+		ProgressHandler: options.ProgressHandler,
+	}
+
 	if scheme, ref, err := referenceutil.ParseIPFSRefWithScheme(rawRef); err == nil {
 		if options.VerifyOptions.Provider != "none" {
 			return nil, errors.New("--verify flag is not supported on IPFS as of now")
@@ -75,8 +87,7 @@ func EnsureImage(ctx context.Context, client *containerd.Client, rawRef string,
 			ipfsPath = dir
 		}
 
-		ensured, err = ipfs.EnsureImage(ctx, client, options.Stdout, options.Stderr, options.GOptions.Snapshotter, scheme, ref,
-			pull, ocispecPlatforms, unpack, quiet, ipfsPath)
+		ensured, err = ipfs.EnsureImage(ctx, client, scheme, ref, ipfsPath, pullCfg)
 		if err != nil {
 			return nil, err
 		}
@@ -88,8 +99,7 @@ func EnsureImage(ctx context.Context, client *containerd.Client, rawRef string,
 		return nil, err
 	}
 
-	ensured, err = imgutil.EnsureImage(ctx, client, options.Stdout, options.Stderr, options.GOptions.Snapshotter, ref,
-		pull, options.GOptions.InsecureRegistry, options.GOptions.HostsDir, ocispecPlatforms, unpack, quiet)
+	ensured, err = imgutil.EnsureImage(ctx, client, ref, pullCfg)
 	if err != nil {
 		return nil, err
 	}