Repository: nats-io/nats-server
Pull request #2815 (2022-01-25 02:22:05)

Added in /healthz endpoint.
Used for health and liveness probes for environments like k8s.

Currently this code returns a 200 and `{ "status": "ok" }` iff all configured ports are open
and if JetStream is configured and we have contact with the meta leader and its current and all streams are up to date.

If we are not current we return a 503 at first error and a `{ "status": "unavailable", "error": "DESCRIPTION" }` body.

We can do much more but this feels like a decent start to test with upgrading servers in a k8s environment that is running JetStream.

Signed-off-by: Derek Collison <derek@nats.io>

/cc @nats-io/core



Patch:
diff --git a/server/monitor.go b/server/monitor.go
index 6f604ea9..cd997a47 100644
--- a/server/monitor.go
+++ b/server/monitor.go
@@ -1260,6 +1260,7 @@ func (s *Server) HandleRoot(w http.ResponseWriter, r *http.Request) {
 	<a href=.%s>subsz</a><br/>
 	<a href=.%s>accountz</a><br/>
 	<a href=.%s>jsz</a><br/>
+	<a href=.%s>healthz</a><br/>
     <br/>
     <a href=https://docs.nats.io/nats-server/configuration/monitoring>help</a>
   </body>
@@ -1272,6 +1273,7 @@ func (s *Server) HandleRoot(w http.ResponseWriter, r *http.Request) {
 		s.basePath(SubszPath),
 		s.basePath(AccountzPath),
 		s.basePath(JszPath),
+		s.basePath(HealthzPath),
 	)
 }
 
@@ -2686,9 +2688,98 @@ func (s *Server) HandleJsz(w http.ResponseWriter, r *http.Request) {
 	}
 	b, err := json.MarshalIndent(l, "", "  ")
 	if err != nil {
-		s.Errorf("Error marshaling response to /leafz request: %v", err)
+		s.Errorf("Error marshaling response to /jsz request: %v", err)
 	}
 
 	// Handle response
 	ResponseHandler(w, r, b)
 }
