From d7c154ea41d9d2c7745666b32bf9c3b8cd131a77 Mon Sep 17 00:00:00 2001 From: Alvaro Saurin Date: Tue, 28 Apr 2015 11:46:13 +0200 Subject: [PATCH] Removed the "now" parameters and use an external clock provider --- nameserver/cache.go | 67 ++++++++++++++++++++++++++++++++----- nameserver/cache_test.go | 71 +++++++++++++++++++++------------------- nameserver/server.go | 13 +++----- 3 files changed, 100 insertions(+), 51 deletions(-) diff --git a/nameserver/cache.go b/nameserver/cache.go index cba1742d..8cfd424c 100644 --- a/nameserver/cache.go +++ b/nameserver/cache.go @@ -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() +} diff --git a/nameserver/cache_test.go b/nameserver/cache_test.go index 3f785fb8..b53d7529 100644 --- a/nameserver/cache_test.go +++ b/nameserver/cache_test.go @@ -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") } diff --git a/nameserver/server.go b/nameserver/server.go index d80846fb..6fc82cf4 100644 --- a/nameserver/server.go +++ b/nameserver/server.go @@ -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) }