Merge pull request #15993 from timstclair/readiness-workers

Auto commit by PR queue bot

pkg/kubelet/kubelet.go

@@ -322,8 +322,7 @@ func NewMainKubelet(
 	procFs := procfs.NewProcFs()
 	imageBackOff := util.NewBackOff(resyncInterval, MaxContainerBackOff)
 
-	readinessManager := proberesults.NewManager()
+	klet.livenessManager = proberesults.NewManager()
-	klet.livenessManager = proberesults.NewManagerWithUpdates()
 
 	// Initialize the runtime.
 	switch containerRuntime {
@@ -419,7 +418,6 @@ func NewMainKubelet(
 
 	klet.probeManager = prober.NewManager(
 		klet.statusManager,
-		readinessManager,
 		klet.livenessManager,
 		klet.runner,
 		containerRefManager,

pkg/kubelet/prober/manager.go

@@ -27,6 +27,7 @@ import (
 	"k8s.io/kubernetes/pkg/kubelet/status"
 	kubeutil "k8s.io/kubernetes/pkg/kubelet/util"
 	"k8s.io/kubernetes/pkg/types"
+	"k8s.io/kubernetes/pkg/util"
 	"k8s.io/kubernetes/pkg/util/sets"
 )
 
@@ -74,19 +75,24 @@ type manager struct {
 
 func NewManager(
 	statusManager status.Manager,
-	readinessManager results.Manager,
 	livenessManager results.Manager,
 	runner kubecontainer.ContainerCommandRunner,
 	refManager *kubecontainer.RefManager,
 	recorder record.EventRecorder) Manager {
 	prober := newProber(runner, refManager, recorder)
-	return &manager{
+	readinessManager := results.NewManager()
+	m := &manager{
 		statusManager:    statusManager,
 		prober:           prober,
 		readinessManager: readinessManager,
 		livenessManager:  livenessManager,
 		workers:          make(map[probeKey]*worker),
 	}
+
+	// Start syncing readiness.
+	go util.Forever(m.updateReadiness, 0)
+
+	return m
 }
 
 // Key uniquely identifying container probes
@@ -211,3 +217,10 @@ func (m *manager) removeWorker(podUID types.UID, containerName string, probeType
 	defer m.workerLock.Unlock()
 	delete(m.workers, probeKey{podUID, containerName, probeType})
 }
+
+func (m *manager) updateReadiness() {
+	update := <-m.readinessManager.Updates()
+
+	ready := update.Result == results.Success
+	m.statusManager.SetContainerReadiness(update.Pod, update.ContainerID, ready)
+}

pkg/kubelet/prober/manager_test.go

@@ -23,17 +23,17 @@ import (
 
 	"github.com/golang/glog"
 	"k8s.io/kubernetes/pkg/api"
-	"k8s.io/kubernetes/pkg/client/record"
-	"k8s.io/kubernetes/pkg/client/unversioned/testclient"
 	kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
-	kubepod "k8s.io/kubernetes/pkg/kubelet/pod"
 	"k8s.io/kubernetes/pkg/kubelet/prober/results"
-	"k8s.io/kubernetes/pkg/kubelet/status"
 	"k8s.io/kubernetes/pkg/probe"
 	"k8s.io/kubernetes/pkg/util"
 	"k8s.io/kubernetes/pkg/util/wait"
 )
 
+func init() {
+	util.ReallyCrash = true
+}
+
 var defaultProbe *api.Probe = &api.Probe{
 	Handler: api.Handler{
 		Exec: &api.ExecAction{},
@@ -172,7 +172,6 @@ func TestCleanupPods(t *testing.T) {
 }
 
 func TestUpdatePodStatus(t *testing.T) {
-	const podUID = "pod_uid"
 	unprobed := api.ContainerStatus{
 		Name:        "unprobed_container",
 		ContainerID: "test://unprobed_container_id",
@@ -218,27 +217,27 @@ func TestUpdatePodStatus(t *testing.T) {
 	m := newTestManager()
 	// Setup probe "workers" and cached results.
 	m.workers = map[probeKey]*worker{
-		probeKey{podUID, unprobed.Name, liveness}:       {},
+		probeKey{testPodUID, unprobed.Name, liveness}:       {},
-		probeKey{podUID, probedReady.Name, readiness}:   {},
+		probeKey{testPodUID, probedReady.Name, readiness}:   {},
-		probeKey{podUID, probedPending.Name, readiness}: {},
+		probeKey{testPodUID, probedPending.Name, readiness}: {},
-		probeKey{podUID, probedUnready.Name, readiness}: {},
+		probeKey{testPodUID, probedUnready.Name, readiness}: {},
-		probeKey{podUID, terminated.Name, readiness}:    {},
+		probeKey{testPodUID, terminated.Name, readiness}:    {},
 	}
-	m.readinessManager.Set(kubecontainer.ParseContainerID(probedReady.ContainerID), results.Success, nil)
+	m.readinessManager.Set(kubecontainer.ParseContainerID(probedReady.ContainerID), results.Success, &api.Pod{})
-	m.readinessManager.Set(kubecontainer.ParseContainerID(probedUnready.ContainerID), results.Failure, nil)
+	m.readinessManager.Set(kubecontainer.ParseContainerID(probedUnready.ContainerID), results.Failure, &api.Pod{})
-	m.readinessManager.Set(kubecontainer.ParseContainerID(terminated.ContainerID), results.Success, nil)
+	m.readinessManager.Set(kubecontainer.ParseContainerID(terminated.ContainerID), results.Success, &api.Pod{})
 
-	m.UpdatePodStatus(podUID, &podStatus)
+	m.UpdatePodStatus(testPodUID, &podStatus)
 
 	expectedReadiness := map[probeKey]bool{
-		probeKey{podUID, unprobed.Name, readiness}:      true,
+		probeKey{testPodUID, unprobed.Name, readiness}:      true,
-		probeKey{podUID, probedReady.Name, readiness}:   true,
+		probeKey{testPodUID, probedReady.Name, readiness}:   true,
-		probeKey{podUID, probedPending.Name, readiness}: false,
+		probeKey{testPodUID, probedPending.Name, readiness}: false,
-		probeKey{podUID, probedUnready.Name, readiness}: false,
+		probeKey{testPodUID, probedUnready.Name, readiness}: false,
-		probeKey{podUID, terminated.Name, readiness}:    false,
+		probeKey{testPodUID, terminated.Name, readiness}:    false,
 	}
 	for _, c := range podStatus.ContainerStatuses {
-		expected, ok := expectedReadiness[probeKey{podUID, c.Name, readiness}]
+		expected, ok := expectedReadiness[probeKey{testPodUID, c.Name, readiness}]
 		if !ok {
 			t.Fatalf("Missing expectation for test case: %v", c.Name)
 		}
@@ -249,6 +248,31 @@ func TestUpdatePodStatus(t *testing.T) {
 	}
 }
 
+func TestUpdateReadiness(t *testing.T) {
+	testPod := getTestPod(readiness, api.Probe{})
+	m := newTestManager()
+	m.statusManager.SetPodStatus(&testPod, getTestRunningStatus())
+
+	m.AddPod(&testPod)
+	probePaths := []probeKey{{testPodUID, testContainerName, readiness}}
+	if err := expectProbes(m, probePaths); err != nil {
+		t.Error(err)
+	}
+
+	// Wait for ready status.
+	if err := waitForReadyStatus(m, true); err != nil {
+		t.Error(err)
+	}
+
+	// Prober fails.
+	m.prober.exec = fakeExecProber{probe.Failure, nil}
+
+	// Wait for failed status.
+	if err := waitForReadyStatus(m, false); err != nil {
+		t.Error(err)
+	}
+}
+
 func expectProbes(m *manager, expectedProbes []probeKey) error {
 	m.workerLock.RLock()
 	defer m.workerLock.RUnlock()
@@ -275,24 +299,10 @@ outer:
 	return fmt.Errorf("Unexpected probes: %v; Missing probes: %v;", unexpected, missing)
 }
 
-func newTestManager() *manager {
+const interval = 100 * time.Millisecond
-	m := NewManager(
-		status.NewManager(&testclient.Fake{}, kubepod.NewBasicPodManager(nil)),
-		results.NewManager(),
-		results.NewManager(),
-		nil, // runner
-		kubecontainer.NewRefManager(),
-		&record.FakeRecorder{},
-	).(*manager)
-	// Don't actually execute probes.
-	m.prober.exec = fakeExecProber{probe.Success, nil}
-	return m
-}
 
 // Wait for the given workers to exit & clean up.
 func waitForWorkerExit(m *manager, workerPaths []probeKey) error {
-	const interval = 100 * time.Millisecond
-
 	for _, w := range workerPaths {
 		condition := func() (bool, error) {
 			_, exists := m.getWorker(w.podUID, w.containerName, w.probeType)
@@ -309,3 +319,27 @@ func waitForWorkerExit(m *manager, workerPaths []probeKey) error {
 
 	return nil
 }
+
+// Wait for the given workers to exit & clean up.
+func waitForReadyStatus(m *manager, ready bool) error {
+	condition := func() (bool, error) {
+		status, ok := m.statusManager.GetPodStatus(testPodUID)
+		if !ok {
+			return false, fmt.Errorf("status not found: %q", testPodUID)
+		}
+		if len(status.ContainerStatuses) != 1 {
+			return false, fmt.Errorf("expected single container, found %d", len(status.ContainerStatuses))
+		}
+		if status.ContainerStatuses[0].ContainerID != testContainerID.String() {
+			return false, fmt.Errorf("expected container %q, found %q",
+				testContainerID, status.ContainerStatuses[0].ContainerID)
+		}
+		return status.ContainerStatuses[0].Ready == ready, nil
+	}
+	glog.Infof("Polling for ready state %v", ready)
+	if err := wait.Poll(interval, util.ForeverTestTimeout, condition); err != nil {
+		return err
+	}
+
+	return nil
+}

pkg/kubelet/prober/prober_test.go

@@ -27,7 +27,6 @@ import (
 	"k8s.io/kubernetes/pkg/kubelet/prober/results"
 	"k8s.io/kubernetes/pkg/probe"
 	"k8s.io/kubernetes/pkg/util"
-	"k8s.io/kubernetes/pkg/util/exec"
 )
 
 func TestFormatURL(t *testing.T) {
@@ -246,12 +245,3 @@ func TestProbe(t *testing.T) {
 		}
 	}
 }
-
-type fakeExecProber struct {
-	result probe.Result
-	err    error
-}
-
-func (p fakeExecProber) Probe(_ exec.Cmd) (probe.Result, string, error) {
-	return p.result, "", p.err
-}

pkg/kubelet/prober/results/results_manager.go

@@ -70,7 +70,7 @@ type manager struct {
 	sync.RWMutex
 	// map of container ID -> probe Result
 	cache map[kubecontainer.ContainerID]Result
-	// channel of updates (may be nil)
+	// channel of updates
 	updates chan Update
 }
 
@@ -78,15 +78,10 @@ var _ Manager = &manager{}
 
 // NewManager creates ane returns an empty results manager.
 func NewManager() Manager {
-	m := &manager{cache: make(map[kubecontainer.ContainerID]Result)}
+	return &manager{
-	return m
+		cache:   make(map[kubecontainer.ContainerID]Result),
-}
+		updates: make(chan Update, 20),
-
+	}
-// NewManager creates ane returns an empty results manager.
-func NewManagerWithUpdates() Manager {
-	m := NewManager().(*manager)
-	m.updates = make(chan Update, 20)
-	return m
 }
 
 func (m *manager) Get(id kubecontainer.ContainerID) (Result, bool) {
@@ -98,7 +93,7 @@ func (m *manager) Get(id kubecontainer.ContainerID) (Result, bool) {
 
 func (m *manager) Set(id kubecontainer.ContainerID, result Result, pod *api.Pod) {
 	if m.setInternal(id, result) {
-		m.pushUpdate(Update{id, result, pod})
+		m.updates <- Update{id, result, pod}
 	}
 }
 
@@ -123,10 +118,3 @@ func (m *manager) Remove(id kubecontainer.ContainerID) {
 func (m *manager) Updates() <-chan Update {
 	return m.updates
 }
-
-// pushUpdates sends an update on the updates channel if it is initialized.
-func (m *manager) pushUpdate(update Update) {
-	if m.updates != nil {
-		m.updates <- update
-	}
-}

pkg/kubelet/prober/results/results_manager_test.go

@@ -46,7 +46,7 @@ func TestCacheOperations(t *testing.T) {
 }
 
 func TestUpdates(t *testing.T) {
-	m := NewManagerWithUpdates()
+	m := NewManager()
 
 	pod := &api.Pod{ObjectMeta: api.ObjectMeta{Name: "test-pod"}}
 	fooID := kubecontainer.ContainerID{"test", "foo"}

pkg/kubelet/prober/testing.go

@@ -0,0 +1,122 @@
+/*
+Copyright 2015 The Kubernetes Authors All rights reserved.
+
+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 prober
+
+import (
+	"reflect"
+
+	"k8s.io/kubernetes/pkg/api"
+	"k8s.io/kubernetes/pkg/api/unversioned"
+	"k8s.io/kubernetes/pkg/client/record"
+	"k8s.io/kubernetes/pkg/client/unversioned/testclient"
+	kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
+	kubepod "k8s.io/kubernetes/pkg/kubelet/pod"
+	"k8s.io/kubernetes/pkg/kubelet/prober/results"
+	"k8s.io/kubernetes/pkg/kubelet/status"
+	"k8s.io/kubernetes/pkg/probe"
+	"k8s.io/kubernetes/pkg/util/exec"
+)
+
+const (
+	testContainerName = "cOnTaInEr_NaMe"
+	testPodUID        = "pOd_UiD"
+)
+
+var testContainerID = kubecontainer.ContainerID{"test", "cOnTaInEr_Id"}
+
+func getTestRunningStatus() api.PodStatus {
+	containerStatus := api.ContainerStatus{
+		Name:        testContainerName,
+		ContainerID: testContainerID.String(),
+	}
+	containerStatus.State.Running = &api.ContainerStateRunning{unversioned.Now()}
+	podStatus := api.PodStatus{
+		Phase:             api.PodRunning,
+		ContainerStatuses: []api.ContainerStatus{containerStatus},
+	}
+	return podStatus
+}
+
+func getTestPod(probeType probeType, probeSpec api.Probe) api.Pod {
+	container := api.Container{
+		Name: testContainerName,
+	}
+
+	// All tests rely on the fake exec prober.
+	probeSpec.Handler = api.Handler{
+		Exec: &api.ExecAction{},
+	}
+
+	// Apply test defaults, overwridden for test speed.
+	defaults := map[string]int64{
+		"TimeoutSeconds":   1,
+		"PeriodSeconds":    1,
+		"SuccessThreshold": 1,
+		"FailureThreshold": 1,
+	}
+	for field, value := range defaults {
+		f := reflect.ValueOf(&probeSpec).Elem().FieldByName(field)
+		if f.Int() == 0 {
+			f.SetInt(value)
+		}
+	}
+
+	switch probeType {
+	case readiness:
+		container.ReadinessProbe = &probeSpec
+	case liveness:
+		container.LivenessProbe = &probeSpec
+	}
+	pod := api.Pod{
+		Spec: api.PodSpec{
+			Containers:    []api.Container{container},
+			RestartPolicy: api.RestartPolicyNever,
+		},
+	}
+	pod.Name = "testPod"
+	pod.UID = testPodUID
+	return pod
+}
+
+func newTestManager() *manager {
+	refManager := kubecontainer.NewRefManager()
+	refManager.SetRef(testContainerID, &api.ObjectReference{}) // Suppress prober warnings.
+	m := NewManager(
+		status.NewManager(&testclient.Fake{}, kubepod.NewBasicPodManager(nil)),
+		results.NewManager(),
+		nil, // runner
+		refManager,
+		&record.FakeRecorder{},
+	).(*manager)
+	// Don't actually execute probes.
+	m.prober.exec = fakeExecProber{probe.Success, nil}
+	return m
+}
+
+func newTestWorker(m *manager, probeType probeType, probeSpec api.Probe) *worker {
+	pod := getTestPod(probeType, probeSpec)
+	return newWorker(m, probeType, &pod, pod.Spec.Containers[0])
+}
+
+type fakeExecProber struct {
+	result probe.Result
+	err    error
+}
+
+func (p fakeExecProber) Probe(_ exec.Cmd) (probe.Result, string, error) {
+	return p.result, "", p.err
+}

pkg/kubelet/prober/worker_test.go

@@ -18,7 +18,6 @@ package prober
 
 import (
 	"fmt"
-	"reflect"
 	"testing"
 	"time"
 
@@ -36,28 +35,25 @@ import (
 	"k8s.io/kubernetes/pkg/util/wait"
 )
 
-const (
+func init() {
-	containerName = "cOnTaInEr_NaMe"
+	util.ReallyCrash = true
-	podUID        = "pOd_UiD"
+}
-)
-
-var containerID = kubecontainer.ContainerID{"test", "cOnTaInEr_Id"}
 
 func TestDoProbe(t *testing.T) {
 	m := newTestManager()
 
 	// Test statuses.
-	runningStatus := getRunningStatus()
+	runningStatus := getTestRunningStatus()
-	pendingStatus := getRunningStatus()
+	pendingStatus := getTestRunningStatus()
 	pendingStatus.ContainerStatuses[0].State.Running = nil
-	terminatedStatus := getRunningStatus()
+	terminatedStatus := getTestRunningStatus()
 	terminatedStatus.ContainerStatuses[0].State.Running = nil
 	terminatedStatus.ContainerStatuses[0].State.Terminated = &api.ContainerStateTerminated{
 		StartedAt: unversioned.Now(),
 	}
-	otherStatus := getRunningStatus()
+	otherStatus := getTestRunningStatus()
 	otherStatus.ContainerStatuses[0].Name = "otherContainer"
-	failedStatus := getRunningStatus()
+	failedStatus := getTestRunningStatus()
 	failedStatus.Phase = api.PodFailed
 
 	tests := []struct {
@@ -112,7 +108,7 @@ func TestDoProbe(t *testing.T) {
 			if c := w.doProbe(); c != test.expectContinue {
 				t.Errorf("[%s-%d] Expected continue to be %v but got %v", probeType, i, test.expectContinue, c)
 			}
-			result, ok := resultsManager(m, probeType).Get(containerID)
+			result, ok := resultsManager(m, probeType).Get(testContainerID)
 			if ok != test.expectSet {
 				t.Errorf("[%s-%d] Expected to have result: %v but got %v", probeType, i, test.expectSet, ok)
 			}
@@ -122,7 +118,7 @@ func TestDoProbe(t *testing.T) {
 
 			// Clean up.
 			m.statusManager = status.NewManager(&testclient.Fake{}, kubepod.NewBasicPodManager(nil))
-			resultsManager(m, probeType).Remove(containerID)
+			resultsManager(m, probeType).Remove(testContainerID)
 		}
 	}
 }
@@ -134,13 +130,13 @@ func TestInitialDelay(t *testing.T) {
 		w := newTestWorker(m, probeType, api.Probe{
 			InitialDelaySeconds: 10,
 		})
-		m.statusManager.SetPodStatus(w.pod, getRunningStatus())
+		m.statusManager.SetPodStatus(w.pod, getTestRunningStatus())
 
 		expectContinue(t, w, w.doProbe(), "during initial delay")
 		expectResult(t, w, results.Result(probeType == liveness), "during initial delay")
 
 		// 100 seconds later...
-		laterStatus := getRunningStatus()
+		laterStatus := getTestRunningStatus()
 		laterStatus.ContainerStatuses[0].State.Running.StartedAt.Time =
 			time.Now().Add(-100 * time.Second)
 		m.statusManager.SetPodStatus(w.pod, laterStatus)
@@ -153,8 +149,8 @@ func TestInitialDelay(t *testing.T) {
 
 func TestFailureThreshold(t *testing.T) {
 	m := newTestManager()
-	w := newTestWorker(m, readiness, api.Probe{})
+	w := newTestWorker(m, readiness, api.Probe{SuccessThreshold: 1, FailureThreshold: 3})
-	m.statusManager.SetPodStatus(w.pod, getRunningStatus())
+	m.statusManager.SetPodStatus(w.pod, getTestRunningStatus())
 
 	for i := 0; i < 2; i++ {
 		// First probe should succeed.
@@ -171,7 +167,7 @@ func TestFailureThreshold(t *testing.T) {
 
 		// Next 2 probes should still be "success".
 		for j := 0; j < 2; j++ {
-			msg := fmt.Sprintf("%d failure (%d)", j+1, i)
+			msg := fmt.Sprintf("%d failing (%d)", j+1, i)
 			expectContinue(t, w, w.doProbe(), msg)
 			expectResult(t, w, results.Success, msg)
 		}
@@ -188,10 +184,10 @@ func TestFailureThreshold(t *testing.T) {
 func TestSuccessThreshold(t *testing.T) {
 	m := newTestManager()
 	w := newTestWorker(m, readiness, api.Probe{SuccessThreshold: 3, FailureThreshold: 1})
-	m.statusManager.SetPodStatus(w.pod, getRunningStatus())
+	m.statusManager.SetPodStatus(w.pod, getTestRunningStatus())
 
 	// Start out failure.
-	w.resultsManager.Set(containerID, results.Failure, nil)
+	w.resultsManager.Set(testContainerID, results.Failure, &api.Pod{})
 
 	for i := 0; i < 2; i++ {
 		// Probe defaults to Failure.
@@ -223,15 +219,15 @@ func TestCleanUp(t *testing.T) {
 	m := newTestManager()
 
 	for _, probeType := range [...]probeType{liveness, readiness} {
-		key := probeKey{podUID, containerName, probeType}
+		key := probeKey{testPodUID, testContainerName, probeType}
 		w := newTestWorker(m, probeType, api.Probe{})
-		m.statusManager.SetPodStatus(w.pod, getRunningStatus())
+		m.statusManager.SetPodStatus(w.pod, getTestRunningStatus())
 		go w.run()
 		m.workers[key] = w
 
 		// Wait for worker to run.
 		condition := func() (bool, error) {
-			ready, _ := resultsManager(m, probeType).Get(containerID)
+			ready, _ := resultsManager(m, probeType).Get(testContainerID)
 			return ready == results.Success, nil
 		}
 		if ready, _ := condition(); !ready {
@@ -245,7 +241,7 @@ func TestCleanUp(t *testing.T) {
 			t.Fatalf("[%s] error waiting for worker exit: %v", probeType, err)
 		}
 
-		if _, ok := resultsManager(m, probeType).Get(containerID); ok {
+		if _, ok := resultsManager(m, probeType).Get(testContainerID); ok {
 			t.Errorf("[%s] Expected result to be cleared.", probeType)
 		}
 		if _, ok := m.workers[key]; ok {
@@ -255,9 +251,11 @@ func TestCleanUp(t *testing.T) {
 }
 
 func TestHandleCrash(t *testing.T) {
+	util.ReallyCrash = false // Test that we *don't* really crash.
+
 	m := newTestManager()
 	w := newTestWorker(m, readiness, api.Probe{})
-	m.statusManager.SetPodStatus(w.pod, getRunningStatus())
+	m.statusManager.SetPodStatus(w.pod, getTestRunningStatus())
 
 	expectContinue(t, w, w.doProbe(), "Initial successful probe.")
 	expectResult(t, w, results.Success, "Initial successful probe.")
@@ -274,64 +272,8 @@ func TestHandleCrash(t *testing.T) {
 	expectResult(t, w, results.Success, "Crashing probe unchanged.")
 }
 
-func newTestWorker(m *manager, probeType probeType, probeSpec api.Probe) *worker {
-	// All tests rely on the fake exec prober.
-	probeSpec.Handler = api.Handler{
-		Exec: &api.ExecAction{},
-	}
-	// Apply default values.
-	defaults := map[string]int64{
-		"TimeoutSeconds":   1,
-		"PeriodSeconds":    10,
-		"SuccessThreshold": 1,
-		"FailureThreshold": 3,
-	}
-	for field, value := range defaults {
-		f := reflect.ValueOf(&probeSpec).Elem().FieldByName(field)
-		if f.Int() == 0 {
-			f.SetInt(value)
-		}
-	}
-
-	pod := getTestPod(probeType, probeSpec)
-	return newWorker(m, probeType, &pod, pod.Spec.Containers[0])
-}
-
-func getRunningStatus() api.PodStatus {
-	containerStatus := api.ContainerStatus{
-		Name:        containerName,
-		ContainerID: containerID.String(),
-	}
-	containerStatus.State.Running = &api.ContainerStateRunning{unversioned.Now()}
-	podStatus := api.PodStatus{
-		Phase:             api.PodRunning,
-		ContainerStatuses: []api.ContainerStatus{containerStatus},
-	}
-	return podStatus
-}
-
-func getTestPod(probeType probeType, probeSpec api.Probe) api.Pod {
-	container := api.Container{
-		Name: containerName,
-	}
-	switch probeType {
-	case readiness:
-		container.ReadinessProbe = &probeSpec
-	case liveness:
-		container.LivenessProbe = &probeSpec
-	}
-	pod := api.Pod{
-		Spec: api.PodSpec{
-			Containers:    []api.Container{container},
-			RestartPolicy: api.RestartPolicyNever,
-		},
-	}
-	pod.UID = podUID
-	return pod
-}
-
 func expectResult(t *testing.T, w *worker, expectedResult results.Result, msg string) {
-	result, ok := resultsManager(w.probeManager, w.probeType).Get(containerID)
+	result, ok := resultsManager(w.probeManager, w.probeType).Get(testContainerID)
 	if !ok {
 		t.Errorf("[%s - %s] Expected result to be set, but was not set", w.probeType, msg)
 	} else if result != expectedResult {

pkg/kubelet/status/manager.go

@@ -27,6 +27,7 @@ import (
 	"k8s.io/kubernetes/pkg/api/errors"
 	"k8s.io/kubernetes/pkg/api/unversioned"
 	client "k8s.io/kubernetes/pkg/client/unversioned"
+	kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
 	kubepod "k8s.io/kubernetes/pkg/kubelet/pod"
 	kubetypes "k8s.io/kubernetes/pkg/kubelet/types"
 	kubeletutil "k8s.io/kubernetes/pkg/kubelet/util"
@@ -77,6 +78,10 @@ type Manager interface {
 	// SetPodStatus caches updates the cached status for the given pod, and triggers a status update.
 	SetPodStatus(pod *api.Pod, status api.PodStatus)
 
+	// SetContainerReadiness updates the cached container status with the given readiness, and
+	// triggers a status update.
+	SetContainerReadiness(pod *api.Pod, containerID kubecontainer.ContainerID, ready bool)
+
 	// TerminatePods resets the container status for the provided pods to terminated and triggers
 	// a status update. This function may not enqueue all the provided pods, in which case it will
 	// return false
@@ -171,19 +176,50 @@ func (m *manager) SetPodStatus(pod *api.Pod, status api.PodStatus) {
 		status.StartTime = &now
 	}
 
-	newStatus := m.updateStatusInternal(pod, status)
+	m.updateStatusInternal(pod, status)
-	if newStatus != nil {
+}
-		select {
+
-		case m.podStatusChannel <- podStatusSyncRequest{pod.UID, *newStatus}:
+func (m *manager) SetContainerReadiness(pod *api.Pod, containerID kubecontainer.ContainerID, ready bool) {
-		default:
+	m.podStatusesLock.Lock()
-			// Let the periodic syncBatch handle the update if the channel is full.
+	defer m.podStatusesLock.Unlock()
-			// We can't block, since we hold the mutex lock.
+
+	oldStatus, found := m.podStatuses[pod.UID]
+	if !found {
+		glog.Warningf("Container readiness changed before pod has synced: %q - %q",
+			kubeletutil.FormatPodName(pod), containerID.String())
+		return
+	}
+	status := oldStatus.status
+
+	// Find the container to update.
+	containerIndex := -1
+	for i, c := range status.ContainerStatuses {
+		if c.ContainerID == containerID.String() {
+			containerIndex = i
+			break
 		}
 	}
+	if containerIndex == -1 {
+		glog.Warningf("Container readiness changed for unknown container: %q - %q",
+			kubeletutil.FormatPodName(pod), containerID.String())
+		return
+	}
+
+	if status.ContainerStatuses[containerIndex].Ready == ready {
+		glog.V(4).Infof("Container readiness unchanged (%v): %q - %q", ready,
+			kubeletutil.FormatPodName(pod), containerID.String())
+		return
+	}
+
+	// Make sure we're not updating the cached version.
+	status.ContainerStatuses = make([]api.ContainerStatus, len(status.ContainerStatuses))
+	copy(status.ContainerStatuses, oldStatus.status.ContainerStatuses)
+	status.ContainerStatuses[containerIndex].Ready = ready
+	m.updateStatusInternal(pod, status)
 }
 
 func (m *manager) TerminatePods(pods []*api.Pod) bool {
-	sent := true
+	allSent := true
 	m.podStatusesLock.Lock()
 	defer m.podStatusesLock.Unlock()
 	for _, pod := range pods {
@@ -192,28 +228,26 @@ func (m *manager) TerminatePods(pods []*api.Pod) bool {
 				Terminated: &api.ContainerStateTerminated{},
 			}
 		}
-		newStatus := m.updateStatusInternal(pod, pod.Status)
+		if sent := m.updateStatusInternal(pod, pod.Status); !sent {
-		if newStatus != nil {
-			select {
-			case m.podStatusChannel <- podStatusSyncRequest{pod.UID, *newStatus}:
-			default:
-				sent = false
 			glog.V(4).Infof("Termination notice for %q was dropped because the status channel is full", kubeletutil.FormatPodName(pod))
-			}
+			allSent = false
-		} else {
-			sent = false
 		}
 	}
-	return sent
+	return allSent
 }
 
-// updateStatusInternal updates the internal status cache, and returns a versioned status if an
+// updateStatusInternal updates the internal status cache, and queues an update to the api server if
-// update is necessary. This method IS NOT THREAD SAFE and must be called from a locked function.
+// necessary. Returns whether an update was triggered.
-func (m *manager) updateStatusInternal(pod *api.Pod, status api.PodStatus) *versionedPodStatus {
+// This method IS NOT THREAD SAFE and must be called from a locked function.
+func (m *manager) updateStatusInternal(pod *api.Pod, status api.PodStatus) bool {
 	// The intent here is to prevent concurrent updates to a pod's status from
 	// clobbering each other so the phase of a pod progresses monotonically.
 	oldStatus, found := m.podStatuses[pod.UID]
-	if !found || !isStatusEqual(&oldStatus.status, &status) || pod.DeletionTimestamp != nil {
+	if found && isStatusEqual(&oldStatus.status, &status) && pod.DeletionTimestamp == nil {
+		glog.V(3).Infof("Ignoring same status for pod %q, status: %+v", kubeletutil.FormatPodName(pod), status)
+		return false // No new status.
+	}
+
 	newStatus := versionedPodStatus{
 		status:       status,
 		version:      oldStatus.version + 1,
@@ -221,10 +255,14 @@ func (m *manager) updateStatusInternal(pod *api.Pod, status api.PodStatus) *vers
 		podNamespace: pod.Namespace,
 	}
 	m.podStatuses[pod.UID] = newStatus
-		return &newStatus
+
-	} else {
+	select {
-		glog.V(3).Infof("Ignoring same status for pod %q, status: %+v", kubeletutil.FormatPodName(pod), status)
+	case m.podStatusChannel <- podStatusSyncRequest{pod.UID, newStatus}:
-		return nil // No new status.
+		return true
+	default:
+		// Let the periodic syncBatch handle the update if the channel is full.
+		// We can't block, since we hold the mutex lock.
+		return false
 	}
 }
 

pkg/kubelet/status/manager_test.go

@@ -30,6 +30,7 @@ import (
 	"k8s.io/kubernetes/pkg/api/unversioned"
 	client "k8s.io/kubernetes/pkg/client/unversioned"
 	"k8s.io/kubernetes/pkg/client/unversioned/testclient"
+	kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
 	kubepod "k8s.io/kubernetes/pkg/kubelet/pod"
 	kubetypes "k8s.io/kubernetes/pkg/kubelet/types"
 	"k8s.io/kubernetes/pkg/runtime"
@@ -499,3 +500,37 @@ func TestStaticPodStatus(t *testing.T) {
 	_, found := m.GetPodStatus(otherPod.UID)
 	assert.False(t, found, "otherPod status should have been deleted")
 }
+
+func TestSetContainerReadiness(t *testing.T) {
+	containerID := kubecontainer.ContainerID{"test", "cOnTaInEr_Id"}
+	containerStatus := api.ContainerStatus{
+		Name:        "cOnTaInEr_NaMe",
+		ContainerID: containerID.String(),
+		Ready:       false,
+	}
+	status := api.PodStatus{
+		ContainerStatuses: []api.ContainerStatus{containerStatus},
+	}
+
+	m := newTestManager(&testclient.Fake{})
+
+	t.Log("Setting readiness before status should fail.")
+	m.SetContainerReadiness(testPod, containerID, true)
+	verifyUpdates(t, m, 0)
+
+	t.Log("Setting initial status.")
+	m.SetPodStatus(testPod, status)
+	verifyUpdates(t, m, 1)
+
+	t.Log("Setting unchanged readiness should do nothing.")
+	m.SetContainerReadiness(testPod, containerID, false)
+	verifyUpdates(t, m, 0)
+
+	t.Log("Setting different readiness should generate update.")
+	m.SetContainerReadiness(testPod, containerID, true)
+	verifyUpdates(t, m, 1)
+
+	t.Log("Setting non-existant container readiness should fail.")
+	m.SetContainerReadiness(testPod, kubecontainer.ContainerID{"test", "foo"}, true)
+	verifyUpdates(t, m, 0)
+}