diff --git a/pkg/kubelet/api/testing/fake_image_service.go b/pkg/kubelet/api/testing/fake_image_service.go
index d53fec9e638..11e7b1c70eb 100644
--- a/pkg/kubelet/api/testing/fake_image_service.go
+++ b/pkg/kubelet/api/testing/fake_image_service.go
@@ -17,7 +17,6 @@ limitations under the License.
 package testing
 
 import (
-	"fmt"
 	"sync"
 
 	runtimeApi "k8s.io/kubernetes/pkg/kubelet/api/v1alpha1/runtime"
@@ -89,11 +88,7 @@ func (r *FakeImageService) ImageStatus(image *runtimeApi.ImageSpec) (*runtimeApi
 
 	r.Called = append(r.Called, "ImageStatus")
 
-	if img, ok := r.Images[image.GetImage()]; ok {
-		return img, nil
-	}
-
-	return nil, fmt.Errorf("image %q not found", image.GetImage())
+	return r.Images[image.GetImage()], nil
 }
 
 func (r *FakeImageService) PullImage(image *runtimeApi.ImageSpec, auth *runtimeApi.AuthConfig) error {
diff --git a/pkg/kubelet/api/v1alpha1/runtime/api.pb.go b/pkg/kubelet/api/v1alpha1/runtime/api.pb.go
index a07e322055b..e4e91b988cf 100644
--- a/pkg/kubelet/api/v1alpha1/runtime/api.pb.go
+++ b/pkg/kubelet/api/v1alpha1/runtime/api.pb.go
@@ -2963,7 +2963,8 @@ var _RuntimeService_serviceDesc = grpc.ServiceDesc{
 type ImageServiceClient interface {
 	// ListImages lists existing images.
 	ListImages(ctx context.Context, in *ListImagesRequest, opts ...grpc.CallOption) (*ListImagesResponse, error)
-	// ImageStatus returns the status of the image.
+	// ImageStatus returns the status of the image. If the image is not
+	// present, returns nil.
 	ImageStatus(ctx context.Context, in *ImageStatusRequest, opts ...grpc.CallOption) (*ImageStatusResponse, error)
 	// PullImage pulls an image with authentication config.
 	PullImage(ctx context.Context, in *PullImageRequest, opts ...grpc.CallOption) (*PullImageResponse, error)
@@ -3021,7 +3022,8 @@ func (c *imageServiceClient) RemoveImage(ctx context.Context, in *RemoveImageReq
 type ImageServiceServer interface {
 	// ListImages lists existing images.
 	ListImages(context.Context, *ListImagesRequest) (*ListImagesResponse, error)
-	// ImageStatus returns the status of the image.
+	// ImageStatus returns the status of the image. If the image is not
+	// present, returns nil.
 	ImageStatus(context.Context, *ImageStatusRequest) (*ImageStatusResponse, error)
 	// PullImage pulls an image with authentication config.
 	PullImage(context.Context, *PullImageRequest) (*PullImageResponse, error)
diff --git a/pkg/kubelet/api/v1alpha1/runtime/api.proto b/pkg/kubelet/api/v1alpha1/runtime/api.proto
index 12e23298160..300a2157dd9 100644
--- a/pkg/kubelet/api/v1alpha1/runtime/api.proto
+++ b/pkg/kubelet/api/v1alpha1/runtime/api.proto
@@ -46,7 +46,8 @@ service RuntimeService {
 service ImageService {
     // ListImages lists existing images.
     rpc ListImages(ListImagesRequest) returns (ListImagesResponse) {}
-    // ImageStatus returns the status of the image.
+    // ImageStatus returns the status of the image. If the image is not
+    // present, returns nil.
     rpc ImageStatus(ImageStatusRequest) returns (ImageStatusResponse) {}
     // PullImage pulls an image with authentication config.
     rpc PullImage(PullImageRequest) returns (PullImageResponse) {}
diff --git a/pkg/kubelet/dockershim/convert.go b/pkg/kubelet/dockershim/convert.go
index 91cdb3c1082..09872982953 100644
--- a/pkg/kubelet/dockershim/convert.go
+++ b/pkg/kubelet/dockershim/convert.go
@@ -36,7 +36,7 @@ const (
 	statusExitedPrefix  = "Exited"
 )
 
-func toRuntimeAPIImage(image *dockertypes.Image) (*runtimeApi.Image, error) {
+func imageToRuntimeAPIImage(image *dockertypes.Image) (*runtimeApi.Image, error) {
 	if image == nil {
 		return nil, fmt.Errorf("unable to convert a nil pointer to a runtime API image")
 	}
@@ -50,6 +50,31 @@ func toRuntimeAPIImage(image *dockertypes.Image) (*runtimeApi.Image, error) {
 	}, nil
 }
 
+func imageInspectToRuntimeAPIImage(image *dockertypes.ImageInspect) (*runtimeApi.Image, error) {
+	if image == nil {
+		return nil, fmt.Errorf("unable to convert a nil pointer to a runtime API image")
+	}
+
+	size := uint64(image.VirtualSize)
+	return &runtimeApi.Image{
+		Id:          &image.ID,
+		RepoTags:    image.RepoTags,
+		RepoDigests: image.RepoDigests,
+		Size_:       &size,
+	}, nil
+
+}
+
+func toPullableImageID(id string, image *dockertypes.ImageInspect) string {
+	// Default to the image ID, but if RepoDigests is not empty, use
+	// the first digest instead.
+	imageID := DockerImageIDPrefix + id
+	if len(image.RepoDigests) > 0 {
+		imageID = DockerPullableImageIDPrefix + image.RepoDigests[0]
+	}
+	return imageID
+}
+
 func toRuntimeAPIContainer(c *dockertypes.Container) (*runtimeApi.Container, error) {
 	state := toRuntimeAPIContainerState(c.Status)
 	metadata, err := parseContainerName(c.Names[0])
diff --git a/pkg/kubelet/dockershim/convert_test.go b/pkg/kubelet/dockershim/convert_test.go
index c6e428ba235..993dffb2451 100644
--- a/pkg/kubelet/dockershim/convert_test.go
+++ b/pkg/kubelet/dockershim/convert_test.go
@@ -19,6 +19,7 @@ package dockershim
 import (
 	"testing"
 
+	dockertypes "github.com/docker/engine-api/types"
 	"github.com/stretchr/testify/assert"
 
 	runtimeApi "k8s.io/kubernetes/pkg/kubelet/api/v1alpha1/runtime"
@@ -40,3 +41,31 @@ func TestConvertDockerStatusToRuntimeAPIState(t *testing.T) {
 		assert.Equal(t, test.expected, actual)
 	}
 }
+
+func TestConvertToPullableImageID(t *testing.T) {
+	testCases := []struct {
+		id       string
+		image    *dockertypes.ImageInspect
+		expected string
+	}{
+		{
+			id: "image-1",
+			image: &dockertypes.ImageInspect{
+				RepoDigests: []string{"digest-1"},
+			},
+			expected: DockerPullableImageIDPrefix + "digest-1",
+		},
+		{
+			id: "image-2",
+			image: &dockertypes.ImageInspect{
+				RepoDigests: []string{},
+			},
+			expected: DockerImageIDPrefix + "image-2",
+		},
+	}
+
+	for _, test := range testCases {
+		actual := toPullableImageID(test.id, test.image)
+		assert.Equal(t, test.expected, actual)
+	}
+}
diff --git a/pkg/kubelet/dockershim/docker_container.go b/pkg/kubelet/dockershim/docker_container.go
index b66b9babac0..a80ce8e660b 100644
--- a/pkg/kubelet/dockershim/docker_container.go
+++ b/pkg/kubelet/dockershim/docker_container.go
@@ -229,6 +229,13 @@ func (ds *dockerService) ContainerStatus(containerID string) (*runtimeApi.Contai
 		return nil, fmt.Errorf("failed to parse timestamp for container %q: %v", containerID, err)
 	}
 
+	// Convert the image id to pullable id.
+	ir, err := ds.client.InspectImageByID(r.Image)
+	if err != nil {
+		return nil, fmt.Errorf("unable to inspect docker image %q while inspecting docker container %q: %v", r.Image, containerID, err)
+	}
+	imageID := toPullableImageID(r.Image, ir)
+
 	// Convert the mounts.
 	mounts := []*runtimeApi.Mount{}
 	for i := range r.Mounts {
@@ -293,7 +300,7 @@ func (ds *dockerService) ContainerStatus(containerID string) (*runtimeApi.Contai
 		Id:          &r.ID,
 		Metadata:    metadata,
 		Image:       &runtimeApi.ImageSpec{Image: &r.Config.Image},
-		ImageRef:    &r.Image,
+		ImageRef:    &imageID,
 		Mounts:      mounts,
 		ExitCode:    &exitCode,
 		State:       &state,
diff --git a/pkg/kubelet/dockershim/docker_container_test.go b/pkg/kubelet/dockershim/docker_container_test.go
index 08a41b11003..fb8962c5d8c 100644
--- a/pkg/kubelet/dockershim/docker_container_test.go
+++ b/pkg/kubelet/dockershim/docker_container_test.go
@@ -105,7 +105,7 @@ func TestContainerStatus(t *testing.T) {
 	ct, st, ft := dt, dt, dt
 	state := runtimeApi.ContainerState_CREATED
 	// The following variables are not set in FakeDockerClient.
-	imageRef := ""
+	imageRef := DockerImageIDPrefix + ""
 	exitCode := int32(0)
 	var reason, message string
 
diff --git a/pkg/kubelet/dockershim/docker_image.go b/pkg/kubelet/dockershim/docker_image.go
index fc0dc76eef7..ec8536f54e4 100644
--- a/pkg/kubelet/dockershim/docker_image.go
+++ b/pkg/kubelet/dockershim/docker_image.go
@@ -17,10 +17,9 @@ limitations under the License.
 package dockershim
 
 import (
-	"fmt"
-
 	dockertypes "github.com/docker/engine-api/types"
 	runtimeApi "k8s.io/kubernetes/pkg/kubelet/api/v1alpha1/runtime"
+	"k8s.io/kubernetes/pkg/kubelet/dockertools"
 )
 
 // This file implements methods in ImageManagerService.
@@ -41,7 +40,7 @@ func (ds *dockerService) ListImages(filter *runtimeApi.ImageFilter) ([]*runtimeA
 
 	result := []*runtimeApi.Image{}
 	for _, i := range images {
-		apiImage, err := toRuntimeAPIImage(&i)
+		apiImage, err := imageToRuntimeAPIImage(&i)
 		if err != nil {
 			// TODO: log an error message?
 			continue
@@ -51,16 +50,16 @@ func (ds *dockerService) ListImages(filter *runtimeApi.ImageFilter) ([]*runtimeA
 	return result, nil
 }
 
-// ImageStatus returns the status of the image.
+// ImageStatus returns the status of the image, returns nil if the image doesn't present.
 func (ds *dockerService) ImageStatus(image *runtimeApi.ImageSpec) (*runtimeApi.Image, error) {
-	images, err := ds.ListImages(&runtimeApi.ImageFilter{Image: image})
+	imageInspect, err := ds.client.InspectImageByRef(image.GetImage())
 	if err != nil {
+		if dockertools.IsImageNotFoundError(err) {
+			return nil, nil
+		}
 		return nil, err
 	}
-	if len(images) != 1 {
-		return nil, fmt.Errorf("ImageStatus returned more than one image: %+v", images)
-	}
-	return images[0], nil
+	return imageInspectToRuntimeAPIImage(imageInspect)
 }
 
 // PullImage pulls an image with authentication config.
@@ -79,6 +78,19 @@ func (ds *dockerService) PullImage(image *runtimeApi.ImageSpec, auth *runtimeApi
 
 // RemoveImage removes the image.
 func (ds *dockerService) RemoveImage(image *runtimeApi.ImageSpec) error {
-	_, err := ds.client.RemoveImage(image.GetImage(), dockertypes.ImageRemoveOptions{PruneChildren: true})
+	// If the image has multiple tags, we need to remove all the tags
+	// TODO: We assume image.Image is image ID here, which is true in the current implementation
+	// of kubelet, but we should still clarify this in CRI.
+	imageInspect, err := ds.client.InspectImageByID(image.GetImage())
+	if err == nil && imageInspect != nil && len(imageInspect.RepoTags) > 1 {
+		for _, tag := range imageInspect.RepoTags {
+			if _, err := ds.client.RemoveImage(tag, dockertypes.ImageRemoveOptions{PruneChildren: true}); err != nil {
+				return err
+			}
+		}
+		return nil
+	}
+
+	_, err = ds.client.RemoveImage(image.GetImage(), dockertypes.ImageRemoveOptions{PruneChildren: true})
 	return err
 }
diff --git a/pkg/kubelet/dockershim/docker_image_test.go b/pkg/kubelet/dockershim/docker_image_test.go
new file mode 100644
index 00000000000..7eff4a4d697
--- /dev/null
+++ b/pkg/kubelet/dockershim/docker_image_test.go
@@ -0,0 +1,45 @@
+/*
+Copyright 2016 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 dockershim
+
+import (
+	"testing"
+
+	dockertypes "github.com/docker/engine-api/types"
+
+	runtimeApi "k8s.io/kubernetes/pkg/kubelet/api/v1alpha1/runtime"
+	"k8s.io/kubernetes/pkg/kubelet/dockertools"
+)
+
+func TestRemoveImage(t *testing.T) {
+	ds, fakeDocker, _ := newTestDockerService()
+	id := "1111"
+	fakeDocker.Image = &dockertypes.ImageInspect{ID: id, RepoTags: []string{"foo"}}
+	ds.RemoveImage(&runtimeApi.ImageSpec{Image: &id})
+	fakeDocker.AssertCallDetails(dockertools.NewCalledDetail("inspect_image", nil),
+		dockertools.NewCalledDetail("remove_image", []interface{}{id, dockertypes.ImageRemoveOptions{PruneChildren: true}}))
+}
+
+func TestRemoveImageWithMultipleTags(t *testing.T) {
+	ds, fakeDocker, _ := newTestDockerService()
+	id := "1111"
+	fakeDocker.Image = &dockertypes.ImageInspect{ID: id, RepoTags: []string{"foo", "bar"}}
+	ds.RemoveImage(&runtimeApi.ImageSpec{Image: &id})
+	fakeDocker.AssertCallDetails(dockertools.NewCalledDetail("inspect_image", nil),
+		dockertools.NewCalledDetail("remove_image", []interface{}{"foo", dockertypes.ImageRemoveOptions{PruneChildren: true}}),
+		dockertools.NewCalledDetail("remove_image", []interface{}{"bar", dockertypes.ImageRemoveOptions{PruneChildren: true}}))
+}
diff --git a/pkg/kubelet/dockershim/naming.go b/pkg/kubelet/dockershim/naming.go
index 86147645064..915fd05d36f 100644
--- a/pkg/kubelet/dockershim/naming.go
+++ b/pkg/kubelet/dockershim/naming.go
@@ -22,6 +22,7 @@ import (
 	"strings"
 
 	runtimeApi "k8s.io/kubernetes/pkg/kubelet/api/v1alpha1/runtime"
+	"k8s.io/kubernetes/pkg/kubelet/dockertools"
 	"k8s.io/kubernetes/pkg/kubelet/leaky"
 )
 
@@ -48,6 +49,10 @@ const (
 	sandboxContainerName = leaky.PodInfraContainerName
 	// Delimiter used to construct docker container names.
 	nameDelimiter = "_"
+	// DockerImageIDPrefix is the prefix of image id in container status.
+	DockerImageIDPrefix = dockertools.DockerPrefix
+	// DockerPullableImageIDPrefix is the prefix of pullable image id in container status.
+	DockerPullableImageIDPrefix = dockertools.DockerPullablePrefix
 )
 
 func makeSandboxName(s *runtimeApi.PodSandboxConfig) string {
diff --git a/pkg/kubelet/dockertools/docker_manager.go b/pkg/kubelet/dockertools/docker_manager.go
index 438d20d92cf..f5b9e9f5f49 100644
--- a/pkg/kubelet/dockertools/docker_manager.go
+++ b/pkg/kubelet/dockertools/docker_manager.go
@@ -401,7 +401,7 @@ func (dm *DockerManager) inspectContainer(id string, podName, podNamespace strin
 	imageID := DockerPrefix + iResult.Image
 	imgInspectResult, err := dm.client.InspectImageByID(iResult.Image)
 	if err != nil {
-		utilruntime.HandleError(fmt.Errorf("unable to inspect docker image %q while inspecting docker container %q: %v", containerName, iResult.Image, err))
+		utilruntime.HandleError(fmt.Errorf("unable to inspect docker image %q while inspecting docker container %q: %v", iResult.Image, containerName, err))
 	} else {
 		if len(imgInspectResult.RepoDigests) > 1 {
 			glog.V(4).Infof("Container %q had more than one associated RepoDigest (%v), only using the first", containerName, imgInspectResult.RepoDigests)
diff --git a/pkg/kubelet/dockertools/docker_manager_test.go b/pkg/kubelet/dockertools/docker_manager_test.go
index 90f1172abd5..a350def7bff 100644
--- a/pkg/kubelet/dockertools/docker_manager_test.go
+++ b/pkg/kubelet/dockertools/docker_manager_test.go
@@ -425,17 +425,17 @@ func TestDeleteImage(t *testing.T) {
 	manager, fakeDocker := newTestDockerManager()
 	fakeDocker.Image = &dockertypes.ImageInspect{ID: "1111", RepoTags: []string{"foo"}}
 	manager.RemoveImage(kubecontainer.ImageSpec{Image: "1111"})
-	fakeDocker.AssertCallDetails([]calledDetail{{name: "inspect_image"}, {name: "remove_image",
-		arguments: []interface{}{"1111", dockertypes.ImageRemoveOptions{PruneChildren: true}}}})
+	fakeDocker.AssertCallDetails(NewCalledDetail("inspect_image", nil), NewCalledDetail("remove_image",
+		[]interface{}{"1111", dockertypes.ImageRemoveOptions{PruneChildren: true}}))
 }
 
 func TestDeleteImageWithMultipleTags(t *testing.T) {
 	manager, fakeDocker := newTestDockerManager()
 	fakeDocker.Image = &dockertypes.ImageInspect{ID: "1111", RepoTags: []string{"foo", "bar"}}
 	manager.RemoveImage(kubecontainer.ImageSpec{Image: "1111"})
-	fakeDocker.AssertCallDetails([]calledDetail{{name: "inspect_image"},
-		{name: "remove_image", arguments: []interface{}{"foo", dockertypes.ImageRemoveOptions{PruneChildren: true}}},
-		{name: "remove_image", arguments: []interface{}{"bar", dockertypes.ImageRemoveOptions{PruneChildren: true}}}})
+	fakeDocker.AssertCallDetails(NewCalledDetail("inspect_image", nil),
+		NewCalledDetail("remove_image", []interface{}{"foo", dockertypes.ImageRemoveOptions{PruneChildren: true}}),
+		NewCalledDetail("remove_image", []interface{}{"bar", dockertypes.ImageRemoveOptions{PruneChildren: true}}))
 }
 
 func TestKillContainerInPod(t *testing.T) {
@@ -1409,6 +1409,7 @@ func TestVerifyNonRoot(t *testing.T) {
 		},
 		"nil image in inspect": {
 			container:     &api.Container{},
+			inspectImage:  nil,
 			expectedError: "unable to inspect image",
 		},
 		"nil config in image inspect": {
diff --git a/pkg/kubelet/dockertools/fake_docker_client.go b/pkg/kubelet/dockertools/fake_docker_client.go
index 0f6908c739e..0c538c50e74 100644
--- a/pkg/kubelet/dockertools/fake_docker_client.go
+++ b/pkg/kubelet/dockertools/fake_docker_client.go
@@ -38,6 +38,11 @@ type calledDetail struct {
 	arguments []interface{}
 }
 
+// NewCalledDetail create a new call detail item.
+func NewCalledDetail(name string, arguments []interface{}) calledDetail {
+	return calledDetail{name: name, arguments: arguments}
+}
+
 // FakeDockerClient is a simple fake docker client, so that kubelet can be run for testing without requiring a real docker setup.
 type FakeDockerClient struct {
 	sync.Mutex
@@ -86,7 +91,6 @@ func newClientWithVersionAndClock(version, apiVersion string, c clock.Clock) *Fa
 		Errors:       make(map[string]error),
 		ContainerMap: make(map[string]*dockertypes.ContainerJSON),
 		Clock:        c,
-
 		// default this to an empty result, so that we never have a nil non-error response from InspectImage
 		Image: &dockertypes.ImageInspect{},
 	}
@@ -213,7 +217,7 @@ func (f *FakeDockerClient) AssertCalls(calls []string) (err error) {
 	return
 }
 
-func (f *FakeDockerClient) AssertCallDetails(calls []calledDetail) (err error) {
+func (f *FakeDockerClient) AssertCallDetails(calls ...calledDetail) (err error) {
 	f.Lock()
 	defer f.Unlock()
 
diff --git a/pkg/kubelet/dockertools/kube_docker_client.go b/pkg/kubelet/dockertools/kube_docker_client.go
index 32297fa28c2..e6d140cf748 100644
--- a/pkg/kubelet/dockertools/kube_docker_client.go
+++ b/pkg/kubelet/dockertools/kube_docker_client.go
@@ -626,3 +626,10 @@ type imageNotFoundError struct {
 func (e imageNotFoundError) Error() string {
 	return fmt.Sprintf("no such image: %q", e.ID)
 }
+
+// IsImageNotFoundError checks whether the error is image not found error. This is exposed
+// to share with dockershim.
+func IsImageNotFoundError(err error) bool {
+	_, ok := err.(imageNotFoundError)
+	return ok
+}
diff --git a/pkg/kubelet/kuberuntime/kuberuntime_image.go b/pkg/kubelet/kuberuntime/kuberuntime_image.go
index b9b9cb1f463..29c57000548 100644
--- a/pkg/kubelet/kuberuntime/kuberuntime_image.go
+++ b/pkg/kubelet/kuberuntime/kuberuntime_image.go
@@ -80,17 +80,12 @@ func (m *kubeGenericRuntimeManager) PullImage(image kubecontainer.ImageSpec, pul
 
 // IsImagePresent checks whether the container image is already in the local storage.
 func (m *kubeGenericRuntimeManager) IsImagePresent(image kubecontainer.ImageSpec) (bool, error) {
-	images, err := m.imageService.ListImages(&runtimeApi.ImageFilter{
-		Image: &runtimeApi.ImageSpec{
-			Image: &image.Image,
-		},
-	})
+	status, err := m.imageService.ImageStatus(&runtimeApi.ImageSpec{Image: &image.Image})
 	if err != nil {
-		glog.Errorf("ListImages failed: %v", err)
+		glog.Errorf("ImageStatus for image %q failed: %v", image, err)
 		return false, err
 	}
-
-	return len(images) > 0, nil
+	return status != nil, nil
 }
 
 // ListImages gets all images currently on the machine.