Merge pull request #610 from inercia/weave-583-1

External clock provider for the cache

nameserver/addrs.go

@@ -15,6 +15,7 @@ type ZoneRecord interface {
 	IP() net.IP    // The IP (v4) address
 	Priority() int // The priority
 	Weight() int   // The weight
+	TTL() int      // TTL
 }
 
 type ZoneLookup interface {
@@ -41,12 +42,14 @@ type Record struct {
 	ip       net.IP
 	priority int
 	weight   int
+	ttl      int
 }
 
 func (r Record) Name() string  { return r.name }
 func (r Record) IP() net.IP    { return r.ip }
 func (r Record) Priority() int { return r.priority }
 func (r Record) Weight() int   { return r.weight }
+func (r Record) TTL() int      { return r.ttl }
 
 func (i Record) String() string {
 	var buf bytes.Buffer
@@ -62,6 +65,9 @@ func (i Record) String() string {
 	if i.Weight() > 0 {
 		fmt.Fprintf(&buf, "/W:%d", i.Weight())
 	}
+	if i.TTL() > 0 {
+		fmt.Fprintf(&buf, "/TTL:%d", i.TTL())
+	}
 
 	return buf.String()
 }
@@ -109,3 +115,11 @@ func raddrToIPv4(addr string) (IPv4, error) {
 	revIP4 := revIP.To4()
 	return IPv4([4]byte{revIP4[3], revIP4[2], revIP4[1], revIP4[0]}), nil
 }
+
+func raddrToIP(addr string) (net.IP, error) {
+	r, err := raddrToIPv4(addr)
+	if err != nil {
+		return net.IP{}, err
+	}
+	return r.toNetIP(), nil
+}

nameserver/addrs_test.go

@@ -9,6 +9,7 @@ import (
 
 func TestAddrs(t *testing.T) {
 	InitDefaultLogging(testing.Verbose())
+	Info.Println("TestAddrs starting")
 
 	ip, err := addrToIPv4("10.13.12.11")
 	wt.AssertNoErr(t, err)

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"
@@ -31,6 +34,11 @@ const (
 	stResolved entryStatus = iota // resolved
 )
 
+var statusToString = map[entryStatus]string{
+	stPending:  "pending",
+	stResolved: "resolved",
+}
+
 const (
 	CacheNoLocalReplies uint8 = 1 << iota // not found in local network (stored in the cache so we skip another local lookup or some time)
 )
@@ -163,6 +171,19 @@ func (e *cacheEntry) setReply(reply *dns.Msg, ttl int, flags uint8, now time.Tim
 	return (prevValidUntil != e.validUntil)
 }
 
+func (e *cacheEntry) String() string {
+	var buf bytes.Buffer
+	q := e.question
+	fmt.Fprintf(&buf, "'%s'[%s]: ", q.Name, dns.TypeToString[q.Qtype])
+	if e.Flags&CacheNoLocalReplies != 0 {
+		fmt.Fprintf(&buf, "neg-local")
+	} else {
+		fmt.Fprintf(&buf, "%s", statusToString[e.Status])
+	}
+	fmt.Fprintf(&buf, "(%d bytes)", e.ReplyLen)
+	return buf.String()
+}
+
 //////////////////////////////////////////////////////////////////////////////////////
 
 // An entriesPtrHeap is a min-heap of cache entries.
@@ -196,26 +217,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 +271,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 +293,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 +308,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 +322,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 +369,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

@@ -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()
@@ -31,12 +33,12 @@ func TestCacheLength(t *testing.T) {
 		question := &questionMsg.Question[0]
 
 		ip := net.ParseIP(fmt.Sprintf("10.0.1.%d", i))
-		records := []ZoneRecord{Record{name, ip, 0, 0}}
+		records := []ZoneRecord{Record{name, ip, 0, 0, 0}}
 
 		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)
@@ -83,22 +87,21 @@ func TestCacheEntries(t *testing.T) {
 	}
 
 	t.Logf("Inserting the reply")
-	records := []ZoneRecord{Record{"some.name", net.ParseIP("10.0.1.1"), 0, 0}}
+	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)
@@ -116,31 +119,30 @@ 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}}
+	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}}
+	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)
-	records = []ZoneRecord{Record{"some.name", net.ParseIP("10.0.1.3"), 0, 0}}
+	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,16 +176,16 @@ 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")
 
 	t.Logf("Checking that an Remove() between Get() and Put() does not break things")
