Merge pull request #329 from kmala/probe

ref(probes): move k8s probe from liveness into readiness

Author: kmala

pkg/healthsrv/healthz_handler.go

@@ -6,14 +6,13 @@ import (
 	"time"
 
 	"github.com/deis/builder/pkg/sshd"
-	"k8s.io/kubernetes/pkg/api"
 )
 
 const (
 	waitTimeout = 2 * time.Second
 )
 
-func healthZHandler(nsLister NamespaceLister, bLister BucketLister, serverCircuit *sshd.Circuit) http.Handler {
+func healthZHandler(bLister BucketLister, serverCircuit *sshd.Circuit) http.Handler {
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		stopCh := make(chan struct{})
 
@@ -28,11 +27,6 @@ func healthZHandler(nsLister NamespaceLister, bLister BucketLister, serverCircui
 		go listBuckets(bLister, listBucketsCh, listBucketsErrCh, stopCh)
 		numChecks++
 
-		namespaceListerCh := make(chan *api.NamespaceList)
-		namespaceListerErrCh := make(chan error)
-		go listNamespaces(nsLister, namespaceListerCh, namespaceListerErrCh, stopCh)
-		numChecks++
-
 		timeoutCh := time.After(waitTimeout)
 		defer close(stopCh)
 		for i := 0; i < numChecks; i++ {
@@ -43,18 +37,14 @@ func healthZHandler(nsLister NamespaceLister, bLister BucketLister, serverCircui
 				log.Printf("Healthcheck error getting server state (%s)", err)
 				w.WriteHeader(http.StatusServiceUnavailable)
 				return
-			// listing S3 buckets
+
+			// listing buckets
 			case <-listBucketsCh:
 			case err := <-listBucketsErrCh:
 				log.Printf("Healthcheck error listing buckets (%s)", err)
 				w.WriteHeader(http.StatusServiceUnavailable)
 				return
-			// listing k8s namespaces
-			case <-namespaceListerCh:
-			case err := <-namespaceListerErrCh:
-				log.Printf("Healthcheck error listing namespaces (%s)", err)
-				w.WriteHeader(http.StatusServiceUnavailable)
-				return
+
 			// timeout for everything all of the above
 			case <-timeoutCh:
 				log.Printf("Healthcheck endpoint timed out after %s", waitTimeout)

pkg/healthsrv/healthz_handler_test.go

@@ -16,11 +16,10 @@ var (
 )
 
 func TestHealthZCircuitOpen(t *testing.T) {
-	nsLister := emptyNamespaceLister{}
 	bLister := emptyBucketLister{}
 	c := sshd.NewCircuit()
 
-	h := healthZHandler(nsLister, bLister, c)
+	h := healthZHandler(bLister, c)
 	w := httptest.NewRecorder()
 	r, err := http.NewRequest("GET", "/healthz", bytes.NewBuffer(nil))
 	assert.NoErr(t, err)
@@ -30,11 +29,11 @@ func TestHealthZCircuitOpen(t *testing.T) {
 }
 
 func TestHealthZBucketListErr(t *testing.T) {
-	nsLister := emptyNamespaceLister{}
 	bLister := errBucketLister{err: errTest}
 	c := sshd.NewCircuit()
 	c.Close()
-	h := healthZHandler(nsLister, bLister, c)
+
+	h := healthZHandler(bLister, c)
 
 	w := httptest.NewRecorder()
 	r, err := http.NewRequest("GET", "/healthz", bytes.NewBuffer(nil))
@@ -44,32 +43,40 @@ func TestHealthZBucketListErr(t *testing.T) {
 	assert.Equal(t, w.Body.Len(), 0, "response body length")
 }
 
-func TestHealthZNamespaceListErr(t *testing.T) {
+func TestReadinessNamespaceListErr(t *testing.T) {
 	nsLister := errNamespaceLister{err: errTest}
-	bLister := emptyBucketLister{}
-	c := sshd.NewCircuit()
-	c.Close()
 
-	h := healthZHandler(nsLister, bLister, c)
+	h := readinessHandler(nsLister)
 	w := httptest.NewRecorder()
-	r, err := http.NewRequest("GET", "/healthz", bytes.NewBuffer(nil))
+	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 TestHealthZSuccess(t *testing.T) {
-	nsLister := emptyNamespaceLister{}
 	bLister := emptyBucketLister{}
 	c := sshd.NewCircuit()
 	c.Close()
 
-	h := healthZHandler(nsLister, bLister, c)
+	h := healthZHandler(bLister, c)
 	w := httptest.NewRecorder()
 	r, err := http.NewRequest("GET", "/healthz", bytes.NewBuffer(nil))
 	assert.NoErr(t, err)
 	h.ServeHTTP(w, r)
 	assert.Equal(t, w.Code, http.StatusOK, "response code")
 	assert.Equal(t, w.Body.Len(), 0, "response body length")
 }
+
+func TestReadinessSuccess(t *testing.T) {
+	nsLister := emptyNamespaceLister{}
+
+	h := readinessHandler(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.StatusOK, "response code")
+	assert.Equal(t, w.Body.Len(), 0, "response body length")
+}

pkg/healthsrv/readiness_handler.go

@@ -0,0 +1,34 @@
+package healthsrv
+
+import (
+	"log"
+	"net/http"
+	"time"
+
+	"k8s.io/kubernetes/pkg/api"
+)
+
+func readinessHandler(nsLister NamespaceLister) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		stopCh := make(chan struct{})
+
+		namespaceListerCh := make(chan *api.NamespaceList)
+		namespaceListerErrCh := make(chan error)
+		go listNamespaces(nsLister, namespaceListerCh, namespaceListerErrCh, stopCh)
+
+		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)
+		}
+		w.WriteHeader(http.StatusOK)
+	})
+}

pkg/healthsrv/server.go

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