Skip to content

Commit

Permalink
Refactor metrics
Browse files Browse the repository at this point in the history 
This patch refactors the metrics support within fabio to
achieve several goals:

 * support labels for DogstatD, Prometheus and others
 * support raw events for statsd and others without aggregation
 * support multiple backends simultaneously to support migration
 * make integration of new metrics backends and/or libraries simple
 * being fully backwards compatible to not break existing setups

One of the main challenges is that fabio generates names for the metrics
targets through templates and that these templates are configurable. A
metrics backend which supports labels needs to generate a different name
for the same metric than one that does not support labels.

Combining this with the need to support multiple different metrics
backends at runtime means that none of the existing metrics libraries
like go-kit/metrics and hashicorp/go-metrics is a good fit. However,
they can be used as drivers which reduces the need for testing and makes
integration of new backends simpler since fabio does not need to rely on
a single metrics library.

This patch is work in progress with the new metrics code in the metrics4
package and the old metrics code in the 'metrics_old' package where it
is kept for reference until it has been migrated or removed. The basic
architecture should be good enough to add more providers and
functionality so that the community can help.

Right now there are two providers "flat" and "label" to demonstrate the
concepts. They provide counter and timers in statsd/dogstatd format and
can be configured with -metrics.target=flat[,label]. Other providers
should be added in the same way.

The metrics_old code should be converted to the new provider scheme.

The go-kit/metrics library currently supports most of the necessary
drivers and is the preferred way of integrating new drivers.

The list below is a probably incomplete list of things that need to be
done:

 * The configuration is not used except for metrics.target
 * The drivers from the metrics_old package need to be migrated
 * Additional drivers from go-kit need to be added
 * The old rcrowley/go-metrics code relies on 'registries' for
   namespacing. After the routing table has changed the registry needs
   to be pruned of services which no longer exist so that go-metrics
   stops reporting them.
 * The RX/TX counters in the TCP and TCP+SNI proxy are probably
   sub-optimal or broken.
 * Some counters may not have been initialized properly, e.g. WSConn
 * WSConn should probably be a gauge which also needs to be set to 0 on
   startup.
 * The approach of injecting a noop metrics provider if none has been
   set has already bitten me once since some metrics are reported while
   others are not. I need to think about this a bit more.

Fixes #126
Fixes #165
Fixes #211
Fixes #253
Fixes #326
Fixes #327
Fixes #371
Closes #329
Closes #331
Closes #334
Closes #335

Remove 'Register' methods and create objects directly.

Add Gauge and use it for websocket connections

Vendoring in github.com/alexcesaro/statsd

First stab at integrating raw statsd from #335

Rebase off master 10/23/2020 - NJ
  • Loading branch information
