plugin/cache: Fix prefetching issues (#1363)

* Improve plugin/cache metrics

* Add coredns_cache_prefetch_total metric to track number of prefetches.
* Remove unnecessary Cache.get() call which would incorrectly increment
  cache counters.
* Initialize all counters and gauges at zero.

* Allow prefetching of a single request per ttl

The original implementation didn't allow prefetching queries which are
only requested once during the duration of a TTL. The minimum amount of
queries which had to be seen was therefore capped at 2.

This change also implements a real prefetch test. The existing test was
a noop and always passed regardless of any prefetch implementation.

* Fix prefetching for items with a short TTL

The default prefetch threshold (percentage) is 10% of the lifetime of a
cache item. With the previous implementation, this disabled prefetching
for all items with a TTL < 10s (the resulting percentage would be 0, at
which point a cached item is already discarded).

This change uses a time based threshold calculation and ensures that
a prefetch is triggered at a TTL of 1 at the latest.

* Fix wrong duration reporting of cached responses

The logging and metrics plugins (among others) included the duration of
a cache prefetch in the request latency of client request. This change
fixes this wrong reporting and executes the prefetch request in a
goroutine in the background.

1 files changed, 35 insertions, 18 deletions

diff --git a/plugin/cache/handler.go b/plugin/cache/handler.go
index df2c74e39..e579aaffc 100644
--- a/plugin/cache/handler.go
+++ b/plugin/cache/handler.go
@@ -1,6 +1,7 @@
 package cache
 
 import (
+	"math"
 	"sync"
 	"time"
 
@@ -25,11 +26,11 @@ func (c *Cache) ServeDNS(ctx context.Context, w dns.ResponseWriter, r *dns.Msg)
 
 	do := state.Do() // TODO(): might need more from OPT record? Like the actual bufsize?
 
-	now := time.Now().UTC()
+	now := c.now().UTC()
 
 	i, ttl := c.get(now, qname, qtype, do)
 	if i != nil && ttl > 0 {
-		resp := i.toMsg(r)
+		resp := i.toMsg(r, now)
 
 		state.SizeAndDo(resp)
 		resp, _ = state.Scrub(resp)
@@ -37,25 +38,23 @@ func (c *Cache) ServeDNS(ctx context.Context, w dns.ResponseWriter, r *dns.Msg)
 
 		if c.prefetch > 0 {
 			i.Freq.Update(c.duration, now)
-		}
-
-		pct := 100
-		if i.origTTL != 0 { // you'll never know
-			pct = int(float64(ttl) / float64(i.origTTL) * 100)
-		}
-
-		if c.prefetch > 0 && i.Freq.Hits() > c.prefetch && pct < c.percentage {
-			// When prefetching we loose the item i, and with it the frequency
-			// that we've gathered sofar. See we copy the frequencies info back
-			// into the new item that was stored in the cache.
-			prr := &ResponseWriter{ResponseWriter: w, Cache: c, prefetch: true}
-			plugin.NextOrFailure(c.Name(), c.Next, ctx, prr, r)
 
-			if i1, _ := c.get(now, qname, qtype, do); i1 != nil {
-				i1.Freq.Reset(now, i.Freq.Hits())
+			threshold := int(math.Ceil(float64(c.percentage) / 100 * float64(i.origTTL)))
+			if i.Freq.Hits() >= c.prefetch && ttl <= threshold {
+				go func() {
+					cachePrefetches.Inc()
+					// When prefetching we loose the item i, and with it the frequency
+					// that we've gathered sofar. See we copy the frequencies info back
+					// into the new item that was stored in the cache.
+					prr := &ResponseWriter{ResponseWriter: w, Cache: c, prefetch: true}
+					plugin.NextOrFailure(c.Name(), c.Next, ctx, prr, r)
+
+					if i1 := c.exists(qname, qtype, do); i1 != nil {
+						i1.Freq.Reset(now, i.Freq.Hits())
+					}
+				}()
 			}
 		}
-
 		return dns.RcodeSuccess, nil
 	}
 
@@ -82,6 +81,17 @@ func (c *Cache) get(now time.Time, qname string, qtype uint16, do bool) (*item,
 	return nil, 0
 }
 
+func (c *Cache) exists(qname string, qtype uint16, do bool) *item {
+	k := hash(qname, qtype, do)
+	if i, ok := c.ncache.Get(k); ok {
+		return i.(*item)
+	}
+	if i, ok := c.pcache.Get(k); ok {
+		return i.(*item)
+	}
+	return nil
+}
+
 var (
 	cacheSize = prometheus.NewGaugeVec(prometheus.GaugeOpts{
 		Namespace: plugin.Namespace,
@@ -110,6 +120,13 @@ var (
 		Name:      "misses_total",
 		Help:      "The count of cache misses.",
 	})
+
+	cachePrefetches = prometheus.NewCounter(prometheus.CounterOpts{
+		Namespace: plugin.Namespace,
+		Subsystem: "cache",
+		Name:      "prefetch_total",
+		Help:      "The number of time the cache has prefetched a cached item.",
+	})
 )
 
 var once sync.Once