Cache queries with globs

For each render and find requests, add a cache item that contains original query.

This should be useful for queries with globs.

Also make cache expire time configurable.

Author: Vladimir Smirnov

main.go

@@ -52,6 +52,7 @@ var Config = struct {
 	MaxIdleConnsPerHost int
 
 	ConcurrencyLimitPerServer int
+	ExpireDelaySec            int32
 }{
 	MaxProcs:    1,
 	IntervalSec: 60,
@@ -63,6 +64,8 @@ var Config = struct {
 
 	MaxIdleConnsPerHost: 100,
 
+	ExpireDelaySec: 10 * 60, // 10 minutes
+
 	pathCache: pathCache{ec: expirecache.New(0)},
 }
 
@@ -81,8 +84,10 @@ var Metrics = struct {
 
 	Timeouts *expvar.Int
 
-	CacheSize  expvar.Func
-	CacheItems expvar.Func
+	CacheSize   expvar.Func
+	CacheItems  expvar.Func
+	CacheMisses *expvar.Int
+	CacheHits   *expvar.Int
 }{
 	FindRequests: expvar.NewInt("find_requests"),
 	FindErrors:   expvar.NewInt("find_errors"),
@@ -96,6 +101,9 @@ var Metrics = struct {
 	InfoErrors:   expvar.NewInt("info_errors"),
 
 	Timeouts: expvar.NewInt("timeouts"),
+
+	CacheHits:   expvar.NewInt("cache_hits"),
+	CacheMisses: expvar.NewInt("cache_misses"),
 }
 
 // BuildVersion is defined at build and reported at startup and as expvar
@@ -360,7 +368,10 @@ func findHandler(w http.ResponseWriter, req *http.Request) {
 		var backends []string
 		var ok bool
 		if backends, ok = Config.pathCache.get(tld); !ok || backends == nil || len(backends) == 0 {
+			Metrics.CacheMisses.Add(1)
 			backends = Config.Backends
+		} else {
+			Metrics.CacheHits.Add(1)
 		}
 
 		responses := multiGet(backends, rewrite.RequestURI())
@@ -375,9 +386,12 @@ func findHandler(w http.ResponseWriter, req *http.Request) {
 		metrics = append(metrics, m...)
 
 		// update our cache of which servers have which metrics
+		allServers := make([]string, 0)
 		for k, v := range paths {
 			Config.pathCache.set(k, v)
+			allServers = append(allServers, v...)
 		}
+		Config.pathCache.set(originalQuery, allServers)
 	}
 
 	encodeFindResponse(format, originalQuery, w, metrics)
@@ -467,7 +481,10 @@ func renderHandler(w http.ResponseWriter, req *http.Request) {
 
 			// lookup the server list for this metric, or use all the servers if it's unknown
 			if serverList, ok = Config.pathCache.get(target); !ok || serverList == nil || len(serverList) == 0 {
+				Metrics.CacheMisses.Add(1)
 				serverList = Config.Backends
+			} else {
+				Metrics.CacheHits.Add(1)
 			}
 
 			responses = append(responses, multiGet(serverList, rewrite.RequestURI())...)
@@ -477,7 +494,10 @@ func renderHandler(w http.ResponseWriter, req *http.Request) {
 
 		// lookup the server list for this metric, or use all the servers if it's unknown
 		if serverList, ok = Config.pathCache.get(target); !ok || serverList == nil || len(serverList) == 0 {
+			Metrics.CacheMisses.Add(1)
 			serverList = Config.Backends
+		} else {
+			Metrics.CacheHits.Add(1)
 		}
 
 		responses = multiGet(serverList, rewrite.RequestURI())
@@ -490,7 +510,7 @@ func renderHandler(w http.ResponseWriter, req *http.Request) {
 		return
 	}
 
-	metrics := mergeResponses(req, responses)
+	servers, metrics := mergeResponses(req, responses)
 	if metrics == nil {
 		Metrics.RenderErrors.Add(1)
 		err := fmt.Sprintf("no decoded responses to merge for req: %s", req.URL.RequestURI())
@@ -499,6 +519,8 @@ func renderHandler(w http.ResponseWriter, req *http.Request) {
 		return
 	}
 
+	Config.pathCache.set(target, servers)
+
 	switch format {
 	case "protobuf3":
 		w.Header().Set("Content-Type", contentTypeProtobuf)
@@ -567,8 +589,9 @@ func createRenderResponse(metrics *pb3.MultiFetchResponse, missing interface{})
 	return response
 }
 
-func mergeResponses(req *http.Request, responses []serverResponse) *pb3.MultiFetchResponse {
+func mergeResponses(req *http.Request, responses []serverResponse) ([]string, *pb3.MultiFetchResponse) {
 
+	servers := make([]string, 0, len(responses))
 	metrics := make(map[string][]pb3.FetchResponse)
 
 	for _, r := range responses {
@@ -583,12 +606,13 @@ func mergeResponses(req *http.Request, responses []serverResponse) *pb3.MultiFet
 		for _, m := range d.Metrics {
 			metrics[m.GetName()] = append(metrics[m.GetName()], *m)
 		}
+		servers = append(servers, r.server)
 	}
 
 	var multi pb3.MultiFetchResponse
 
 	if len(metrics) == 0 {
-		return nil
+		return servers, nil
 	}
 
 	for name, decoded := range metrics {
@@ -617,7 +641,7 @@ func mergeResponses(req *http.Request, responses []serverResponse) *pb3.MultiFet
 		multi.Metrics = append(multi.Metrics, &metric)
 	}
 
-	return &multi
+	return servers, &multi
 }
 
 func mergeValues(req *http.Request, metric *pb3.FetchResponse, decoded []pb3.FetchResponse) {
@@ -698,7 +722,10 @@ func infoHandler(w http.ResponseWriter, req *http.Request) {
 
 	// lookup the server list for this metric, or use all the servers if it's unknown
 	if serverList, ok = Config.pathCache.get(target); !ok || serverList == nil || len(serverList) == 0 {
+		Metrics.CacheMisses.Add(1)
 		serverList = Config.Backends
+	} else {
+		Metrics.CacheHits.Add(1)
 	}
 
 	format := req.FormValue("format")
@@ -923,6 +950,8 @@ func main() {
 
 		graphite.Register(fmt.Sprintf("carbon.zipper.%s.cache_size", hostname), Metrics.CacheSize)
 		graphite.Register(fmt.Sprintf("carbon.zipper.%s.cache_items", hostname), Metrics.CacheItems)
+		graphite.Register(fmt.Sprintf("carbon.zipper.%s.cache_hits", hostname), Metrics.CacheHits)
+		graphite.Register(fmt.Sprintf("carbon.zipper.%s.cache_misses", hostname), Metrics.CacheMisses)
 
 		go mstats.Start(*interval)
 
@@ -1022,14 +1051,13 @@ type pathCache struct {
 }
 
 func (p *pathCache) set(k string, v []string) {
-	// expire cache entries after 10 minutes
-	const expireDelay = 60 * 10
+	// expire cache entries after Config.ExpireCache minutes
 	var size uint64
 	for _, vv := range v {
 		size += uint64(len(vv))
 	}
 
-	p.ec.Set(k, v, size, expireDelay)
+	p.ec.Set(k, v, size, Config.ExpireDelaySec)
 }
 
 func (p *pathCache) get(k string) ([]string, bool) {