Merge pull request #1391 from fuweid/feature_add_image_load

feature: add pouch load functionality

apis/server/image_bridge.go

@@ -134,3 +134,18 @@ func (s *Server) postImageTag(ctx context.Context, rw http.ResponseWriter, req *
 	rw.WriteHeader(http.StatusCreated)
 	return nil
 }
+
+// loadImage loads an image by http tar stream.
+func (s *Server) loadImage(ctx context.Context, rw http.ResponseWriter, req *http.Request) error {
+	imageName := req.FormValue("name")
+	if imageName == "" {
+		imageName = "unknown/unknown"
+	}
+
+	if err := s.ImageMgr.LoadImage(ctx, imageName, req.Body); err != nil {
+		return err
+	}
+
+	rw.WriteHeader(http.StatusOK)
+	return nil
+}

apis/server/router.go

@@ -61,6 +61,7 @@ func initRoute(s *Server) http.Handler {
 	s.addRoute(r, http.MethodDelete, "/images/{name:.*}", s.removeImage)
 	s.addRoute(r, http.MethodGet, "/images/{name:.*}/json", s.getImage)
 	s.addRoute(r, http.MethodPost, "/images/{name:.*}/tag", s.postImageTag)
+	s.addRoute(r, http.MethodPost, "/images/load", withCancelHandler(s.loadImage))
 
 	// volume
 	s.addRoute(r, http.MethodGet, "/volumes", s.listVolume)

apis/swagger.yml

@@ -152,6 +152,32 @@ paths:
           description: "A base64-encoded auth configuration. [See the authentication section for details.](#section/Authentication)"
           type: "string"
 
+  /images/load:
+     post:
+      summary: "Import images"
+      description: |
+        Load a set of images by oci.v1 format tar stream
+      consumes:
+        - application/x-tar
+      responses:
+        200:
+          description: "no error"
+        500:
+          description: "server error"
+          schema:
+            $ref: "#/responses/500ErrorResponse"
+      parameters:
+        - name: "imageTarStream"
+          in: "body"
+          description: "tar stream containing images"
+          schema:
+            type: "string"
+            format: "binary"
+        - name: "name"
+          in: "query"
+          description: "set the image name for the tar stream, default unknown/unknown"
+          type: "string"
+
   /images/{imageid}/json:
     get:
       summary: "Inspect a image"

cli/load.go

@@ -0,0 +1,71 @@
+package main
+
+import (
+	"context"
+	"io"
+	"os"
+
+	"github.com/spf13/cobra"
+)
+
+// loadDescription is used to describe load command in detail and auto generate command doc.
+var loadDescription = "load a set of images by tar stream"
+
+// LoadCommand use to implement 'load' command.
+type LoadCommand struct {
+	baseCommand
+	input string
+}
+
+// Init initialize load command.
+func (l *LoadCommand) Init(c *Cli) {
+	l.cli = c
+	l.cmd = &cobra.Command{
+		Use:   "load [OPTIONS] [IMAGE_NAME]",
+		Short: "load a set of images from a tar archive or STDIN",
+		Long:  loadDescription,
+		Args:  cobra.MaximumNArgs(1),
+		RunE: func(_ *cobra.Command, args []string) error {
+			return l.runLoad(args)
+		},
+		Example: loadExample(),
+	}
+	l.addFlags()
+}
+
+// addFlags adds flags for specific command.
+func (l *LoadCommand) addFlags() {
+	flagSet := l.cmd.Flags()
+	flagSet.StringVarP(&l.input, "input", "i", "", "Read from tar archive file, instead of STDIN")
+}
+
+// runLoad is the entry of load command.
+func (l *LoadCommand) runLoad(args []string) error {
+	ctx := context.Background()
+	apiClient := l.cli.Client()
+
+	var (
+		in        io.Reader = os.Stdin
+		imageName           = ""
+	)
+
+	if l.input != "" {
+		file, err := os.Open(l.input)
+		if err != nil {
+			return err
+		}
+
+		defer file.Close()
+		in = file
+	}
+
+	if len(args) > 0 {
+		imageName = args[0]
+	}
+	return apiClient.ImageLoad(ctx, imageName, in)
+}
+
+// loadExample shows examples in load command, and is used in auto-generated cli docs.
+func loadExample() string {
+	return `$ pouch load -i busybox.tar busybox`
+}

cli/main.go

@@ -30,6 +30,7 @@ func main() {
 	cli.AddCommand(base, &VolumeCommand{})
 	cli.AddCommand(base, &NetworkCommand{})
 	cli.AddCommand(base, &TagCommand{})
+	cli.AddCommand(base, &LoadCommand{})
 
 	cli.AddCommand(base, &InspectCommand{})
 	cli.AddCommand(base, &RenameCommand{})

client/image_load.go

@@ -0,0 +1,26 @@
+package client
+
+import (
+	"context"
+	"io"
+	"net/url"
+)
+
+// ImageLoad requests daemon to load an image from tarstream.
+func (client *APIClient) ImageLoad(ctx context.Context, imageName string, reader io.Reader) error {
+	q := url.Values{}
+	if imageName != "" {
+		q.Set("name", imageName)
+	}
+
+	headers := map[string][]string{}
+	headers["Content-Type"] = []string{"application/x-tar"}
+
+	resp, err := client.postRawData(ctx, "/images/load", q, reader, headers)
+	if err != nil {
+		return err
+	}
+
+	ensureCloseReader(resp)
+	return nil
+}

client/image_load_test.go

@@ -0,0 +1,57 @@
+package client
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"strings"
+	"testing"
+)
+
+func TestImageLoadServerError(t *testing.T) {
+	expectedError := "Server error"
+
+	client := &APIClient{
+		HTTPCli: newMockClient(errorMockResponse(http.StatusInternalServerError, expectedError)),
+	}
+
+	err := client.ImageLoad(context.Background(), "test_image_load_500", nil)
+	if err == nil || !strings.Contains(err.Error(), expectedError) {
+		t.Fatalf("expected (%v), got (%v)", expectedError, err)
+	}
+}
+
+func TestImageLoadOK(t *testing.T) {
+	expectedURL := "/images/load"
+	expectedImageName := "test_image_load_ok"
+
+	httpClient := newMockClient(func(req *http.Request) (*http.Response, error) {
+		if !strings.HasPrefix(req.URL.Path, expectedURL) {
+			return nil, fmt.Errorf("expected URL '%s', got '%s'", expectedURL, req.URL)
+		}
+
+		if req.Method != "POST" {
+			return nil, fmt.Errorf("expected POST method, got %s", req.Method)
+		}
+
+		if got := req.FormValue("name"); got != expectedImageName {
+			return nil, fmt.Errorf("expected (%s), got %s", expectedImageName, got)
+		}
+
+		return &http.Response{
+			StatusCode: http.StatusOK,
+			Body:       ioutil.NopCloser(bytes.NewReader([]byte(""))),
+		}, nil
+	})
+
+	client := &APIClient{
+		HTTPCli: httpClient,
+	}
+
+	if err := client.ImageLoad(context.Background(), expectedImageName, nil); err != nil {
+		t.Fatal(err)
+	}
+
+}

client/interface.go

@@ -48,6 +48,7 @@ type ImageAPIClient interface {
 	ImagePull(ctx context.Context, name, tag, encodedAuth string) (io.ReadCloser, error)
 	ImageRemove(ctx context.Context, name string, force bool) error
 	ImageTag(ctx context.Context, image string, tag string) error
+	ImageLoad(ctx context.Context, name string, r io.Reader) error
 }
 
 // VolumeAPIClient defines methods of Volume client.

client/request.go

@@ -42,15 +42,29 @@ func (client *APIClient) get(ctx context.Context, path string, query url.Values,
 }
 
 func (client *APIClient) post(ctx context.Context, path string, query url.Values, obj interface{}, headers map[string][]string) (*Response, error) {
-	return client.sendRequest(ctx, "POST", path, query, obj, headers)
+	body, err := objectToJSONStream(obj)
+	if err != nil {
+		return nil, err
+	}
+
+	return client.sendRequest(ctx, "POST", path, query, body, headers)
+}
+
+func (client *APIClient) postRawData(ctx context.Context, path string, query url.Values, data io.Reader, headers map[string][]string) (*Response, error) {
+	return client.sendRequest(ctx, "POST", path, query, data, headers)
 }
 
 func (client *APIClient) delete(ctx context.Context, path string, query url.Values, headers map[string][]string) (*Response, error) {
 	return client.sendRequest(ctx, "DELETE", path, query, nil, headers)
 }
 
 func (client *APIClient) hijack(ctx context.Context, path string, query url.Values, obj interface{}, header map[string][]string) (net.Conn, *bufio.Reader, error) {
-	req, err := client.newRequest("POST", path, query, obj, header)
+	body, err := objectToJSONStream(obj)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	req, err := client.newRequest("POST", path, query, body, header)
 	if err != nil {
 		return nil, nil, err
 	}
@@ -81,20 +95,7 @@ func (client *APIClient) hijack(ctx context.Context, path string, query url.Valu
 	return rwc, br, nil
 }
 
-func (client *APIClient) newRequest(method, path string, query url.Values, obj interface{}, header map[string][]string) (*http.Request, error) {
-	var body io.Reader
-	if method == "POST" {
-		if obj != nil {
-			b, err := json.Marshal(obj)
-			if err != nil {
-				return nil, err
-			}
-			body = bytes.NewReader(b)
-		} else {
-			body = bytes.NewReader([]byte{})
-		}
-	}
-
+func (client *APIClient) newRequest(method, path string, query url.Values, body io.Reader, header map[string][]string) (*http.Request, error) {
 	fullPath := client.baseURL + client.GetAPIPath(path, query)
 	req, err := http.NewRequest(method, fullPath, body)
 	if err != nil {
@@ -110,8 +111,8 @@ func (client *APIClient) newRequest(method, path string, query url.Values, obj i
 	return req, err
 }
 
-func (client *APIClient) sendRequest(ctx context.Context, method, path string, query url.Values, obj interface{}, headers map[string][]string) (*Response, error) {
-	req, err := client.newRequest(method, path, query, obj, headers)
+func (client *APIClient) sendRequest(ctx context.Context, method, path string, query url.Values, body io.Reader, headers map[string][]string) (*Response, error) {
+	req, err := client.newRequest(method, path, query, body, headers)
 	if err != nil {
 		return nil, err
 	}
@@ -164,3 +165,15 @@ func cancellableDo(ctx context.Context, client *http.Client, req *http.Request)
 		return resp.response, resp.err
 	}
 }
+
+func objectToJSONStream(obj interface{}) (io.Reader, error) {
+	if obj != nil {
+		b, err := json.Marshal(obj)
+		if err != nil {
+			return nil, err
+		}
+		return bytes.NewReader(b), nil
+	}
+
+	return nil, nil
+}

ctrd/image.go

@@ -3,6 +3,7 @@ package ctrd
 import (
 	"context"
 	"fmt"
+	"io"
 	"net/http"
 	"sync"
 	"time"
@@ -64,6 +65,31 @@ func (c *Client) RemoveImage(ctx context.Context, ref string) error {
 	return nil
 }
 
+// ImportImage creates a set of images by tarstream.
+//
+// NOTE: One tar may have several manifests.
+func (c *Client) ImportImage(ctx context.Context, importer ctrdmetaimages.Importer, reader io.Reader) ([]containerd.Image, error) {
+	wrapperCli, err := c.Get(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get a containerd grpc client: %v", err)
+	}
+
+	// NOTE: The import will store the data into boltdb. But the unpack may
+	// fail. It is not transaction.
+	imgs, err := wrapperCli.client.Import(ctx, importer, reader)
+	if err != nil {
+		return nil, err
+	}
+
+	for _, img := range imgs {
+		err = img.Unpack(ctx, containerd.DefaultSnapshotter)
+		if err != nil {
+			return nil, err
+		}
+	}
+	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) {
 	wrapperCli, err := c.Get(ctx)

ctrd/interface.go

@@ -2,6 +2,7 @@ package ctrd
 
 import (
 	"context"
+	"io"
 	"time"
 
 	"github.com/alibaba/pouch/apis/types"
@@ -67,6 +68,8 @@ type ImageAPIClient interface {
 	PullImage(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.
+	ImportImage(ctx context.Context, importer ctrdmetaimages.Importer, reader io.Reader) ([]containerd.Image, error)
 }
 
 // SnapshotAPIClient provides access to containerd snapshot features

daemon/mgr/image.go

@@ -46,6 +46,9 @@ type ImageMgr interface {
 
 	// CheckReference returns imageID, actual reference and primary reference.
 	CheckReference(ctx context.Context, idOrRef string) (digest.Digest, reference.Named, reference.Named, error)
+
+	// LoadImage creates a set of images by tarstream.
+	LoadImage(ctx context.Context, imageName string, tarstream io.ReadCloser) error
 }
 
 // ImageManager is an implementation of interface ImageMgr.
@@ -345,7 +348,7 @@ func (mgr *ImageManager) updateLocalStore() error {
 
 	for _, img := range imgs {
 		if err := mgr.storeImageReference(ctx, img); err != nil {
-			return err
+			logrus.Warnf("failed to load the image reference into local store: %v", err)
 		}
 	}
 	return nil