Use containerd importer

api/server/router/image/backend.go

@@ -30,7 +30,7 @@ type imageBackend interface {
 
 type importExportBackend interface {
 	LoadImage(ctx context.Context, inTar io.ReadCloser, outStream io.Writer, quiet bool) error
-	ImportImage(ctx context.Context, src string, repository, platform string, tag string, msg string, inConfig io.ReadCloser, outStream io.Writer, changes []string) error
+	ImportImage(ctx context.Context, src, repository, tag string, platform *specs.Platform, msg string, inConfig io.ReadCloser, outStream io.Writer, changes []string) error
 	ExportImage(ctx context.Context, names []string, outStream io.Writer) error
 }
 

api/server/router/image/image_routes.go

@@ -81,11 +81,7 @@ func (s *imageRouter) postImagesCreate(ctx context.Context, w http.ResponseWrite
 		// 'err' MUST NOT be defined within this block, we need any error
 		// generated from the download to be available to the output
 		// stream processing below
-		os := ""
-		if platform != nil {
-			os = platform.OS
-		}
-		err = s.backend.ImportImage(ctx, src, repo, os, tag, message, r.Body, output, r.Form["changes"])
+		err = s.backend.ImportImage(ctx, src, repo, tag, platform, message, r.Body, output, r.Form["changes"])
 	}
 	if err != nil {
 		if !output.Flushed() {

daemon/images/image_import.go

@@ -1,128 +1,152 @@
 package images // import "github.com/docker/docker/daemon/images"
 
 import (
+	"bytes"
 	"context"
 	"io"
+	"net/http"
+	"net/url"
+	"strings"
+	"time"
 
+	"github.com/containerd/containerd/archive/compression"
+	"github.com/containerd/containerd/images"
+	"github.com/containerd/containerd/images/archive"
+	"github.com/docker/distribution/reference"
+	"github.com/docker/docker/builder/remotecontext"
 	"github.com/docker/docker/errdefs"
+	"github.com/docker/docker/layer"
+	"github.com/docker/docker/pkg/progress"
+	"github.com/docker/docker/pkg/streamformatter"
+	ocispec "github.com/opencontainers/image-spec/specs-go/v1"
 	"github.com/pkg/errors"
 )
 
 // ImportImage imports an image, getting the archived layer data either from
 // inConfig (if src is "-"), or from a URI specified in src. Progress output is
 // written to outStream. Repository and tag names can optionally be given in
 // the repo and tag arguments, respectively.
-func (i *ImageService) ImportImage(ctx context.Context, src string, repository, os string, tag string, msg string, inConfig io.ReadCloser, outStream io.Writer, changes []string) error {
-	return errdefs.NotImplemented(errors.New("import image not implemented"))
-	/*
-		var (
-			rc     io.ReadCloser
-			resp   *http.Response
-			newRef reference.Named
-		)
-
-		// Default the operating system if not supplied.
-		if os == "" {
-			os = runtime.GOOS
-		}
-
-		if repository != "" {
-			var err error
-			newRef, err = reference.ParseNormalizedNamed(repository)
-			if err != nil {
-				return errdefs.InvalidParameter(err)
-			}
-			if _, isCanonical := newRef.(reference.Canonical); isCanonical {
-				return errdefs.InvalidParameter(errors.New("cannot import digest reference"))
-			}
-
-			if tag != "" {
-				//set newRef
-				_, err = reference.WithTag(newRef, tag)
-				if err != nil {
-					return errdefs.InvalidParameter(err)
-				}
-			}
-		}
+func (i *ImageService) ImportImage(
+	ctx context.Context,
+	src string,
+	repository string,
+	tag string,
+	platform *ocispec.Platform,
+	msg string,
+	inConfig io.ReadCloser,
+	outStream io.Writer,
+	changes []string,
+) error {
+	var (
+		rc     io.ReadCloser
+		resp   *http.Response
+		newRef reference.Named
+	)
 
-		config, err := dockerfile.BuildFromConfig(&container.Config{}, changes, os)
+	if repository != "" {
+		var err error
+		newRef, err = reference.ParseNormalizedNamed(repository)
 		if err != nil {
-			return err
+			return errdefs.InvalidParameter(err)
+		}
+		if _, isCanonical := newRef.(reference.Canonical); isCanonical {
+			return errdefs.InvalidParameter(errors.New("cannot import digest reference"))
 		}
-		if src == "-" {
-			rc = inConfig
-		} else {
-			inConfig.Close()
-			if len(strings.Split(src, "://")) == 1 {
-				src = "http://" + src
-			}
-			u, err := url.Parse(src)
-			if err != nil {
-				return errdefs.InvalidParameter(err)
-			}
 
-			resp, err = remotecontext.GetWithStatusError(u.String())
+		if tag != "" {
+			_, err = reference.WithTag(newRef, tag)
 			if err != nil {
-				return err
+				return errdefs.InvalidParameter(err)
 			}
-			outStream.Write(streamformatter.FormatStatus("", "Downloading from %s", u))
-			progressOutput := streamformatter.NewJSONProgressOutput(outStream, true)
-			rc = progress.NewProgressReader(resp.Body, progressOutput, resp.ContentLength, "", "Importing")
 		}
+	}
 
-		defer rc.Close()
-		if len(msg) == 0 {
-			msg = "Imported from " + src
+	if src == "-" {
+		rc = inConfig
+	} else {
+		inConfig.Close()
+		if strings.Count(src, "://") == 0 {
+			src = "http://" + src
 		}
-
-		inflatedLayerData, err := archive.DecompressStream(rc)
+		u, err := url.Parse(src)
 		if err != nil {
-			return err
-		}
-		l, err := i.layerStores[os].Register(inflatedLayerData, "")
-		if err != nil {
-			return err
-		}
-		defer layer.ReleaseAndLog(i.layerStores[os], l)
-
-		created := time.Now().UTC()
-		imgConfig, err := json.Marshal(&image.Image{
-			V1Image: image.V1Image{
-				DockerVersion: dockerversion.Version,
-				Config:        config,
-				Architecture:  runtime.GOARCH,
-				OS:            os,
-				Created:       created,
-				Comment:       msg,
-			},
-			RootFS: &image.RootFS{
-				Type:    "layers",
-				DiffIDs: []layer.DiffID{l.DiffID()},
-			},
-			History: []image.History{{
-				Created: created,
-				Comment: msg,
-			}},
-		})
-		if err != nil {
-			return err
+			return errdefs.InvalidParameter(err)
 		}
 
-		// TODO(containerd): Use content store + image store
-		id, err := i.imageStore.Create(imgConfig)
+		resp, err = remotecontext.GetWithStatusError(u.String())
 		if err != nil {
 			return err
 		}
+		outStream.Write(streamformatter.FormatStatus("", "Downloading from %s", u))
+		progressOutput := streamformatter.NewJSONProgressOutput(outStream, true)
+		rc = progress.NewProgressReader(resp.Body, progressOutput, resp.ContentLength, "", "Importing")
+	}
 
-		// FIXME: connect with commit code and call refstore directly
-		if newRef != nil {
-			if err := i.TagImageWithReference(dgst, newRef); err != nil {
-				return err
-			}
+	defer rc.Close()
+	if len(msg) == 0 {
+		msg = "Imported from " + src
+	}
+
+	// TODO: doing this so that I can quickly test how this works, but we probably
+	// don't want to buffer this entire thing into memory. Use a pipe or similar.
+	buf := &bytes.Buffer{}
+	tee := io.TeeReader(rc, buf)
+
+	inflatedLayerData, err := compression.DecompressStream(tee)
+	if err != nil {
+		return err
+	}
+	defer inflatedLayerData.Close()
+
+	if platform == nil {
+		platform = &i.defaultPlatform
+	}
+
+	// Tee into layerstore and content store
+	layerStore, err := i.GetLayerStore(*platform)
+	if err != nil {
+		return err
+	}
+	l, err := layerStore.Register(inflatedLayerData, "")
+	if err != nil {
+		return err
+	}
+	defer layer.ReleaseAndLog(layerStore, l)
+
+	ctx, done, err := i.client.WithLease(ctx)
+	if err != nil {
+		return err
+	}
+	defer done(ctx)
+
+	desc, err := archive.ImportIndex(ctx, i.client.ContentStore(), buf)
+	if err != nil {
+		return err
+	}
+
+	created := time.Now().UTC()
+	img := images.Image{
+		Name: src,
+		// Labels: map[string]string
+		Target:    desc,
+		CreatedAt: created,
+	}
+
+	img, err = i.client.ImageService().Create(ctx, img)
+	if err != nil {
+		return err
+	}
+
+	// FIXME: connect with commit code and call refstore directly
+	if newRef != nil {
+		if err := i.TagImageWithReference(ctx, img.Target, newRef); err != nil {
+			return err
 		}
+	}
 
-		i.LogImageEvent(ctx, id.String(), id.String(), "import")
-		outStream.Write(streamformatter.FormatStatus("", id.String()))
-		return nil
-	*/
+	// TODO: is the ID the digest or the name?
+	id := img.Target.Digest.String()
+	i.LogImageEvent(ctx, id, img.Name, "import")
+	outStream.Write(streamformatter.FormatStatus("", id))
+	return nil
 }