Add health check functions.
diff --git a/handlers.go b/handlers.go
new file mode 100644
index 0000000..c0d11a9
--- /dev/null
+++ b/handlers.go
@@ -0,0 +1,128 @@
+package goscaffold
+
+import (
+ "encoding/json"
+ "errors"
+ "net/http"
+)
+
+/*
+requestHandler handles all requests and stops them if we are marked down.
+*/
+type requestHandler struct {
+ s *HTTPScaffold
+ child http.Handler
+}
+
+func (h *requestHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
+ startErr := h.s.tracker.start()
+ if startErr == nil {
+ h.child.ServeHTTP(resp, req)
+ h.s.tracker.end()
+ } else {
+ writeUnavailable(resp, req, NotReady, startErr)
+ }
+}
+
+/*
+managementHandler adds support for health checks and diagnostics.
+*/
+type managementHandler struct {
+ s *HTTPScaffold
+ mux *http.ServeMux
+ child http.Handler
+}
+
+func (h *managementHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
+ handler, pattern := h.mux.Handler(req)
+ if pattern == "" && h.child != nil {
+ // Fall through for stuff that's not a management call
+ h.child.ServeHTTP(resp, req)
+ } else {
+ // Handler may be one of ours, or a built-in not found handler
+ handler.ServeHTTP(resp, req)
+ }
+}
+
+func (s *HTTPScaffold) createManagementHandler() *managementHandler {
+ h := &managementHandler{
+ s: s,
+ mux: http.NewServeMux(),
+ }
+ if s.healthPath != "" {
+ h.mux.HandleFunc(s.healthPath, s.handleHealth)
+ }
+ if s.readyPath != "" {
+ h.mux.HandleFunc(s.readyPath, s.handleReady)
+ }
+ return h
+}
+
+func (s *HTTPScaffold) callHealthCheck() (HealthStatus, error) {
+ if s.healthCheck == nil {
+ return OK, nil
+ }
+ status, err := s.healthCheck()
+ if status == OK {
+ return OK, nil
+ }
+ if err == nil {
+ return status, errors.New(status.String())
+ }
+ return status, err
+}
+
+/*
+handleHealth only fails if the user's health check function tells us.
+*/
+func (s *HTTPScaffold) handleHealth(resp http.ResponseWriter, req *http.Request) {
+ status, healthErr := s.callHealthCheck()
+
+ if status == Failed {
+ writeUnavailable(resp, req, status, healthErr)
+ } else {
+ resp.WriteHeader(http.StatusOK)
+ }
+}
+
+/*
+handleReady fails if we are marked down and also if the user's health function
+tells us.
+*/
+func (s *HTTPScaffold) handleReady(resp http.ResponseWriter, req *http.Request) {
+ status, healthErr := s.callHealthCheck()
+ if status == OK {
+ healthErr = s.tracker.markedDown()
+ if healthErr != nil {
+ status = NotReady
+ }
+ }
+
+ if status == OK {
+ resp.WriteHeader(http.StatusOK)
+ } else {
+ writeUnavailable(resp, req, status, healthErr)
+ }
+}
+
+func writeUnavailable(
+ resp http.ResponseWriter, req *http.Request,
+ stat HealthStatus, err error) {
+
+ mt := SelectMediaType(req, []string{"text/plain", "application/json"})
+
+ resp.WriteHeader(http.StatusServiceUnavailable)
+ switch mt {
+ case "application/json":
+ re := map[string]string{
+ "status": stat.String(),
+ "reason": err.Error(),
+ }
+ buf, _ := json.Marshal(&re)
+ resp.Header().Set("Content-Type", mt)
+ resp.Write(buf)
+ default:
+ resp.Header().Set("Content-Type", "text/plain")
+ resp.Write([]byte(err.Error()))
+ }
+}
diff --git a/healthstatus_string.go b/healthstatus_string.go
new file mode 100644
index 0000000..f89a363
--- /dev/null
+++ b/healthstatus_string.go
@@ -0,0 +1,16 @@
+// Code generated by "stringer -type HealthStatus ."; DO NOT EDIT
+
+package goscaffold
+
+import "fmt"
+
+const _HealthStatus_name = "OKNotReadyFailed"
+
+var _HealthStatus_index = [...]uint8{0, 2, 10, 16}
+
+func (i HealthStatus) String() string {
+ if i < 0 || i >= HealthStatus(len(_HealthStatus_index)-1) {
+ return fmt.Sprintf("HealthStatus(%d)", i)
+ }
+ return _HealthStatus_name[_HealthStatus_index[i]:_HealthStatus_index[i+1]]
+}
diff --git a/scaffold.go b/scaffold.go
index 44fbcb2..1a07320 100644
--- a/scaffold.go
+++ b/scaffold.go
@@ -1,7 +1,6 @@
package goscaffold
import (
- "encoding/json"
"errors"
"fmt"
"net"
@@ -27,16 +26,53 @@
var ErrSignalCaught = errors.New("Caught shutdown signal")
/*
+ErrManualStop is used when the user doesn't have a reason.
+*/
+var ErrManualStop = errors.New("Shutdown called")
+
+/*
+HealthStatus is a type of response from a health check.
+*/
+type HealthStatus int
+
+//go:generate stringer -type HealthStatus .
+
+const (
+ // OK denotes that everything is good
+ OK HealthStatus = iota
+ // NotReady denotes that the server is OK, but cannot process requests now
+ NotReady HealthStatus = iota
+ // Failed denotes that the server is bad
+ Failed HealthStatus = iota
+)
+
+/*
+HealthChecker is a type of function that an implementer may
+implement in order to customize what we return from the "health"
+and "ready" URLs. It must return either "OK", which means that everything
+is fine, "not ready," which means that the "ready" check will fail but
+the health check is OK, and "failed," which means that both are bad.
+The function may return an optional error, which will be returned as
+a reason for the status and will be placed in responses.
+*/
+type HealthChecker func() (HealthStatus, error)
+
+/*
An HTTPScaffold provides a set of features on top of a standard HTTP
listener. It includes an HTTP handler that may be plugged in to any
standard Go HTTP server. It is intended to be placed before any other
handlers.
*/
type HTTPScaffold struct {
- insecurePort int
- open bool
- tracker *requestTracker
- insecureListener net.Listener
+ insecurePort int
+ managementPort int
+ open bool
+ tracker *requestTracker
+ insecureListener net.Listener
+ managementListener net.Listener
+ healthCheck HealthChecker
+ healthPath string
+ readyPath string
}
/*
@@ -44,7 +80,11 @@
do nothing.
*/
func CreateHTTPScaffold() *HTTPScaffold {
- return &HTTPScaffold{}
+ return &HTTPScaffold{
+ insecurePort: 0,
+ managementPort: -1,
+ open: false,
+ }
}
/*
@@ -62,11 +102,68 @@
called after "Listen."
*/
func (s *HTTPScaffold) InsecureAddress() string {
+ if s.insecureListener == nil {
+ return ""
+ }
return s.insecureListener.Addr().String()
}
/*
-Open opens up the port that was created when the scaffold was set up.
+SetManagementPort sets the port number for management operations, including
+health checks and diagnostic operations. If not set, then these operations
+happen on the other ports. If set, then they only happen on this port.
+*/
+func (s *HTTPScaffold) SetManagementPort(p int) {
+ s.managementPort = p
+}
+
+/*
+ManagementAddress returns the actual address (including the port if an
+ephemeral port was used) where we are listening for management
+operations. If "SetManagementPort" was not set, then it returns null.
+*/
+func (s *HTTPScaffold) ManagementAddress() string {
+ if s.managementListener == nil {
+ return ""
+ }
+ return s.managementListener.Addr().String()
+}
+
+/*
+SetHealthPath sets up a health check on the management port (if set) or
+otherwise the main port. If a health check function has been supplied,
+it will return 503 if the function returns "Failed" and 200 otherwise.
+This path is intended to be used by systems like Kubernetes as the
+"health check." These systems will shut down the server if we return
+a non-200 URL.
+*/
+func (s *HTTPScaffold) SetHealthPath(p string) {
+ s.healthPath = p
+}
+
+/*
+SetReadyPath sets up a readines check on the management port (if set) or
+otherwise the main port. If a health check function has been supplied,
+it will return 503 if the function returns "Failed" or "Not Ready".
+It will also return 503 if the "Shutdown" function was called
+(or caught by signal handler). This path is intended to be used by
+load balancers that will decide whether to route calls, but not by
+systems like Kubernetes that will decide to shut down this server.
+*/
+func (s *HTTPScaffold) SetReadyPath(p string) {
+ s.readyPath = p
+}
+
+/*
+SetHealthChecker specifies a function that the scaffold will call every time
+"HealthPath" or "ReadyPath" is invoked.
+*/
+func (s *HTTPScaffold) SetHealthChecker(c HealthChecker) {
+ s.healthCheck = c
+}
+
+/*
+Open opens up the ports that were created when the scaffold was set up.
This method is optional. It may be called before Listen so that we can
retrieve the actual address where the server is listening before we actually
start to listen.
@@ -81,6 +178,27 @@
return err
}
s.insecureListener = il
+ defer func() {
+ if !s.open {
+ il.Close()
+ }
+ }()
+
+ if s.managementPort >= 0 {
+ ml, err := net.ListenTCP("tcp", &net.TCPAddr{
+ Port: s.managementPort,
+ })
+ if err != nil {
+ return err
+ }
+ s.managementListener = ml
+ defer func() {
+ if !s.open {
+ ml.Close()
+ }
+ }()
+ }
+
s.open = true
return nil
}
@@ -111,23 +229,48 @@
s.open = true
}
- handler := &httpHandler{
- s: s,
- handler: baseHandler,
+ // This is the handler that wraps customer API calls with tracking
+ trackingHandler := &requestHandler{
+ s: s,
+ child: baseHandler,
}
- go http.Serve(s.insecureListener, handler)
+ mgmtHandler := s.createManagementHandler()
+
+ var mainHandler http.Handler
+ if s.managementPort >= 0 {
+ // Management on separate port
+ mainHandler = trackingHandler
+ go http.Serve(s.managementListener, mgmtHandler)
+ } else {
+ // Management on same port
+ mgmtHandler.child = trackingHandler
+ mainHandler = mgmtHandler
+ }
+
+ go http.Serve(s.insecureListener, mainHandler)
+
err := <-s.tracker.C
+
s.insecureListener.Close()
+ if s.managementListener != nil {
+ s.managementListener.Close()
+ }
+
return err
}
/*
Shutdown indicates that the server should stop handling incoming requests
and exit from the "Serve" call. This may be called automatically by
-calling "CatchSignals," or automatically using this call.
+calling "CatchSignals," or automatically using this call. If
+"reason" is nil, a default reason will be assigned.
*/
func (s *HTTPScaffold) Shutdown(reason error) {
- s.tracker.shutdown(reason)
+ if reason == nil {
+ s.tracker.shutdown(ErrManualStop)
+ } else {
+ s.tracker.shutdown(reason)
+ }
}
/*
@@ -173,31 +316,3 @@
fmt.Fprint(os.Stderr, string(stackBuf[:w]))
}
-
-type httpHandler struct {
- s *HTTPScaffold
- handler http.Handler
-}
-
-func (h *httpHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
- startErr := h.s.tracker.start()
- if startErr == nil {
- h.handler.ServeHTTP(resp, req)
- h.s.tracker.end()
- } else {
- mt := SelectMediaType(req, []string{"text/plain", "application/json"})
- resp.Header().Set("Content-Type", mt)
- resp.WriteHeader(http.StatusServiceUnavailable)
- switch mt {
- case "application/json":
- re := map[string]string{
- "error": "Stopping",
- "message": startErr.Error(),
- }
- buf, _ := json.Marshal(&re)
- resp.Write(buf)
- default:
- resp.Write([]byte(startErr.Error()))
- }
- }
-}
diff --git a/scaffold_example_test.go b/scaffold_example_test.go
new file mode 100644
index 0000000..92f5be5
--- /dev/null
+++ b/scaffold_example_test.go
@@ -0,0 +1,39 @@
+package goscaffold
+
+import (
+ "fmt"
+ "net/http"
+)
+
+func Example() {
+ // Create a new scaffold that will listen for HTTP on port 8080
+ scaf := CreateHTTPScaffold()
+ scaf.SetInsecurePort(8080)
+
+ // Direct the scaffold to catch common signals and trigger a
+ // graceful shutdown.
+ scaf.CatchSignals()
+
+ listener := &TestListener{}
+
+ // Listen now. The listener will return when the server is actually
+ // shut down.
+ err := scaf.Listen(listener)
+
+ // If we get here, and if we care to know, then the error will tell
+ // us why we were shut down.
+ fmt.Printf("HTTP server shut down: %s\n", err.Error())
+}
+
+/*
+TestListener is an HTTP listener used for the example code. It just returns
+200 and "Hello, World!"
+*/
+type TestListener struct {
+}
+
+func (l *TestListener) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
+ resp.Header().Set("Content-Type", "text/plain")
+ resp.WriteHeader(http.StatusOK)
+ resp.Write([]byte("Hello, World!"))
+}
diff --git a/scaffold_test.go b/scaffold_test.go
index 8ba7c46..e4fe194 100644
--- a/scaffold_test.go
+++ b/scaffold_test.go
@@ -1,9 +1,12 @@
package goscaffold
import (
+ "encoding/json"
"errors"
"fmt"
+ "io/ioutil"
"net/http"
+ "sync/atomic"
"time"
. "github.com/onsi/ginkgo"
@@ -29,14 +32,49 @@
}, 5*time.Second).Should(BeTrue())
resp, err := http.Get(fmt.Sprintf("http://%s", s.InsecureAddress()))
Expect(err).Should(Succeed())
+ resp.Body.Close()
Expect(resp.StatusCode).Should(Equal(200))
shutdownErr := errors.New("Validate")
s.Shutdown(shutdownErr)
Eventually(stopChan).Should(Receive(Equal(shutdownErr)))
})
+ It("Separate management port", func() {
+ s := CreateHTTPScaffold()
+ s.SetManagementPort(0)
+ stopChan := make(chan error)
+ err := s.Open()
+ Expect(err).Should(Succeed())
+
+ go func() {
+ fmt.Fprintf(GinkgoWriter, "Gonna listen on %s and %s\n",
+ s.InsecureAddress(), s.ManagementAddress())
+ stopErr := s.Listen(&testHandler{})
+ fmt.Fprintf(GinkgoWriter, "Done listening\n")
+ stopChan <- stopErr
+ }()
+
+ // Just make sure that it's up
+ Eventually(func() bool {
+ return testGet(s, "")
+ }, 5*time.Second).Should(BeTrue())
+ resp, err := http.Get(fmt.Sprintf("http://%s", s.InsecureAddress()))
+ Expect(err).Should(Succeed())
+ resp.Body.Close()
+ Expect(resp.StatusCode).Should(Equal(200))
+ resp, err = http.Get(fmt.Sprintf("http://%s", s.ManagementAddress()))
+ Expect(err).Should(Succeed())
+ resp.Body.Close()
+ Expect(resp.StatusCode).Should(Equal(404))
+ shutdownErr := errors.New("Validate")
+ s.Shutdown(shutdownErr)
+ Eventually(stopChan).Should(Receive(Equal(shutdownErr)))
+ })
+
It("Shutdown", func() {
s := CreateHTTPScaffold()
+ s.SetHealthPath("/health")
+ s.SetReadyPath("/ready")
stopChan := make(chan bool)
err := s.Open()
Expect(err).Should(Succeed())
@@ -47,9 +85,8 @@
}()
go func() {
- resp2, err2 := http.Get(fmt.Sprintf("http://%s?delay=1s", s.InsecureAddress()))
- Expect(err2).Should(Succeed())
- Expect(resp2.StatusCode).Should(Equal(200))
+ code, _ := getText(fmt.Sprintf("http://%s?delay=1s", s.InsecureAddress()))
+ Expect(code).Should(Equal(200))
}()
// Just make sure server is listening
@@ -57,15 +94,27 @@
return testGet(s, "")
}, 5*time.Second).Should(BeTrue())
+ // Ensure that we are healthy and ready
+ code, _ := getText(fmt.Sprintf("http://%s/health", s.InsecureAddress()))
+ Expect(code).Should(Equal(200))
+ code, _ = getText(fmt.Sprintf("http://%s/ready", s.InsecureAddress()))
+ Expect(code).Should(Equal(200))
+
// Previous call prevents server from exiting
Consistently(stopChan, 250*time.Millisecond).ShouldNot(Receive())
// Tell the server to try and exit
s.Shutdown(errors.New("Stop"))
- // Should take one second -- in the meantime, calls should fail with 503
- resp, err := http.Get(fmt.Sprintf("http://%s?", s.InsecureAddress()))
- Expect(err).Should(Succeed())
- Expect(resp.StatusCode).Should(Equal(503))
+
+ // Should take one second -- in the meantime, calls should fail with 503,
+ // health should be good, but ready should be bad
+ code, _ = getText(fmt.Sprintf("http://%s", s.InsecureAddress()))
+ Expect(code).Should(Equal(503))
+ code, _ = getText(fmt.Sprintf("http://%s/ready", s.InsecureAddress()))
+ Expect(code).Should(Equal(503))
+ code, _ = getText(fmt.Sprintf("http://%s/health", s.InsecureAddress()))
+ Expect(code).Should(Equal(200))
+
// But in less than two seconds, server should be down
Eventually(stopChan, 2*time.Second).Should(Receive(BeTrue()))
// Calls should now fail
@@ -73,14 +122,124 @@
return testGet(s, "")
}, time.Second).Should(BeFalse())
})
+
+ It("Health Check Functions", func() {
+ status := int32(OK)
+ var healthErr = &atomic.Value{}
+
+ statusFunc := func() (HealthStatus, error) {
+ stat := HealthStatus(atomic.LoadInt32(&status))
+ av := healthErr.Load()
+ if av != nil {
+ errPtr := av.(*error)
+ return stat, *errPtr
+ }
+ return stat, nil
+ }
+
+ s := CreateHTTPScaffold()
+ s.SetManagementPort(0)
+ s.SetHealthPath("/health")
+ s.SetReadyPath("/ready")
+ s.SetHealthChecker(statusFunc)
+ stopChan := make(chan error)
+ err := s.Open()
+ Expect(err).Should(Succeed())
+
+ go func() {
+ fmt.Fprintf(GinkgoWriter, "Gonna listen on %s and %s\n",
+ s.InsecureAddress(), s.ManagementAddress())
+ stopErr := s.Listen(&testHandler{})
+ fmt.Fprintf(GinkgoWriter, "Done listening\n")
+ stopChan <- stopErr
+ }()
+
+ // Just make sure that it's up
+ Eventually(func() bool {
+ return testGet(s, "")
+ }, 5*time.Second).Should(BeTrue())
+
+ // Health should be good
+ code, _ := getText(fmt.Sprintf("http://%s/health", s.ManagementAddress()))
+ Expect(code).Should(Equal(200))
+ code, _ = getText(fmt.Sprintf("http://%s/ready", s.ManagementAddress()))
+ Expect(code).Should(Equal(200))
+
+ // Mark down to "unhealthy" state. Should be bad.
+ atomic.StoreInt32(&status, int32(Failed))
+ code, bod := getText(fmt.Sprintf("http://%s/health", s.ManagementAddress()))
+ Expect(code).Should(Equal(503))
+ Expect(bod).Should(Equal("Failed"))
+ code, _ = getText(fmt.Sprintf("http://%s/ready", s.ManagementAddress()))
+ Expect(code).Should(Equal(503))
+
+ // Change to merely "not ready" state. Should be healthy but not ready.
+ atomic.StoreInt32(&status, int32(NotReady))
+ code, _ = getText(fmt.Sprintf("http://%s/health", s.ManagementAddress()))
+ Expect(code).Should(Equal(200))
+ code, bod = getText(fmt.Sprintf("http://%s/ready", s.ManagementAddress()))
+ Expect(code).Should(Equal(503))
+ Expect(bod).Should(Equal("NotReady"))
+
+ // Customize the error message.
+ customErr := errors.New("Custom")
+ healthErr.Store(&customErr)
+ code, bod = getText(fmt.Sprintf("http://%s/ready", s.ManagementAddress()))
+ Expect(code).Should(Equal(503))
+ Expect(bod).Should(Equal("Custom"))
+
+ // Check it in JSON
+ code, js := getJSON(fmt.Sprintf("http://%s/ready", s.ManagementAddress()))
+ Expect(code).Should(Equal(503))
+ Expect(js["status"]).Should(Equal("NotReady"))
+ Expect(js["reason"]).Should(Equal("Custom"))
+
+ // Mark back up. Should be all good
+ atomic.StoreInt32(&status, int32(OK))
+ code, _ = getText(fmt.Sprintf("http://%s/health", s.ManagementAddress()))
+ Expect(code).Should(Equal(200))
+ code, _ = getText(fmt.Sprintf("http://%s/ready", s.ManagementAddress()))
+ Expect(code).Should(Equal(200))
+
+ s.Shutdown(nil)
+ Eventually(stopChan).Should(Receive(Equal(ErrManualStop)))
+ })
})
+func getText(url string) (int, string) {
+ req, err := http.NewRequest("GET", url, nil)
+ Expect(err).Should(Succeed())
+ req.Header.Set("Accept", "text/plain")
+ resp, err := http.DefaultClient.Do(req)
+ Expect(err).Should(Succeed())
+ defer resp.Body.Close()
+ bod, err := ioutil.ReadAll(resp.Body)
+ Expect(err).Should(Succeed())
+ return resp.StatusCode, string(bod)
+}
+
+func getJSON(url string) (int, map[string]string) {
+ req, err := http.NewRequest("GET", url, nil)
+ Expect(err).Should(Succeed())
+ req.Header.Set("Accept", "application/json")
+ resp, err := http.DefaultClient.Do(req)
+ Expect(err).Should(Succeed())
+ defer resp.Body.Close()
+ bod, err := ioutil.ReadAll(resp.Body)
+ Expect(err).Should(Succeed())
+ var vals map[string]string
+ err = json.Unmarshal(bod, &vals)
+ Expect(err).Should(Succeed())
+ return resp.StatusCode, vals
+}
+
func testGet(s *HTTPScaffold, path string) bool {
resp, err := http.Get(fmt.Sprintf("http://%s", s.InsecureAddress()))
if err != nil {
fmt.Fprintf(GinkgoWriter, "Get %s = %s\n", path, err)
return false
}
+ resp.Body.Close()
if resp.StatusCode != 200 {
fmt.Fprintf(GinkgoWriter, "Get %s = %d\n", path, resp.StatusCode)
return false
diff --git a/tracker.go b/tracker.go
index 0b624ea..0f9c9ea 100644
--- a/tracker.go
+++ b/tracker.go
@@ -53,16 +53,11 @@
shutting down.
*/
func (t *requestTracker) start() error {
- sd := atomic.LoadInt32(&t.shuttingDown)
- if sd != 0 {
- reason := t.shutdownReason.Load().(*error)
- if reason == nil {
- return nil
- }
- return *reason
+ md := t.markedDown()
+ if md == nil {
+ t.commandChan <- startRequest
}
- t.commandChan <- startRequest
- return nil
+ return md
}
/*
@@ -74,6 +69,23 @@
}
/*
+markedDown returns nil if everything is good, and an error if the server
+has been marked down. The error is the one that was sent to the
+"Shutdown" method.
+*/
+func (t *requestTracker) markedDown() error {
+ sd := atomic.LoadInt32(&t.shuttingDown)
+ if sd != 0 {
+ reason := t.shutdownReason.Load().(*error)
+ if reason == nil {
+ return nil
+ }
+ return *reason
+ }
+ return nil
+}
+
+/*
shutdown indicates that the tracker should start counting down until
the number of running requests reaches zero. The "reason" will be returned
as the result of the "start" call.
