Don't wait for sync to update readiness

pkg/kubelet/kubelet.go

@@ -322,8 +322,7 @@ func NewMainKubelet(
 	procFs := procfs.NewProcFs()
 	imageBackOff := util.NewBackOff(resyncInterval, MaxContainerBackOff)
 
-	readinessManager := proberesults.NewManager()
-	klet.livenessManager = proberesults.NewManagerWithUpdates()
+	klet.livenessManager = proberesults.NewManager()
 
 	// 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{podUID, probedReady.Name, readiness}:   {},
-		probeKey{podUID, probedPending.Name, readiness}: {},
-		probeKey{podUID, probedUnready.Name, readiness}: {},
-		probeKey{podUID, terminated.Name, readiness}:    {},
+		probeKey{testPodUID, unprobed.Name, liveness}:       {},
+		probeKey{testPodUID, probedReady.Name, readiness}:   {},
+		probeKey{testPodUID, probedPending.Name, readiness}: {},
+		probeKey{testPodUID, probedUnready.Name, readiness}: {},
+		probeKey{testPodUID, terminated.Name, readiness}:    {},
 	}
-	m.readinessManager.Set(kubecontainer.ParseContainerID(probedReady.ContainerID), results.Success, nil)
-	m.readinessManager.Set(kubecontainer.ParseContainerID(probedUnready.ContainerID), results.Failure, nil)
-	m.readinessManager.Set(kubecontainer.ParseContainerID(terminated.ContainerID), results.Success, nil)
+	m.readinessManager.Set(kubecontainer.ParseContainerID(probedReady.ContainerID), results.Success, &api.Pod{})
+	m.readinessManager.Set(kubecontainer.ParseContainerID(probedUnready.ContainerID), results.Failure, &api.Pod{})
+	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{podUID, probedReady.Name, readiness}:   true,
-		probeKey{podUID, probedPending.Name, readiness}: false,
-		probeKey{podUID, probedUnready.Name, readiness}: false,
-		probeKey{podUID, terminated.Name, readiness}:    false,
+		probeKey{testPodUID, unprobed.Name, readiness}:      true,
+		probeKey{testPodUID, probedReady.Name, readiness}:   true,
+		probeKey{testPodUID, probedPending.Name, readiness}: false,
+		probeKey{testPodUID, probedUnready.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 {
-	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
-}
+const interval = 100 * time.Millisecond
 
 // 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,23 +70,18 @@ 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
 }
 
 var _ Manager = &manager{}
 
 // NewManager creates ane returns an empty results manager.
 func NewManager() Manager {
-	m := &manager{cache: make(map[kubecontainer.ContainerID]Result)}
-	return m
-}
-
-// NewManager creates ane returns an empty results manager.
-func NewManagerWithUpdates() Manager {
-	m := NewManager().(*manager)
-	m.updates = make(chan Update, 20)
-	return m
+	return &manager{
+		cache:   make(map[kubecontainer.ContainerID]Result),
+		updates: make(chan Update, 20),
+	}
 }
 
 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"}