Merge pull request #9452 from yifan-gu/img_pulling

kubelet: Add 'image pulling' event. Refactor dockerManager.createPodInfraContainer()

pkg/client/record/fake.go

@@ -17,16 +17,24 @@ limitations under the License.
 package record
 
 import (
+	"fmt"
+
 	"k8s.io/kubernetes/pkg/runtime"
 	"k8s.io/kubernetes/pkg/util"
 )
 
 // FakeRecorder is used as a fake during tests.
-type FakeRecorder struct{}
+type FakeRecorder struct {
+	Events []string
+}
 
-func (f *FakeRecorder) Event(object runtime.Object, reason, message string) {}
+func (f *FakeRecorder) Event(object runtime.Object, reason, message string) {
+	f.Events = append(f.Events, fmt.Sprintf("%s %s", reason, message))
+}
 
-func (f *FakeRecorder) Eventf(object runtime.Object, reason, messageFmt string, args ...interface{}) {}
+func (f *FakeRecorder) Eventf(object runtime.Object, reason, messageFmt string, args ...interface{}) {
+	f.Events = append(f.Events, fmt.Sprintf(reason+" "+messageFmt, args...))
+}
 
 func (f *FakeRecorder) PastEventf(object runtime.Object, timestamp util.Time, reason, messageFmt string, args ...interface{}) {
 }

pkg/kubelet/container/runtime.go

@@ -105,9 +105,12 @@ type RuntimeHooks interface {
 	// Determines whether the runtime should pull the specified container's image.
 	ShouldPullImage(pod *api.Pod, container *api.Container, imagePresent bool) bool
 
+	// Runs when we start to pull an image.
+	ReportImagePulling(pod *api.Pod, container *api.Container)
+
 	// Runs after an image is pulled reporting its status. Error may be nil
 	// for a successful pull.
-	ReportImagePull(pod *api.Pod, container *api.Container, err error)
+	ReportImagePulled(pod *api.Pod, container *api.Container, err error)
 }
 
 // Pod is a group of containers, with the status of the pod.

pkg/kubelet/dockertools/manager.go

@@ -67,6 +67,9 @@ const (
 // DockerManager implements the Runtime interface.
 var _ kubecontainer.Runtime = &DockerManager{}
 
+// TODO: make this a TTL based pull (if image older than X policy, pull)
+var podInfraContainerImagePullPolicy = api.PullIfNotPresent
+
 type DockerManager struct {
 	client              DockerInterface
 	recorder            record.EventRecorder
@@ -866,9 +869,10 @@ func (dm *DockerManager) podInfraContainerChanged(pod *api.Pod, podInfraContaine
 		}
 	}
 	expectedPodInfraContainer := &api.Container{
-		Name:  PodInfraContainerName,
-		Image: dm.podInfraContainerImage,
-		Ports: ports,
+		Name:            PodInfraContainerName,
+		Image:           dm.podInfraContainerImage,
+		Ports:           ports,
+		ImagePullPolicy: podInfraContainerImagePullPolicy,
 	}
 	return podInfraContainer.Hash != kubecontainer.HashContainer(expectedPodInfraContainer), nil
 }
@@ -1357,37 +1361,16 @@ func (dm *DockerManager) createPodInfraContainer(pod *api.Pod) (kubeletTypes.Doc
 	}
 
 	container := &api.Container{
-		Name:  PodInfraContainerName,
-		Image: dm.podInfraContainerImage,
-		Ports: ports,
+		Name:            PodInfraContainerName,
+		Image:           dm.podInfraContainerImage,
+		Ports:           ports,
+		ImagePullPolicy: podInfraContainerImagePullPolicy,
 	}
