kubelet/dockertools: Refactor image pulling for pod infra container.

Replace the trunk of pull image code with dockerManagner.pullImage().
Also add tests to verify the image pulling/pulled events.

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/dockertools/manager.go

@@ -69,6 +69,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
@@ -835,6 +838,7 @@ func (dm *DockerManager) podInfraContainerChanged(pod *api.Pod, podInfraContaine
 		Name:            PodInfraContainerName,
 		Image:           dm.podInfraContainerImage,
 		Ports:           ports,
+		ImagePullPolicy: podInfraContainerImagePullPolicy,
 	}
 	return podInfraContainer.Hash != kubecontainer.HashContainer(expectedPodInfraContainer), nil
 }
@@ -1316,35 +1320,13 @@ func (dm *DockerManager) createPodInfraContainer(pod *api.Pod) (kubeletTypes.Doc
 		Name:            PodInfraContainerName,
 		Image:           dm.podInfraContainerImage,
 		Ports:           ports,
+		ImagePullPolicy: podInfraContainerImagePullPolicy,
 	}
-	ref, err := kubecontainer.GenerateContainerRef(pod, container)
+
-	if err != nil {
+	// No pod secrets for the infra container.
-		glog.Errorf("Couldn't make a ref to pod %v, container %v: '%v'", pod.Name, container.Name, err)
+	if err := dm.pullImage(pod, container, nil); err != nil {
-	}
-	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)
-		}
 		return "", err
 	}
-	if ok {
-		if ref != nil {
-			dm.recorder.Eventf(ref, "pulled", "Pod container image %q already present on machine", container.Image)
-		}
-	} else {
-		dm.runtimeHooks.ReportImagePulling(pod, container)
-		err := dm.PullImage(spec, nil /* no pod secrets for the infra container */)
-		dm.runtimeHooks.ReportImagePulled(pod, container, err)
-		if err != nil {
-			return "", err
-		}
-		if ref != nil {
-			dm.recorder.Eventf(ref, "pulled", "Successfully pulled Pod container image %q", container.Image)
-		}
-	}
 
 	id, err := dm.runContainerInPod(pod, container, netNamespace, "")
 	if err != nil {
@@ -1501,20 +1483,22 @@ 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 {
-	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)
 	}
+	spec := kubecontainer.ImageSpec{container.Image}
+	present, err := dm.IsImagePresent(spec)
+	if err != nil {
 		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
 	}
 

pkg/kubelet/dockertools/manager_test.go

@@ -79,7 +79,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{}
@@ -866,6 +871,7 @@ func generatePodInfraContainerHash(pod *api.Pod) uint64 {
 		Name:            PodInfraContainerName,
 		Image:           PodInfraContainerImage,
 		Ports:           ports,
+		ImagePullPolicy: podInfraContainerImagePullPolicy,
 	}
 	return kubecontainer.HashContainer(container)
 }
@@ -891,7 +897,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{
@@ -918,7 +924,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
 		}
 	}
@@ -936,10 +942,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{
@@ -965,7 +971,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)
 	}
 
@@ -1296,10 +1302,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{
@@ -1323,13 +1330,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

@@ -69,5 +69,5 @@ func (kr *kubeletRuntimeHooks) ReportImagePulling(pod *api.Pod, container *api.C
 		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 for container: %v", container.Image, container.Name)
+	kr.recorder.Eventf(ref, "pulling", "Pulling image %q", container.Image)
 }