feature: split image pull interface

Actually, there is no real Pull interface in containerd, image pull
do not need to unpack but only run a container need. Split Pull
operation into two operations, fetch and unpack. Let pouchd deal with
pulled image.

Signed-off-by: Ace-Tang <aceapril@126.com>

ctrd/image.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"fmt"
 	"io"
-	"net/http"
 	"sync"
 	"time"
 
@@ -177,8 +176,8 @@ func (c *Client) importImage(ctx context.Context, importer ctrdmetaimages.Import
 	return imgs, nil
 }
 
-// PullImage downloads an image from the remote repository.
-func (c *Client) PullImage(ctx context.Context, ref string, authConfig *types.AuthConfig, stream *jsonstream.JSONStream) (containerd.Image, error) {
+// FetchImage fetchs image content from the remote repository.
+func (c *Client) FetchImage(ctx context.Context, ref string, authConfig *types.AuthConfig, stream *jsonstream.JSONStream) (containerd.Image, error) {
 	wrapperCli, err := c.Get(ctx)
 	if err != nil {
 		return nil, fmt.Errorf("failed to get a containerd grpc client: %v", err)
@@ -192,10 +191,8 @@ func (c *Client) PullImage(ctx context.Context, ref string, authConfig *types.Au
 	ongoing := newJobs(ref)
 
 	options := []containerd.RemoteOpt{
-		containerd.WithPullUnpack,
 		containerd.WithSchema1Conversion,
 		containerd.WithResolver(resolver),
-		containerd.WithPullSnapshotter(CurrentSnapshotterName()),
 		containerd.WithPullLabel(snapshots.TypeLabelKey, snapshots.ImageType),
 	}
 
@@ -221,33 +218,23 @@ func (c *Client) PullImage(ctx context.Context, ref string, authConfig *types.Au
 	}()
 
 	// start to pull image.
-	img, err := c.pullImage(ctx, wrapperCli, ref, options)
+	img, err := c.fetchImage(ctx, wrapperCli, ref, options)
 
 	// cancel fetch progress before handle error.
 	cancelProgress()
-	defer stream.Close()
 
 	// wait fetch progress to finish.
 	<-wait
 
 	if err != nil {
-		// Send Error information to client through stream
-		message := jsonstream.JSONMessage{
-			Error: &jsonstream.JSONError{
-				Code:    http.StatusInternalServerError,
-				Message: err.Error(),
-			},
-			ErrorMessage: err.Error(),
-		}
-		stream.WriteObject(message)
 		return nil, err
 	}
 
-	logrus.Infof("success to pull image: %s", img.Name())
+	logrus.Infof("success to fetch image: %s", img.Name())
 	return img, nil
 }
 
-func (c *Client) pullImage(ctx context.Context, wrapperCli *WrapperClient, ref string, options []containerd.RemoteOpt) (containerd.Image, error) {
+func (c *Client) fetchImage(ctx context.Context, wrapperCli *WrapperClient, ref string, options []containerd.RemoteOpt) (containerd.Image, error) {
 	img, err := wrapperCli.client.Pull(ctx, ref, options...)
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to pull image")

ctrd/interface.go

@@ -79,8 +79,8 @@ type ImageAPIClient interface {
 	GetImage(ctx context.Context, ref string) (containerd.Image, error)
 	// ListImages returns the list of containerd.Image filtered by the given conditions.
 	ListImages(ctx context.Context, filter ...string) ([]containerd.Image, error)
-	// PullImage pulls image by the given reference.
-	PullImage(ctx context.Context, ref string, authConfig *types.AuthConfig, stream *jsonstream.JSONStream) (containerd.Image, error)
+	// FetchImage fetchs image content by the given reference.
+	FetchImage(ctx context.Context, ref string, authConfig *types.AuthConfig, stream *jsonstream.JSONStream) (containerd.Image, error)
 	// RemoveImage removes the image by the given reference.
 	RemoveImage(ctx context.Context, ref string) error
 	// ImportImage creates a set of images by tarstream.

daemon/mgr/image.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"net/http"
 	"strings"
 	"time"
 
@@ -132,23 +133,42 @@ func (mgr *ImageManager) PullImage(ctx context.Context, ref string, authConfig *
 
 	pctx, cancel := context.WithCancel(ctx)
 	stream := jsonstream.New(out, nil)
-	wait := make(chan struct{})
 
-	go func() {
-		// wait stream to finish.
-		defer cancel()
+	closeStream := func() {
+		// close and wait stream
+		stream.Close()
 		stream.Wait()
-		close(wait)
-	}()
+		cancel()
+	}
+
+	writeStream := func(err error) {
+		// Send Error information to client through stream
+		message := jsonstream.JSONMessage{
+			Error: &jsonstream.JSONError{
+				Code:    http.StatusInternalServerError,
+				Message: err.Error(),
+			},
+			ErrorMessage: err.Error(),
+		}
+		stream.WriteObject(message)
+		closeStream()
+	}
 
 	namedRef = reference.TrimTagForDigest(reference.WithDefaultTagIfMissing(namedRef))
-	img, err := mgr.client.PullImage(pctx, namedRef.String(), authConfig, stream)
-	// wait goroutine to exit.
-	<-wait
+	img, err := mgr.client.FetchImage(pctx, namedRef.String(), authConfig, stream)
 	if err != nil {
+		writeStream(err)
 		return err
 	}
 
+	// unpack image
+	if err = img.Unpack(ctx, ctrd.CurrentSnapshotterName()); err != nil {
+		writeStream(err)
+		return err
+	}
+
+	closeStream()
+
 	mgr.LogImageEvent(ctx, img.Name(), namedRef.String(), "pull")
 
 	return mgr.StoreImageReference(ctx, img)