Make the debug metrics endpoints leader aware

pkg/http/api.go

@@ -2,7 +2,9 @@ package http
 
 import (
 	"encoding/json"
+	"expvar"
 	"fmt"
+	"io"
 	"net/http"
 	"net/http/pprof"
 	"os"
@@ -42,11 +44,13 @@ type API interface {
 	RecentApps(w http.ResponseWriter, r *http.Request, _ httprouter.Params)
 	Help(w http.ResponseWriter, r *http.Request, _ httprouter.Params)
 	MemcacheConfig(w http.ResponseWriter, r *http.Request, _ httprouter.Params)
+	Metrics(w http.ResponseWriter, r *http.Request, _ httprouter.Params)
 }
 
 var endpoints = []string{} // known API URIs
 
 var okIfNotExistsFlags = &throttle.CheckFlags{OKIfNotExists: true}
+var metricsHandler = exp.ExpHandler(metrics.DefaultRegistry)
 
 type GeneralResponse struct {
 	StatusCode int
@@ -355,9 +359,26 @@ func register(router *httprouter.Router, path string, f httprouter.Handle) {
 	endpoints = append(endpoints, path)
 }
 
-func metricsHandle(w http.ResponseWriter, r *http.Request, p httprouter.Params) {
-	handler := exp.ExpHandler(metrics.DefaultRegistry)
-	handler.ServeHTTP(w, r)
+type discardWriter struct{ io.Writer }
+
+func (discardWriter) Header() http.Header { return make(http.Header) }
+func (discardWriter) WriteHeader(int)     {}
+
+// Metrics exports process metrics, excluding aggregate and probe metrics on followers.
+// Filtering happens here because published expvar values cannot be removed.
+func (api *APIImpl) Metrics(w http.ResponseWriter, r *http.Request, _ httprouter.Params) {
+	metricsHandler.ServeHTTP(discardWriter{Writer: io.Discard}, r)
+	isLeader := api.consensusService != nil && api.consensusService.IsLeader()
+
+	w.Header().Set("Content-Type", "application/json; charset=utf-8")
+	metricValues := make(map[string]json.RawMessage)
+	expvar.Do(func(metric expvar.KeyValue) {
+		if !isLeader && (strings.HasPrefix(metric.Key, "aggregated.") || strings.HasPrefix(metric.Key, "probes.")) {
+			return
+		}
+		metricValues[metric.Key] = json.RawMessage(metric.Value.String())
+	})
+	json.NewEncoder(w).Encode(metricValues)
 }
 
 func (api *APIImpl) SkipHost(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
@@ -431,8 +452,8 @@ func ConfigureRoutes(api API) *httprouter.Router {
 	register(router, "/skipped-hosts", api.SkippedHosts)
 	register(router, "/recover-host/:hostName", api.RecoverHost)
 
-	register(router, "/debug/vars", metricsHandle)
-	register(router, "/debug/metrics", metricsHandle)
+	register(router, "/debug/vars", api.Metrics)
+	register(router, "/debug/metrics", api.Metrics)
 
 	if config.Settings().EnableProfiling {
 		router.HandlerFunc(http.MethodGet, "/debug/pprof/", pprof.Index)

pkg/http/api_test.go

@@ -2,14 +2,26 @@ package http
 
 import (
 	"encoding/json"
+	"expvar"
 	"net/http"
 	"net/http/httptest"
 	"reflect"
 	"testing"
 
 	"github.com/github/freno/pkg/config"
+	"github.com/github/freno/pkg/group"
+	metrics "github.com/rcrowley/go-metrics"
 )
 
+type metricsConsensusService struct {
+	group.ConsensusService
+	isLeader bool
+}
+
+func (s *metricsConsensusService) IsLeader() bool {
+	return s.isLeader
+}
+
 func TestLbCheck(t *testing.T) {
 	api := NewAPIImpl(nil, nil)
 	recorder := httptest.NewRecorder()
@@ -37,6 +49,7 @@ func TestRoutes(t *testing.T) {
 	}{
 		{http.MethodGet, "/lb-check", http.StatusOK},
 		{http.MethodGet, "/config/memcache", http.StatusOK},
+		{http.MethodGet, "/debug/metrics", http.StatusOK},
 	}
 	for _, route := range expectedRoutes {
 		r, _ := http.NewRequest(route.verb, route.path, nil)
@@ -101,3 +114,50 @@ func TestMemcacheConfigWhenDefault(t *testing.T) {
 		t.Errorf("Expected MemcacheConfig body to be %s, but it's %s", expected, body)
 	}
 }
+
+func TestMetricsFiltersAggregateAndProbeMetricsFromFollowers(t *testing.T) {
+	const (
+		aggregatedMetricName = "aggregated.mysql.metrics-export-test"
+		probeMetricName      = "probes.total.metrics-export-test"
+		processMetricName    = "consensus.metrics-export-test"
+	)
+
+	metricNames := []string{aggregatedMetricName, probeMetricName, processMetricName}
+	for _, metricName := range metricNames {
+		metrics.DefaultRegistry.Unregister(metricName)
+		defer metrics.DefaultRegistry.Unregister(metricName)
+	}
+
+	metrics.GetOrRegisterGaugeFloat64(aggregatedMetricName, nil).Update(42)
+	metrics.GetOrRegisterCounter(probeMetricName, nil).Inc(7)
+	metrics.GetOrRegisterGauge(processMetricName, nil).Update(1)
+
+	readMetrics := func(isLeader bool) map[string]interface{} {
+		recorder := httptest.NewRecorder()
+		api := NewAPIImpl(nil, &metricsConsensusService{isLeader: isLeader})
+		api.Metrics(recorder, httptest.NewRequest(http.MethodGet, "/debug/metrics", nil), nil)
+		if contentType := recorder.Header().Get("Content-Type"); contentType != "application/json; charset=utf-8" {
+			t.Errorf("Unexpected content type: %s", contentType)
+		}
+		result := make(map[string]interface{})
+		if err := json.Unmarshal(recorder.Body.Bytes(), &result); err != nil {
+			t.Fatal(err)
+		}
+		return result
+	}
+
+	leaderMetrics := readMetrics(true)
+	metrics.DefaultRegistry.Unregister(aggregatedMetricName)
+	metrics.DefaultRegistry.Unregister(probeMetricName)
+	if expvar.Get(aggregatedMetricName) == nil || expvar.Get(probeMetricName) == nil {
+		t.Fatal("Expected aggregate and probe metrics to remain in expvar after unregister")
+	}
+
+	followerMetrics := readMetrics(false)
+	if leaderMetrics[aggregatedMetricName] == nil || leaderMetrics[probeMetricName] == nil || leaderMetrics[processMetricName] == nil {
+		t.Errorf("Unexpected leader metrics: %v", leaderMetrics)
+	}
+	if followerMetrics[aggregatedMetricName] != nil || followerMetrics[probeMetricName] != nil || followerMetrics[processMetricName] == nil {
+		t.Errorf("Unexpected follower metrics: %v", followerMetrics)
+	}
+}