Merge pull request #135 from iksaif/healthy-ready

Add /-/ready and /-/healty

handler/handler_test.go

@@ -47,6 +47,36 @@ func (m *MockMetricStore) Shutdown() error {
 	return nil
 }
 
+func (m *MockMetricStore) Healthy() error {
+	return nil
+}
+
+func (m *MockMetricStore) Ready() error {
+	return nil
+}
+
+func TestHealthyReady(t *testing.T) {
+	mms := MockMetricStore{}
+	req, err := http.NewRequest("GET", "http://example.org/", &bytes.Buffer{})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	healthyHandler := Healthy(&mms)
+	readyHandler := Ready(&mms)
+
+	w := httptest.NewRecorder()
+	healthyHandler(w, req)
+	if expected, got := http.StatusOK, w.Code; expected != got {
+		t.Errorf("Wanted status code %v, got %v.", expected, got)
+	}
+
+	readyHandler(w, req)
+	if expected, got := http.StatusOK, w.Code; expected != got {
+		t.Errorf("Wanted status code %v, got %v.", expected, got)
+	}
+}
+
 func TestPush(t *testing.T) {
 	mms := MockMetricStore{}
 	handler := Push(&mms, false)

handler/misc.go

@@ -0,0 +1,47 @@
+// Copyright 2017 The Prometheus Authors
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package handler
+
+import (
+	"io"
+	"net/http"
+
+	"github.com/prometheus/pushgateway/storage"
+)
+
+func Healthy(
+	ms storage.MetricStore,
+) func(http.ResponseWriter, *http.Request) {
+	return func(w http.ResponseWriter, _ *http.Request) {
+		err := ms.Healthy()
+		if err == nil {
+			io.WriteString(w, "OK")
+		} else {
+			http.Error(w, err.Error(), 500)
+		}
+	}
+}
+
+func Ready(
+	ms storage.MetricStore,
+) func(http.ResponseWriter, *http.Request) {
+	return func(w http.ResponseWriter, _ *http.Request) {
+		err := ms.Ready()
+		if err == nil {
+			io.WriteString(w, "OK")
+		} else {
+			http.Error(w, err.Error(), 500)
+		}
+	}
+}

main.go

@@ -67,6 +67,9 @@ func main() {
 	// prometheus.EnableCollectChecks(true)
 
 	r := httprouter.New()
+	r.Handler("GET", "/-/healthy", prometheus.InstrumentHandlerFunc("healthy", handler.Healthy(ms)))
+	r.Handler("GET", "/-/ready", prometheus.InstrumentHandlerFunc("ready", handler.Ready(ms)))
+
 	r.Handler("GET", *metricsPath, prometheus.Handler())
 
 	// Handlers for pushing and deleting metrics.

storage/diskmetricstore.go

@@ -15,6 +15,7 @@ package storage
 
 import (
 	"encoding/gob"
+	"fmt"
 	"io"
 	"io/ioutil"
 	"os"
@@ -61,6 +62,8 @@ func NewDiskMetricStore(
 	persistenceFile string,
 	persistenceInterval time.Duration,
 ) *DiskMetricStore {
+	// TODO: Do that outside of the constructor to allow the HTTP server to
+	//  serve /-/healthy and /-/ready earlier.
 	dms := &DiskMetricStore{
 		writeQueue:      make(chan WriteRequest, writeQueueCapacity),
 		drain:           make(chan struct{}),
@@ -134,6 +137,26 @@ func (dms *DiskMetricStore) Shutdown() error {
 	return <-dms.done
 }
 
+// Healthy implements the MetricStore interface.
+func (dms *DiskMetricStore) Healthy() error {
+	// By taking the lock we check that there is no deadlock.
+	dms.lock.Lock()
+	defer dms.lock.Unlock()
+
+	// A pushgateway that cannot be written to should not be
+	// considered as healthy.
+	if len(dms.writeQueue) == cap(dms.writeQueue) {
+		return fmt.Errorf("write queue is full")
+	}
+
+	return nil
+}
+
+// Ready implements the MetricStore interface.
+func (dms *DiskMetricStore) Ready() error {
+	return dms.Healthy()
+}
+
 func (dms *DiskMetricStore) loop(persistenceInterval time.Duration) {
 	lastPersist := time.Now()
 	persistScheduled := false

storage/diskmetricstore_test.go

@@ -538,6 +538,14 @@ func TestNoPersistence(t *testing.T) {
 	if err := checkMetricFamilies(dms); err != nil {
 		t.Error(err)
 	}
+
+	if err := dms.Ready(); err != nil {
+		t.Error(err)
+	}
+
+	if err := dms.Healthy(); err != nil {
+		t.Error(err)
+	}
 }
 
 func checkMetricFamilies(dms *DiskMetricStore, expectedMFs ...*dto.MetricFamily) error {

storage/interface.go

@@ -56,6 +56,13 @@ type MetricStore interface {
 	// undefinded state). If nil is returned, the MetricStore cannot be
 	// "restarted" again, but it can still be used for read operations.
 	Shutdown() error
+	// Healthy returns nil if the MetricStore is currently working as
+	// expected or false, Error if it is not.
+	Healthy() error
+	// Ready returns nil if the MetricStore is ready to be used (all files
+	// are opened and checkpoints have been restored) or false, Error if it
+	// is not.
+	Ready() error
 }
 
 // WriteRequest is a request to change the MetricStore, i.e. to process it, a