Merge pull request G-Research#40 from G-Research/healthchecks

Makes the existing Info healthcheck used by Thanos call the query API to assert that OpenTSDB is ready to query.
Adds grpc healthcheck service endpoint.
Adds grpc_health_probe to docker image, using wget.

Dockerfile

@@ -7,6 +7,9 @@ RUN go build -mod=vendor -ldflags '-extldflags "-static"' -o geras ./cmd/geras/m
 
 # final stage
 FROM alpine
-WORKDIR /app
-COPY --from=build-env /src/geras /app/
+WORKDIR /bin
+COPY --from=build-env /src/geras /bin/
+RUN GRPC_HEALTH_PROBE_VERSION=v0.3.0 && \
+    wget -qO/bin/grpc_health_probe https://github.com/grpc-ecosystem/grpc-health-probe/releases/download/${GRPC_HEALTH_PROBE_VERSION}/grpc_health_probe-linux-amd64 && \
+    chmod +x /bin/grpc_health_probe
 ENTRYPOINT ["./geras"]

README.md

@@ -33,6 +33,8 @@ After the build you will have a self-contained binary (`geras`). It writes logs
         Log format. One of [logfmt, json] (default "logfmt")
   -log.level string
         Log filtering level. One of [debug, info, warn, error] (default "error")
+  -healthcheck-metric
+        A metric to query as a readiness health check (default "tsd.rpc.recieved")
   -metrics-refresh-interval duration
         Time between metric name refreshes. Use negative duration to disable refreshes. (default 15m0s)
   -metrics-suggestions

cmd/geras/main.go

@@ -22,6 +22,7 @@ import (
 	"github.com/thanos-io/thanos/pkg/store/storepb"
 	"golang.org/x/net/trace"
 	"google.golang.org/grpc"
+	healthpb "google.golang.org/grpc/health/grpc_health_v1"
 )
 
 func NewConfiguredLogger(format string, logLevel string) (log.Logger, error) {
@@ -97,6 +98,7 @@ func main() {
 	traceDumpBody := flag.Bool("trace-dumpbody", false, "Include TSDB request and response bodies in traces (can be expensive)")
 	logFormat := flag.String("log.format", "logfmt", "Log format. One of [logfmt, json]")
 	logLevel := flag.String("log.level", "error", "Log filtering level. One of [debug, info, warn, error]")
+	healthcheckMetric := flag.String("healthcheck-metric", "tsd.rpc.received", "A metric to query as a readiness health check.")
 	openTSDBAddress := flag.String("opentsdb-address", "", "<host>:<port>")
 	refreshInterval := flag.Duration("metrics-refresh-interval", time.Minute*15,
 		"Time between metric name refreshes. Use negative duration to disable refreshes.")
@@ -170,9 +172,10 @@ func main() {
 	http.Handle("/metrics", promhttp.Handler())
 
 	// create openTSDBStore and expose its api on a grpc server
-	srv := store.NewOpenTSDBStore(logger, client, prometheus.DefaultRegisterer, *refreshInterval, storeLabels, allowedMetricNames, blockedMetricNames, *enableMetricSuggestions)
+	srv := store.NewOpenTSDBStore(logger, client, prometheus.DefaultRegisterer, *refreshInterval, storeLabels, allowedMetricNames, blockedMetricNames, *enableMetricSuggestions, *healthcheckMetric)
 	grpcSrv := grpc.NewServer()
 	storepb.RegisterStoreServer(grpcSrv, srv)
+	healthpb.RegisterHealthServer(grpcSrv, srv)
 	l, err := net.Listen("tcp", *grpcListenAddr)
 	if err != nil {
 		level.Error(logger).Log("err", err)

pkg/store/store.go

@@ -18,6 +18,7 @@ import (
 	"github.com/thanos-io/thanos/pkg/store/storepb"
 	"golang.org/x/net/trace"
 	"google.golang.org/grpc/codes"
+	healthpb "google.golang.org/grpc/health/grpc_health_v1"
 	"google.golang.org/grpc/status"
 
 	opentsdb "github.com/G-Research/opentsdb-goclient/client"
@@ -33,9 +34,10 @@ type OpenTSDBStore struct {
 	allowedMetricNames, blockedMetricNames *regexp.Regexp
 	enableMetricSuggestions                bool
 	storeLabels                            []storepb.Label
+	healthcheckMetric                      string
 }
 
-func NewOpenTSDBStore(logger log.Logger, client opentsdb.ClientContext, reg prometheus.Registerer, interval time.Duration, storeLabels []storepb.Label, allowedMetricNames, blockedMetricNames *regexp.Regexp, enableMetricSuggestions bool) *OpenTSDBStore {
+func NewOpenTSDBStore(logger log.Logger, client opentsdb.ClientContext, reg prometheus.Registerer, interval time.Duration, storeLabels []storepb.Label, allowedMetricNames, blockedMetricNames *regexp.Regexp, enableMetricSuggestions bool, healthcheckMetric string) *OpenTSDBStore {
 	store := &OpenTSDBStore{
 		logger:                  log.With(logger, "component", "opentsdb"),
 		openTSDBClient:          client,
@@ -45,6 +47,7 @@ func NewOpenTSDBStore(logger log.Logger, client opentsdb.ClientContext, reg prom
 		storeLabels:             storeLabels,
 		allowedMetricNames:      allowedMetricNames,
 		blockedMetricNames:      blockedMetricNames,
+		healthcheckMetric:       healthcheckMetric,
 	}
 	store.updateMetrics(context.Background(), logger)
 	return store
@@ -119,13 +122,30 @@ func (store *OpenTSDBStore) Info(
 		Labels:  store.storeLabels,
 	}
 	var err error
-	store.timedTSDBOp("version", func() error {
-		_, err = store.openTSDBClient.WithContext(ctx).Version()
+	store.timedTSDBOp("query", func() error {
+		now := time.Now().Unix()
+		q := opentsdb.QueryParam{
+			Start: now,
+			End:   now + 1,
+			Queries: []opentsdb.SubQuery{{
+				Metric:     store.healthcheckMetric,
+				Aggregator: "sum",
+			}},
+		}
+		_, err = store.openTSDBClient.WithContext(ctx).Query(q)
 		return err
 	})
 	return &res, err
 }
 
+func (store OpenTSDBStore) Check(ctx context.Context, req *healthpb.HealthCheckRequest) (*healthpb.HealthCheckResponse, error) {
+	return &healthpb.HealthCheckResponse{Status: healthpb.HealthCheckResponse_SERVING}, nil
+}
+
+func (store OpenTSDBStore) Watch(req *healthpb.HealthCheckRequest, srv healthpb.Health_WatchServer) error {
+	return status.Errorf(codes.Unimplemented, "method Watch not implemented")
+}
+
 func (store *OpenTSDBStore) Series(
 	req *storepb.SeriesRequest,
 	server storepb.Store_SeriesServer) error {