+
+type HealthStatus struct {
+	Status string `json:"status"`
+	Error  string `json:"error,omitempty"`
+}
+
+// https://tools.ietf.org/id/draft-inadarei-api-health-check-05.html
+func (s *Server) HandleHealthz(w http.ResponseWriter, r *http.Request) {
+	s.mu.Lock()
+	s.httpReqStats[HealthzPath]++
+	s.mu.Unlock()
+
+	var health = &HealthStatus{Status: "ok"}
+
+	if err := s.readyForConnections(time.Millisecond); err != nil {
+		health.Status = "error"
+		health.Error = err.Error()
+		w.WriteHeader(http.StatusServiceUnavailable)
+	} else if js := s.getJetStream(); js != nil {
+		// Check JetStream status here.
+		js.mu.RLock()
+		clustered, cc := !js.standAlone, js.cluster
+		js.mu.RUnlock()
+		if clustered {
+			// We do more checking for clustered mode to allow for proper rolling updates.
+			// We will make sure that we have seen the meta leader and that we are current with all assets.
+			node := js.getMetaGroup()
+			if node.GroupLeader() == _EMPTY_ {
+				health.Status = "unavailable"
+				health.Error = "JetStream has not established contact with a meta leader"
+				w.WriteHeader(http.StatusServiceUnavailable)
+			} else if !node.Current() {
+				health.Status = "unavailable"
+				health.Error = "JetStream is not current with the meta leader"
+				w.WriteHeader(http.StatusServiceUnavailable)
+			} else {
+				// If we are here we are current and have seen our meta leader.
+				// Now check assets.
+				var _a [512]*jsAccount
+				accounts := _a[:0]
+				js.mu.RLock()
+				// Collect accounts.
+				for _, jsa := range js.accounts {
+					accounts = append(accounts, jsa)
+				}
+				js.mu.RUnlock()
+
+				var streams []*stream
+			Err:
+				// Walk our accounts and assets.
+				for _, jsa := range accounts {
+					if len(streams) > 0 {
+						streams = streams[:0]
+					}
+					jsa.mu.RLock()
+					accName := jsa.account.Name
+					for _, stream := range jsa.streams {
+						streams = append(streams, stream)
+					}
+					jsa.mu.RUnlock()
+					// Now walk the streams themselves.
+					js.mu.RLock()
+					for _, stream := range streams {
+						// Skip non-replicated.
+						if stream.cfg.Replicas <= 1 {
+							continue
+						}
+						sname := stream.name()
+						if !cc.isStreamCurrent(accName, sname) {
+							health.Status = "unavailable"
+							health.Error = fmt.Sprintf("JetStream stream %q for account %q is not current", sname, accName)
+							w.WriteHeader(http.StatusServiceUnavailable)
+							js.mu.RUnlock()
+							break Err
+						}
+					}
+					js.mu.RUnlock()
+				}
+			}
+		}
+	}
+
+	b, err := json.Marshal(health)
+	if err != nil {
+		s.Errorf("Error marshaling response to /healthz request: %v", err)
+	}
+
+	ResponseHandler(w, r, b)
+}
diff --git a/server/server.go b/server/server.go
index 3310db2e..47afdacb 100644
--- a/server/server.go
+++ b/server/server.go
@@ -2265,6 +2265,7 @@ const (
 	StackszPath  = "/stacksz"
 	AccountzPath = "/accountz"
 	JszPath      = "/jsz"
+	HealthzPath  = "/healthz"
 )
 
 func (s *Server) basePath(p string) string {
@@ -2340,8 +2341,8 @@ func (s *Server) startMonitoring(secure bool) error {
 		return fmt.Errorf("can't listen to the monitor port: %v", err)
 	}
 
-	s.Noticef("Starting %s monitor on %s", monitorProtocol,
-		net.JoinHostPort(opts.HTTPHost, strconv.Itoa(httpListener.Addr().(*net.TCPAddr).Port)))
+	rport := httpListener.Addr().(*net.TCPAddr).Port
+	s.Noticef("Starting %s monitor on %s", monitorProtocol, net.JoinHostPort(opts.HTTPHost, strconv.Itoa(rport)))
 
 	mux := http.NewServeMux()
 
@@ -2367,6 +2368,9 @@ func (s *Server) startMonitoring(secure bool) error {
 	mux.HandleFunc(s.basePath(AccountzPath), s.HandleAccountz)
 	// Jsz
 	mux.HandleFunc(s.basePath(JszPath), s.HandleJsz)
+	// Healthz
+	mux.HandleFunc(s.basePath(HealthzPath), s.HandleHealthz)
+
 	// Do not set a WriteTimeout because it could cause cURL/browser
 	// to return empty response or unable to display page if the
 	// server needs more time to build the response.
@@ -2942,7 +2946,7 @@ func (s *Server) readyForConnections(d time.Duration) error {
 		s.mu.Lock()
 		chk["server"] = info{ok: s.listener != nil, err: s.listenerErr}
 		chk["route"] = info{ok: (opts.Cluster.Port == 0 || s.routeListener != nil), err: s.routeListenerErr}
-		chk["gateway"] = info{ok: (opts.Gateway.Name == "" || s.gatewayListener != nil), err: s.gatewayListenerErr}
+		chk["gateway"] = info{ok: (opts.Gateway.Name == _EMPTY_ || s.gatewayListener != nil), err: s.gatewayListenerErr}
 		chk["leafNode"] = info{ok: (opts.LeafNode.Port == 0 || s.leafNodeListener != nil), err: s.leafNodeListenerErr}
 		chk["websocket"] = info{ok: (opts.Websocket.Port == 0 || s.websocket.listener != nil), err: s.websocket.listenerErr}
 		chk["mqtt"] = info{ok: (opts.MQTT.Port == 0 || s.mqtt.listener != nil), err: s.mqtt.listenerErr}
@@ -2957,7 +2961,9 @@ func (s *Server) readyForConnections(d time.Duration) error {
 		if numOK == len(chk) {
 			return nil
 		}
-		time.Sleep(25 * time.Millisecond)
+		if d > 25*time.Millisecond {
+			time.Sleep(25 * time.Millisecond)
+		}
 	}
 
 	failed := make([]string, 0, len(chk))


Test patch:
diff --git a/server/jetstream_cluster_test.go b/server/jetstream_cluster_test.go
index 64e78afb..fe44c833 100644
--- a/server/jetstream_cluster_test.go
+++ b/server/jetstream_cluster_test.go
@@ -10383,6 +10383,7 @@ func (sc *supercluster) randomServer() *Server {
 
 var jsClusterAccountsTempl = `
 	listen: 127.0.0.1:-1
+
 	server_name: %s
 	jetstream: {max_mem_store: 256MB, max_file_store: 2GB, store_dir: "%s"}
 
diff --git a/server/norace_test.go b/server/norace_test.go
index 2fb04142..39552aff 100644
--- a/server/norace_test.go
+++ b/server/norace_test.go
@@ -27,6 +27,7 @@ import (
 	"io/ioutil"
 	"math/rand"
 	"net"
+	"net/http"
 	"net/url"
 	"runtime"
 	"runtime/debug"
@@ -4024,3 +4025,79 @@ func TestNoRaceConsumerFileStoreConcurrentDiskIO(t *testing.T) {
 	close(startCh)
 	wg.Wait()
 }
+
+func TestNoRaceJetStreamClusterHealthz(t *testing.T) {
+	c := createJetStreamCluster(t, jsClusterAccountsTempl, "HZ", _EMPTY_, 3, 23033, true)
+	defer c.shutdown()
+
+	nc1, js1 := jsClientConnect(t, c.randomServer(), nats.UserInfo("one", "p"))
+	defer nc1.Close()
+
+	nc2, js2 := jsClientConnect(t, c.randomServer(), nats.UserInfo("two", "p"))
+	defer nc2.Close()
+
+	var err error
+	for _, sname := range []string{"foo", "bar", "baz"} {
+		_, err = js1.AddStream(&nats.StreamConfig{Name: sname, Replicas: 3})
+		require_NoError(t, err)
+		_, err = js2.AddStream(&nats.StreamConfig{Name: sname, Replicas: 3})
+		require_NoError(t, err)
+	}
+	// R1
+	_, err = js1.AddStream(&nats.StreamConfig{Name: "r1", Replicas: 1})
+	require_NoError(t, err)
+
+	// Now shutdown then send a bunch of data.
+	s := c.servers[0]
+	s.Shutdown()
+
+	for i := 0; i < 5_000; i++ {
+		_, err = js1.PublishAsync("foo", []byte("OK"))
+		require_NoError(t, err)
+		_, err = js2.PublishAsync("bar", []byte("OK"))
+		require_NoError(t, err)
+	}
+	select {
+	case <-js1.PublishAsyncComplete():
+	case <-time.After(5 * time.Second):
+		t.Fatalf("Did not receive completion signal")
+	}
+	select {
+	case <-js2.PublishAsyncComplete():
+	case <-time.After(5 * time.Second):
+		t.Fatalf("Did not receive completion signal")
+	}
+
+	s = c.restartServer(s)
+	opts := s.getOpts()
+	opts.HTTPHost = "127.0.0.1"
+	opts.HTTPPort = 11222
+	err = s.StartMonitoring()
+	require_NoError(t, err)
+	url := fmt.Sprintf("http://127.0.0.1:%d/healthz", opts.HTTPPort)
+
+	getHealth := func() (int, *HealthStatus) {
+		resp, err := http.Get(url)
+		require_NoError(t, err)
+		defer resp.Body.Close()
+		body, err := ioutil.ReadAll(resp.Body)
+		require_NoError(t, err)
+		var hs HealthStatus
+		err = json.Unmarshal(body, &hs)
+		require_NoError(t, err)
+		return resp.StatusCode, &hs
+	}
+
+	errors := 0
+	checkFor(t, 20*time.Second, 100*time.Millisecond, func() error {
+		code, hs := getHealth()
+		if code >= 200 && code < 300 {
+			return nil
+		}
+		errors++
+		return fmt.Errorf("Got %d status with %+v", code, hs)
+	})
+	if errors == 0 {
+		t.Fatalf("Expected to have some errors until we became current, got none")
+	}
+}