Merge pull request #6560 from yifan-gu/get_recent

pkg/kubelet/dockertools/fake_docker_client.go

@@ -31,19 +31,20 @@ import (
 // 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
-	ContainerList []docker.APIContainers
-	Container     *docker.Container
-	ContainerMap  map[string]*docker.Container
-	Image         *docker.Image
-	Images        []docker.APIImages
-	Err           error
-	called        []string
-	Stopped       []string
-	pulled        []string
-	Created       []string
-	Removed       []string
-	RemovedImages util.StringSet
-	VersionInfo   docker.Env
+	ContainerList       []docker.APIContainers
+	ExitedContainerList []docker.APIContainers
+	Container           *docker.Container
+	ContainerMap        map[string]*docker.Container
+	Image               *docker.Image
+	Images              []docker.APIImages
+	Err                 error
+	called              []string
+	Stopped             []string
+	pulled              []string
+	Created             []string
+	Removed             []string
+	RemovedImages       util.StringSet
+	VersionInfo         docker.Env
 }
 
 func (f *FakeDockerClient) ClearCalls() {
@@ -67,17 +68,48 @@ func (f *FakeDockerClient) AssertCalls(calls []string) (err error) {
 	return
 }
 
+func (f *FakeDockerClient) AssertCreated(created []string) error {
+	f.Lock()
+	defer f.Unlock()
+
+	actualCreated := []string{}
+	for _, c := range f.Created {
+		dockerName, _, err := ParseDockerName(c)
+		if err != nil {
+			return fmt.Errorf("unexpected error: %v", err)
+		}
+		actualCreated = append(actualCreated, dockerName.ContainerName)
+	}
+	sort.StringSlice(created).Sort()
+	sort.StringSlice(actualCreated).Sort()
+	if !reflect.DeepEqual(created, actualCreated) {
+		return fmt.Errorf("expected %#v, got %#v", created, actualCreated)
+	}
+	return nil
+}
+
+func (f *FakeDockerClient) AssertStopped(stopped []string) error {
+	f.Lock()
+	defer f.Unlock()
+	sort.StringSlice(stopped).Sort()
+	sort.StringSlice(f.Stopped).Sort()
+	if !reflect.DeepEqual(stopped, f.Stopped) {
+		return fmt.Errorf("expected %#v, got %#v", stopped, f.Stopped)
+	}
+	return nil
+}
+
 func (f *FakeDockerClient) AssertUnorderedCalls(calls []string) (err error) {
 	f.Lock()
 	defer f.Unlock()
 
-	actual := make([]string, len(calls))
-	expected := make([]string, len(f.called))
-	copy(actual, calls)
-	copy(expected, f.called)
+	expected := make([]string, len(calls))
+	actual := make([]string, len(f.called))
+	copy(expected, calls)
+	copy(actual, f.called)
 
-	sort.StringSlice(actual).Sort()
 	sort.StringSlice(expected).Sort()
+	sort.StringSlice(actual).Sort()
 
 	if !reflect.DeepEqual(actual, expected) {
 		err = fmt.Errorf("expected(sorted) %#v, got(sorted) %#v", expected, actual)
@@ -91,6 +123,10 @@ func (f *FakeDockerClient) ListContainers(options docker.ListContainersOptions)
 	f.Lock()
 	defer f.Unlock()
 	f.called = append(f.called, "list")
+
+	if options.All {
+		return append(f.ContainerList, f.ExitedContainerList...), f.Err
+	}
 	return f.ContainerList, f.Err
 }
 

pkg/kubelet/dockertools/manager.go

@@ -30,7 +30,6 @@ import (
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/capabilities"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/client/record"
 	kubecontainer "github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet/container"
-	"github.com/GoogleCloudPlatform/kubernetes/pkg/types"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
 	"github.com/fsouza/go-dockerclient"
 	"github.com/golang/glog"
@@ -51,40 +50,6 @@ func NewDockerManager(client DockerInterface, recorder record.EventRecorder) *Do
 	return &DockerManager{client: client, recorder: recorder}
 }
 
-// GetRecentDockerContainersWithNameAndUUID returns a list of dead docker containers which matches the name
-// and uid given.
-func (self *DockerManager) GetRecentDockerContainersWithNameAndUUID(podFullName string, uid types.UID,
-	containerName string) ([]*docker.Container, error) {
-	var result []*docker.Container
-	containers, err := self.client.ListContainers(docker.ListContainersOptions{All: true})
-	if err != nil {
-		return nil, err
-	}
-	for _, dockerContainer := range containers {
-		if len(dockerContainer.Names) == 0 {
-			continue
-		}
-		dockerName, _, err := ParseDockerName(dockerContainer.Names[0])
-		if err != nil {
-			continue
-		}
-		if dockerName.PodFullName != podFullName {
-			continue
-		}
-		if uid != "" && dockerName.PodUID != uid {
-			continue
-		}
-		if dockerName.ContainerName != containerName {
-			continue
-		}
-		inspectResult, _ := self.client.InspectContainer(dockerContainer.ID)
-		if inspectResult != nil && !inspectResult.State.Running && !inspectResult.State.Paused {
-			result = append(result, inspectResult)
-		}
-	}
-	return result, nil
-}
-
 // GetKubeletDockerContainerLogs returns logs of a specific container. By
 // default, it returns a snapshot of the container log. Set |follow| to true to
 // stream the log. Set |follow| to false and specify the number of lines (e.g.

pkg/kubelet/kubelet.go

@@ -999,28 +999,33 @@ func (kl *Kubelet) makePodDataDirs(pod *api.Pod) error {
 	return nil
 }
 
-func (kl *Kubelet) shouldContainerBeRestarted(container *api.Container, pod *api.Pod) bool {
+func (kl *Kubelet) shouldContainerBeRestarted(container *api.Container, pod *api.Pod, podStatus *api.PodStatus) bool {
 	podFullName := kubecontainer.GetPodFullName(pod)
-	// Check RestartPolicy for dead container
-	recentContainers, err := kl.containerManager.GetRecentDockerContainersWithNameAndUUID(podFullName, pod.UID, container.Name)
-	if err != nil {
-		glog.Errorf("Error listing recent containers for pod %q: %v", podFullName, err)
-		// TODO(dawnchen): error handling here?
+
+	// Get all dead container status.
+	var resultStatus []*api.ContainerStatus
+	for i, containerStatus := range podStatus.ContainerStatuses {
+		if containerStatus.Name == container.Name && containerStatus.State.Termination != nil {
+			resultStatus = append(resultStatus, &podStatus.ContainerStatuses[i])
+		}
 	}
-	// set dead containers to unready state
-	for _, c := range recentContainers {
-		kl.readinessManager.RemoveReadiness(c.ID)
+
+	// Set dead containers to unready state.
+	for _, c := range resultStatus {
+		// TODO(yifan): Unify the format of container ID. (i.e. including docker:// as prefix).
+		kl.readinessManager.RemoveReadiness(strings.TrimPrefix(c.ContainerID, dockertools.DockerPrefix))
 	}
 
-	if len(recentContainers) > 0 {
+	// Check RestartPolicy for dead container.
+	if len(resultStatus) > 0 {
 		if pod.Spec.RestartPolicy == api.RestartPolicyNever {
 			glog.Infof("Already ran container %q of pod %q, do nothing", container.Name, podFullName)
 			return false
-
 		}
 		if pod.Spec.RestartPolicy == api.RestartPolicyOnFailure {
-			// Check the exit code of last run
-			if recentContainers[0].State.ExitCode == 0 {
+			// Check the exit code of last run. Note: This assumes the result is sorted
+			// by the created time in reverse order.
+			if resultStatus[0].State.Termination.ExitCode == 0 {
 				glog.Infof("Already successfully ran container %q of pod %q, do nothing", container.Name, podFullName)
 				return false
 			}
@@ -1106,7 +1111,6 @@ func (kl *Kubelet) computePodContainerChanges(pod *api.Pod, runningPod kubeconta
 	containersToStart := make(map[int]empty)
 	containersToKeep := make(map[dockertools.DockerID]int)
 	createPodInfraContainer := false
-	var podStatus api.PodStatus
 
 	var podInfraContainerID dockertools.DockerID
 	podInfraContainer := runningPod.FindContainerByName(dockertools.PodInfraContainerName)
@@ -1114,15 +1118,20 @@ func (kl *Kubelet) computePodContainerChanges(pod *api.Pod, runningPod kubeconta
 		glog.V(4).Infof("Found infra pod for %q", podFullName)
 		podInfraContainerID = dockertools.DockerID(podInfraContainer.ID)
 		containersToKeep[podInfraContainerID] = -1
-		podStatus, err = kl.GetPodStatus(podFullName)
-		if err != nil {
-			glog.Errorf("Unable to get pod with name %q and uid %q info with error(%v)", podFullName, uid, err)
-		}
+
 	} else {
 		glog.V(2).Infof("No Infra Container for %q found. All containers will be restarted.", podFullName)
 		createPodInfraContainer = true
 	}
 
+	// Do not use the cache here since we need the newest status to check
+	// if we need to restart the container below.
+	podStatus, err := kl.generatePodStatus(podFullName)
+	if err != nil {
+		glog.Errorf("Unable to get pod with name %q and uid %q info with error(%v)", podFullName, uid, err)
+		return podContainerChangesSpec{}, err
+	}
+
 	for index, container := range pod.Spec.Containers {
 		expectedHash := dockertools.HashContainer(&container)
 
@@ -1176,7 +1185,7 @@ func (kl *Kubelet) computePodContainerChanges(pod *api.Pod, runningPod kubeconta
 				continue
 			}
 		} else {
-			if kl.shouldContainerBeRestarted(&container, pod) {
+			if kl.shouldContainerBeRestarted(&container, pod, &podStatus) {
 				// If we are here it means that the container is dead and sould be restarted, or never existed and should
 				// be created. We may be inserting this ID again if the container has changed and it has
 				// RestartPolicy::Always, but it's not a big deal.

pkg/kubelet/kubelet_test.go

@@ -671,7 +671,7 @@ func TestSyncPodsWithPodInfraCreatesContainer(t *testing.T) {
 	waitGroup.Wait()
 
 	verifyCalls(t, fakeDocker, []string{
-		"list", "list", "list", "inspect_container", "list", "create", "start", "list", "inspect_container", "inspect_container"})
+		"list", "list", "list", "inspect_container", "create", "start", "list", "inspect_container", "inspect_container"})
 
 	fakeDocker.Lock()
 	if len(fakeDocker.Created) != 1 ||
@@ -730,7 +730,7 @@ func TestSyncPodsWithPodInfraCreatesContainerCallsHandler(t *testing.T) {
 	waitGroup.Wait()
 
 	verifyCalls(t, fakeDocker, []string{
-		"list", "list", "list", "inspect_container", "list", "create", "start", "list", "inspect_container", "inspect_container"})
+		"list", "list", "list", "inspect_container", "create", "start", "list", "inspect_container", "inspect_container"})
 
 	fakeDocker.Lock()
 	if len(fakeDocker.Created) != 1 ||
@@ -801,7 +801,13 @@ func TestSyncPodsDeletesWithNoPodInfraContainer(t *testing.T) {
 	waitGroup.Wait()
 
 	verifyUnorderedCalls(t, fakeDocker, []string{
-		"list", "list", "list", "list", "inspect_container", "inspect_container", "list", "inspect_container", "inspect_container", "stop", "create", "start", "inspect_container", "create", "start", "list", "inspect_container", "inspect_container"})
+		"list",
+		// foo1
+		"list", "list", "inspect_container", "stop", "create", "start", "inspect_container", "create", "start", "list", "inspect_container", "inspect_container",
+
+		// foo2
+		"list", "list", "inspect_container", "inspect_container", "list", "inspect_container", "inspect_container",
+	})
 
 	// A map iteration is used to delete containers, so must not depend on
 	// order here.
@@ -1632,7 +1638,7 @@ func TestSyncPodEventHandlerFails(t *testing.T) {
 		t.Errorf("unexpected error: %v", err)
 	}
 
-	verifyCalls(t, fakeDocker, []string{"list", "list", "create", "start", "stop", "list"})
+	verifyCalls(t, fakeDocker, []string{"list", "create", "start", "stop", "list"})
 
 	if len(fakeDocker.Stopped) != 1 {
 		t.Errorf("Wrong containers were stopped: %v", fakeDocker.Stopped)
@@ -1649,8 +1655,8 @@ func TestSyncPodsWithPullPolicy(t *testing.T) {
 	puller.HasImages = []string{"existing_one", "want:latest"}
 	kubelet.podInfraContainerImage = "custom_image_name"
 	fakeDocker.ContainerList = []docker.APIContainers{}
-	waitGroup.Add(1)
-	err := kubelet.SyncPods([]api.Pod{
+
+	pods := []api.Pod{
 		{
 			ObjectMeta: api.ObjectMeta{
 				UID:       "12345678",
@@ -1667,7 +1673,10 @@ func TestSyncPodsWithPullPolicy(t *testing.T) {
 				},
 			},
 		},
-	}, emptyPodUIDs, map[string]api.Pod{}, time.Now())
+	}
+	kubelet.podManager.SetPods(pods)
+	waitGroup.Add(1)
+	err := kubelet.SyncPods(pods, emptyPodUIDs, map[string]api.Pod{}, time.Now())
 	if err != nil {
 		t.Errorf("unexpected error: %v", err)
 	}
@@ -3442,6 +3451,7 @@ func TestHostNetworkAllowed(t *testing.T) {
 			HostNetwork: true,
 		},
 	}
+	kubelet.podManager.SetPods([]api.Pod{*pod})
 	err := kubelet.syncPod(pod, nil, container.Pod{})
 	if err != nil {
 		t.Errorf("expected pod infra creation to succeed: %v", err)
@@ -3476,3 +3486,122 @@ func TestHostNetworkDisallowed(t *testing.T) {
 		t.Errorf("expected pod infra creation to fail")
 	}
 }
+
+func TestSyncPodsWithRestartPolicy(t *testing.T) {
+	testKubelet := newTestKubelet(t)
+	testKubelet.fakeCadvisor.On("MachineInfo").Return(&cadvisorApi.MachineInfo{}, nil)
+	kubelet := testKubelet.kubelet
+	fakeDocker := testKubelet.fakeDocker
+	waitGroup := testKubelet.waitGroup
+
+	containers := []api.Container{
+		{Name: "succeeded"},
+		{Name: "failed"},
+	}
+
+	runningAPIContainers := []docker.APIContainers{
+		{
+			// pod infra container
+			Names: []string{"/k8s_POD_foo_new_12345678_0"},
+			ID:    "9876",
+		},
+	}
+	exitedAPIContainers := []docker.APIContainers{
+		{
+			// format is // k8s_<container-id>_<pod-fullname>_<pod-uid>
+			Names: []string{"/k8s_succeeded." + strconv.FormatUint(dockertools.HashContainer(&containers[0]), 16) + "_foo_new_12345678_0"},
+			ID:    "1234",
+		},
+		{
+			// format is // k8s_<container-id>_<pod-fullname>_<pod-uid>
+			Names: []string{"/k8s_failed." + strconv.FormatUint(dockertools.HashContainer(&containers[1]), 16) + "_foo_new_12345678_0"},
+			ID:    "5678",
+		},
+	}
+
+	containerMap := map[string]*docker.Container{
+		"1234": {
+			ID:     "1234",
+			Name:   "succeeded",
+			Config: &docker.Config{},
+			State: docker.State{
+				ExitCode:   0,
+				StartedAt:  time.Now(),
+				FinishedAt: time.Now(),
+			},
+		},
+		"5678": {
+			ID:     "5678",
+			Name:   "failed",
+			Config: &docker.Config{},
+			State: docker.State{
+				ExitCode:   42,
+				StartedAt:  time.Now(),
+				FinishedAt: time.Now(),
+			},
+		},
+	}
+
+	tests := []struct {
+		policy  api.RestartPolicy
+		calls   []string
+		created []string
+		stopped []string
+	}{
+		{
+			api.RestartPolicyAlways,
+			[]string{"list", "list", "list", "inspect_container", "inspect_container", "inspect_container", "create", "start", "create", "start", "list", "inspect_container", "inspect_container", "inspect_container"},
+			[]string{"succeeded", "failed"},
+			[]string{},
+		},
+		{
+			api.RestartPolicyOnFailure,
+			[]string{"list", "list", "list", "inspect_container", "inspect_container", "inspect_container", "create", "start", "list", "inspect_container", "inspect_container", "inspect_container"},
+			[]string{"failed"},
+			[]string{},
+		},
+		{
+			api.RestartPolicyNever,
+			[]string{"list", "list", "list", "inspect_container", "inspect_container", "inspect_container", "stop", "list", "inspect_container", "inspect_container"},
+			[]string{},
+			[]string{"9876"},
+		},
+	}
+
+	for i, tt := range tests {
+		fakeDocker.ContainerList = runningAPIContainers
+		fakeDocker.ExitedContainerList = exitedAPIContainers
+		fakeDocker.ContainerMap = containerMap
+		fakeDocker.ClearCalls()
+		pods := []api.Pod{
+			{
+				ObjectMeta: api.ObjectMeta{
+					UID:       "12345678",
+					Name:      "foo",
+					Namespace: "new",
+				},
+				Spec: api.PodSpec{
+					Containers:    containers,
+					RestartPolicy: tt.policy,
+				},
+			},
+		}
+		kubelet.podManager.SetPods(pods)
+		waitGroup.Add(1)
+		err := kubelet.SyncPods(pods, emptyPodUIDs, map[string]api.Pod{}, time.Now())
+		if err != nil {
+			t.Errorf("%d: unexpected error: %v", i, err)
+		}
+		waitGroup.Wait()
+
+		// 'stop' is because the pod infra container is killed when no container is running.
+		verifyCalls(t, fakeDocker, tt.calls)
+
+		if err := fakeDocker.AssertCreated(tt.created); err != nil {
+			t.Errorf("%d: %v", i, err)
+		}
+		if err := fakeDocker.AssertStopped(tt.stopped); err != nil {
+			t.Errorf("%d: %v", i, err)
+		}
+	}
+}

pkg/kubelet/runonce_test.go

@@ -74,6 +74,9 @@ func (d *testDocker) InspectContainer(id string) (*docker.Container, error) {
 func TestRunOnce(t *testing.T) {
 	cadvisor := &cadvisor.Mock{}
 	cadvisor.On("MachineInfo").Return(&cadvisorApi.MachineInfo{}, nil)
+
+	podManager, _ := newFakePodManager()
+
 	kb := &Kubelet{
 		rootDirectory:       "/tmp/kubelet",
 		recorder:            &record.FakeRecorder{},
@@ -81,6 +84,8 @@ func TestRunOnce(t *testing.T) {
 		nodeLister:          testNodeLister{},
 		statusManager:       newStatusManager(nil),
 		containerRefManager: kubecontainer.NewRefManager(),
+		readinessManager:    kubecontainer.NewReadinessManager(),
+		podManager:          podManager,
 	}
 
 	kb.networkPlugin, _ = network.InitNetworkPlugin([]network.NetworkPlugin{}, "", network.NewFakeHost(nil))
@@ -141,7 +146,8 @@ func TestRunOnce(t *testing.T) {
 	}
 	kb.dockerPuller = &dockertools.FakeDockerPuller{}
 	kb.containerManager = dockertools.NewDockerManager(kb.dockerClient, kb.recorder)
-	results, err := kb.runOnce([]api.Pod{
+
+	pods := []api.Pod{
 		{
 			ObjectMeta: api.ObjectMeta{
 				UID:       "12345678",
@@ -154,7 +160,9 @@ func TestRunOnce(t *testing.T) {
 				},
 			},
 		},
-	}, time.Millisecond)
+	}
+	podManager.SetPods(pods)
+	results, err := kb.runOnce(pods, time.Millisecond)
 	if err != nil {
 		t.Errorf("unexpected error: %v", err)
 	}