Repository: atlassian/gostatsd
Pull request #620 (2023-04-20 03:52:38)

Adds an actual healthcheck
I don't love this, but it's good enough for now.


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 68f1d5b..80fa667 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,3 +1,7 @@
+35.2.0
+------
+- Adds more meaningful healthchecks
+
 35.1.6 - 35.1.19
 ------
 - These versions were just developing and debugging the new build and release process on Github Actions
diff --git a/HTTP.md b/HTTP.md
index 8dc8675..00457a2 100644
--- a/HTTP.md
+++ b/HTTP.md
@@ -11,9 +11,22 @@ Only one prof will be allowed to run at any point, and requesting multiple will
 - `/expvar`, routes directly to the [expvar handler](https://golang.org/pkg/expvar/#Handler)
 
 ### `healthcheck` endpoints
-- `/healthcheck`, reports if the server is internally healthy.  This is what should be used for health checking by an LB.
-- `/deepcheck`, reports the status of downstream services.  This should not be used for system healthcheck, as a bad
-  dependency should not cause an otherwise healthy server to cycle, because it will likely fail again.
+- `/healthcheck`, reports if the server is internally healthy.  This is what should be used for health checking by a
+  load-balancer.  It does not report on downstream dependencies.
+- `/deepcheck`, reports the status of downstream dependencies.  This should not be used for system healthcheck, as a bad
+  dependency should not cause an otherwise healthy server to cycle, because it will likely fail again.  It is generally
+  checked on startup, not continuously, but is designed with the assumption that it will be continuously polled.
+
+Both endpoints will return 200 if healthy, or 500 if not healthy.  Currently the body is:
+```
+{
+  "ok": ["results of checks"],
+  "failed": ["results of checks"]
+}
+```
+
+however this is not contractual and may change.  The contract is the error code, not the body.  A 500 is returned if
+the `failed` field is non-empty.
 
 ### `ingestion` endpoint
 - `/vN/raw` and `/vN/event`, takes in protobuf formatted raw metrics.  This endpoint is intended for gostatsd to
diff --git a/pkg/healthcheck/healthcheck.go b/pkg/healthcheck/healthcheck.go
new file mode 100644
index 0000000..c5868df
--- /dev/null
+++ b/pkg/healthcheck/healthcheck.go
@@ -0,0 +1,31 @@
+package healthcheck
+
+// HealthcheckFunc is a function that returns a status message, and if the check if healthy or not (false).
+// healthchecks must not block, and downstream dependencies should be reported on via a watchdog style, and not by
+// making a roundtrip.
+type HealthcheckFunc func() (string, HealthyStatus)
+
+type HealthyStatus bool
+
+const (
+	Healthy   = HealthyStatus(true)
+	Unhealthy = HealthyStatus(false)
+)
+
+type HealthCheckProvider interface {
+	HealthChecks() []HealthcheckFunc
+}
+
+type DeepCheckProvider interface {
+	DeepChecks() []HealthcheckFunc
+}
+
+func MaybeAppendHealthChecks(healthChecks []HealthcheckFunc, deepChecks []HealthcheckFunc, maybeProvider interface{}) ([]HealthcheckFunc, []HealthcheckFunc) {
+	if hcp, ok := maybeProvider.(HealthCheckProvider); ok {
+		healthChecks = append(healthChecks, hcp.HealthChecks()...)
+	}
+	if dcp, ok := maybeProvider.(DeepCheckProvider); ok {
+		deepChecks = append(deepChecks, dcp.DeepChecks()...)
+	}
+	return healthChecks, deepChecks
+}
diff --git a/pkg/statsd/handler_http_forwarder_v2.go b/pkg/statsd/handler_http_forwarder_v2.go
index 6a16838..bf0b45e 100644
--- a/pkg/statsd/handler_http_forwarder_v2.go
+++ b/pkg/statsd/handler_http_forwarder_v2.go
@@ -23,6 +23,7 @@ import (
 	"github.com/atlassian/gostatsd"
 	"github.com/atlassian/gostatsd/internal/util"
 	"github.com/atlassian/gostatsd/pb"
+	"github.com/atlassian/gostatsd/pkg/healthcheck"
 	"github.com/atlassian/gostatsd/pkg/stats"
 	"github.com/atlassian/gostatsd/pkg/transport"
 )
@@ -39,12 +40,13 @@ const (
 
 // HttpForwarderHandlerV2 is a PipelineHandler which sends metrics to another gostatsd instance
 type HttpForwarderHandlerV2 struct {
-	postId          uint64 // atomic - used for an id in logs
-	messagesInvalid uint64 // atomic - messages which failed to be created
-	messagesCreated uint64 // atomic - messages which were created
-	messagesSent    uint64 // atomic - messages successfully sent
-	messagesRetried uint64 // atomic - retries (first send is not a retry, final failure is not a retry)
-	messagesDropped uint64 // atomic - final failure
+	postId             uint64 // atomic - used for an id in logs
+	messagesInvalid    uint64 // atomic - messages which failed to be created
+	messagesCreated    uint64 // atomic - messages which were created
+	messagesSent       uint64 // atomic - messages successfully sent
+	messagesRetried    uint64 // atomic - retries (first send is not a retry, final failure is not a retry)
+	messagesDropped    uint64 // atomic - final failure
+	lastSuccessfulSend atomic.Int64
 
 	logger                logrus.FieldLogger
 	apiEndpoint           string
@@ -62,6 +64,10 @@ type HttpForwarderHandlerV2 struct {
 	consolidatedMetrics chan []*gostatsd.MetricMap
 }
 
+var (
+	_ healthcheck.DeepCheckProvider = &HttpForwarderHandlerV2{}
+)
+
 // NewHttpForwarderHandlerV2FromViper returns a new http API client.
 func NewHttpForwarderHandlerV2FromViper(logger logrus.FieldLogger, v *viper.Viper, pool *transport.TransportPool) (*HttpForwarderHandlerV2, error) {
 	subViper := util.GetSubViper(v, "http-transport")
@@ -193,6 +199,20 @@ func NewHttpForwarderHandlerV2(
 	}, nil
 }
 
+func (hfh *HttpForwarderHandlerV2) DeepChecks() []healthcheck.HealthcheckFunc {
+	return []healthcheck.HealthcheckFunc{
+		hfh.dcSentRecently,
+	}
+}
+
+func (hfh *HttpForwarderHandlerV2) dcSentRecently() (string, healthcheck.HealthyStatus) {
+	if lastSend := hfh.lastSuccessfulSend.Load(); lastSend == 0 {
+		return "sentRecently: never sent", healthcheck.Unhealthy
+	} else {
+		return fmt.Sprintf("sentRecently: lastSend=%s", time.Unix(0, lastSend).String()), healthcheck.Healthy
+	}
+}
+
 func (hfh *HttpForwarderHandlerV2) EstimatedTags() int {
 	return 0
 }
@@ -239,8 +259,14 @@ func (hfh *HttpForwarderHandlerV2) emitMetrics(statser stats.Statser) {
 	statser.Count("http.forwarder.dropped", float64(messagesDropped), nil)
 }
 
+// sendNop sends an empty metric map downstream.  It's used to "prime the pump" for the deepcheck.
+func (hfh *HttpForwarderHandlerV2) sendNop(ctx context.Context) {
+	hfh.postMetrics(ctx, gostatsd.NewMetricMap(false), "", 0)
+}
+
 func (hfh *HttpForwarderHandlerV2) Run(ctx context.Context) {
 	var wg wait.Group
+	wg.StartWithContext(ctx, hfh.sendNop)
 	wg.Start(func() {
 		for metricMaps := range hfh.consolidatedMetrics {
 			hfh.acquireMergingSem()
@@ -386,6 +412,7 @@ func (hfh *HttpForwarderHandlerV2) post(ctx context.Context, message proto.Messa
 	for {
 		if err = post(); err == nil {
 			atomic.AddUint64(&hfh.messagesSent, 1)
+			hfh.lastSuccessfulSend.Store(clock.Now(ctx).UnixNano())
 			return
 		}
 
diff --git a/pkg/statsd/statsd.go b/pkg/statsd/statsd.go
index 529ea69..cdd5ad7 100644
--- a/pkg/statsd/statsd.go
+++ b/pkg/statsd/statsd.go
@@ -14,6 +14,7 @@ import (
 	"golang.org/x/time/rate"
 
 	"github.com/atlassian/gostatsd"
+	"github.com/atlassian/gostatsd/pkg/healthcheck"
 	"github.com/atlassian/gostatsd/pkg/stats"
 	"github.com/atlassian/gostatsd/pkg/transport"
 	"github.com/atlassian/gostatsd/pkg/web"
@@ -153,11 +154,15 @@ func (s *Server) createFinalSink(logger logrus.FieldLogger) (gostatsd.PipelineHa
 func (s *Server) RunWithCustomSocket(ctx context.Context, sf SocketFactory) error {
 	logger := logrus.StandardLogger()
 
+	var healthChecks []healthcheck.HealthcheckFunc
+	var deepChecks []healthcheck.HealthcheckFunc
+
 	handler, runnables, err := s.createFinalSink(logger)
 	if err != nil {
 		return err
 	}
 
+	healthChecks, deepChecks = healthcheck.MaybeAppendHealthChecks(healthChecks, deepChecks, handler)
 	runnables = append(append(make([]gostatsd.Runnable, 0, len(s.Runnables)), s.Runnables...), runnables...)
 
 	// Create the tag processor
@@ -196,7 +201,7 @@ func (s *Server) RunWithCustomSocket(ctx context.Context, sf SocketFactory) erro
 	runnables = gostatsd.MaybeAppendRunnable(runnables, statser)
 
 	// Create any http servers
-	httpServers, err := web.NewHttpServersFromViper(s.Viper, logger, handler)
+	httpServers, err := web.NewHttpServersFromViper(s.Viper, logger, handler, healthChecks, deepChecks)
 	if err != nil {
 		return err
 	}
diff --git a/pkg/web/http_healthchecker.go b/pkg/web/http_healthchecker.go
index 619ad33..ad2eec0 100644
--- a/pkg/web/http_healthchecker.go
+++ b/pkg/web/http_healthchecker.go
@@ -3,22 +3,57 @@ package web
 import (
 	"net/http"
 
+	jsoniter "github.com/json-iterator/go"
 	"github.com/sirupsen/logrus"
+
+	"github.com/atlassian/gostatsd/pkg/healthcheck"
 )
 
 type healthChecker struct {
-	logger logrus.FieldLogger
+	logger       logrus.FieldLogger
+	healthChecks []healthcheck.HealthcheckFunc
+	deepChecks   []healthcheck.HealthcheckFunc
+}
+
+func runHealthChecks(checks []healthcheck.HealthcheckFunc) (good []string, bad []string) {
+	// Force it render as an array, not null
+	good = []string{}
+	bad = []string{}
+	for _, check := range checks {
+		report, isHealthy := check()
+		if isHealthy == healthcheck.Healthy {
+			good = append(good, report)
+		} else {
+			bad = append(bad, report)
+		}
+	}
+	return good, bad
+}
+
+func respondToHealthChecks(resp http.ResponseWriter, checks []healthcheck.HealthcheckFunc) {
+	good, bad := runHealthChecks(checks)
+	resp.Header().Set("content-type", "application/json")
+	if len(bad) > 0 {
+		resp.WriteHeader(http.StatusInternalServerError)
+	} else {
+		resp.WriteHeader(http.StatusOK)
+	}
+
+	enc := jsoniter.NewEncoder(resp)
+	_ = enc.Encode(map[string][]string{
+		"ok":     good,
+		"failed": bad,
+	})
 }
 
-// healthCheck reports if the server is ready to process traffic.  It does not validate downstream dependencies.
-func (hc *healthChecker) healthCheck(w http.ResponseWriter, req *http.Request) {
+// healthCheck reports if the server is ready to process traffic.
+func (hc *healthChecker) healthCheck(resp http.ResponseWriter, req *http.Request) {
 	hc.logger.Info("healthCheck")
-	_, _ = w.Write([]byte("OK"))
+	respondToHealthChecks(resp, hc.healthChecks)
 }
 
-// deepCheck reports on the status of downstream dependencies.  It should be non-blocking (ie, report on the status
-// of watchdogs, rather than make roundtrips).  For now, it just reports that it's always happy.
-func (hc *healthChecker) deepCheck(w http.ResponseWriter, req *http.Request) {
+// deepCheck reports on the status of downstream dependencies.
+func (hc *healthChecker) deepCheck(resp http.ResponseWriter, req *http.Request) {
 	hc.logger.Info("deepCheck")
-	_, _ = w.Write([]byte("OK"))
+	respondToHealthChecks(resp, hc.deepChecks)
 }
diff --git a/pkg/web/httpservers.go b/pkg/web/httpservers.go
index 6ac7098..0252ce8 100644
--- a/pkg/web/httpservers.go
+++ b/pkg/web/httpservers.go
@@ -16,6 +16,7 @@ import (
 
 	"github.com/atlassian/gostatsd"
 	"github.com/atlassian/gostatsd/internal/util"
+	"github.com/atlassian/gostatsd/pkg/healthcheck"
 )
 
 type httpServer struct {
@@ -34,11 +35,17 @@ type route struct {
 
 var done = struct{}{}
 
-func NewHttpServersFromViper(v *viper.Viper, logger logrus.FieldLogger, handler gostatsd.PipelineHandler) ([]*httpServer, error) {
+func NewHttpServersFromViper(
+	v *viper.Viper,
+	logger logrus.FieldLogger,
+	handler gostatsd.PipelineHandler,
+	healthChecks []healthcheck.HealthcheckFunc,
+	deepChecks []healthcheck.HealthcheckFunc,
+) ([]*httpServer, error) {
 	httpServerNames := v.GetStringSlice("http-servers")
 	servers := make([]*httpServer, 0, len(httpServerNames))
 	for _, httpServerName := range httpServerNames {
-		server, err := newHttpServerFromViper(logger, v, httpServerName, handler)
+		server, err := newHttpServerFromViper(logger, v, httpServerName, handler, healthChecks, deepChecks)
 		if err != nil {
 			return nil, fmt.Errorf("failed to make http-server %s: %v", httpServerName, err)
 		}
@@ -52,6 +59,8 @@ func newHttpServerFromViper(
 	vMain *viper.Viper,
 	serverName string,
 	handler gostatsd.PipelineHandler,
+	healthChecks []healthcheck.HealthcheckFunc,
+	deepChecks []healthcheck.HealthcheckFunc,
 ) (*httpServer, error) {
 	vSub := util.GetSubViper(vMain, "http."+serverName)
 	vSub.SetDefault("address", "127.0.0.1:8080")
@@ -69,6 +78,8 @@ func newHttpServerFromViper(
 		vSub.GetBool("enable-expvar"),
 		vSub.GetBool("enable-ingestion"),
 		vSub.GetBool("enable-healthcheck"),
+		healthChecks,
+		deepChecks,
 	)
 }
 
@@ -80,6 +91,8 @@ func NewHttpServer(
 	enableExpVar,
 	enableIngestion,
 	enableHealthcheck bool,
+	healthChecks []healthcheck.HealthcheckFunc,
+	deepChecks []healthcheck.HealthcheckFunc,
 ) (*httpServer, error) {
 	var routes []route
 
@@ -113,7 +126,10 @@ func NewHttpServer(
 	}
 
 	if enableHealthcheck {
-		hc := &healthChecker{logger}
+		hc := &healthChecker{logger: logger,
+			healthChecks: healthChecks,
+			deepChecks:   deepChecks,
+		}
 		routes = append(routes,
 			route{path: "/healthcheck", handler: hc.healthCheck, methods: []string{"GET"}, name: "healthcheck_get"},
 			route{path: "/deepcheck", handler: hc.deepCheck, methods: []string{"GET"}, name: "deepcheck_get"},


Test patch:
diff --git a/pkg/statsd/handler_http_forwarder_v2_test.go b/pkg/statsd/handler_http_forwarder_v2_test.go
index 1f406c4..a4ab745 100644
--- a/pkg/statsd/handler_http_forwarder_v2_test.go
+++ b/pkg/statsd/handler_http_forwarder_v2_test.go
@@ -20,9 +20,47 @@ import (
 	"github.com/atlassian/gostatsd"
 	"github.com/atlassian/gostatsd/internal/fixtures"
 	"github.com/atlassian/gostatsd/pb"
+	"github.com/atlassian/gostatsd/pkg/healthcheck"
 	"github.com/atlassian/gostatsd/pkg/transport"
 )
 
+func TestHttpForwarderDeepCheck(t *testing.T) {
+	t.Parallel()
+
+	logger := logrus.New()
+	s := httptest.NewServer(http.HandlerFunc(func(rw http.ResponseWriter, r *http.Request) {}))
+	t.Cleanup(s.Close)
+	hfh, err := NewHttpForwarderHandlerV2(
+		logger,
+		"",
+		s.URL,
+		1,
+		1,
+		1,
+		false,
+		1*time.Second,
+		1*time.Second,
+		nil,
+		nil,
+		transport.NewTransportPool(logger, viper.New()),
+	)
+	require.NoError(t, err)
+	require.NotNil(t, hfh)
+
+	deepChecks := hfh.DeepChecks()
+	assert.Len(t, deepChecks, 1)
+
+	msg, isHealthy := deepChecks[0]()
+	t.Logf("%s\n", msg)
+	assert.Equal(t, healthcheck.Unhealthy, isHealthy)
+
+	hfh.sendNop(context.Background())
+
+	msg, isHealthy = deepChecks[0]()
+	t.Logf("%s\n", msg)
+	assert.Equal(t, healthcheck.Healthy, isHealthy)
+}
+
 func TestHttpForwarderV2Translation(t *testing.T) {
 	t.Parallel()
 
diff --git a/pkg/web/http_receiver_v2_test.go b/pkg/web/http_receiver_v2_test.go
index 673571c..25fb3f3 100644
--- a/pkg/web/http_receiver_v2_test.go
+++ b/pkg/web/http_receiver_v2_test.go
@@ -41,6 +41,8 @@ func TestForwardingEndToEndV2(t *testing.T) {
 		false,
 		true,
 		false,
+		nil,
+		nil,
 	)
 	require.NoError(t, err)
 
diff --git a/pkg/web/web_test.go b/pkg/web/web_test.go
index 6af746a..22cd86a 100644
--- a/pkg/web/web_test.go
+++ b/pkg/web/web_test.go
@@ -24,6 +24,8 @@ func TestHttpServerShutsdown(t *testing.T) {
 		false,
 		false,
 		true,
+		nil,
+		nil,
 	)
 	require.NoError(t, err)