ref(probes): add controller check to the readiness probe

Author: Keerthan Mala

pkg/controller/utils.go

@@ -27,7 +27,7 @@ type UserInfo struct {
 	Apps        []string
 }
 
-func controllerURLStr(additionalPath ...string) (string, error) {
+func ControllerURLStr(additionalPath ...string) (string, error) {
 	host := os.Getenv(hostEnvName)
 	port := os.Getenv(portEnvName)
 
@@ -65,7 +65,7 @@ func fingerprint(key ssh.PublicKey) string {
 // UserInfoFromKey makes a request to the controller to get the user info from they given key.
 func UserInfoFromKey(key ssh.PublicKey) (*UserInfo, error) {
 	fp := fingerprint(key)
-	url, err := controllerURLStr("v2", "hooks", "key", fp)
+	url, err := ControllerURLStr("v2", "hooks", "key", fp)
 	if err != nil {
 		return nil, err
 	}

pkg/healthsrv/controller_state.go

@@ -0,0 +1,76 @@
+package healthsrv
+
+import (
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"strings"
+
+	"github.com/deis/builder/pkg/controller"
+)
+
+// GetClient is an (*net/http).Client compatible interface that provides just the Get cross-section of functionality.
+// It can also be implemented for unit tests.
+type GetClient interface {
+	Get(string) (*http.Response, error)
+}
+
+type successGetClient struct{}
+
+func (e successGetClient) Get(url string) (*http.Response, error) {
+	resp := &http.Response{
+		Body:       ioutil.NopCloser(strings.NewReader("")),
+		StatusCode: http.StatusOK,
+	}
+	return resp, nil
+}
+
+type failureGetClient struct{}
+
+func (e failureGetClient) Get(url string) (*http.Response, error) {
+	resp := &http.Response{
+		Body:       ioutil.NopCloser(strings.NewReader("")),
+		StatusCode: http.StatusServiceUnavailable,
+	}
+	return resp, nil
+}
+
+type errGetClient struct {
+	err error
+}
+
+func (e errGetClient) Get(url string) (*http.Response, error) {
+	return nil, e.err
+}
+
+func controllerState(client GetClient, succCh chan<- string, errCh chan<- error, stopCh <-chan struct{}) {
+	url, err := controller.ControllerURLStr("healthz")
+	if err != nil {
+		select {
+		case errCh <- err:
+		case <-stopCh:
+		}
+		return
+	}
+	resp, err := client.Get(url)
+	if err != nil {
+		select {
+		case errCh <- err:
+		case <-stopCh:
+		}
+		return
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode < 200 || resp.StatusCode > 299 {
+		select {
+		case errCh <- fmt.Errorf("Failed to get controller health status"):
+		case <-stopCh:
+		}
+		return
+	}
+	select {
+	case succCh <- "controller is healthy":
+	case <-stopCh:
+		return
+	}
+}

pkg/healthsrv/healthz_handler_test.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"net/http"
 	"net/http/httptest"
+	"os"
 	"testing"
 
 	"github.com/arschles/assert"
@@ -45,8 +46,37 @@ func TestHealthZBucketListErr(t *testing.T) {
 
 func TestReadinessNamespaceListErr(t *testing.T) {
 	nsLister := errNamespaceLister{err: errTest}
+	client := successGetClient{}
+	os.Setenv("DEIS_CONTROLLER_SERVICE_HOST", "127.0.0.1")
+	os.Setenv("DEIS_CONTROLLER_SERVICE_PORT", "8000")
 
-	h := readinessHandler(nsLister)
+	h := readinessHandler(client, nsLister)
+	w := httptest.NewRecorder()
+	r, err := http.NewRequest("GET", "/readiness", bytes.NewBuffer(nil))
+	assert.NoErr(t, err)
+	h.ServeHTTP(w, r)
+	assert.Equal(t, w.Code, http.StatusServiceUnavailable, "response code")
+	assert.Equal(t, w.Body.Len(), 0, "response body length")
+}
+
+func TestReadinessControllerErr(t *testing.T) {
+	nsLister := emptyNamespaceLister{}
+	client := failureGetClient{}
+
+	h := readinessHandler(client, nsLister)
+	w := httptest.NewRecorder()
+	r, err := http.NewRequest("GET", "/readiness", bytes.NewBuffer(nil))
+	assert.NoErr(t, err)
+	h.ServeHTTP(w, r)
+	assert.Equal(t, w.Code, http.StatusServiceUnavailable, "response code")
+	assert.Equal(t, w.Body.Len(), 0, "response body length")
+}
+
+func TestReadinessControllerGetErr(t *testing.T) {
+	nsLister := emptyNamespaceLister{}
+	client := errGetClient{err: errTest}
+
+	h := readinessHandler(client, nsLister)
 	w := httptest.NewRecorder()
 	r, err := http.NewRequest("GET", "/readiness", bytes.NewBuffer(nil))
 	assert.NoErr(t, err)
@@ -71,8 +101,9 @@ func TestHealthZSuccess(t *testing.T) {
 
 func TestReadinessSuccess(t *testing.T) {
 	nsLister := emptyNamespaceLister{}
+	client := successGetClient{}
 
-	h := readinessHandler(nsLister)
+	h := readinessHandler(client, nsLister)
 	w := httptest.NewRecorder()
 	r, err := http.NewRequest("GET", "/readiness", bytes.NewBuffer(nil))
 	assert.NoErr(t, err)

pkg/healthsrv/readiness_handler.go

@@ -8,26 +8,46 @@ import (
 	"k8s.io/kubernetes/pkg/api"
 )
 
-func readinessHandler(nsLister NamespaceLister) http.Handler {
+func readinessHandler(client GetClient, nsLister NamespaceLister) http.Handler {
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		stopCh := make(chan struct{})
 
+		numChecks := 0
 		namespaceListerCh := make(chan *api.NamespaceList)
 		namespaceListerErrCh := make(chan error)
 		go listNamespaces(nsLister, namespaceListerCh, namespaceListerErrCh, stopCh)
+		numChecks++
+
+		controllerStateCh := make(chan string)
+		controllerStateErrCh := make(chan error)
+		go controllerState(client, controllerStateCh, controllerStateErrCh, stopCh)
+		numChecks++
 
 		timeoutCh := time.After(waitTimeout)
 		defer close(stopCh)
-		select {
-		// listing k8s namespaces
-		case <-namespaceListerCh:
-		case err := <-namespaceListerErrCh:
-			log.Printf("Healthcheck error listing namespaces (%s)", err)
-			w.WriteHeader(http.StatusServiceUnavailable)
-		// timeout for everything all of the above
-		case <-timeoutCh:
-			log.Printf("Healthcheck endpoint timed out after %s", waitTimeout)
-			w.WriteHeader(http.StatusServiceUnavailable)
+		for i := 0; i < numChecks; i++ {
+			select {
+			// listing k8s namespaces
+			case <-namespaceListerCh:
+			case err := <-namespaceListerErrCh:
+				log.Printf("Readinesscheck error listing namespaces (%s)", err)
+				w.WriteHeader(http.StatusServiceUnavailable)
+				return
+
+			// listing k8s namespaces
+			case <-controllerStateCh:
+			case err := <-controllerStateErrCh:
+				log.Printf("Readinesscheck error listning controller (%s)", err)
+				w.WriteHeader(http.StatusServiceUnavailable)
+				return
+
+			// timeout for everything all of the above
+			case <-timeoutCh:
+				log.Printf("Readinesscheck endpoint timed out after %s", waitTimeout)
+				w.WriteHeader(http.StatusServiceUnavailable)
+				return
+
+			}
 		}
 		w.WriteHeader(http.StatusOK)
 	})

pkg/healthsrv/server.go

@@ -11,8 +11,9 @@ import (
 // with the indicative error.
 func Start(port int, nsLister NamespaceLister, bLister BucketLister, sshServerCircuit *sshd.Circuit) error {
 	mux := http.NewServeMux()
+	client := &http.Client{}
 	mux.Handle("/healthz", healthZHandler(bLister, sshServerCircuit))
-	mux.Handle("/readiness", readinessHandler(nsLister))
+	mux.Handle("/readiness", readinessHandler(client, nsLister))
 
 	hostStr := fmt.Sprintf(":%d", port)
 	return http.ListenAndServe(hostStr, mux)