Issue #41: Cleanup metrics for deleted routes

fabio.properties

@@ -233,6 +233,7 @@
 # The default is
 #
 # metrics.target =
+metrics.target = stdout
 
 
 # metrics.prefix configures the prefix of all reported metrics.
@@ -254,6 +255,7 @@
 # The default is
 #
 # metrics.interval = 30s
+metrics.interval = 5s
 
 
 # metrics.graphite.addr configures the host:port of the Graphite

metrics/metrics.go

@@ -18,6 +18,11 @@ import (
 
 var pfx string
 
+// ServiceRegistry contains a separate metrics registry for
+// the timers for all targets to avoid conflicts
+// with globally registered timers.
+var ServiceRegistry = gometrics.NewRegistry()
+
 func Init(cfgs []config.Metrics) error {
 	for _, cfg := range cfgs {
 		if err := initMetrics(cfg); err != nil {
@@ -85,6 +90,7 @@ func defaultPrefix() string {
 func initStdout(interval time.Duration) error {
 	logger := log.New(os.Stderr, "localhost: ", log.Lmicroseconds)
 	go gometrics.Log(gometrics.DefaultRegistry, interval, logger)
+	go gometrics.Log(ServiceRegistry, interval, logger)
 	return nil
 }
 
@@ -95,5 +101,6 @@ func initGraphite(addr string, interval time.Duration) error {
 	}
 
 	go graphite.Graphite(gometrics.DefaultRegistry, interval, pfx, a)
+	go graphite.Graphite(ServiceRegistry, interval, pfx, a)
 	return nil
 }

route/route.go

@@ -49,9 +49,9 @@ func (r *Route) addTarget(service string, targetURL *url.URL, fixedWeight float6
 	}
 
 	name := metrics.TargetName(service, r.Host, r.Path, targetURL)
-	timer := gometrics.GetOrRegisterTimer(name, gometrics.DefaultRegistry)
+	timer := gometrics.GetOrRegisterTimer(name, metrics.ServiceRegistry)
 
-	t := &Target{Service: service, Tags: tags, URL: targetURL, FixedWeight: fixedWeight, Timer: timer}
+	t := &Target{Service: service, Tags: tags, URL: targetURL, FixedWeight: fixedWeight, Timer: timer, timerName: name}
 	r.Targets = append(r.Targets, t)
 	r.weighTargets()
 }

route/table.go

@@ -8,33 +8,81 @@ import (
 	"net/url"
 	"sort"
 	"strings"
+	"sync"
 	"sync/atomic"
-)
 
-// active stores the current routing table. Should never be nil.
-var active atomic.Value
+	"github.com/eBay/fabio/metrics"
+)
 
 var errInvalidPrefix = errors.New("route: prefix must not be empty")
 var errInvalidTarget = errors.New("route: target must not be empty")
 var errNoMatch = errors.New("route: no target match")
 
+// table stores the active routing table. Must never be nil.
+var table atomic.Value
+
+// init initializes the routing table.
 func init() {
-	active.Store(make(Table))
+	table.Store(make(Table))
 }
 
+// GetTable returns the active routing table. The function
+// is safe to be called from multiple goroutines and the
+// value is never nil.
 func GetTable() Table {
-	return active.Load().(Table)
+	return table.Load().(Table)
 }
 
+// mu guards table and registry in SetTable.
+var mu sync.Mutex
+
+// SetTable sets the active routing table. A nil value
+// logs a warning and is ignored. The function is safe
+// to be called from multiple goroutines.
 func SetTable(t Table) {
 	if t == nil {
 		log.Print("[WARN] Ignoring nil routing table")
 		return
 	}
-	active.Store(t)
+	mu.Lock()
+	table.Store(t)
+	syncRegistry(t)
+	mu.Unlock()
 	log.Printf("[INFO] Updated config to\n%s", t)
 }
 
+// syncRegistry unregisters all inactive timers.
+// It assumes that all timers of the table have
+// already been registered.
+func syncRegistry(t Table) {
+	timers := map[string]bool{}
+
+	// get all registered timers
+	metrics.ServiceRegistry.Each(func(name string, m interface{}) {
+		timers[name] = false
+	})
+
+	// mark the ones from this table as active.
+	// this can also add new entries but we do not
+	// really care since we are only interested in the
+	// inactive ones.
+	for _, routes := range t {
+		for _, r := range routes {
+			for _, tg := range r.Targets {
+				timers[tg.timerName] = true
+			}
+		}
+	}
+
+	// unregister inactive timers
+	for name, active := range timers {
+		if !active {
+			metrics.ServiceRegistry.Unregister(name)
+			log.Printf("[INFO] Unregistered timer %s", name)
+		}
+	}
+}
+
 // Table contains a set of routes grouped by host.
 // The host routes are sorted from most to least specific
 // by sorting the routes in reverse order by path.

route/table_registry_test.go

@@ -0,0 +1,35 @@
+package route
+
+import (
+	"reflect"
+	"sort"
+	"testing"
+
+	"github.com/eBay/fabio/metrics"
+)
+
+func TestSyncRegistry(t *testing.T) {
+	names := func() []string {
+		var n []string
+		metrics.ServiceRegistry.Each(func(name string, x interface{}) {
+			n = append(n, name)
+		})
+		sort.Strings(n)
+		return n
+	}
+
+	metrics.ServiceRegistry.UnregisterAll()
+
+	tbl := make(Table)
+	tbl.AddRoute("svc-a", "/aaa", "http://localhost:1234", 1, nil)
+	tbl.AddRoute("svc-b", "/bbb", "http://localhost:5678", 1, nil)
+	if got, want := names(), []string{"svc-a._./aaa.localhost_1234", "svc-b._./bbb.localhost_5678"}; !reflect.DeepEqual(got, want) {
+		t.Fatalf("got %v want %v", got, want)
+	}
+
+	tbl.DelRoute("svc-b", "/bbb", "http://localhost:5678")
+	syncRegistry(tbl)
+	if got, want := names(), []string{"svc-a._./aaa.localhost_1234"}; !reflect.DeepEqual(got, want) {
+		t.Fatalf("got %v want %v", got, want)
+	}
+}

route/target.go

@@ -25,4 +25,7 @@ type Target struct {
 
 	// Timer measures throughput and latency of this target
 	Timer gometrics.Timer
+
+	// timerName is the name of the timer in the metrics registry
+	timerName string
 }