initial implementation for image management

Signed-off-by: Mike Brown <brownwm@us.ibm.com>

.travis.yml

@@ -1,6 +1,5 @@
 language: go
 go:
-    - 1.6.x
     - 1.7.x
     - 1.8.x
     - tip

Makefile

@@ -69,7 +69,7 @@ cri-containerd: check-gopath
 	   $(PROJECT)/cmd/cri-containerd
 
 test:
-	go test -timeout=1m -v -race ./pkg/... $(BUILD_TAGS)
+	go test -timeout=10m -v -race ./pkg/... $(BUILD_TAGS)
 
 clean:
 	rm -f $(BUILD_DIR)/cri-containerd

pkg/metadata/image_metadata.go

@@ -0,0 +1,150 @@
+/*
+Copyright 2017 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package metadata
+
+import (
+	"encoding/json"
+
+	"github.com/kubernetes-incubator/cri-containerd/pkg/metadata/store"
+)
+
+// The code is very similar to sandbox.go, but there is no template support
+// in golang, thus similar files for different types.
+// TODO(random-liu): Figure out a way to simplify this.
+// TODO(random-liu): Handle versioning
+
+// imageMetadataVersion is current version of image metadata.
+const imageMetadataVersion = "v1" // nolint
+
+// versionedImageMetadata is the internal struct representing the versioned
+// image metadata
+// nolint
+type versionedImageMetadata struct {
+	// Version indicates the version of the versioned image metadata.
+	Version string `json:"version,omitempty"`
+	ImageMetadata
+}
+
+// ImageMetadata is the unversioned image metadata.
+type ImageMetadata struct {
+	// Id of the image. Normally the Digest
+	ID string `json:"id,omitempty"`
+	// Other names by which this image is known.
+	RepoTags []string `json:"repo_tags,omitempty"`
+	// Digests by which this image is known.
+	RepoDigests []string `json:"repo_digests,omitempty"`
+	// Size of the image in bytes. Must be > 0.
+	Size uint64 `json:"size,omitempty"`
+}
+
+// ImageMetadataUpdateFunc is the function used to update ImageMetadata.
+type ImageMetadataUpdateFunc func(ImageMetadata) (ImageMetadata, error)
+
+// imageMetadataToStoreUpdateFunc generates a metadata store UpdateFunc from ImageMetadataUpdateFunc.
+func imageMetadataToStoreUpdateFunc(u ImageMetadataUpdateFunc) store.UpdateFunc {
+	return func(data []byte) ([]byte, error) {
+		meta := &ImageMetadata{}
+		if err := json.Unmarshal(data, meta); err != nil {
+			return nil, err
+		}
+		newMeta, err := u(*meta)
+		if err != nil {
+			return nil, err
+		}
+		return json.Marshal(newMeta)
+	}
+}
+
+// ImageMetadataStore is the store for metadata of all images.
+type ImageMetadataStore interface {
+	// Create creates an image's metadata from ImageMetadata in the store.
+	Create(ImageMetadata) error
+	// Get gets the specified image metadata.
+	Get(string) (*ImageMetadata, error)
+	// Update updates a specified image metatdata.
+	Update(string, ImageMetadataUpdateFunc) error
+	// List lists all image metadatas.
+	List() ([]*ImageMetadata, error)
+	// Delete deletes the image's metatdata from the store.
+	Delete(string) error
+}
+
+// imageMetadataStore is an implmentation of ImageMetadataStore.
+type imageMetadataStore struct {
+	store store.MetadataStore
+}
+
+// NewImageMetadataStore creates an ImageMetadataStore from a basic MetadataStore.
+func NewImageMetadataStore(store store.MetadataStore) ImageMetadataStore {
+	return &imageMetadataStore{store: store}
+}
+
+// Create creates a image's metadata from ImageMetadata in the store.
+func (s *imageMetadataStore) Create(metadata ImageMetadata) error {
+	data, err := json.Marshal(&metadata)
+	if err != nil {
+		return err
+	}
+	return s.store.Create(metadata.ID, data)
+}
+
+// Get gets the specified image metadata.
+func (s *imageMetadataStore) Get(digest string) (*ImageMetadata, error) {
+	data, err := s.store.Get(digest)
+	if err != nil {
+		return nil, err
+	}
+	// Return nil without error if the corresponding metadata
+	// does not exist.
+	if data == nil {
+		return nil, nil
+	}
+	imageMetadata := &ImageMetadata{}
+	if err := json.Unmarshal(data, imageMetadata); err != nil {
+		return nil, err
+	}
+	return imageMetadata, nil
+}
+
+// Update updates a specified image's metadata. The function is running in a
+// transaction. Update will not be applied when the update function
+// returns error.
+func (s *imageMetadataStore) Update(digest string, u ImageMetadataUpdateFunc) error {
+	return s.store.Update(digest, imageMetadataToStoreUpdateFunc(u))
+}
+
+// List lists all image metadata.
+func (s *imageMetadataStore) List() ([]*ImageMetadata, error) {
+	allData, err := s.store.List()
+	if err != nil {
+		return nil, err
+	}
+	var imageMetadataA []*ImageMetadata
+	for _, data := range allData {
+		imageMetadata := &ImageMetadata{}
+		if err := json.Unmarshal(data, imageMetadata); err != nil {
+			return nil, err
+		}
+		imageMetadataA = append(imageMetadataA, imageMetadata)
+	}
+	return imageMetadataA, nil
+}
+
+// Delete deletes the image metadata from the store.
+func (s *imageMetadataStore) Delete(digest string) error {
+	return s.store.Delete(digest)
+}

pkg/metadata/image_metadata_test.go

@@ -0,0 +1,87 @@
+/*
+Copyright 2017 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package metadata
+
+import (
+	"testing"
+
+	assertlib "github.com/stretchr/testify/assert"
+
+	"github.com/kubernetes-incubator/cri-containerd/pkg/metadata/store"
+)
+
+func TestImageMetadataStore(t *testing.T) {
+	imageMetadataMap := map[string]*ImageMetadata{
+		"1": {
+			ID:   "1", // TODO(mikebrow): fill
+			Size: 10,
+		},
+		"2": {
+			ID:   "2",
+			Size: 20,
+		},
+		"3": {
+			ID:   "3",
+			Size: 30,
+		},
+	}
+	assert := assertlib.New(t)
+
+	s := NewImageMetadataStore(store.NewMetadataStore())
+
+	t.Logf("should be able to create image metadata")
+	for _, meta := range imageMetadataMap {
+		assert.NoError(s.Create(*meta))
+	}
+
+	t.Logf("should be able to get image metadata")
+	for id, expectMeta := range imageMetadataMap {
+		meta, err := s.Get(id)
+		assert.NoError(err)
+		assert.Equal(expectMeta, meta)
+	}
+
+	t.Logf("should be able to list image metadata")
+	imgs, err := s.List()
+	assert.NoError(err)
+	assert.Len(imgs, 3)
+
+	t.Logf("should be able to update image metadata")
+	testID := "2"
+	newSize := uint64(200)
+	expectMeta := *imageMetadataMap[testID]
+	expectMeta.Size = newSize
+	err = s.Update(testID, func(o ImageMetadata) (ImageMetadata, error) {
+		o.Size = newSize
+		return o, nil
+	})
+	assert.NoError(err)
+	newMeta, err := s.Get(testID)
+	assert.NoError(err)
+	assert.Equal(&expectMeta, newMeta)
+
+	t.Logf("should be able to delete image metadata")
+	assert.NoError(s.Delete(testID))
+	imgs, err = s.List()
+	assert.NoError(err)
+	assert.Len(imgs, 2)
+
+	t.Logf("get should return nil without error after deletion")
+	meta, err := s.Get(testID)
+	assert.NoError(err)
+	assert.Nil(meta)
+}

pkg/server/image_list.go

@@ -17,14 +17,34 @@ limitations under the License.
 package server
 
 import (
-	"errors"
+	"fmt"
-
 	"golang.org/x/net/context"
 
 	"k8s.io/kubernetes/pkg/kubelet/api/v1alpha1/runtime"
 )
 
 // ListImages lists existing images.
+// TODO(mikebrow): add filters
+// TODO(mikebrow): harden api
 func (c *criContainerdService) ListImages(ctx context.Context, r *runtime.ListImagesRequest) (*runtime.ListImagesResponse, error) {
-	return nil, errors.New("not implemented")
+	imageMetadataA, err := c.imageMetadataStore.List()
+	if err != nil {
+		return nil, fmt.Errorf("failed to list image metadata from store %v", err)
+	}
+	// TODO(mikebrow): Get the id->tags, and id->digest mapping from our checkpoint;
+	// Get other information from containerd image/content store
+
+	var images []*runtime.Image
+	for _, image := range imageMetadataA { // TODO(mikebrow): write a ImageMetadata to runtime.Image converter
+		ri := &runtime.Image{
+			Id:          image.ID,
+			RepoTags:    image.RepoTags,
+			RepoDigests: image.RepoDigests,
+			Size_:       image.Size,
+			// TODO(mikebrow): Uid and Username?
+		}
+		images = append(images, ri)
+	}
+
+	return &runtime.ListImagesResponse{Images: images}, nil
 }

pkg/server/image_pull.go

@@ -17,14 +17,179 @@ limitations under the License.
 package server
 
 import (
-	"errors"
+	"encoding/json"
+	"fmt"
 
+	"github.com/kubernetes-incubator/cri-containerd/pkg/metadata"
+
+	"github.com/containerd/containerd/content"
+	containerdimages "github.com/containerd/containerd/images"
+	"github.com/containerd/containerd/remotes"
+	"github.com/containerd/containerd/remotes/docker"
+	imagespec "github.com/opencontainers/image-spec/specs-go/v1"
 	"golang.org/x/net/context"
-
 	"k8s.io/kubernetes/pkg/kubelet/api/v1alpha1/runtime"
 )
 
 // PullImage pulls an image with authentication config.
+// TODO(mikebrow): add authentication
+// TODO(mikebrow): harden api (including figuring out at what layer we should be blocking on duplicate requests.)
 func (c *criContainerdService) PullImage(ctx context.Context, r *runtime.PullImageRequest) (*runtime.PullImageResponse, error) {
-	return nil, errors.New("not implemented")
+	var (
+		err  error
+		size int64
+		desc imagespec.Descriptor
+	)
+
+	image := r.GetImage().Image
+
+	if desc, size, err = c.pullImage(ctx, image); err != nil {
+		return nil, fmt.Errorf("failed to pull image %q: %v", image, err)
+	}
+	digest := desc.Digest.String() // TODO(mikebrow): add truncIndex for image id
+
+	// TODO(mikebrow): pass a metadata struct to pullimage and fill in the tags/digests
+	// store the image metadata
+	// TODO(mikebrow): consider what to do if pullimage was called and metadata already exists (error? udpate?)
+	meta := &metadata.ImageMetadata{
+		ID:          digest,
+		RepoTags:    []string{image},
+		RepoDigests: []string{digest},
+		Size:        uint64(size), // TODO(mikebrow):  compressed or uncompressed size? using compressed
+	}
+	if err = c.imageMetadataStore.Create(*meta); err != nil {
+		return &runtime.PullImageResponse{ImageRef: digest},
+			fmt.Errorf("pulled image `%q` but failed to store metadata for digest: %s err: %v", image, digest, err)
+	}
+
+	// Return the image digest
+	return &runtime.PullImageResponse{ImageRef: digest}, err
+}
+
+// imageReferenceResolver attempts to resolve the image reference into a name
+// and manifest via the containerd library call..
+//
+// The argument `ref` should be a scheme-less URI representing the remote.
+// Structurally, it has a host and path. The "host" can be used to directly
+// reference a specific host or be matched against a specific handler.
+//
+// The returned name should be used to identify the referenced entity.
+// Dependending on the remote namespace, this may be immutable or mutable.
+// While the name may differ from ref, it should itself be a valid ref.
+//
+// If the resolution fails, an error will be returned.
+// TODO(mikebrow) add config.json based image.Config as an additional return value from this resolver fn()
+func (c *criContainerdService) imageReferenceResolver(ctx context.Context, ref string) (resolvedImageName string, manifest imagespec.Manifest, compressedSize uint64, err error) {
+	var (
+		size    int64
+		desc    imagespec.Descriptor
+		fetcher remotes.Fetcher
+	)
+
+	// Resolve the image name; place that in the image store; then dispatch
+	// a handler to fetch the object for the manifest
+	resolver := docker.NewResolver()
+	resolvedImageName, desc, fetcher, err = resolver.Resolve(ctx, ref)
+	if err != nil {
+		return resolvedImageName, manifest, compressedSize, fmt.Errorf("failed to resolve ref %q: err: %v", ref, err)
+	}
+
+	err = c.imageStore.Put(ctx, resolvedImageName, desc)
+	if err != nil {
+		return resolvedImageName, manifest, compressedSize, fmt.Errorf("failed to put %q: desc: %v err: %v", resolvedImageName, desc, err)
+	}
+
+	err = containerdimages.Dispatch(
+		ctx,
+		remotes.FetchHandler(c.contentIngester, fetcher),
+		desc)
+	if err != nil {
+		return resolvedImageName, manifest, compressedSize, fmt.Errorf("failed to fetch %q: desc: %v err: %v", resolvedImageName, desc, err)
+	}
+
+	image, err := c.imageStore.Get(ctx, resolvedImageName)
+	if err != nil {
+		return resolvedImageName, manifest, compressedSize,
+			fmt.Errorf("get failed for image:%q err: %v", resolvedImageName, err)
+	}
+	p, err := content.ReadBlob(ctx, c.contentProvider, image.Target.Digest)
+	if err != nil {
+		return resolvedImageName, manifest, compressedSize,
+			fmt.Errorf("readblob failed for digest:%q err: %v", image.Target.Digest, err)
+	}
+	err = json.Unmarshal(p, &manifest)
+	if err != nil {
+		return resolvedImageName, manifest, compressedSize,
+			fmt.Errorf("unmarshal blob to manifest failed for digest:%q err: %v", image.Target.Digest, err)
+	}
+	size, err = image.Size(ctx, c.contentProvider)
+	if err != nil {
+		return resolvedImageName, manifest, compressedSize,
+			fmt.Errorf("size failed for image:%q %v", image.Target.Digest, err)
+	}
+	compressedSize = uint64(size)
+	return resolvedImageName, manifest, compressedSize, nil
+}
+
+func (c *criContainerdService) pullImage(ctx context.Context, ref string) (imagespec.Descriptor, int64, error) {
+	var (
+		err               error
+		size              int64
+		desc              imagespec.Descriptor
+		resolvedImageName string
+		fetcher           remotes.Fetcher
+	)
+
+	// Resolve the image name; place that in the image store; then dispatch
+	// a handler for a sequence of handlers which: 1) fetch the object using a
+	// FetchHandler; and 3) recurse through any sub-layers via a ChildrenHandler
+	resolver := docker.NewResolver()
+
+	resolvedImageName, desc, fetcher, err = resolver.Resolve(ctx, ref)
+	if err != nil {
+		return desc, size, fmt.Errorf("failed to resolve ref %q: err: %v", ref, err)
+	}
+
+	err = c.imageStore.Put(ctx, resolvedImageName, desc)
+	if err != nil {
+		return desc, size, fmt.Errorf("failed to put %q: desc: %v err: %v", resolvedImageName, desc, err)
+	}
+
+	err = containerdimages.Dispatch(
+		ctx,
+		containerdimages.Handlers(
+			remotes.FetchHandler(c.contentIngester, fetcher),
+			containerdimages.ChildrenHandler(c.contentProvider)),
+		desc)
+	if err != nil {
+		return desc, size, fmt.Errorf("failed to fetch %q: desc: %v err: %v", resolvedImageName, desc, err)
+	}
+
+	image, err := c.imageStore.Get(ctx, resolvedImageName)
+	if err != nil {
+		return desc, size,
+			fmt.Errorf("get failed for image:%q err: %v", resolvedImageName, err)
+	}
+	p, err := content.ReadBlob(ctx, c.contentProvider, image.Target.Digest)
+	if err != nil {
+		return desc, size,
+			fmt.Errorf("readblob failed for digest:%q err: %v", image.Target.Digest, err)
+	}
+	var manifest imagespec.Manifest
+	err = json.Unmarshal(p, &manifest)
+	if err != nil {
+		return desc, size,
+			fmt.Errorf("unmarshal blob to manifest failed for digest:%q %v", image.Target.Digest, err)
+	}
+	_, err = c.rootfsUnpacker.Unpack(ctx, manifest.Layers) // ignoring returned chainID for now
+	if err != nil {
+		return desc, size,
+			fmt.Errorf("unpack failed for manifest layers:%v %v", manifest.Layers, err)
+	}
+	size, err = image.Size(ctx, c.contentProvider)
+	if err != nil {
+		return desc, size,
+			fmt.Errorf("size failed for image:%q %v", image.Target.Digest, err)
+	}
+	return desc, size, nil
 }

pkg/server/image_remove.go

@@ -17,14 +17,17 @@ limitations under the License.
 package server
 
 import (
-	"errors"
-
 	"golang.org/x/net/context"
 
 	"k8s.io/kubernetes/pkg/kubelet/api/v1alpha1/runtime"
 )
 
 // RemoveImage removes the image.
+// TODO(mikebrow): harden api
 func (c *criContainerdService) RemoveImage(ctx context.Context, r *runtime.RemoveImageRequest) (*runtime.RemoveImageResponse, error) {
-	return nil, errors.New("not implemented")
+	// Only remove image from the internal metadata store for now.
+	// Note that the image must be digest here in current implementation.
+	// TODO(mikebrow): remove the image via containerd
+	err := c.imageMetadataStore.Delete(r.GetImage().GetImage())
+	return &runtime.RemoveImageResponse{}, err
 }

pkg/server/image_status.go

@@ -17,14 +17,48 @@ limitations under the License.
 package server
 
 import (
-	"errors"
-
 	"golang.org/x/net/context"
 
 	"k8s.io/kubernetes/pkg/kubelet/api/v1alpha1/runtime"
 )
 
-// ImageStatus returns the status of the image, returns nil if the image doesn't present.
+// ImageStatus returns the status of the image, returns nil if the image isn't present.
 func (c *criContainerdService) ImageStatus(ctx context.Context, r *runtime.ImageStatusRequest) (*runtime.ImageStatusResponse, error) {
-	return nil, errors.New("not implemented")
+	ref := r.GetImage().GetImage()
+	// TODO(mikebrow): Get the id->tags, and id->digest mapping from our checkpoint;
+	// Get other information from containerd image/content store
+
+	// if the passed ref is a digest.. and is stored the following get should work
+	// note: get returns nil with no err
+	meta, _ := c.imageMetadataStore.Get(ref)
+	if meta != nil {
+		return &runtime.ImageStatusResponse{Image: &runtime.Image{ // TODO(mikebrow): write a ImageMetadata to runtim.Image converter
+			Id:          meta.ID,
+			RepoTags:    meta.RepoTags,
+			RepoDigests: meta.RepoDigests,
+			Size_:       meta.Size,
+			// TODO(mikebrow): Uid and Username?
+		}}, nil
+	}
+
+	// Search for image by ref in repo tags if found the ID matching ref
+	// is our digest.
+	imageMetadataA, err := c.imageMetadataStore.List()
+	if err == nil {
+		for _, meta := range imageMetadataA {
+			for _, tag := range meta.RepoTags {
+				if ref == tag {
+					return &runtime.ImageStatusResponse{Image: &runtime.Image{ // TODO(mikebrow): write a ImageMetadata to runtim.Image converter
+						Id:          meta.ID,
+						RepoTags:    meta.RepoTags,
+						RepoDigests: meta.RepoDigests,
+						Size_:       meta.Size,
+						// TODO(mikebrow): Uid and Username?
+					}}, nil
+				}
+			}
+		}
+	}
+
+	return nil, nil
 }

pkg/server/service.go

@@ -19,15 +19,23 @@ package server
 import (
 	"google.golang.org/grpc"
 
+	contentapi "github.com/containerd/containerd/api/services/content"
+	"github.com/containerd/containerd/api/services/execution"
+	imagesapi "github.com/containerd/containerd/api/services/images"
+	rootfsapi "github.com/containerd/containerd/api/services/rootfs"
+	"github.com/containerd/containerd/content"
+	"github.com/containerd/containerd/images"
+	"github.com/containerd/containerd/rootfs"
+	contentservice "github.com/containerd/containerd/services/content"
+	imagesservice "github.com/containerd/containerd/services/images"
+	rootfsservice "github.com/containerd/containerd/services/rootfs"
+	"github.com/kubernetes-incubator/cri-containerd/pkg/metadata"
+	"github.com/kubernetes-incubator/cri-containerd/pkg/metadata/store"
 	"k8s.io/kubernetes/pkg/kubelet/api/v1alpha1/runtime"
 
 	// TODO remove the underscores from the following imports as the services are
 	// implemented. "_" is being used to hold the reference to keep autocomplete
 	// from deleting them until referenced below.
-	_ "github.com/containerd/containerd/api/services/content"
-	_ "github.com/containerd/containerd/api/services/execution"
-	_ "github.com/containerd/containerd/api/services/images"
-	_ "github.com/containerd/containerd/api/services/rootfs"
 	_ "github.com/containerd/containerd/api/types/container"
 	_ "github.com/containerd/containerd/api/types/descriptor"
 	_ "github.com/containerd/containerd/api/types/mount"
@@ -42,10 +50,24 @@ type CRIContainerdService interface {
 }
 
 // criContainerdService implements CRIContainerdService.
-type criContainerdService struct{}
+type criContainerdService struct {
+	containerService   execution.ContainerServiceClient
+	imageStore         images.Store
+	contentIngester    content.Ingester
+	contentProvider    content.Provider
+	rootfsUnpacker     rootfs.Unpacker
+	imageMetadataStore metadata.ImageMetadataStore
+}
 
 // NewCRIContainerdService returns a new instance of CRIContainerdService
 func NewCRIContainerdService(conn *grpc.ClientConn) CRIContainerdService {
 	// TODO: Initialize different containerd clients.
-	return &criContainerdService{}
+	return &criContainerdService{
+		containerService:   execution.NewContainerServiceClient(conn),
+		imageStore:         imagesservice.NewStoreFromClient(imagesapi.NewImagesClient(conn)),
+		contentIngester:    contentservice.NewIngesterFromClient(contentapi.NewContentClient(conn)),
+		contentProvider:    contentservice.NewProviderFromClient(contentapi.NewContentClient(conn)),
+		rootfsUnpacker:     rootfsservice.NewUnpackerFromClient(rootfsapi.NewRootFSClient(conn)),
+		imageMetadataStore: metadata.NewImageMetadataStore(store.NewMetadataStore()),
+	}
 }