Move health checking logic out to a utility.

Add a minion registry that health checks.

pkg/kubelet/health_check.go

@@ -22,23 +22,12 @@ import (
 	"strconv"
 
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
 	"github.com/golang/glog"
 )
 
-type HealthCheckStatus int
-
-const (
-	CheckHealthy   HealthCheckStatus = 0
-	CheckUnhealthy HealthCheckStatus = 1
-	CheckUnknown   HealthCheckStatus = 2
-)
-
 type HealthChecker interface {
-	HealthCheck(container api.Container) (HealthCheckStatus, error)
-}
-
-type httpDoInterface interface {
-	Get(string) (*http.Response, error)
+	HealthCheck(container api.Container) (util.HealthCheckStatus, error)
 }
 
 // MakeHealthChecker creates a new HealthChecker.
@@ -57,18 +46,18 @@ type MuxHealthChecker struct {
 	checkers map[string]HealthChecker
 }
 
-func (m *MuxHealthChecker) HealthCheck(container api.Container) (HealthCheckStatus, error) {
+func (m *MuxHealthChecker) HealthCheck(container api.Container) (util.HealthCheckStatus, error) {
 	checker, ok := m.checkers[container.LivenessProbe.Type]
 	if !ok || checker == nil {
 		glog.Warningf("Failed to find health checker for %s %s", container.Name, container.LivenessProbe.Type)
-		return CheckUnknown, nil
+		return util.CheckUnknown, nil
 	}
 	return checker.HealthCheck(container)
 }
 
 // HTTPHealthChecker is an implementation of HealthChecker which checks container health by sending HTTP Get requests.
 type HTTPHealthChecker struct {
-	client httpDoInterface
+	client util.HTTPGetInterface
 }
 
 func (h *HTTPHealthChecker) findPort(container api.Container, portName string) int64 {
@@ -81,17 +70,17 @@ func (h *HTTPHealthChecker) findPort(container api.Container, portName string) i
 	return -1
 }
 
-func (h *HTTPHealthChecker) HealthCheck(container api.Container) (HealthCheckStatus, error) {
+func (h *HTTPHealthChecker) HealthCheck(container api.Container) (util.HealthCheckStatus, error) {
 	params := container.LivenessProbe.HTTPGet
 	if params == nil {
-		return CheckUnknown, fmt.Errorf("Error, no HTTP parameters specified: %v", container)
+		return util.CheckUnknown, fmt.Errorf("Error, no HTTP parameters specified: %v", container)
 	}
 	port := h.findPort(container, params.Port)
 	if port == -1 {
 		var err error
 		port, err = strconv.ParseInt(params.Port, 10, 0)
 		if err != nil {
-			return CheckUnknown, err
+			return util.CheckUnknown, err
 		}
 	}
 	var host string
@@ -101,17 +90,5 @@ func (h *HTTPHealthChecker) HealthCheck(container api.Container) (HealthCheckSta
 		host = "localhost"
 	}
 	url := fmt.Sprintf("http://%s:%d%s", host, port, params.Path)
-	res, err := h.client.Get(url)
-	if res != nil && res.Body != nil {
-		defer res.Body.Close()
-	}
-	if err != nil {
-		// At this point, if it fails, its either a policy (unlikely) or HTTP protocol (likely) error.
-		return CheckUnhealthy, nil
-	}
-	if res.StatusCode == http.StatusOK {
-		return CheckHealthy, nil
-	}
-	glog.V(1).Infof("Health check failed for %v, Response: %v", container, *res)
-	return CheckUnhealthy, nil
+	return util.IsHealthy(url, h.client)
 }

pkg/kubelet/health_check_test.go

@@ -21,6 +21,7 @@ import (
 	"testing"
 
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
 )
 
 type fakeHTTPClient struct {
@@ -56,7 +57,7 @@ func TestHttpHealth(t *testing.T) {
 	}
 
 	ok, err := check.HealthCheck(container)
-	if ok != CheckHealthy {
+	if ok != util.CheckHealthy {
 		t.Error("Unexpected unhealthy")
 	}
 	if err != nil {

pkg/kubelet/kubelet.go

@@ -736,7 +736,7 @@ func (kl *Kubelet) syncManifest(manifest *api.ContainerManifest, keepChannel cha
 				glog.V(1).Infof("health check errored: %v", err)
 				continue
 			}
-			if healthy != CheckHealthy {
+			if healthy != util.CheckHealthy {
 				glog.V(1).Infof("manifest %s container %s is unhealthy.", manifest.ID, container.Name)
 				if err != nil {
 					glog.V(1).Infof("Failed to get container info %v, for %s", err, containerID)
@@ -993,16 +993,16 @@ func (kl *Kubelet) GetMachineStats() (*api.ContainerStats, error) {
 	return kl.statsFromContainerPath("/")
 }
 
-func (kl *Kubelet) healthy(container api.Container, dockerContainer *docker.APIContainers) (HealthCheckStatus, error) {
+func (kl *Kubelet) healthy(container api.Container, dockerContainer *docker.APIContainers) (util.HealthCheckStatus, error) {
 	// Give the container 60 seconds to start up.
 	if container.LivenessProbe == nil {
-		return CheckHealthy, nil
+		return util.CheckHealthy, nil
 	}
 	if time.Now().Unix()-dockerContainer.Created < container.LivenessProbe.InitialDelaySeconds {
-		return CheckHealthy, nil
+		return util.CheckHealthy, nil
 	}
 	if kl.HealthChecker == nil {
-		return CheckHealthy, nil
+		return util.CheckHealthy, nil
 	}
 	return kl.HealthChecker.HealthCheck(container)
 }

pkg/kubelet/kubelet_test.go

@@ -424,8 +424,8 @@ func TestSyncManifestsDeletes(t *testing.T) {
 
 type FalseHealthChecker struct{}
 
-func (f *FalseHealthChecker) HealthCheck(container api.Container) (HealthCheckStatus, error) {
-	return CheckUnhealthy, nil
+func (f *FalseHealthChecker) HealthCheck(container api.Container) (util.HealthCheckStatus, error) {
+	return util.CheckUnhealthy, nil
 }
 
 func TestSyncManifestsUnhealthy(t *testing.T) {

pkg/registry/healthy_minion_registry.go

@@ -0,0 +1,86 @@
+/*
+Copyright 2014 Google Inc. 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 registry
+
+import (
+	"fmt"
+	"net/http"
+
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
+)
+
+type HealthyMinionRegistry struct {
+	delegate MinionRegistry
+	client   util.HTTPGetInterface
+	port     int
+}
+
+func NewHealthyMinionRegistry(delegate MinionRegistry, client *http.Client) MinionRegistry {
+	return &HealthyMinionRegistry{
+		delegate: delegate,
+		client:   client,
+		port:     10250,
+	}
+}
+
+func (h *HealthyMinionRegistry) makeMinionURL(minion string) string {
+	return fmt.Sprintf("http://%s:%d/healthz", minion, h.port)
+}
+
+func (h *HealthyMinionRegistry) List() (currentMinions []string, err error) {
+	var result []string
+	list, err := h.delegate.List()
+	if err != nil {
+		return result, err
+	}
+	for _, minion := range list {
+		status, err := util.IsHealthy(h.makeMinionURL(minion), h.client)
+		if err != nil {
+			return result, err
+		}
+		if status == util.CheckHealthy {
+			result = append(result, minion)
+		}
+	}
+	return result, nil
+}
+
+func (h *HealthyMinionRegistry) Insert(minion string) error {
+	return h.delegate.Insert(minion)
+}
+
+func (h *HealthyMinionRegistry) Delete(minion string) error {
+	return h.delegate.Delete(minion)
+}
+
+func (h *HealthyMinionRegistry) Contains(minion string) (bool, error) {
+	contains, err := h.delegate.Contains(minion)
+	if err != nil {
+		return false, err
+	}
+	if !contains {
+		return false, nil
+	}
+	status, err := util.IsHealthy(h.makeMinionURL(minion), h.client)
+	if err != nil {
+		return false, err
+	}
+	if status == util.CheckUnhealthy {
+		return false, nil
+	}
+	return true, nil
+}

pkg/registry/healthy_minion_registry_test.go

@@ -0,0 +1,96 @@
+/*
+Copyright 2014 Google Inc. 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 registry
+
+import (
+	"net/http"
+	"reflect"
+	"testing"
+)
+
+type alwaysYes struct{}
+
+func (alwaysYes) Get(url string) (*http.Response, error) {
+	return &http.Response{
+		StatusCode: http.StatusOK,
+	}, nil
+}
+
+func TestBasicDelegation(t *testing.T) {
+	mockMinionRegistry := MockMinionRegistry{
+		minions: []string{"m1", "m2", "m3"},
+	}
+	healthy := HealthyMinionRegistry{
+		delegate: &mockMinionRegistry,
+		client:   alwaysYes{},
+	}
+
+	list, err := healthy.List()
+	expectNoError(t, err)
+	if !reflect.DeepEqual(list, mockMinionRegistry.minions) {
+		t.Errorf("Expected %v, Got %v", mockMinionRegistry.minions, list)
+	}
+	err = healthy.Insert("foo")
+	expectNoError(t, err)
+
+	ok, err := healthy.Contains("m1")
+	expectNoError(t, err)
+	if !ok {
+		t.Errorf("Unexpected absence of 'm1'")
+	}
+
+	ok, err = healthy.Contains("m5")
+	expectNoError(t, err)
+	if ok {
+		t.Errorf("Unexpected presence of 'm5'")
+	}
+}
+
+type notMinion struct {
+	minion string
+}
+
+func (n *notMinion) Get(url string) (*http.Response, error) {
+	if url != "http://"+n.minion+":10250/healthz" {
+		return &http.Response{StatusCode: http.StatusOK}, nil
+	} else {
+		return &http.Response{StatusCode: http.StatusInternalServerError}, nil
+	}
+}
+
+func TestFiltering(t *testing.T) {
+	mockMinionRegistry := MockMinionRegistry{
+		minions: []string{"m1", "m2", "m3"},
+	}
+	healthy := HealthyMinionRegistry{
+		delegate: &mockMinionRegistry,
+		client:   &notMinion{minion: "m1"},
+		port:     10250,
+	}
+
+	expected := []string{"m2", "m3"}
+	list, err := healthy.List()
+	expectNoError(t, err)
+	if !reflect.DeepEqual(list, expected) {
+		t.Errorf("Expected %v, Got %v", expected, list)
+	}
+	ok, err := healthy.Contains("m1")
+	expectNoError(t, err)
+	if ok {
+		t.Errorf("Unexpected presence of 'm1'")
+	}
+}

pkg/registry/mock_registry.go

@@ -75,3 +75,53 @@ func (registry *MockPodRegistry) DeletePod(podId string) error {
 	defer registry.Unlock()
 	return registry.err
 }
+
+type MockMinionRegistry struct {
+	err     error
+	minion  string
+	minions []string
+	sync.Mutex
+}
+
+func MakeMockMinionRegistry(minions []string) *MockMinionRegistry {
+	return &MockMinionRegistry{
+		minions: minions,
+	}
+}
+
+func (registry *MockMinionRegistry) List() ([]string, error) {
+	registry.Lock()
+	defer registry.Unlock()
+	return registry.minions, registry.err
+}
+
+func (registry *MockMinionRegistry) Insert(minion string) error {
+	registry.Lock()
+	defer registry.Unlock()
+	registry.minion = minion
+	return registry.err
+}
+
+func (registry *MockMinionRegistry) Contains(minion string) (bool, error) {
+	registry.Lock()
+	defer registry.Unlock()
+	for _, name := range registry.minions {
+		if name == minion {
+			return true, registry.err
+		}
+	}
+	return false, registry.err
+}
+
+func (registry *MockMinionRegistry) Delete(minion string) error {
+	registry.Lock()
+	defer registry.Unlock()
+	var newList []string
+	for _, name := range registry.minions {
+		if name != minion {
+			newList = append(newList, name)
+		}
+	}
+	registry.minions = newList
+	return registry.err
+}

pkg/util/health.go

@@ -0,0 +1,51 @@
+/*
+Copyright 2014 Google Inc. 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 util
+
+import (
+	"net/http"
+
+	"github.com/golang/glog"
+)
+
+type HealthCheckStatus int
+
+const (
+	CheckHealthy   HealthCheckStatus = 0
+	CheckUnhealthy HealthCheckStatus = 1
+	CheckUnknown   HealthCheckStatus = 2
+)
+
+type HTTPGetInterface interface {
+	Get(url string) (*http.Response, error)
+}
+
+func IsHealthy(url string, client HTTPGetInterface) (HealthCheckStatus, error) {
+	res, err := client.Get(url)
+	if res.Body != nil {
+		defer res.Body.Close()
+	}
+	if err != nil {
+		return CheckUnknown, err
+	}
+	if res.StatusCode >= http.StatusOK && res.StatusCode < http.StatusBadRequest {
+		return CheckHealthy, nil
+	} else {
+		glog.V(1).Infof("Health check failed for %s, Response: %v", url, *res)
+		return CheckUnhealthy, nil
+	}
+}