-	records = []ZoneRecord{Record{"some.name", net.ParseIP("10.0.9.9"), 0, 0}}
+	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/mdns_client.go

@@ -1,6 +1,8 @@
 package nameserver
 
 import (
+	"bytes"
+	"fmt"
 	"github.com/miekg/dns"
 	"math"
 	"net"
@@ -29,6 +31,7 @@ var (
 type Response struct {
 	name string
 	addr net.IP
+	ttl  int
 	err  error
 }
 
@@ -36,6 +39,7 @@ func (r Response) Name() string  { return r.name }
 func (r Response) IP() net.IP    { return r.addr }
 func (r Response) Priority() int { return 0 }
 func (r Response) Weight() int   { return 0 }
+func (r Response) TTL() int      { return r.ttl }
 
 func (r Response) Equal(r2 *Response) bool {
 	if r.name != r2.name {
@@ -50,6 +54,24 @@ func (r Response) Equal(r2 *Response) bool {
 	return true
 }
 
+func (i Response) String() string {
+	var buf bytes.Buffer
+	if i.err != nil {
+		fmt.Fprintf(&buf, "%s", i.err)
+	} else {
+		if len(i.Name()) > 0 {
+			fmt.Fprintf(&buf, "%s", i.Name())
+		}
+		if !i.IP().IsUnspecified() {
+			fmt.Fprintf(&buf, "[%s]", i.IP())
+		}
+		if i.ttl > 0 {
+			fmt.Fprintf(&buf, "(TTL:%d)", i.TTL())
+		}
+	}
+	return buf.String()
+}
+
 type responseInfo struct {
 	timeout   time.Time // if no answer by this time, give up
 	insistent bool      // insistent queries are not removed on the first reply
@@ -182,10 +204,11 @@ func (c *MDNSClient) ResponseCallback(r *dns.Msg) {
 			switch rr := answer.(type) {
 			case *dns.A:
 				name = rr.Hdr.Name
-				res = &Response{addr: rr.A}
+				res = &Response{name: name, addr: rr.A, ttl: int(rr.Hdr.Ttl)}
 			case *dns.PTR:
 				name = rr.Hdr.Name
-				res = &Response{name: rr.Ptr}
+				raddr, _ := raddrToIP(name)
+				res = &Response{name: rr.Ptr, addr: raddr, ttl: int(rr.Hdr.Ttl)}
 			default:
 				return
 			}

nameserver/mdns_client_test.go

@@ -76,6 +76,8 @@ func RunLocalMulticastServer() (*dns.Server, error) {
 }
 
 func setup(t *testing.T) (*MDNSClient, *dns.Server, error) {
+	InitDefaultLogging(testing.Verbose())
+
 	server, err := RunLocalMulticastServer()
 	if err != nil {
 		t.Fatalf("Unable to run test server: %s. No default multicast interface?", err)
@@ -109,7 +111,7 @@ func (c *testContext) checkResponse(t *testing.T, channelOk bool, resp *Response
 		return
 	}
 	wt.AssertNoErr(t, resp.err)
-	log.Printf("Got address response %s addr %s", resp.Name, resp.IP())
+	log.Printf("Got address response %s addr %s", resp.Name(), resp.IP())
 	c.receivedAddr = resp.IP()
 	c.receivedCount++
 }

nameserver/mdns_server.go

@@ -116,16 +116,16 @@ func (s *MDNSServer) makeHandler(qtype uint16, lookup LookupFunc) dns.HandlerFun
 		if len(r.Answer) == 0 && len(r.Question) > 0 {
 			q := &r.Question[0]
 			if q.Qtype == qtype {
-				Debug.Printf("[mdns msgid %d] Trying to answer to mDNS query '%s'", r.MsgHdr.Id, q.Name)
+				Debug.Printf("[mdns msgid %d] srv: trying to answer to mDNS query '%s'", r.MsgHdr.Id, q.Name)
 				if m := lookup(s.zone, r, q); m != nil {
-					Debug.Printf("[mdns msgid %d] - found local answer to mDNS query '%s'", r.MsgHdr.Id, q.Name)
+					Debug.Printf("[mdns msgid %d] srv: found local answer to mDNS query '%s'", r.MsgHdr.Id, q.Name)
 					if err := s.sendResponse(m); err != nil {
-						Warning.Printf("[mdns msgid %d] - error writing to %v", r.MsgHdr.Id, s.sendconn)
+						Warning.Printf("[mdns msgid %d] srv: error writing mDNS response to %v", r.MsgHdr.Id, s.sendconn)
 					} else {
-						Debug.Printf("[mdns msgid %d] - response sent: %d answers", r.MsgHdr.Id, len(m.Answer))
+						Debug.Printf("[mdns msgid %d] srv: response sent: %d answers", r.MsgHdr.Id, len(m.Answer))
 					}
 				} else {
-					Debug.Printf("[mdns msgid %d] - no local answer for mDNS query '%s'",
+					Debug.Printf("[mdns msgid %d] srv: no local answer for answering mDNS query '%s'",
 						r.MsgHdr.Id, q.Name)
 				}
 			}

nameserver/mdns_server_test.go

@@ -10,23 +10,25 @@ import (
 )
 
 func TestServerSimpleQuery(t *testing.T) {
-	InitDefaultLogging(testing.Verbose())
-
 	var (
-		testRecord1 = Record{"test.weave.local.", net.ParseIP("10.20.20.10"), 0, 0}
+		testRecord1 = Record{"test.weave.local.", net.ParseIP("10.20.20.10"), 0, 0, 0}
+		testRecord2 = Record{"test.weave.local.", net.ParseIP("10.20.20.20"), 0, 0, 0}
 		testInAddr1 = "10.20.20.10.in-addr.arpa."
 	)
 
-	mzone := NewMockedZone(testRecord1)
+	InitDefaultLogging(testing.Verbose())
+	Info.Println("TestServerSimpleQuery starting")
+
+	mzone := newMockedZoneWithRecords([]ZoneRecord{testRecord1, testRecord2})
 	mdnsServer, err := NewMDNSServer(mzone)
 	wt.AssertNoErr(t, err)
 	err = mdnsServer.Start(nil)
 	wt.AssertNoErr(t, err)
 	defer mdnsServer.Stop()
 
-	var receivedAddr net.IP
-	var receivedName string
-	var recvChan chan interface{}
+	receivedAddrs := make([]net.IP, 0)
+	receivedName := ""
+	recvChan := make(chan interface{})
 	receivedCount := 0
 
 	// Implement a minimal listener for responses
@@ -41,7 +43,7 @@ func TestServerSimpleQuery(t *testing.T) {
 				switch rr := answer.(type) {
 				case *dns.A:
 					t.Logf("... A:\n%+v", rr)
-					receivedAddr = rr.A
+					receivedAddrs = append(receivedAddrs, rr.A)
 					receivedCount++
 				case *dns.PTR:
 					t.Logf("... PTR:\n%+v", rr)
@@ -54,7 +56,7 @@ func TestServerSimpleQuery(t *testing.T) {
 	}
 
 	sendQuery := func(name string, querytype uint16) {
-		receivedAddr = nil
+		receivedAddrs = make([]net.IP, 0)
 		receivedName = ""
 		receivedCount = 0
 		recvChan = make(chan interface{})
@@ -90,22 +92,25 @@ func TestServerSimpleQuery(t *testing.T) {
 
 	time.Sleep(100 * time.Millisecond) // Allow for server to get going
 
-	Debug.Printf("Query: %s dns.TypeA", testRecord1.Name())
+	Debug.Printf("Checking that we get 2 IPs fo name '%s' [A]", testRecord1.Name())
 	sendQuery(testRecord1.Name(), dns.TypeA)
-	if receivedCount != 1 {
+	if receivedCount != 2 {
 		t.Fatalf("Unexpected result count %d for %s", receivedCount, testRecord1.Name())
 	}
-	if !receivedAddr.Equal(testRecord1.IP()) {
-		t.Fatalf("Unexpected result %s for %s", receivedAddr, testRecord1.Name())
+	if !(receivedAddrs[0].Equal(testRecord1.IP()) || receivedAddrs[0].Equal(testRecord2.IP())) {
+		t.Fatalf("Unexpected result %s for %s", receivedAddrs, testRecord1.Name())
+	}
+	if !(receivedAddrs[1].Equal(testRecord1.IP()) || receivedAddrs[1].Equal(testRecord2.IP())) {
+		t.Fatalf("Unexpected result %s for %s", receivedAddrs, testRecord1.Name())
 	}
 
-	Debug.Printf("Query: testfail.weave. dns.TypeA")
+	Debug.Printf("Checking that 'testfail.weave.' [A] gets no answers")
 	sendQuery("testfail.weave.", dns.TypeA)
 	if receivedCount != 0 {
 		t.Fatalf("Unexpected result count %d for testfail.weave", receivedCount)
 	}
 
-	Debug.Printf("Query: %s dns.TypePTR", testInAddr1)
+	Debug.Printf("Checking that '%s' [PTR] gets one name", testInAddr1)
 	sendQuery(testInAddr1, dns.TypePTR)
 	if receivedCount != 1 {
 		t.Fatalf("Expected an answer to %s, got %d answers", testInAddr1, receivedCount)

nameserver/mocks_test.go

@@ -3,51 +3,89 @@ package nameserver
 import (
 	. "github.com/weaveworks/weave/common"
 	"net"
+	"sync"
+)
+
+const (
+	testSocketTimeout = 100 // in millisecs
 )
 
 // Warn about some methods that some day should be implemented...
 func notImplWarn() { Warning.Printf("Mocked method. Not implemented.") }
 
-// A mocked Zone that always returns the same record
-type mockedZone struct {
-	record ZoneRecord
+// A mocked Zone that always returns the same records
+// * it does not send/receive any mDNS query
+type mockedZoneWithRecords struct {
+	sync.RWMutex
+
+	records []ZoneRecord
 
 	// Statistics
 	NumLookupsName   int
 	NumLookupsInaddr int
 }
 
-func NewMockedZone(zr ZoneRecord) *mockedZone { return &mockedZone{record: zr} }
-func (mz mockedZone) Domain() string          { return DefaultLocalDomain }
-func (mz mockedZone) LookupName(name string) ([]ZoneRecord, error) {
-	Debug.Printf("[mocked zone]: LookupName: returning record %s", mz.record)
-	mz.NumLookupsName += 1
-	return []ZoneRecord{mz.record}, nil
+func newMockedZoneWithRecords(zr []ZoneRecord) *mockedZoneWithRecords {
+	return &mockedZoneWithRecords{records: zr}
 }
-func (mz mockedZone) LookupInaddr(inaddr string) ([]ZoneRecord, error) {
-	Debug.Printf("[mocked zone]: LookupInaddr: returning record %s", mz.record)
+func (mz *mockedZoneWithRecords) Domain() string { return DefaultLocalDomain }
+func (mz *mockedZoneWithRecords) LookupName(name string) ([]ZoneRecord, error) {
+	Debug.Printf("[mocked zone]: LookupName: returning records %s", mz.records)
+	mz.Lock()
+	defer mz.Unlock()
+
+	mz.NumLookupsName += 1
+	res := make([]ZoneRecord, 0)
+	for _, r := range mz.records {
+		if r.Name() == name {
+			res = append(res, r)
+		}
+	}
+	return res, nil
+}
+
+func (mz *mockedZoneWithRecords) LookupInaddr(inaddr string) ([]ZoneRecord, error) {
+	Debug.Printf("[mocked zone]: LookupInaddr: returning records %s", mz.records)
+	mz.Lock()
+	defer mz.Unlock()
+
 	mz.NumLookupsInaddr += 1
-	return []ZoneRecord{mz.record}, nil
+	res := make([]ZoneRecord, 0)
+	for _, r := range mz.records {
+		revIP, err := raddrToIP(inaddr)
+		if err != nil {
+			return nil, newParseError("lookup address", inaddr)
+		}
+		if r.IP().Equal(revIP) {
+			res = append(res, r)
+		}
+	}
+	return res, nil
+}
+
+func (mz *mockedZoneWithRecords) DomainLookupName(name string) ([]ZoneRecord, error) {
+	return mz.LookupName(name)
+}
+func (mz *mockedZoneWithRecords) DomainLookupInaddr(inaddr string) ([]ZoneRecord, error) {
+	return mz.LookupInaddr(inaddr)
 }
 
 // the following methods are not currently needed...
-func (mz mockedZone) Status() string                                       { notImplWarn(); return "nothing" }
-func (mz mockedZone) AddRecord(ident string, name string, ip net.IP) error { notImplWarn(); return nil }
-func (mz mockedZone) DeleteRecord(ident string, ip net.IP) error           { notImplWarn(); return nil }
-func (mz mockedZone) DeleteRecordsFor(ident string) error                  { notImplWarn(); return nil }
-func (mz mockedZone) DomainLookupName(name string) ([]ZoneRecord, error) {
-	notImplWarn()
-	return nil, nil
-}
-func (mz mockedZone) DomainLookupInaddr(inaddr string) ([]ZoneRecord, error) {
-	notImplWarn()
-	return nil, nil
-}
-func (mz mockedZone) ObserveName(name string, observer ZoneRecordObserver) error {
+func (mz *mockedZoneWithRecords) AddRecord(ident string, name string, ip net.IP) error {
 	notImplWarn()
 	return nil
 }
-func (mz mockedZone) ObserveInaddr(inaddr string, observer ZoneRecordObserver) error {
+func (mz *mockedZoneWithRecords) DeleteRecord(ident string, ip net.IP) error {
+	notImplWarn()
+	return nil
+}
+func (mz *mockedZoneWithRecords) DeleteRecordsFor(ident string) error { notImplWarn(); return nil }
+func (mz *mockedZoneWithRecords) Status() string                      { notImplWarn(); return "nothing" }
+func (mz *mockedZoneWithRecords) ObserveName(name string, observer ZoneRecordObserver) error {
+	notImplWarn()
+	return nil
+}
+func (mz *mockedZoneWithRecords) ObserveInaddr(inaddr string, observer ZoneRecordObserver) error {
 	notImplWarn()
 	return nil
 }

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)
 	}

nameserver/server_test.go

@@ -133,7 +133,7 @@ func TestTCPDNSServer(t *testing.T) {
 	bs := make([]byte, 4)
 	for i := 0; i < numAnswers; i++ {
 		binary.LittleEndian.PutUint32(bs, uint32(i))
-		addrs = append(addrs, Record{"", net.IPv4(bs[0], bs[1], bs[2], bs[3]), 0, 0})
+		addrs = append(addrs, Record{"", net.IPv4(bs[0], bs[1], bs[2], bs[3]), 0, 0, 0})
 	}
 
 	// handler for the fallback server: it will just return a very long response

nameserver/zone.go

@@ -85,7 +85,7 @@ func (zone *ZoneDb) LookupName(name string) ([]ZoneRecord, error) {
 	defer zone.mx.RUnlock()
 	for _, r := range zone.recs {
 		if r.Name == name {
-			return []ZoneRecord{Record{r.Name, r.IP, 0, 0}}, nil
+			return []ZoneRecord{Record{r.Name, r.IP, 0, 0, 0}}, nil
 		}
 	}
 	return nil, LookupError(name)
@@ -100,7 +100,7 @@ func (zone *ZoneDb) LookupInaddr(inaddr string) ([]ZoneRecord, error) {
 		defer zone.mx.RUnlock()
 		for _, r := range zone.recs {
 			if r.IP.Equal(ip) {
-				return []ZoneRecord{Record{r.Name, r.IP, 0, 0}}, nil
+				return []ZoneRecord{Record{r.Name, r.IP, 0, 0, 0}}, nil
 			}
 		}
 		return nil, LookupError(inaddr)