Merge pull request #24126 from Random-Liu/fix-pull-image

Automatic merge from submit-queue

Fix PullImage and add corresponding node e2e test

Fixes #24101. This is a bug introduced by #23506, since ref #23563.

The root cause of #24101 is described [here](https://github.com/kubernetes/kubernetes/issues/24101#issuecomment-208547623).

This PR
1) Fixes #24101 by decoding the messages returned during pulling image, and return error if any of the messages contains error.
2) Add the node e2e test to detect this kind of failure.
3) Get present check out of `ConformanceImage.Remove()` and `ConformanceImage.Pull()`. Because sometimes we may expect error to occur in `PullImage()` and `RemoveImage()`, but even that doesn't happen, the `Present()` check will still return error and let the test pass.

@yujuhong @freehan @liangchenye 

Also /cc @resouer, because he is doing the image related functions refactoring.

pkg/kubelet/dockertools/kube_docker_client.go

@@ -26,7 +26,8 @@ import (
 	"strconv"
 	"time"
 
-	"github.com/docker/docker/pkg/stdcopy"
+	dockermessage "github.com/docker/docker/pkg/jsonmessage"
+	dockerstdcopy "github.com/docker/docker/pkg/stdcopy"
 	dockerapi "github.com/docker/engine-api/client"
 	dockertypes "github.com/docker/engine-api/types"
 	dockerfilters "github.com/docker/engine-api/types/filters"
@@ -201,8 +202,21 @@ func (d *kubeDockerClient) PullImage(opts docker.PullImageOptions, auth docker.A
 	}
 	defer resp.Close()
 	// TODO(random-liu): Use the image pulling progress information.
-	_, err = io.Copy(ioutil.Discard, resp)
+	decoder := json.NewDecoder(resp)
-	return err
+	for {
+		var msg dockermessage.JSONMessage
+		err := decoder.Decode(&msg)
+		if err == io.EOF {
+			break
+		}
+		if err != nil {
+			return err
+		}
+		if msg.Error != nil {
+			return msg.Error
+		}
+	}
+	return nil
 }
 
 func (d *kubeDockerClient) RemoveImage(image string) error {
@@ -326,7 +340,7 @@ func (d *kubeDockerClient) redirectResponseToOutputStream(tty bool, outputStream
 	if tty {
 		_, err = io.Copy(outputStream, resp)
 	} else {
-		_, err = stdcopy.StdCopy(outputStream, errorStream, resp)
+		_, err = dockerstdcopy.StdCopy(outputStream, errorStream, resp)
 	}
 	return err
 }

test/e2e_node/conformance_test.go

@@ -51,7 +51,9 @@ var _ = Describe("Container Conformance Test", func() {
 				"gcr.io/google_containers/node-conformance:v3",
 				"gcr.io/google_containers/node-conformance:v4",
 			}
-			It("it should pull successfully [Conformance]", func() {
+			// TODO(random-liu): Each It should be independent, we should not let them depend on
+			// each other.
+			It("should pull successfully [Conformance]", func() {
 				for _, imageTag := range conformImageTags {
 					image, _ := NewConformanceImage("docker", imageTag)
 					conformImages = append(conformImages, image)
@@ -61,46 +63,62 @@ var _ = Describe("Container Conformance Test", func() {
 					Eventually(func() error {
 						return image.Pull()
 					}, imageRetryTimeout, imagePullInterval).ShouldNot(HaveOccurred())
+					present, err := image.Present()
+					Expect(err).ShouldNot(HaveOccurred())
+					Expect(present).Should(BeTrue())
 				}
 			})
-			It("it should list pulled images [Conformance]", func() {
+			It("should list pulled images [Conformance]", func() {
 				image, _ := NewConformanceImage("docker", "")
 				tags, _ := image.List()
 				for _, tag := range conformImageTags {
 					Expect(tags).To(ContainElement(tag))
 				}
 			})
-			It("it should remove successfully [Conformance]", func() {
+			It("should remove successfully [Conformance]", func() {
 				for _, image := range conformImages {
-					if err := image.Remove(); err != nil {
+					err := image.Remove()
-						Expect(err).NotTo(HaveOccurred())
+					Expect(err).NotTo(HaveOccurred())
-						break
+					present, err := image.Present()
-					}
+					Expect(err).NotTo(HaveOccurred())
+					Expect(present).To(BeFalse())
 				}
 			})
 		})
 		Context("when testing image that does not exist", func() {
-			var invalidImage ConformanceImage
+			It("should not pull successfully [Conformance]", func() {
-			var invalidImageTag string
+				invalidImageTags := []string{
-			It("it should not pull successfully [Conformance]", func() {
+					// nonexistent image registry
-				invalidImageTag = "foo.com/foo/foo"
+					"foo.com/foo/foo",
-				invalidImage, _ = NewConformanceImage("docker", invalidImageTag)
+					// nonexistent image
-				err := invalidImage.Pull()
+					"gcr.io/google_containers/not_exist",
-				Expect(err).To(HaveOccurred())
+					// TODO(random-liu): Add test for image pulling credential
-			})
+				}
-			It("it should not list pulled images [Conformance]", func() {
+				for _, invalidImageTag := range invalidImageTags {
-				image, _ := NewConformanceImage("docker", "")
+					invalidImage, _ := NewConformanceImage("docker", invalidImageTag)
-				tags, _ := image.List()
+					By("start pulling image")
-				Expect(tags).NotTo(ContainElement(invalidImageTag))
+					err := invalidImage.Pull()
-			})
+					Expect(err).Should(HaveOccurred())
-			It("it should not remove successfully [Conformance]", func() {
+
-				err := invalidImage.Remove()
+					By("check image present")
-				Expect(err).To(HaveOccurred())
+					present, err := invalidImage.Present()
+					Expect(err).ShouldNot(HaveOccurred())
+					Expect(present).To(BeFalse())
+
+					By("listing image")
+					tags, err := invalidImage.List()
+					Expect(err).ShouldNot(HaveOccurred())
+					Expect(tags).NotTo(ContainElement(invalidImage))
+
+					By("removing image")
+					err = invalidImage.Remove()
+					Expect(err).Should(HaveOccurred())
+				}
 			})
 		})
 		Context("when running a container that terminates", func() {
 			var terminateCase ConformanceContainer
-			It("it should run successfully to completion [Conformance]", func() {
+			It("should run successfully to completion [Conformance]", func() {
 				terminateCase = ConformanceContainer{
 					Container: api.Container{
 						Image:           "gcr.io/google_containers/busybox",
@@ -121,19 +139,19 @@ var _ = Describe("Container Conformance Test", func() {
 					return pod.Phase, err
 				}, retryTimeout, pollInterval).Should(Equal(terminateCase.Phase))
 			})
-			It("it should report its phase as 'succeeded' [Conformance]", func() {
+			It("should report its phase as 'succeeded' [Conformance]", func() {
 				ccontainer, err := terminateCase.Get()
 				Expect(err).NotTo(HaveOccurred())
 				Expect(ccontainer).Should(CContainerEqual(terminateCase))
 			})
-			It("it should be possible to delete [Conformance]", func() {
+			It("should be possible to delete [Conformance]", func() {
 				err := terminateCase.Delete()
 				Expect(err).NotTo(HaveOccurred())
 			})
 		})
 		Context("when running a container with invalid image", func() {
 			var invalidImageCase ConformanceContainer
-			It("it should not start successfully [Conformance]", func() {
+			It("should not start successfully [Conformance]", func() {
 				invalidImageCase = ConformanceContainer{
 					Container: api.Container{
 						Image:           "foo.com/foo/foo",
@@ -152,12 +170,12 @@ var _ = Describe("Container Conformance Test", func() {
 					return pod.Phase, err
 				}, retryTimeout, pollInterval).Should(Equal(invalidImageCase.Phase))
 			})
-			It("it should report its phase as 'pending' [Conformance]", func() {
+			It("should report its phase as 'pending' [Conformance]", func() {
 				ccontainer, err := invalidImageCase.Get()
 				Expect(err).NotTo(HaveOccurred())
 				Expect(ccontainer).Should(CContainerEqual(invalidImageCase))
 			})
-			It("it should be possible to delete [Conformance]", func() {
+			It("should be possible to delete [Conformance]", func() {
 				err := invalidImageCase.Delete()
 				Expect(err).NotTo(HaveOccurred())
 			})

test/e2e_node/image.go

@@ -18,7 +18,6 @@ package e2e_node
 
 import (
 	"fmt"
-	"time"
 
 	kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
 	"k8s.io/kubernetes/pkg/kubelet/dockertools"
@@ -55,18 +54,11 @@ func dockerRuntime() kubecontainer.Runtime {
 }
 
 func (ci *ConformanceImage) Pull() error {
-	err := ci.Runtime.PullImage(ci.Image, nil)
+	return ci.Runtime.PullImage(ci.Image, nil)
-	if err != nil {
+}
-		return err
-	}
 
-	if present, err := ci.Runtime.IsImagePresent(ci.Image); err != nil {
+func (ci *ConformanceImage) Present() (bool, error) {
-		return err
+	return ci.Runtime.IsImagePresent(ci.Image)
-	} else if !present {
-		return fmt.Errorf("Failed to detect the pulled image :%s.", ci.Image.Image)
-	}
-
-	return nil
 }
 
 func (ci *ConformanceImage) List() ([]string, error) {
@@ -82,23 +74,5 @@ func (ci *ConformanceImage) List() ([]string, error) {
 }
 
 func (ci *ConformanceImage) Remove() error {
-	ci.Runtime.GarbageCollect(kubecontainer.ContainerGCPolicy{MinAge: time.Second * 30, MaxPerPodContainer: 1, MaxContainers: 0})
+	return ci.Runtime.RemoveImage(ci.Image)
-
-	var err error
-	for start := time.Now(); time.Since(start) < time.Minute*2; time.Sleep(time.Second * 30) {
-		if err = ci.Runtime.RemoveImage(ci.Image); err == nil {
-			break
-		}
-	}
-	if err != nil {
-		return err
-	}
-
-	if present, err := ci.Runtime.IsImagePresent(ci.Image); err != nil {
-		return err
-	} else if present {
-		return fmt.Errorf("Failed to remove the pulled image %s.", ci.Image.Image)
-	}
-
-	return nil
 }