Merge pull request kubernetes#8 from Random-Liu/add-image-poc-impleme…

…ntation

Add image POC implementation.

pkg/kubelet/BUILD

@@ -109,11 +109,13 @@ go_library(
         "//pkg/volume/util/volumehelper:go_default_library",
         "//plugin/pkg/scheduler/algorithm/predicates:go_default_library",
         "//third_party/forked/golang/expansion:go_default_library",
+        "//vendor:github.com/docker/containerd/api/services/execution",
         "//vendor:github.com/golang/glog",
         "//vendor:github.com/golang/groupcache/lru",
         "//vendor:github.com/google/cadvisor/events",
         "//vendor:github.com/google/cadvisor/info/v1",
         "//vendor:github.com/google/cadvisor/info/v2",
+        "//vendor:google.golang.org/grpc",
         "//vendor:k8s.io/apimachinery/pkg/api/errors",
         "//vendor:k8s.io/apimachinery/pkg/api/resource",
         "//vendor:k8s.io/apimachinery/pkg/apis/meta/v1",

pkg/kubelet/containerdshim/BUILD

@@ -11,101 +11,36 @@ load(
 go_library(
     name = "go_default_library",
     srcs = [
-        "checkpoint_store.go",
-        "convert.go",
+        "containerd_container.go",
+        "containerd_image.go",
+        "containerd_sandbox.go",
+        "containerd_service.go",
+        "containerd_streaming.go",
         "doc.go",
-        "docker_checkpoint.go",
-        "docker_container.go",
-        "docker_image.go",
-        "docker_legacy.go",
-        "docker_sandbox.go",
-        "docker_service.go",
-        "docker_streaming.go",
-        "dummy.go",
-        "helpers.go",
-        "naming.go",
-        "security_context.go",
     ],
     tags = ["automanaged"],
     deps = [
-        "//pkg/api/v1:go_default_library",
-        "//pkg/apis/componentconfig:go_default_library",
         "//pkg/kubelet/api:go_default_library",
         "//pkg/kubelet/api/v1alpha1/runtime:go_default_library",
-        "//pkg/kubelet/cm:go_default_library",
-        "//pkg/kubelet/container:go_default_library",
-        "//pkg/kubelet/containerdshim/cm:go_default_library",
-        "//pkg/kubelet/containerdshim/errors:go_default_library",
-        "//pkg/kubelet/dockertools:go_default_library",
-        "//pkg/kubelet/dockertools/securitycontext:go_default_library",
-        "//pkg/kubelet/leaky:go_default_library",
-        "//pkg/kubelet/network:go_default_library",
-        "//pkg/kubelet/network/cni:go_default_library",
-        "//pkg/kubelet/network/hostport:go_default_library",
-        "//pkg/kubelet/network/kubenet:go_default_library",
-        "//pkg/kubelet/qos:go_default_library",
-        "//pkg/kubelet/server/streaming:go_default_library",
-        "//pkg/kubelet/types:go_default_library",
-        "//pkg/kubelet/util/cache:go_default_library",
-        "//pkg/kubelet/util/ioutils:go_default_library",
-        "//pkg/util/hash:go_default_library",
-        "//pkg/util/term:go_default_library",
-        "//vendor:github.com/blang/semver",
         "//vendor:github.com/docker/containerd/api/services/content",
         "//vendor:github.com/docker/containerd/api/services/execution",
         "//vendor:github.com/docker/containerd/api/services/shim",
         "//vendor:github.com/docker/containerd/api/types/container",
         "//vendor:github.com/docker/containerd/api/types/mount",
-        "//vendor:github.com/docker/engine-api/types",
-        "//vendor:github.com/docker/engine-api/types/container",
-        "//vendor:github.com/docker/engine-api/types/filters",
-        "//vendor:github.com/docker/engine-api/types/strslice",
-        "//vendor:github.com/docker/go-connections/nat",
         "//vendor:github.com/golang/glog",
-        "//vendor:k8s.io/apimachinery/pkg/util/errors",
-        "//vendor:k8s.io/apimachinery/pkg/util/sets",
-        "//vendor:k8s.io/apimachinery/pkg/util/wait",
+        "//vendor:github.com/opencontainers/image-spec/specs-go",
+        "//vendor:github.com/opencontainers/runtime-spec/specs-go",
     ],
 )
 
 go_test(
     name = "go_default_test",
-    srcs = [
-        "checkpoint_store_test.go",
-        "convert_test.go",
-        "docker_checkpoint_test.go",
-        "docker_container_test.go",
-        "docker_image_test.go",
-        "docker_legacy_test.go",
-        "docker_sandbox_test.go",
-        "docker_service_test.go",
-        "helpers_test.go",
-        "naming_test.go",
-        "security_context_test.go",
-    ],
+    srcs = ["containerd_image_test.go"],
     library = ":go_default_library",
     tags = ["automanaged"],
     deps = [
-        "//pkg/api/v1:go_default_library",
         "//pkg/kubelet/api/v1alpha1/runtime:go_default_library",
-        "//pkg/kubelet/container:go_default_library",
-        "//pkg/kubelet/container/testing:go_default_library",
-        "//pkg/kubelet/containerdshim/errors:go_default_library",
-        "//pkg/kubelet/containerdshim/testing:go_default_library",
-        "//pkg/kubelet/dockertools:go_default_library",
-        "//pkg/kubelet/dockertools/securitycontext:go_default_library",
-        "//pkg/kubelet/network:go_default_library",
-        "//pkg/kubelet/network/testing:go_default_library",
-        "//pkg/kubelet/types:go_default_library",
-        "//pkg/kubelet/util/cache:go_default_library",
-        "//pkg/security/apparmor:go_default_library",
-        "//vendor:github.com/blang/semver",
-        "//vendor:github.com/docker/engine-api/types",
-        "//vendor:github.com/docker/engine-api/types/container",
-        "//vendor:github.com/golang/mock/gomock",
         "//vendor:github.com/stretchr/testify/assert",
-        "//vendor:github.com/stretchr/testify/require",
-        "//vendor:k8s.io/client-go/util/clock",
     ],
 )
 
@@ -120,10 +55,7 @@ filegroup(
     name = "all-srcs",
     srcs = [
         ":package-srcs",
-        "//pkg/kubelet/containerdshim/cm:all-srcs",
-        "//pkg/kubelet/containerdshim/errors:all-srcs",
         "//pkg/kubelet/containerdshim/remote:all-srcs",
-        "//pkg/kubelet/containerdshim/testing:all-srcs",
     ],
     tags = ["automanaged"],
 )

pkg/kubelet/containerdshim/containerd_image.go

@@ -18,26 +18,107 @@ package containerdshim
 
 import (
 	"fmt"
+	"os/exec"
+	"strings"
 
 	runtimeapi "k8s.io/kubernetes/pkg/kubelet/api/v1alpha1/runtime"
 )
 
+// containerd doesn't have metadata store now, so save the metadata ourselves.
+// imageStore is a map from image digest to image metadata.
+// No need to store detailed image information layers now, because for the POC
+// we'll re-fetch it when creating the rootfs.
+var imageStore map[string]*runtimeapi.Image = map[string]*runtimeapi.Image{}
+
 // P0
+// Ignore the filter for now.
 func (cs *containerdService) ListImages(filter *runtimeapi.ImageFilter) ([]*runtimeapi.Image, error) {
-	return nil, fmt.Errorf("not implemented")
+	var images []*runtimeapi.Image
+	for _, image := range imageStore {
+		images = append(images, image)
+	}
+	return images, nil
 }
 
 // P0
+// The image here could be either digest or reference in current implementation.
 func (cs *containerdService) ImageStatus(image *runtimeapi.ImageSpec) (*runtimeapi.Image, error) {
-	return nil, fmt.Errorf("not implemented")
+	// Try digest first.
+	if img, ok := imageStore[image.Image]; ok {
+		return img, nil
+	}
+	// Try image reference.
+	for _, img := range imageStore {
+		for _, t := range img.RepoTags {
+			if image.Image == t {
+				return img, nil
+			}
+		}
+	}
+	return nil, nil
 }
 
 // P0
+// For the POC code, docker image must be `docker.io/library/image:tag` or `docker.io/library/image`.
 func (cs *containerdService) PullImage(image *runtimeapi.ImageSpec, auth *runtimeapi.AuthConfig) (string, error) {
-	return "", fmt.Errorf("not implemented")
+	repo, tag := repoAndTag(image.Image)
+	digest, err := imageDigest(repo, tag)
+	if err != nil {
+		return "", fmt.Errorf("failed to get image digest %q: %v", image.Image, err)
+	}
+	if err := pullImage(repo, tag); err != nil {
+		return "", fmt.Errorf("failed to pull image %q: %v", image.Image, err)
+	}
+	if _, ok := imageStore[digest]; !ok {
+		imageStore[digest] = &runtimeapi.Image{
+			Id:          digest,
+			RepoDigests: []string{digest},
+			// Use fake image size, because we don't care about it in the POC.
+			Size_: 1024,
+		}
+	}
+	img := imageStore[digest]
+	// Add new image tag
+	for _, t := range img.RepoTags {
+		if image.Image == t {
+			return digest, nil
+		}
+	}
+	img.RepoTags = append(img.RepoTags, image.Image)
+	// Return the image digest
+	return digest, nil
 }
 
 // P1
 func (cs *containerdService) RemoveImage(image *runtimeapi.ImageSpec) error {
-	return fmt.Errorf("not implemented")
+	// Only remove image from the internal metadata for now. Note that the image
+	// must be digest here in current implementation.
+	delete(imageStore, image.Image)
+	return nil
+}
+
+const mediaType = "mediatype:application/vnd.docker.distribution.manifest.v2+json"
+
+func repoAndTag(image string) (string, string) {
+	repoAndTag := strings.Split(image, ":")
+	if len(repoAndTag) < 2 {
+		return image, "latest"
+	}
+	return repoAndTag[0], repoAndTag[1]
+}
+
+func imageDigest(repo, tag string) (string, error) {
+	output, err := exec.Command("sh", "-c", fmt.Sprintf("dist fetch %s %s %s | shasum -a256", repo, tag, mediaType)).Output()
+	if err != nil {
+		return "", fmt.Errorf("failed to get image digest %s:%s, output: %s, err: %v", repo, tag, output, err)
+	}
+	return "sha256:" + string(output), nil
+}
+
+func pullImage(repo, tag string) error {
+	output, err := exec.Command("sh", "-c", fmt.Sprintf("dist fetch %s %s %s | jq -r '.layers[] | \"dist fetch %s \"+.digest + \"| dist ingest --expected-digest \"+.digest+\" --expected-size \"+(.size | tostring) +\" %s@\"+.digest' | xargs -I{} -P10 -n1 sh -c \"{}\"", repo, tag, mediaType, repo, repo)).Output()
+	if err != nil {
+		return fmt.Errorf("failed to pull image %s:%s, output: %s, err: %v", repo, tag, output, err)
+	}
+	return nil
 }

pkg/kubelet/containerdshim/containerd_image_test.go

@@ -0,0 +1,90 @@
+/*
+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 containerdshim
+
+import (
+	"fmt"
+	"os"
+	"os/exec"
+	"strconv"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	runtimeapi "k8s.io/kubernetes/pkg/kubelet/api/v1alpha1/runtime"
+)
+
+// NOTE: To run the test, please make sure `dist` is in $PATH.
+func TestXxx(t *testing.T) {
+	const (
+		storePath = ".content"
+		image     = "docker.io/library/redis"
+	)
+	// Make sure there is no exsiting image.
+	if _, err := os.Stat(storePath); err == nil {
+		os.RemoveAll(storePath)
+	}
+	cs := NewContainerdService(nil)
+
+	t.Logf("Should be able to pull image")
+	digest, err := cs.PullImage(&runtimeapi.ImageSpec{Image: image}, nil)
+	assert.NoError(t, err)
+	t.Logf("Should be able to list new images")
+	imgs, err := cs.ListImages(nil)
+	assert.NoError(t, err)
+	assert.Len(t, imgs, 1)
+	assert.Equal(t, digest, imgs[0].Id)
+	t.Logf("Should be able to get new image status with name")
+	img, err := cs.ImageStatus(&runtimeapi.ImageSpec{Image: image})
+	assert.NoError(t, err)
+	assert.Equal(t, imgs[0], img)
+	t.Logf("Should be able to get new image status with digest")
+	img, err = cs.ImageStatus(&runtimeapi.ImageSpec{Image: digest})
+	assert.NoError(t, err)
+	assert.Equal(t, imgs[0], img)
+	t.Logf("Should be able to see the image in content store")
+	_, err = os.Stat(storePath)
+	assert.NoError(t, err)
+	output, err := exec.Command("sh", "-c", fmt.Sprintf("ls %s/blobs | wc -l", storePath)).Output()
+	assert.NoError(t, err)
+	layerNum, err := strconv.Atoi(strings.TrimSpace(string(output)))
+	assert.NoError(t, err)
+	assert.NotZero(t, layerNum)
+
+	t.Logf("Should has the same digest and pull no new layer if we pull the same image")
+	newDigest, err := cs.PullImage(&runtimeapi.ImageSpec{Image: image}, nil)
+	assert.NoError(t, err)
+	assert.Equal(t, digest, newDigest)
+	output, err = exec.Command("sh", "-c", fmt.Sprintf("ls %s/blobs | wc -l", storePath)).Output()
+	assert.NoError(t, err)
+	newLayerNum, err := strconv.Atoi(strings.TrimSpace(string(output)))
+	assert.NoError(t, err)
+	assert.Equal(t, layerNum, newLayerNum)
+
+	t.Logf("Should be able to remove image")
+	err = cs.RemoveImage(&runtimeapi.ImageSpec{Image: digest})
+	assert.NoError(t, err)
+	imgs, err = cs.ListImages(nil)
+	assert.NoError(t, err)
+	assert.Empty(t, imgs)
+	img, err = cs.ImageStatus(&runtimeapi.ImageSpec{Image: image})
+	assert.NoError(t, err)
+	assert.Nil(t, img)
+
+	os.RemoveAll(storePath)
+}

pkg/kubelet/containerdshim/containerd_service.go

@@ -29,6 +29,8 @@ import (
 	_ "github.com/docker/containerd/api/services/shim"
 	_ "github.com/docker/containerd/api/types/container"
 	_ "github.com/docker/containerd/api/types/mount"
+	_ "github.com/opencontainers/image-spec/specs-go"
+	_ "github.com/opencontainers/runtime-spec/specs-go"
 )
 
 type ContainerdService interface {

pkg/kubelet/containerdshim/remote/BUILD

@@ -10,8 +10,8 @@ load(
 go_library(
     name = "go_default_library",
     srcs = [
-        "docker_server.go",
-        "docker_service.go",
+        "containerd_server.go",
+        "containerd_service.go",
     ],
     tags = ["automanaged"],
     deps = [