Merge pull request #1302 from brendandburns/validatez

Add a validator for validating components in the cluster infrastructure.

pkg/apiserver/apiserver.go

@@ -91,6 +91,16 @@ func (g *APIGroup) InstallREST(mux mux, paths ...string) {
 	redirectHandler := &RedirectHandler{g.handler.storage, g.handler.codec}
 	opHandler := &OperationHandler{g.handler.ops, g.handler.codec}
 
+	servers := map[string]string{
+		"controller-manager": "127.0.0.1:10252",
+		"scheduler":          "127.0.0.1:10251",
+		// TODO: Add minion health checks here too.
+	}
+	validator, err := NewValidator(servers)
+	if err != nil {
+		glog.Errorf("failed to set up validator: %v", err)
+		validator = nil
+	}
 	for _, prefix := range paths {
 		prefix = strings.TrimRight(prefix, "/")
 		proxyHandler := &ProxyHandler{prefix + "/proxy/", g.handler.storage, g.handler.codec}
@@ -100,6 +110,9 @@ func (g *APIGroup) InstallREST(mux mux, paths ...string) {
 		mux.Handle(prefix+"/redirect/", http.StripPrefix(prefix+"/redirect/", redirectHandler))
 		mux.Handle(prefix+"/operations", http.StripPrefix(prefix+"/operations", opHandler))
 		mux.Handle(prefix+"/operations/", http.StripPrefix(prefix+"/operations/", opHandler))
+		if validator != nil {
+			mux.Handle(prefix+"/validate", validator)
+		}
 	}
 }
 

pkg/apiserver/validator.go

@@ -0,0 +1,122 @@
+/*
+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 apiserver
+
+import (
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+	"log"
+	"net"
+	"net/http"
+	"strconv"
+
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/health"
+)
+
+// TODO: this basic interface is duplicated in N places.  consolidate?
+type httpGet interface {
+	Get(url string) (*http.Response, error)
+}
+
+type server struct {
+	addr string
+	port int
+}
+
+// validator is responsible for validating the cluster and serving
+type validator struct {
+	// a list of servers to health check
+	servers map[string]server
+	client  httpGet
+}
+
+func (s *server) check(client httpGet) (health.Status, string, error) {
+	resp, err := client.Get("http://" + net.JoinHostPort(s.addr, strconv.Itoa(s.port)) + "/healthz")
+	if err != nil {
+		return health.Unknown, "", err
+	}
+	defer resp.Body.Close()
+	data, err := ioutil.ReadAll(resp.Body)
+	if err != nil {
+		return health.Unknown, string(data), err
+	}
+	if resp.StatusCode != http.StatusOK {
+		return health.Unhealthy, string(data),
+			fmt.Errorf("unhealthy http status code: %d (%s)", resp.StatusCode, resp.Status)
+	}
+	return health.Healthy, string(data), nil
+}
+
+type ServerStatus struct {
+	Component  string        `json:"component,omitempty"`
+	Health     string        `json:"health,omitempty"`
+	HealthCode health.Status `json:"healthCode,omitempty"`
+	Msg        string        `json:"msg,omitempty"`
+	Err        string        `json:"err,omitempty"`
+}
+
+func (v *validator) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	reply := []ServerStatus{}
+	for name, server := range v.servers {
+		status, msg, err := server.check(v.client)
+		var errorMsg string
+		if err != nil {
+			errorMsg = err.Error()
+		} else {
+			errorMsg = "nil"
+		}
+		reply = append(reply, ServerStatus{name, status.String(), status, msg, errorMsg})
+	}
+	data, err := json.Marshal(reply)
+	log.Printf("FOO: %s", string(data))
+	if err != nil {
+		w.WriteHeader(http.StatusInternalServerError)
+		w.Write([]byte(err.Error()))
+		return
+	}
+	w.WriteHeader(http.StatusOK)
+	w.Write(data)
+}
+
+// NewValidator creates a validator for a set of servers.
+func NewValidator(servers map[string]string) (http.Handler, error) {
+	result := map[string]server{}
+	for name, value := range servers {
+		host, port, err := net.SplitHostPort(value)
+		if err != nil {
+			return nil, fmt.Errorf("invalid server spec: %s (%v)", value, err)
+		}
+		val, err := strconv.Atoi(port)
+		if err != nil {
+			return nil, fmt.Errorf("invalid server spec: %s (%v)", port, err)
+		}
+		result[name] = server{host, val}
+	}
+	return &validator{
+		servers: result,
+		client:  &http.Client{},
+	}, nil
+}
+
+func makeTestValidator(servers map[string]string, get httpGet) (http.Handler, error) {
+	v, e := NewValidator(servers)
+	if e == nil {
+		v.(*validator).client = get
+	}
+	return v, e
+}

pkg/apiserver/validator_test.go

@@ -0,0 +1,128 @@
+/*
+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 apiserver
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/health"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
+)
+
+type fakeHttpGet struct {
+	err  error
+	resp *http.Response
+	url  string
+}
+
+func (f *fakeHttpGet) Get(url string) (*http.Response, error) {
+	f.url = url
+	return f.resp, f.err
+}
+
+func makeFake(data string, statusCode int, err error) *fakeHttpGet {
+	return &fakeHttpGet{
+		err: err,
+		resp: &http.Response{
+			Body:       ioutil.NopCloser(bytes.NewBufferString(data)),
+			StatusCode: statusCode,
+		},
+	}
+}
+
+func TestValidate(t *testing.T) {
+	tests := []struct {
+		err            error
+		data           string
+		expectedStatus health.Status
+		code           int
+		expectErr      bool
+	}{
+		{fmt.Errorf("test error"), "", health.Unknown, 500 /*ignored*/, true},
+		{nil, "foo", health.Healthy, 200, false},
+		{nil, "foo", health.Unhealthy, 500, true},
+	}
+
+	s := server{addr: "foo.com", port: 8080}
+
+	for _, test := range tests {
+		fake := makeFake(test.data, test.code, test.err)
+		status, data, err := s.check(fake)
+		expect := fmt.Sprintf("http://%s:%d/healthz", s.addr, s.port)
+		if fake.url != expect {
+			t.Errorf("expected %s, got %s", expect, fake.url)
+		}
+		if test.expectErr && err == nil {
+			t.Errorf("unexpected non-error")
+		}
+		if !test.expectErr && err != nil {
+			t.Errorf("unexpected error: %v", err)
+		}
+		if data != test.data {
+			t.Errorf("expected empty string, got %s", status)
+		}
+		if status != test.expectedStatus {
+			t.Errorf("expected %s, got %s", test.expectedStatus.String(), status.String())
+		}
+	}
+}
+
+func TestValidator(t *testing.T) {
+	fake := makeFake("foo", 200, nil)
+	validator, err := makeTestValidator(map[string]string{
+		"foo": "foo.com:80",
+		"bar": "bar.com:8080",
+	}, fake)
+	if err != nil {
+		t.Errorf("unexpected error: %v", err)
+	}
+
+	testServer := httptest.NewServer(validator)
+	defer testServer.Close()
+
+	resp, err := http.Get(testServer.URL + "/validatez")
+
+	if err != nil {
+		t.Errorf("unexpected error: %v", err)
+	}
+	if resp.StatusCode != http.StatusOK {
+		t.Errorf("unexpected response: %v", resp.StatusCode)
+	}
+	defer resp.Body.Close()
+	data, err := ioutil.ReadAll(resp.Body)
+	if err != nil {
+		t.Errorf("unexpected error: %v", err)
+	}
+	status := []ServerStatus{}
+	err = json.Unmarshal(data, &status)
+	if err != nil {
+		t.Errorf("unexpected error: %v", err)
+	}
+	components := util.StringSet{}
+	for _, s := range status {
+		components.Insert(s.Component)
+	}
+	if len(status) != 2 || !components.Has("foo") || !components.Has("bar") {
+		t.Errorf("unexpected status: %#v", status)
+	}
+}

pkg/health/health.go

@@ -94,3 +94,12 @@ func findPortByName(container api.Container, portName string) int {
 	}
 	return -1
 }
+
+func (s Status) String() string {
+	if s == Healthy {
+		return "healthy"
+	} else if s == Unhealthy {
+		return "unhealthy"
+	}
+	return "unknown"
+}