-	ref, err := kubecontainer.GenerateContainerRef(pod, container)
-	if err != nil {
-		glog.Errorf("Couldn't make a ref to pod %v, container %v: '%v'", pod.Name, container.Name, err)
-	}
-	spec := kubecontainer.ImageSpec{container.Image}
-	// TODO: make this a TTL based pull (if image older than X policy, pull)
-	ok, err := dm.IsImagePresent(spec)
-	if err != nil {
-		if ref != nil {
-			dm.recorder.Eventf(ref, "failed", "Failed to inspect image %q: %v", container.Image, err)
-		}
+
+	// No pod secrets for the infra container.
+	if err := dm.pullImage(pod, container, nil); err != nil {
 		return "", err
 	}
-	if !ok {
-		if err := dm.PullImage(spec, nil /* no pod secrets for the infra container */); err != nil {
-			if ref != nil {
-				dm.recorder.Eventf(ref, "failed", "Failed to pull image %q: %v", container.Image, err)
-			}
-			return "", err
-		}
-		if ref != nil {
-			dm.recorder.Eventf(ref, "pulled", "Successfully pulled Pod container image %q", container.Image)
-		}
-	}
-	if ok && ref != nil {
-		dm.recorder.Eventf(ref, "pulled", "Pod container image %q already present on machine", container.Image)
-	}
 
 	id, err := dm.runContainerInPod(pod, container, netNamespace, "")
 	if err != nil {
@@ -1544,25 +1527,28 @@ func (dm *DockerManager) clearReasonCache(pod *api.Pod, container *api.Container
 
 // Pull the image for the specified pod and container.
 func (dm *DockerManager) pullImage(pod *api.Pod, container *api.Container, pullSecrets []api.Secret) error {
+	ref, err := kubecontainer.GenerateContainerRef(pod, container)
+	if err != nil {
+		glog.Errorf("Couldn't make a ref to pod %v, container %v: '%v'", pod.Name, container.Name, err)
+	}
 	spec := kubecontainer.ImageSpec{container.Image}
 	present, err := dm.IsImagePresent(spec)
-
 	if err != nil {
-		ref, err := kubecontainer.GenerateContainerRef(pod, container)
-		if err != nil {
-			glog.Errorf("Couldn't make a ref to pod %v, container %v: '%v'", pod.Name, container.Name, err)
-		}
 		if ref != nil {
 			dm.recorder.Eventf(ref, "failed", "Failed to inspect image %q: %v", container.Image, err)
 		}
 		return fmt.Errorf("failed to inspect image %q: %v", container.Image, err)
 	}
 	if !dm.runtimeHooks.ShouldPullImage(pod, container, present) {
+		if present && ref != nil {
+			dm.recorder.Eventf(ref, "pulled", "Container image %q already present on machine", container.Image)
+		}
 		return nil
 	}
 
+	dm.runtimeHooks.ReportImagePulling(pod, container)
 	err = dm.PullImage(spec, pullSecrets)
-	dm.runtimeHooks.ReportImagePull(pod, container, err)
+	dm.runtimeHooks.ReportImagePulled(pod, container, err)
 	return err
 }
 

pkg/kubelet/dockertools/manager_test.go

@@ -80,7 +80,12 @@ func (fr *fakeRuntimeHooks) ShouldPullImage(pod *api.Pod, container *api.Contain
 	return false
 }
 
-func (fr *fakeRuntimeHooks) ReportImagePull(pod *api.Pod, container *api.Container, pullError error) {
+func (fr *fakeRuntimeHooks) ReportImagePulling(pod *api.Pod, container *api.Container) {
+	fr.recorder.Eventf(nil, "pulling", fmt.Sprintf("%s:%s:%s", pod.Name, container.Name, container.Image))
+}
+
+func (fr *fakeRuntimeHooks) ReportImagePulled(pod *api.Pod, container *api.Container, pullError error) {
+	fr.recorder.Eventf(nil, "pulled", fmt.Sprintf("%s:%s:%s", pod.Name, container.Name, container.Image))
 }
 
 type fakeOptionGenerator struct{}
@@ -865,9 +870,10 @@ func generatePodInfraContainerHash(pod *api.Pod) uint64 {
 	}
 
 	container := &api.Container{
-		Name:  PodInfraContainerName,
-		Image: PodInfraContainerImage,
-		Ports: ports,
+		Name:            PodInfraContainerName,
+		Image:           PodInfraContainerImage,
+		Ports:           ports,
+		ImagePullPolicy: podInfraContainerImagePullPolicy,
 	}
 	return kubecontainer.HashContainer(container)
 }
@@ -893,7 +899,7 @@ func runSyncPod(t *testing.T, dm *DockerManager, fakeDocker *FakeDockerClient, p
 
 func TestSyncPodCreateNetAndContainer(t *testing.T) {
 	dm, fakeDocker := newTestDockerManager()
-	dm.podInfraContainerImage = "custom_image_name"
+	dm.podInfraContainerImage = "pod_infra_image"
 	fakeDocker.ContainerList = []docker.APIContainers{}
 	pod := &api.Pod{
 		ObjectMeta: api.ObjectMeta{
@@ -920,7 +926,7 @@ func TestSyncPodCreateNetAndContainer(t *testing.T) {
 
 	found := false
 	for _, c := range fakeDocker.ContainerList {
-		if c.Image == "custom_image_name" && strings.HasPrefix(c.Names[0], "/k8s_POD") {
+		if c.Image == "pod_infra_image" && strings.HasPrefix(c.Names[0], "/k8s_POD") {
 			found = true
 		}
 	}
@@ -938,10 +944,10 @@ func TestSyncPodCreateNetAndContainer(t *testing.T) {
 
 func TestSyncPodCreatesNetAndContainerPullsImage(t *testing.T) {
 	dm, fakeDocker := newTestDockerManager()
-	dm.podInfraContainerImage = "custom_image_name"
+	dm.podInfraContainerImage = "pod_infra_image"
 	puller := dm.puller.(*FakeDockerPuller)
 	puller.HasImages = []string{}
-	dm.podInfraContainerImage = "custom_image_name"
+	dm.podInfraContainerImage = "pod_infra_image"
 	fakeDocker.ContainerList = []docker.APIContainers{}
 	pod := &api.Pod{
 		ObjectMeta: api.ObjectMeta{
@@ -967,7 +973,7 @@ func TestSyncPodCreatesNetAndContainerPullsImage(t *testing.T) {
 
 	fakeDocker.Lock()
 
-	if !reflect.DeepEqual(puller.ImagesPulled, []string{"custom_image_name", "something"}) {
+	if !reflect.DeepEqual(puller.ImagesPulled, []string{"pod_infra_image", "something"}) {
 		t.Errorf("Unexpected pulled containers: %v", puller.ImagesPulled)
 	}
 
@@ -1298,10 +1304,11 @@ func TestSyncPodsDoesNothing(t *testing.T) {
 }
 
 func TestSyncPodWithPullPolicy(t *testing.T) {
+	api.ForTesting_ReferencesAllowBlankSelfLinks = true
 	dm, fakeDocker := newTestDockerManager()
 	puller := dm.puller.(*FakeDockerPuller)
 	puller.HasImages = []string{"existing_one", "want:latest"}
-	dm.podInfraContainerImage = "custom_image_name"
+	dm.podInfraContainerImage = "pod_infra_image"
 	fakeDocker.ContainerList = []docker.APIContainers{}
 
 	pod := &api.Pod{
@@ -1325,13 +1332,39 @@ func TestSyncPodWithPullPolicy(t *testing.T) {
 
 	fakeDocker.Lock()
 
+	eventSet := []string{
+		"pulling foo:POD:pod_infra_image",
+		"pulled foo:POD:pod_infra_image",
+		"pulling foo:bar:pull_always_image",
+		"pulled foo:bar:pull_always_image",
+		"pulling foo:bar2:pull_if_not_present_image",
+		"pulled foo:bar2:pull_if_not_present_image",
+		`pulled Container image "existing_one" already present on machine`,
+		`pulled Container image "want:latest" already present on machine`,
+	}
+
+	runtimeHooks := dm.runtimeHooks.(*fakeRuntimeHooks)
+	recorder := runtimeHooks.recorder.(*record.FakeRecorder)
+
+	var actualEvents []string
+	for _, ev := range recorder.Events {
+		if strings.HasPrefix(ev, "pull") {
+			actualEvents = append(actualEvents, ev)
+		}
+	}
+	sort.StringSlice(actualEvents).Sort()
+	sort.StringSlice(eventSet).Sort()
+	if !reflect.DeepEqual(actualEvents, eventSet) {
+		t.Errorf("Expected: %#v, Actual: %#v", eventSet, actualEvents)
+	}
+
 	pulledImageSet := make(map[string]empty)
 	for v := range puller.ImagesPulled {
 		pulledImageSet[puller.ImagesPulled[v]] = empty{}
 	}
 
 	if !reflect.DeepEqual(pulledImageSet, map[string]empty{
-		"custom_image_name":         {},
+		"pod_infra_image":           {},
 		"pull_always_image":         {},
 		"pull_if_not_present_image": {},
 	}) {

pkg/kubelet/runtime_hooks.go

@@ -49,7 +49,7 @@ func (kr *kubeletRuntimeHooks) ShouldPullImage(pod *api.Pod, container *api.Cont
 	return false
 }
 
-func (kr *kubeletRuntimeHooks) ReportImagePull(pod *api.Pod, container *api.Container, pullError error) {
+func (kr *kubeletRuntimeHooks) ReportImagePulled(pod *api.Pod, container *api.Container, pullError error) {
 	ref, err := kubecontainer.GenerateContainerRef(pod, container)
 	if err != nil {
 		glog.Errorf("Couldn't make a ref to pod %q, container %q: '%v'", pod.Name, container.Name, err)
@@ -62,3 +62,12 @@ func (kr *kubeletRuntimeHooks) ReportImagePull(pod *api.Pod, container *api.Cont
 		kr.recorder.Eventf(ref, "pulled", "Successfully pulled image %q", container.Image)
 	}
 }
+
+func (kr *kubeletRuntimeHooks) ReportImagePulling(pod *api.Pod, container *api.Container) {
+	ref, err := kubecontainer.GenerateContainerRef(pod, container)
+	if err != nil {
+		glog.Errorf("Couldn't make a ref to pod %q, container %q: '%v'", pod.Name, container.Name, err)
+		return
+	}
+	kr.recorder.Eventf(ref, "pulling", "Pulling image %q", container.Image)
+}