reweave/weave
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Removed the "now" parameters and use an external clock provider

Browse Source
This commit is contained in:
Alvaro Saurin committed 2015-04-28 12:02:49 +02:00
1 parent 7de2080b9e
commit d7c154ea41
3 files changed
+100 -51
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

nameserver/cache.go
+58 -9
View File
@@ -1,8 +1,11 @@
package nameserver
import (
"bytes"
"container/heap"
"errors"
"fmt"
"github.com/benbjohnson/clock"
"github.com/miekg/dns"
. "github.com/weaveworks/weave/common"
"math"
@@ -196,26 +199,49 @@ func (h *entriesPtrsHeap) Pop() interface{} {
//////////////////////////////////////////////////////////////////////////////////////
// The cache interface
type ZoneCache interface {
// Get from the cache
Get(request *dns.Msg, maxLen int) (reply *dns.Msg, err error)
// Put in the cache
Put(request *dns.Msg, reply *dns.Msg, ttl int, flags uint8) int
// Remove from the cache
Remove(question *dns.Question)
// Purge all expired entries from the cache
Purge()
// Remove all entries
Clear()
// Return the cache length
Len() int
// Return the max capacity
Capacity() int
}
type cacheKey dns.Question
type entries map[cacheKey]*cacheEntry
// Cache is a thread-safe fixed capacity LRU cache.
type Cache struct {
Capacity int
capacity int
entries entries
entriesH entriesPtrsHeap // len(entriesH) <= len(entries), as pending entries can be in entries but not in entriesH
clock clock.Clock
lock sync.RWMutex
}
// NewCache creates a cache of the given capacity
func NewCache(capacity int) (*Cache, error) {
func NewCache(capacity int, clk clock.Clock) (*Cache, error) {
if capacity <= 0 {
return nil, errInvalidCapacity
}
c := &Cache{
Capacity: capacity,
capacity: capacity,
entries: make(entries, capacity),
clock: clk,
}
if c.clock == nil {
c.clock = clock.New()
}
heap.Init(&c.entriesH)
@@ -227,15 +253,16 @@ func (c *Cache) Clear() {
c.lock.Lock()
defer c.lock.Unlock()
c.entries = make(entries, c.Capacity)
c.entries = make(entries, c.capacity)
heap.Init(&c.entriesH)
}
// Purge removes the old elements in the cache
func (c *Cache) Purge(now time.Time) {
func (c *Cache) Purge() {
c.lock.Lock()
defer c.lock.Unlock()
now := c.clock.Now()
for i, entry := range c.entriesH {
if entry.hasExpired(now) {
heap.Remove(&c.entriesH, i)
@@ -248,10 +275,11 @@ func (c *Cache) Purge(now time.Time) {
// Add adds a reply to the cache.
// When `ttl` is equal to `nullTTL`, the cache entry will be valid until the closest TTL in the `reply`
func (c *Cache) Put(request *dns.Msg, reply *dns.Msg, ttl int, flags uint8, now time.Time) int {
func (c *Cache) Put(request *dns.Msg, reply *dns.Msg, ttl int, flags uint8) int {
c.lock.Lock()
defer c.lock.Unlock()
now := c.clock.Now()
question := request.Question[0]
key := cacheKey(question)
ent, found := c.entries[key]
@@ -262,7 +290,7 @@ func (c *Cache) Put(request *dns.Msg, reply *dns.Msg, ttl int, flags uint8, now
}
} else {
// If we will add a new item and the capacity has been exceeded, make some room...
if len(c.entriesH) >= c.Capacity {
if len(c.entriesH) >= c.capacity {
lowestEntry := heap.Pop(&c.entriesH).(*cacheEntry)
delete(c.entries, cacheKey(lowestEntry.question))
}
@@ -276,10 +304,11 @@ func (c *Cache) Put(request *dns.Msg, reply *dns.Msg, ttl int, flags uint8, now
// Look up for a question's reply from the cache.
// If no reply is stored in the cache, it returns a `nil` reply and no error.
func (c *Cache) Get(request *dns.Msg, maxLen int, now time.Time) (reply *dns.Msg, err error) {
func (c *Cache) Get(request *dns.Msg, maxLen int) (reply *dns.Msg, err error) {
c.lock.Lock()
defer c.lock.Unlock()
now := c.clock.Now()
question := request.Question[0]
key := cacheKey(question)
if ent, found := c.entries[key]; found {
@@ -322,3 +351,23 @@ func (c *Cache) Len() int {
return len(c.entries)
}
// Return the max capacity
func (c Cache) Capacity() int {
c.lock.RLock()
defer c.lock.RUnlock()
return c.capacity
}
// Return the max capacity
func (c Cache) String() string {
c.lock.RLock()
defer c.lock.RUnlock()
var buf bytes.Buffer
for _, entry := range c.entries {
fmt.Fprintf(&buf, "%s\n", entry)
}
return buf.String()
}
nameserver/cache_test.go
+37 -34
View File
@@ -2,6 +2,7 @@ package nameserver
import (
"fmt"
"github.com/benbjohnson/clock"
"github.com/miekg/dns"
. "github.com/weaveworks/weave/common"
wt "github.com/weaveworks/weave/testing"
@@ -12,11 +13,12 @@ import (
// Check that the cache keeps its intended capacity constant
func TestCacheLength(t *testing.T) {
InitDefaultLogging(true)
InitDefaultLogging(testing.Verbose())
Info.Println("TestCacheLength starting")
const cacheLen = 128
l, err := NewCache(cacheLen)
l, err := NewCache(cacheLen, nil)
wt.AssertNoErr(t, err)
insTime := time.Now()
@@ -36,7 +38,7 @@ func TestCacheLength(t *testing.T) {
reply := makeAddressReply(questionMsg, question, records)
reply.Answer[0].Header().Ttl = uint32(i)
l.Put(questionMsg, reply, 0, 0, insTime)
l.Put(questionMsg, reply, 0, 0)
}
wt.AssertEqualInt(t, l.Len(), cacheLen, "cache length")
@@ -52,13 +54,15 @@ func TestCacheLength(t *testing.T) {
// Check that the cache entries are ok
func TestCacheEntries(t *testing.T) {
InitDefaultLogging(true)
InitDefaultLogging(testing.Verbose())
Info.Println("TestCacheEntries starting")
Info.Println("Checking cache consistency")
const cacheLen = 128
clk := clock.NewMock()
l, err := NewCache(cacheLen)
l, err := NewCache(cacheLen, clk)
wt.AssertNoErr(t, err)
questionMsg := new(dns.Msg)
@@ -68,14 +72,14 @@ func TestCacheEntries(t *testing.T) {
question := &questionMsg.Question[0]
t.Logf("Trying to get a name")
resp, err := l.Get(questionMsg, minUDPSize, time.Now())
resp, err := l.Get(questionMsg, minUDPSize)
wt.AssertNoErr(t, err)
if resp != nil {
t.Logf("Got\n%s", resp)
t.Fatalf("ERROR: Did not expect a reponse from Get() yet")
}
t.Logf("Trying to get it again")
resp, err = l.Get(questionMsg, minUDPSize, time.Now())
resp, err = l.Get(questionMsg, minUDPSize)
wt.AssertNoErr(t, err)
if resp != nil {
t.Logf("Got\n%s", resp)
@@ -85,20 +89,19 @@ func TestCacheEntries(t *testing.T) {
t.Logf("Inserting the reply")
records := []ZoneRecord{Record{"some.name", net.ParseIP("10.0.1.1"), 0, 0, 0}}
reply1 := makeAddressReply(questionMsg, question, records)
l.Put(questionMsg, reply1, nullTTL, 0, time.Now())
l.Put(questionMsg, reply1, nullTTL, 0)
timeGet1 := time.Now()
t.Logf("Checking we can Get() the reply now")
resp, err = l.Get(questionMsg, minUDPSize, timeGet1)
resp, err = l.Get(questionMsg, minUDPSize)
wt.AssertNoErr(t, err)
wt.AssertTrue(t, resp != nil, "reponse from Get()")
t.Logf("Received '%s'", resp.Answer[0])
wt.AssertType(t, resp.Answer[0], (*dns.A)(nil), "DNS record")
ttlGet1 := resp.Answer[0].Header().Ttl
timeGet2 := timeGet1.Add(time.Duration(1) * time.Second)
clk.Add(time.Duration(1) * time.Second)
t.Logf("Checking that a second Get(), after 1 second, gets the same result, but with reduced TTL")
resp, err = l.Get(questionMsg, minUDPSize, timeGet2)
resp, err = l.Get(questionMsg, minUDPSize)
wt.AssertNoErr(t, err)
wt.AssertTrue(t, resp != nil, "reponse from a second Get()")
t.Logf("Received '%s'", resp.Answer[0])
@@ -106,9 +109,9 @@ func TestCacheEntries(t *testing.T) {
ttlGet2 := resp.Answer[0].Header().Ttl
wt.AssertEqualInt(t, int(ttlGet1-ttlGet2), 1, "TTL difference")
timeGet3 := timeGet1.Add(time.Duration(localTTL) * time.Second)
clk.Add(time.Duration(localTTL) * time.Second)
t.Logf("Checking that a third Get(), after %d second, gets no result", localTTL)
resp, err = l.Get(questionMsg, minUDPSize, timeGet3)
resp, err = l.Get(questionMsg, minUDPSize)
wt.AssertNoErr(t, err)
if resp != nil {
t.Logf("Got\n%s", resp)
@@ -118,29 +121,28 @@ func TestCacheEntries(t *testing.T) {
t.Logf("Checking that an Remove() results in Get() returning nothing")
records = []ZoneRecord{Record{"some.name", net.ParseIP("10.0.9.9"), 0, 0, 0}}
replyTemp := makeAddressReply(questionMsg, question, records)
l.Put(questionMsg, replyTemp, nullTTL, 0, time.Now())
l.Put(questionMsg, replyTemp, nullTTL, 0)
lenBefore := l.Len()
l.Remove(question)
wt.AssertEqualInt(t, l.Len(), lenBefore-1, "cache length")
l.Remove(question) // do it again: should have no effect...
wt.AssertEqualInt(t, l.Len(), lenBefore-1, "cache length")
resp, err = l.Get(questionMsg, minUDPSize, timeGet1)
resp, err = l.Get(questionMsg, minUDPSize)
wt.AssertNoErr(t, err)
wt.AssertTrue(t, resp == nil, "reponse from the Get() after a Remove()")
t.Logf("Inserting a two replies for the same query")
timePut2 := time.Now()
records = []ZoneRecord{Record{"some.name", net.ParseIP("10.0.1.2"), 0, 0, 0}}
reply2 := makeAddressReply(questionMsg, question, records)
l.Put(questionMsg, reply2, nullTTL, 0, timePut2)
timePut3 := timePut2.Add(time.Duration(1) * time.Second)
l.Put(questionMsg, reply2, nullTTL, 0)
clk.Add(time.Duration(1) * time.Second)
records = []ZoneRecord{Record{"some.name", net.ParseIP("10.0.1.3"), 0, 0, 0}}
reply3 := makeAddressReply(questionMsg, question, records)
l.Put(questionMsg, reply3, nullTTL, 0, timePut3)
l.Put(questionMsg, reply3, nullTTL, 0)
t.Logf("Checking we get the last one...")
resp, err = l.Get(questionMsg, minUDPSize, timePut3)
resp, err = l.Get(questionMsg, minUDPSize)
wt.AssertNoErr(t, err)
wt.AssertTrue(t, resp != nil, "reponse from the Get()")
t.Logf("Received '%s'", resp.Answer[0])
@@ -148,7 +150,8 @@ func TestCacheEntries(t *testing.T) {
wt.AssertEqualString(t, resp.Answer[0].(*dns.A).A.String(), "10.0.1.3", "IP address")
wt.AssertEqualInt(t, int(resp.Answer[0].Header().Ttl), int(localTTL), "TTL")
resp, err = l.Get(questionMsg, minUDPSize, timePut3.Add(time.Duration(localTTL-1)*time.Second))
clk.Add(time.Duration(localTTL-1) * time.Second)
resp, err = l.Get(questionMsg, minUDPSize)
wt.AssertNoErr(t, err)
wt.AssertTrue(t, resp != nil, "reponse from the Get()")
t.Logf("Received '%s'", resp.Answer[0])
@@ -158,7 +161,8 @@ func TestCacheEntries(t *testing.T) {
t.Logf("Checking we get empty replies when they are expired...")
lenBefore = l.Len()
resp, err = l.Get(questionMsg, minUDPSize, timePut3.Add(time.Duration(localTTL)*time.Second))
clk.Add(time.Duration(localTTL) * time.Second)
resp, err = l.Get(questionMsg, minUDPSize)
wt.AssertNoErr(t, err)
if resp != nil {
t.Logf("Got\n%s", resp.Answer[0])
@@ -172,7 +176,7 @@ func TestCacheEntries(t *testing.T) {
question2 := &questionMsg2.Question[0]
t.Logf("Trying to Get() a name")
resp, err = l.Get(questionMsg2, minUDPSize, time.Now())
resp, err = l.Get(questionMsg2, minUDPSize)
wt.AssertNoErr(t, err)
wt.AssertNil(t, resp, "reponse from Get() yet")
@@ -180,8 +184,8 @@ func TestCacheEntries(t *testing.T) {
records = []ZoneRecord{Record{"some.name", net.ParseIP("10.0.9.9"), 0, 0, 0}}
replyTemp2 := makeAddressReply(questionMsg2, question2, records)
l.Remove(question2)
l.Put(questionMsg2, replyTemp2, nullTTL, 0, time.Now())
resp, err = l.Get(questionMsg2, minUDPSize, time.Now())
l.Put(questionMsg2, replyTemp2, nullTTL, 0)
resp, err = l.Get(questionMsg2, minUDPSize)
wt.AssertNoErr(t, err)
wt.AssertNotNil(t, resp, "reponse from Get()")
@@ -191,23 +195,22 @@ func TestCacheEntries(t *testing.T) {
question3 := &questionMsg3.Question[0]
t.Logf("Checking that a entry with CacheNoLocalReplies return an error")
timePut3 = time.Now()
l.Put(questionMsg3, nil, nullTTL, CacheNoLocalReplies, timePut3)
resp, err = l.Get(questionMsg3, minUDPSize, timePut3)
l.Put(questionMsg3, nil, nullTTL, CacheNoLocalReplies)
resp, err = l.Get(questionMsg3, minUDPSize)
wt.AssertNil(t, resp, "Get() response with CacheNoLocalReplies")
wt.AssertNotNil(t, err, "Get() error with CacheNoLocalReplies")
timeExpiredGet3 := timePut3.Add(time.Second * time.Duration(negLocalTTL+1))
t.Logf("Checking that we get an expired response after %f seconds", timeExpiredGet3.Sub(timePut3).Seconds())
resp, err = l.Get(questionMsg3, minUDPSize, timeExpiredGet3)
clk.Add(time.Second * time.Duration(negLocalTTL+1))
t.Logf("Checking that we get an expired response after %f seconds", negLocalTTL)
resp, err = l.Get(questionMsg3, minUDPSize)
wt.AssertNil(t, resp, "expired Get() response with CacheNoLocalReplies")
wt.AssertNil(t, err, "expired Get() error with CacheNoLocalReplies")
l.Remove(question3)
t.Logf("Checking that Put&Get with CacheNoLocalReplies with a Remove in the middle returns nothing")
l.Put(questionMsg3, nil, nullTTL, CacheNoLocalReplies, time.Now())
l.Put(questionMsg3, nil, nullTTL, CacheNoLocalReplies)
l.Remove(question3)
resp, err = l.Get(questionMsg3, minUDPSize, time.Now())
resp, err = l.Get(questionMsg3, minUDPSize)
wt.AssertNil(t, resp, "Get() reponse with CacheNoLocalReplies")
wt.AssertNil(t, err, "Get() error with CacheNoLocalReplies")
}
nameserver/server.go
+5 -8
View File
@@ -7,7 +7,6 @@ import (
. "github.com/weaveworks/weave/common"
"net"
"sync"
"time"
)
const (
@@ -130,7 +129,7 @@ func NewDNSServer(config DNSServerConfig, zone Zone, iface *net.Interface) (s *D
cacheLen = config.CacheLen
}
Debug.Printf("[dns] Initializing cache: %d entries", config.CacheLen)
s.cache, err = NewCache(cacheLen)
s.cache, err = NewCache(cacheLen, nil)
if err != nil {
return
}
@@ -254,11 +253,10 @@ func (s *DNSServer) commonQueryHandler(proto dnsProtocol, kind string, qtype uin
return func(w dns.ResponseWriter, r *dns.Msg) {
q := r.Question[0]
Debug.Printf("[dns msgid %d] %s: %+v", r.MsgHdr.Id, kind, q)
now := time.Now()
maxLen := getMaxReplyLen(r, proto)
lookups := []ZoneLookup{s.Zone, s.mdnsCli}
reply, err := s.cache.Get(r, maxLen, time.Now())
reply, err := s.cache.Get(r, maxLen)
if err != nil {
if err == errNoLocalReplies {
Debug.Printf("[dns msgid %d] Cached 'no local replies' - skipping local lookup", r.MsgHdr.Id)
@@ -280,7 +278,7 @@ func (s *DNSServer) commonQueryHandler(proto dnsProtocol, kind string, qtype uin
if q.Qtype != qtype {
Debug.Printf("[dns msgid %d] Unsupported query type %s", r.MsgHdr.Id, dns.TypeToString[q.Qtype])
m := makeDNSNotImplResponse(r)
s.cache.Put(r, m, negLocalTTL, 0, now)
s.cache.Put(r, m, negLocalTTL, 0)
w.WriteMsg(m)
return
}
@@ -290,16 +288,15 @@ func (s *DNSServer) commonQueryHandler(proto dnsProtocol, kind string, qtype uin
m.Authoritative = true
Debug.Printf("[dns msgid %d] Caching response for type %s query for '%s': %s [code:%s]",
m.MsgHdr.Id, dns.TypeToString[q.Qtype], q.Name, answers, dns.RcodeToString[m.Rcode])
s.cache.Put(r, m, nullTTL, 0, now)
s.cache.Put(r, m, nullTTL, 0)
w.WriteMsg(m)
return
}
now = time.Now()
}
Info.Printf("[dns msgid %d] No results for type %s query for '%s' [caching no-local]",
r.MsgHdr.Id, dns.TypeToString[q.Qtype], q.Name)
s.cache.Put(r, nil, negLocalTTL, CacheNoLocalReplies, now)
s.cache.Put(r, nil, negLocalTTL, CacheNoLocalReplies)
fallback(w, r)
}