@magiconair @nathanejohnson
magiconair authored and nathanejohnson committed Oct 23, 2020
1 parent 231e7f2 commit c34282f
Show file tree
Hide file tree
Showing 36 changed files with 1,474 additions and 242 deletions.
4 changes: 2 additions & 2 deletions admin/api/routes.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -59,8 +59,8 @@ func (h *RoutesHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
Weight: tg.Weight,
Tags: tg.Tags,
Cmd: "route add",
Rate1: tg.Timer.Rate1(),
Pct99: tg.Timer.Percentile(0.99),
// Rate1: tg.Timer.Rate1(),
// Pct99: tg.Timer.Percentile(0.99),
}
routes = append(routes, ar)
}
Expand Down
2 changes: 2 additions & 0 deletions go.mod
View file
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@ require (
github.com/DataDog/datadog-go v0.0.0-20180822151419-281ae9f2d895 // indirect
github.com/Shopify/sarama v1.19.0 // indirect
github.com/Shopify/toxiproxy v2.1.3+incompatible // indirect
github.com/alexcesaro/statsd v2.0.0+incompatible
github.com/apache/thrift v0.0.0-20181028152738-da1169d75b15 // indirect
github.com/armon/go-metrics v0.0.0-20180917152333-f0300d1749da // indirect
github.com/armon/go-proxyproto v0.0.0-20180202201750-5b7edb60ff5f
Expand Down Expand Up @@ -102,6 +103,7 @@ require (
google.golang.org/appengine v1.3.0 // indirect
google.golang.org/genproto v0.0.0-20190219182410-082222b4a5c5 // indirect
google.golang.org/grpc v1.16.0
gopkg.in/alexcesaro/statsd.v2 v2.0.0 // indirect
gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 // indirect
gopkg.in/vmihailenco/msgpack.v2 v2.9.1 // indirect
labix.org/v2/mgo v0.0.0-20140701140051-000000000287 // indirect
Expand Down
4 changes: 4 additions & 0 deletions go.sum
View file
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,8 @@ github.com/Shopify/sarama v1.19.0 h1:9oksLxC6uxVPHPVYUmq6xhr1BOF/hHobWH2UzO67z1s
github.com/Shopify/sarama v1.19.0/go.mod h1:FVkBWblsNy7DGZRfXLU0O9RCGt5g3g3yEuWXgklEdEo=
github.com/Shopify/toxiproxy v2.1.3+incompatible h1:awiJqUYH4q4OmoBiRccJykjd7B+w0loJi2keSna4X/M=
github.com/Shopify/toxiproxy v2.1.3+incompatible/go.mod h1:OXgGpZ6Cli1/URJOF1DMxUHB2q5Ap20/P/eIdh4G0pI=
github.com/alexcesaro/statsd v2.0.0+incompatible h1:HG17k1Qk8V1F4UOoq6tx+IUoAbOcI5PHzzEUGeDD72w=
github.com/alexcesaro/statsd v2.0.0+incompatible/go.mod h1:vNepIbQAiyLe1j480173M6NYYaAsGwEcvuDTU3OCUGY=
github.com/apache/thrift v0.0.0-20181028152738-da1169d75b15 h1:GGI45yUJ942DZsI3uodzVabeRUbxPsioScY/EmmE5ek=
github.com/apache/thrift v0.0.0-20181028152738-da1169d75b15/go.mod h1:cp2SuWMxlEZw2r+iP2GNCdIi4C1qmUzdZFSVb+bacwQ=
github.com/armon/go-metrics v0.0.0-20180917152333-f0300d1749da h1:8GUt8eRujhVEGZFFEjBj46YV4rDjvGrNxb0KMWYkL2I=
Expand Down Expand Up @@ -281,6 +283,8 @@ google.golang.org/grpc v1.14.0/go.mod h1:yo6s7OP7yaDglbqo1J04qKzAhqBH6lvTonzMVmE
google.golang.org/grpc v1.15.0/go.mod h1:0JHn/cJsOMiMfNA9+DeHDlAU7KAAB5GDlYFpa9MZMio=
google.golang.org/grpc v1.16.0 h1:dz5IJGuC2BB7qXR5AyHNwAUBhZscK2xVez7mznh72sY=
google.golang.org/grpc v1.16.0/go.mod h1:0JHn/cJsOMiMfNA9+DeHDlAU7KAAB5GDlYFpa9MZMio=
gopkg.in/alexcesaro/statsd.v2 v2.0.0 h1:FXkZSCZIH17vLCO5sO2UucTHsH9pc+17F6pl3JVCwMc=
gopkg.in/alexcesaro/statsd.v2 v2.0.0/go.mod h1:i0ubccKGzBVNBpdGV5MocxyA/XlLUJzA7SLonnE4drU=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 h1:qIbj1fsPNlZgppZ+VLlY7N33q108Sa+fhmuc+sWQYwY=
gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
Expand Down
159 changes: 97 additions & 62 deletions main.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,10 @@ import (
"github.com/fabiolb/fabio/config"
"github.com/fabiolb/fabio/exit"
"github.com/fabiolb/fabio/logger"
"github.com/fabiolb/fabio/metrics"
"github.com/fabiolb/fabio/metrics4"
"github.com/fabiolb/fabio/metrics4/flat"
"github.com/fabiolb/fabio/metrics4/label"
"github.com/fabiolb/fabio/metrics4/statsdraw"
"github.com/fabiolb/fabio/noroute"
"github.com/fabiolb/fabio/proxy"
"github.com/fabiolb/fabio/proxy/tcp"
Expand Down Expand Up @@ -120,9 +123,7 @@ func main() {
registry.Default.DeregisterAll()
})

// init metrics early since that create the global metric registries
// that are used by other parts of the code.
initMetrics(cfg)
metrics := initMetrics(cfg)
initRuntime(cfg)
initBackend(cfg)

Expand All @@ -134,12 +135,12 @@ func main() {
go watchNoRouteHTML(cfg)

first := make(chan bool)
go watchBackend(cfg, first)
go watchBackend(cfg, metrics, first)
log.Print("[INFO] Waiting for first routing table")
<-first

// create proxies after metrics since they use the metrics registry.
startServers(cfg)
startServers(cfg, metrics)

// warn again so that it is visible in the terminal
WarnIfRunAsRoot(cfg.Insecure)
Expand All @@ -148,15 +149,16 @@ func main() {
log.Print("[INFO] Down")
}

func newGrpcProxy(cfg *config.Config, tlscfg *tls.Config) []grpc.ServerOption {
func newGrpcProxy(cfg *config.Config, tlscfg *tls.Config, stats metrics4.Provider) []grpc.ServerOption {

//Init Glob Cache
globCache := route.NewGlobCache(cfg.GlobCacheSize)

statsHandler := &proxy.GrpcStatsHandler{
Connect: metrics.DefaultRegistry.GetCounter("grpc.conn"),
Request: metrics.DefaultRegistry.GetTimer("grpc.requests"),
NoRoute: metrics.DefaultRegistry.GetCounter("grpc.noroute"),
Connect: stats.NewCounter("grpc.conn"),
Request: stats.NewTimer("grpc.requests"),
NoRoute: stats.NewCounter("grpc.noroute"),
Metrics: stats,
}

proxyInterceptor := proxy.GrpcProxyInterceptor{
Expand All @@ -175,7 +177,7 @@ func newGrpcProxy(cfg *config.Config, tlscfg *tls.Config) []grpc.ServerOption {
}
}

func newHTTPProxy(cfg *config.Config) http.Handler {
func newHTTPProxy(cfg *config.Config, stats metrics4.Provider) *proxy.HTTPProxy {
var w io.Writer

//Init Glob Cache
Expand Down Expand Up @@ -206,7 +208,7 @@ func newHTTPProxy(cfg *config.Config) http.Handler {

pick := route.Picker[cfg.Proxy.Strategy]
match := route.Matcher[cfg.Proxy.Matcher]
notFound := metrics.DefaultRegistry.GetCounter("notfound")
notFound := stats.NewCounter("notfound")
log.Printf("[INFO] Using routing strategy %q", cfg.Proxy.Strategy)
log.Printf("[INFO] Using route matching %q", cfg.Proxy.Matcher)

Expand Down Expand Up @@ -235,26 +237,27 @@ func newHTTPProxy(cfg *config.Config) http.Handler {
Lookup: func(r *http.Request) *route.Target {
t := route.GetTable().Lookup(r, r.Header.Get("trace"), pick, match, globCache, cfg.GlobMatchingDisabled)
if t == nil {
notFound.Inc(1)
notFound.Count(1)
log.Print("[WARN] No route for ", r.Host, r.URL)
}
return t
},
Requests: metrics.DefaultRegistry.GetTimer("requests"),
Noroute: metrics.DefaultRegistry.GetCounter("notfound"),
Requests: stats.NewTimer("requests"),
Noroute: stats.NewCounter("notfound"),
Logger: l,
TracerCfg: cfg.Tracing,
AuthSchemes: authSchemes,
WSConn: stats.NewGauge("ws.conn"),
Metrics: stats,
}
}

func lookupHostFn(cfg *config.Config) func(string) *route.Target {
func lookupHostFn(cfg *config.Config, notFound metrics4.Counter) func(string) *route.Target {
pick := route.Picker[cfg.Proxy.Strategy]
notFound := metrics.DefaultRegistry.GetCounter("notfound")
return func(host string) *route.Target {
t := route.GetTable().LookupHost(host, pick)
if t == nil {
notFound.Inc(1)
notFound.Count(1)
log.Print("[WARN] No route for ", host)
}
return t
Expand Down Expand Up @@ -321,7 +324,7 @@ func startAdmin(cfg *config.Config) {
}()
}

func startServers(cfg *config.Config) {
func startServers(cfg *config.Config, stats metrics4.Provider) {
for _, l := range cfg.Listen {
l := l // capture loop var for go routines below
tlscfg, err := makeTLSConfig(l)
Expand All @@ -334,17 +337,20 @@ func startServers(cfg *config.Config) {
log.Printf("[INFO] Client certificate authentication enabled on %s", l.Addr)
}

notFound := stats.NewCounter("notfound")
switch l.Proto {
case "http", "https":
go func() {
h := newHTTPProxy(cfg)
h := newHTTPProxy(cfg, stats)
// reset the ws.conn gauge
h.WSConn.Update(0)
if err := proxy.ListenAndServeHTTP(l, h, tlscfg); err != nil {
exit.Fatal("[FATAL] ", err)
}
}()
case "grpc", "grpcs":
go func() {
h := newGrpcProxy(cfg, tlscfg)
h := newGrpcProxy(cfg, tlscfg, stats)
if err := proxy.ListenAndServeGRPC(l, h, tlscfg); err != nil {
exit.Fatal("[FATAL] ", err)
}
Expand All @@ -353,10 +359,11 @@ func startServers(cfg *config.Config) {
go func() {
h := &tcp.Proxy{
DialTimeout: cfg.Proxy.DialTimeout,
Lookup: lookupHostFn(cfg),
Conn: metrics.DefaultRegistry.GetCounter("tcp.conn"),
ConnFail: metrics.DefaultRegistry.GetCounter("tcp.connfail"),
Noroute: metrics.DefaultRegistry.GetCounter("tcp.noroute"),
Lookup: lookupHostFn(cfg, notFound),
Conn: stats.NewCounter("tcp.conn"),
ConnFail: stats.NewCounter("tcp.connfail"),
Noroute: stats.NewCounter("tcp.noroute"),
Metrics: stats,
}
if err := proxy.ListenAndServeTCP(l, h, tlscfg); err != nil {
exit.Fatal("[FATAL] ", err)
Expand All @@ -366,10 +373,11 @@ func startServers(cfg *config.Config) {
go func() {
h := &tcp.SNIProxy{
DialTimeout: cfg.Proxy.DialTimeout,
Lookup: lookupHostFn(cfg),
Conn: metrics.DefaultRegistry.GetCounter("tcp_sni.conn"),
ConnFail: metrics.DefaultRegistry.GetCounter("tcp_sni.connfail"),
Noroute: metrics.DefaultRegistry.GetCounter("tcp_sni.noroute"),
Lookup: lookupHostFn(cfg, notFound),
Conn: stats.NewCounter("tcp_sni.conn"),
ConnFail: stats.NewCounter("tcp_sni.connfail"),
Noroute: stats.NewCounter("tcp_sni.noroute"),
Metrics: stats,
}
if err := proxy.ListenAndServeTCP(l, h, tlscfg); err != nil {
exit.Fatal("[FATAL] ", err)
Expand Down Expand Up @@ -408,10 +416,11 @@ func startServers(cfg *config.Config) {
go func() {
h := &tcp.DynamicProxy{
DialTimeout: cfg.Proxy.DialTimeout,
Lookup: lookupHostFn(cfg),
Conn: metrics.DefaultRegistry.GetCounter("tcp.conn"),
ConnFail: metrics.DefaultRegistry.GetCounter("tcp.connfail"),
Noroute: metrics.DefaultRegistry.GetCounter("tcp.noroute"),
Lookup: lookupHostFn(cfg, notFound),
Conn: stats.NewCounter("tcp.conn"),
ConnFail: stats.NewCounter("tcp.connfail"),
Noroute: stats.NewCounter("tcp.noroute"),
Metrics: stats,
}
l.Addr = port
if err := proxy.ListenAndServeTCP(l, h, tlscfg); err != nil {
Expand All @@ -423,13 +432,14 @@ func startServers(cfg *config.Config) {
}()
case "https+tcp+sni":
go func() {
hp := newHTTPProxy(cfg)
hp := newHTTPProxy(cfg, stats)
tp := &tcp.SNIProxy{
DialTimeout: cfg.Proxy.DialTimeout,
Lookup: lookupHostFn(cfg),
Conn: metrics.DefaultRegistry.GetCounter("tcp_sni.conn"),
ConnFail: metrics.DefaultRegistry.GetCounter("tcp_sni.connfail"),
Noroute: metrics.DefaultRegistry.GetCounter("tcp_sni.noroute"),
Lookup: lookupHostFn(cfg, notFound),
Conn: stats.NewCounter("tcp_sni.conn"),
ConnFail: stats.NewCounter("tcp_sni.connfail"),
Noroute: stats.NewCounter("tcp_sni.noroute"),
Metrics: stats,
}
if err := proxy.ListenAndServeHTTPSTCPSNI(l, hp, tp, tlscfg, lookupHostMatcher(cfg)); err != nil {
exit.Fatal("[FATAL] ", err)
Expand All @@ -441,31 +451,33 @@ func startServers(cfg *config.Config) {
}
}

func initMetrics(cfg *config.Config) {
if cfg.Metrics.Target == "" {
log.Printf("[INFO] Metrics disabled")
return
}

var deadline = time.Now().Add(cfg.Metrics.Timeout)
var err error
for {
metrics.DefaultRegistry, err = metrics.NewRegistry(cfg.Metrics)
if err == nil {
route.ServiceRegistry, err = metrics.NewRegistry(cfg.Metrics)
}
if err == nil {
return
}
if time.Now().After(deadline) {
exit.Fatal("[FATAL] ", err)
}
log.Print("[WARN] Error initializing metrics. ", err)
time.Sleep(cfg.Metrics.Retry)
if atomic.LoadInt32(&shuttingDown) > 0 {
exit.Exit(1)
func initMetrics(cfg *config.Config) metrics4.Provider {
var p []metrics4.Provider
for _, x := range strings.Split(cfg.Metrics.Target, ",") {
x = strings.TrimSpace(x)
switch x {
case "flat":
p = append(p, &flat.Provider{})
case "label":
p = append(p, &label.Provider{})
case "statsd_raw":
// prefix := cfg.Metrics.Prefix // prefix is a template and needs to be expanded
prefix := ""
pp, err := statsdraw.NewProvider(prefix, cfg.Metrics.StatsDAddr, cfg.Metrics.Interval)
if err != nil {
exit.Fatalf("[FATAL] Cannot initialize statsd metrics: %s", err)
}
p = append(p, pp)
default:
log.Printf("[WARN] Skipping unknown metrics provider %q", x)
continue
}
log.Printf("[INFO] Registering metrics provider %q", x)
}
if len(p) == 0 {
log.Printf("[INFO] Metrics disabled")
}
return metrics4.NewMultiProvider(p)
}

func initRuntime(cfg *config.Config) {
Expand Down Expand Up @@ -519,7 +531,7 @@ func initBackend(cfg *config.Config) {
}
}

func watchBackend(cfg *config.Config, first chan bool) {
func watchBackend(cfg *config.Config, p metrics4.Provider, first chan bool) {
var (
nextTable string
lastTable string
Expand Down Expand Up @@ -580,6 +592,29 @@ func watchBackend(cfg *config.Config, first chan bool) {
}
}

func unregisterMetrics(p metrics4.Provider, oldTable, newTable route.Table) {
names := func(t route.Table) map[string]bool {
m := map[string]bool{}
for _, routes := range t {
for _, r := range routes {
for _, t := range r.Targets {
m[t.TimerName.String()] = true
}
}
}
return m
}

oldNames := names(oldTable)
newNames := names(newTable)
for n := range oldNames {
if !newNames[n] {
log.Printf("[INFO] Unregistering metric %s", n)
p.Unregister(n)
}
}
}

func watchNoRouteHTML(cfg *config.Config) {
html := registry.Default.WatchNoRouteHTML()
for {
Expand Down
Loading

0 comments on commit c34282f

Please sign in to comment.