Merge pull request #615 from inercia/weave-583-2

Background updates for names in the Zone database. Fixes #226.

nameserver/addrs.go

@@ -25,13 +25,16 @@ type ZoneLookup interface {
 	LookupInaddr(inaddr string) ([]ZoneRecord, error)
 }
 
-type ZoneObserver interface {
+type ZoneLookupFunc func(target string) ([]ZoneRecord, error)
+
+type ZoneObservable interface {
 	// Observe anything that affects a particular name in the zone
 	ObserveName(name string, observer ZoneRecordObserver) error
 	// Observe anything that affects a particular IP in the zone
 	ObserveInaddr(inaddr string, observer ZoneRecordObserver) error
 }
 
+type ZoneObserverFunc func(string, ZoneRecordObserver) error
 type ZoneRecordObserver func()
 
 /////////////////////////////////////////////////////////////

nameserver/cache.go

@@ -9,7 +9,6 @@ import (
 	"github.com/miekg/dns"
 	. "github.com/weaveworks/weave/common"
 	"math"
-	"math/rand"
 	"sync"
 	"time"
 )
@@ -43,21 +42,6 @@ const (
 	CacheNoLocalReplies uint8 = 1 << iota // not found in local network (stored in the cache so we skip another local lookup or some time)
 )
 
-// shuffleAnswers reorders answers for very basic load balancing
-func shuffleAnswers(answers []dns.RR) []dns.RR {
-	n := len(answers)
-	if n > 1 {
-		rand.Seed(time.Now().UTC().UnixNano())
-
-		for i := 0; i < n; i++ {
-			r := i + rand.Intn(n-i)
-			answers[r], answers[i] = answers[i], answers[r]
-		}
-	}
-
-	return answers
-}
-
 // a cache entry
 type cacheEntry struct {
 	Status   entryStatus // status of the entry
@@ -124,9 +108,6 @@ func (e *cacheEntry) getReply(request *dns.Msg, maxLen int, now time.Time) (*dns
 	reply.Rcode = e.reply.Rcode
 	reply.Authoritative = true
 
-	// shuffle the values, etc...
-	reply.Answer = shuffleAnswers(reply.Answer)
-
 	return reply, nil
 }
 

nameserver/dns.go

@@ -2,6 +2,8 @@ package nameserver
 
 import (
 	"github.com/miekg/dns"
+	"math/rand"
+	"time"
 )
 
 const (
@@ -33,6 +35,8 @@ func makeTruncatedReply(r *dns.Msg) *dns.Msg {
 	return reply
 }
 
+type DNSResponseBuilder func(r *dns.Msg, q *dns.Question, addrs []ZoneRecord) *dns.Msg
+
 func makeAddressReply(r *dns.Msg, q *dns.Question, addrs []ZoneRecord) *dns.Msg {
 	answers := make([]dns.RR, len(addrs))
 	header := makeHeader(r, q)
@@ -84,3 +88,28 @@ func getMaxReplyLen(r *dns.Msg, proto dnsProtocol) int {
 	}
 	return maxLen
 }
+
+func init() {
+	rand.Seed(time.Now().UTC().UnixNano())
+}
+
+// shuffleAnswers reorders answers for very basic load balancing
+func shuffleAnswers(answers []dns.RR) []dns.RR {
+	if len(answers) > 1 {
+		for i := range answers {
+			j := rand.Intn(i + 1)
+			answers[i], answers[j] = answers[j], answers[i]
+		}
+	}
+
+	return answers
+}
+
+// only take the first `num` answers
+func pruneAnswers(answers []dns.RR, num int) []dns.RR {
+	if num > 0 && len(answers) > num {
+		// TODO: we should have some prefer locally-introduced answers, etc...
+		return answers[:num]
+	}
+	return answers
+}

nameserver/dns_test.go

@@ -0,0 +1,38 @@
+package nameserver
+
+import (
+	"github.com/miekg/dns"
+	. "github.com/weaveworks/weave/common"
+	wt "github.com/weaveworks/weave/testing"
+	"net"
+	"testing"
+)
+
+// Check that we can prune an answer
+func TestPrune(t *testing.T) {
+	InitDefaultLogging(testing.Verbose())
+	Info.Println("TestPrune starting")
+
+	questionMsg := new(dns.Msg)
+	questionMsg.SetQuestion("name", dns.TypeA)
+	questionMsg.RecursionDesired = true
+	question := &questionMsg.Question[0]
+	records := []ZoneRecord{
+		Record{"name", net.ParseIP("10.0.1.1"), 0, 0, 0},
+		Record{"name", net.ParseIP("10.0.1.2"), 0, 0, 0},
+		Record{"name", net.ParseIP("10.0.1.3"), 0, 0, 0},
+		Record{"name", net.ParseIP("10.0.1.4"), 0, 0, 0},
+	}
+
+	reply := makeAddressReply(questionMsg, question, records)
+	reply.Answer[0].Header().Ttl = localTTL
+
+	pruned := pruneAnswers(reply.Answer, 1)
+	wt.AssertEqualInt(t, len(pruned), 1, "wrong number of answers")
+
+	pruned = pruneAnswers(reply.Answer, 2)
+	wt.AssertEqualInt(t, len(pruned), 2, "wrong number of answers")
+
+	pruned = pruneAnswers(reply.Answer, 0)
+	wt.AssertEqualInt(t, len(pruned), len(records), "wrong number of answers")
+}

nameserver/http_test.go

@@ -24,10 +24,15 @@ func TestHttp(t *testing.T) {
 		testDomain      = "weave.local."
 		successTestName = "test1." + testDomain
 		testAddr1       = "10.2.2.1/24"
+		testAddr2       = "10.2.2.2/24"
 		dockerIP        = "9.8.7.6"
 	)
 
-	zone := NewZoneDb(DefaultLocalDomain)
+	zone, err := NewZoneDb(ZoneConfig{})
+	wt.AssertNoErr(t, err)
+	err = zone.Start()
+	wt.AssertNoErr(t, err)
+	defer zone.Stop()
 
 	port := rand.Intn(10000) + 32768
 	fmt.Println("Http test on port", port)
@@ -65,6 +70,31 @@ func TestHttp(t *testing.T) {
 	wt.AssertNoErr(t, err)
 	wt.AssertStatus(t, resp.StatusCode, http.StatusOK, "http response")
 
+	// Adding a new IP for the same name should be OK
+	addrParts2 := strings.Split(testAddr2, "/")
+	addrURL2 := fmt.Sprintf("http://localhost:%d/name/%s/%s", port, containerID, addrParts2[0])
+	resp, err = genForm("PUT", addrURL2,
+		url.Values{"fqdn": {successTestName}, "local_ip": {dockerIP}, "routing_prefix": {addrParts2[1]}})
+	wt.AssertNoErr(t, err)
+	wt.AssertStatus(t, resp.StatusCode, http.StatusOK, "http success response for second IP")
+
+	// Check that we can get two IPs for that name
+	ip2, _, _ := net.ParseCIDR(testAddr2)
+	foundIP, err = zone.LookupName(successTestName)
+	wt.AssertNoErr(t, err)
+	if len(foundIP) != 2 {
+		t.Logf("IPs found: %s", foundIP)
+		t.Fatalf("Unexpected result length: received %d responses", len(foundIP))
+	}
+	if !(foundIP[0].IP().Equal(ip) || foundIP[0].IP().Equal(ip2)) {
+		t.Logf("IPs found: %s", foundIP)
+		t.Fatalf("Unexpected result for %s: received %s, expected %s", successTestName, foundIP, ip)
+	}
+	if !(foundIP[1].IP().Equal(ip) || foundIP[1].IP().Equal(ip2)) {
+		t.Logf("IPs found: %s", foundIP)
+		t.Fatalf("Unexpected result for %s: received %s, expected %s", successTestName, foundIP, ip)
+	}
+
 	// Delete the address
 	resp, err = genForm("DELETE", addrURL, nil)
 	wt.AssertNoErr(t, err)
@@ -80,6 +110,11 @@ func TestHttp(t *testing.T) {
 	wt.AssertNoErr(t, err)
 	wt.AssertStatus(t, resp.StatusCode, http.StatusOK, "http response")
 
+	// Delete the second IP
+	resp, err = genForm("DELETE", addrURL2, nil)
+	wt.AssertNoErr(t, err)
+	wt.AssertStatus(t, resp.StatusCode, http.StatusOK, "http response")
+
 	// Check that the address is gone
 	x, err = zone.LookupName(successTestName)
 	t.Logf("Got %s", x)

nameserver/mdns.go

@@ -0,0 +1,28 @@
+package nameserver
+
+import "net"
+
+// A basic mDNS service
+type ZoneMDNS interface {
+	// Start the service
+	Start(*net.Interface) error
+	// Stop the service
+	Stop() error
+}
+
+// A mDNS server
+type ZoneMDNSServer interface {
+	ZoneMDNS
+	// Return the Zone database used by the server
+	Zone() Zone
+}
+
+// A mDNS client interface
+type ZoneMDNSClient interface {
+	ZoneMDNS
+	ZoneLookup
+	// Perform an insistent lookup for a name
+	InsistentLookupName(name string) ([]ZoneRecord, error)
+	// Perform an insistent lookup for a reverse address
+	InsistentLookupInaddr(inaddr string) ([]ZoneRecord, error)
+}

nameserver/mdns_test.go

@@ -0,0 +1,169 @@
+package nameserver
+
+import (
+	"github.com/miekg/dns"
+	. "github.com/weaveworks/weave/common"
+	wt "github.com/weaveworks/weave/testing"
+	"net"
+	"testing"
+	"time"
+)
+
+// Check that we can use a regular mDNS server with a regular mDNS client
+func TestClientServerSimpleQuery(t *testing.T) {
+	InitDefaultLogging(testing.Verbose())
+
+	testRecord1 := Record{"test.weave.local.", net.ParseIP("10.2.2.1"), 0, 0, 0}
+	testInAddr1 := "1.2.2.10.in-addr.arpa."
+
+	mzone := newMockedZoneWithRecords([]ZoneRecord{testRecord1})
+	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
+	receivedCount := 0
+
+	mdnsCli, err := NewMDNSClient()
+	wt.AssertNoErr(t, err)
+	err = mdnsCli.Start(nil)
+	wt.AssertNoErr(t, err)
+
+	sendQuery := func(name string, querytype uint16) {
+		receivedAddr = nil
+		receivedName = ""
+		receivedCount = 0
+		Debug.Printf("Sending query...")
+		switch querytype {
+		case dns.TypeA:
+			r, err := mdnsCli.LookupName(name)
+			if len(r) > 0 {
+				receivedAddr = r[0].IP()
+				if err == nil {
+					receivedCount++
+				}
+			}
+		case dns.TypePTR:
+			r, err := mdnsCli.LookupInaddr(name)
+			if len(r) > 0 {
+				receivedName = r[0].Name()
+				if err == nil {
+					receivedCount++
+				}
+			}
+		}
+	}
+
+	time.Sleep(100 * time.Millisecond) // Allow for server to get going
+
+	Debug.Printf("Query: %s dns.TypeA", testRecord1.Name())
+	sendQuery(testRecord1.Name(), dns.TypeA)
+	if receivedCount != 1 {
+		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())
+	}
+
+	Debug.Printf("Query: testfail.weave. dns.TypeA")
+	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)
+	sendQuery(testInAddr1, dns.TypePTR)
+	if receivedCount != 1 {
+		t.Fatalf("Expected an answer to %s, got %d answers", testInAddr1, receivedCount)
+	} else if !(testRecord1.Name() == receivedName) {
+		t.Fatalf("Expected answer %s to query for %s, got %s", testRecord1.Name(), testInAddr1, receivedName)
+	}
+}
+
+// Check that we can use a use "insistent" queries
+func TestClientServerInsistentQuery(t *testing.T) {
+	InitDefaultLogging(testing.Verbose())
+
+	testRecord1 := Record{"test.weave.local.", net.ParseIP("10.2.2.1"), 0, 0, 0}
+	testInAddr1 := "1.2.2.10.in-addr.arpa."
+	testRecord2 := Record{"test.weave.local.", net.ParseIP("10.2.2.2"), 0, 0, 0}
+	testInAddr2 := "2.2.2.10.in-addr.arpa."
+
+	mzone1 := newMockedZoneWithRecords([]ZoneRecord{testRecord1})
+	mdnsServer1, err := NewMDNSServer(mzone1)
+	wt.AssertNoErr(t, err)
+	err = mdnsServer1.Start(nil)
+	wt.AssertNoErr(t, err)
+	defer mdnsServer1.Stop()
+
+	mzone2 := newMockedZoneWithRecords([]ZoneRecord{testRecord2})
+	mdnsServer2, err := NewMDNSServer(mzone2)
+	wt.AssertNoErr(t, err)
+	err = mdnsServer2.Start(nil)
+	wt.AssertNoErr(t, err)
+	defer mdnsServer2.Stop()
+
+	// create a third server with exactly the same info as the second server (so we can test duplicates removals)
+	mdnsServer3, err := NewMDNSServer(mzone2)
+	wt.AssertNoErr(t, err)
+	err = mdnsServer3.Start(nil)
+	wt.AssertNoErr(t, err)
+	defer mdnsServer3.Stop()
+
+	var receivedAddrs []ZoneRecord
+	var receivedNames []ZoneRecord
+	receivedCount := 0
+
+	mdnsCli, err := NewMDNSClient()
+	wt.AssertNoErr(t, err)
+	err = mdnsCli.Start(nil)
+	wt.AssertNoErr(t, err)
+
+	sendQuery := func(name string, querytype uint16) {
+		receivedAddrs = nil
+		receivedNames = nil
+		receivedCount = 0
+		Debug.Printf("Sending query...")
+		switch querytype {
+		case dns.TypeA:
+			receivedAddrs, err = mdnsCli.InsistentLookupName(name)
+			if err == nil {
+				receivedCount = len(receivedAddrs)
+			}
+		case dns.TypePTR:
+			receivedNames, err = mdnsCli.InsistentLookupInaddr(name)
+			if err == nil {
+				receivedCount = len(receivedNames)
+			}
+		}
+	}
+
+	time.Sleep(100 * time.Millisecond) // Allow for server to get going
+
+	Debug.Printf("Query: %s dns.TypeA", testRecord1.Name())
+	sendQuery(testRecord1.Name(), dns.TypeA)
+	if receivedCount != 2 {
+		t.Fatalf("Unexpected result count %d for %s", receivedCount, testRecord1.Name())
+	}
+
+	Debug.Printf("Query: testfail.weave. dns.TypeA")
+	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)
+	sendQuery(testInAddr1, dns.TypePTR)
+	if receivedCount != 1 {
+		t.Fatalf("Expected an answer to %s, got %d answers", testInAddr1, receivedCount)
+	}
+
+	Debug.Printf("Query: %s dns.TypePTR", testInAddr2)
+	sendQuery(testInAddr2, dns.TypePTR)
+	if receivedCount != 1 {
+		t.Fatalf("Expected an answer to %s, got %d answers", testInAddr2, receivedCount)
+	}
+}

nameserver/mocks_test.go

@@ -1,9 +1,16 @@
 package nameserver
 
 import (
+	"fmt"
+	"github.com/benbjohnson/clock"
+	"github.com/miekg/dns"
 	. "github.com/weaveworks/weave/common"
+	wt "github.com/weaveworks/weave/testing"
 	"net"
+	"strconv"
 	"sync"
+	"testing"
+	"time"
 )
 
 const (
@@ -89,3 +96,403 @@ func (mz *mockedZoneWithRecords) ObserveInaddr(inaddr string, observer ZoneRecor
 	notImplWarn()
 	return nil
 }
+
+//////////////////////////////////////////////////////////////////
+
+// A mocked mDNS server
+type mockedMDNSServer struct {
+	zone Zone
+}
+
+func newMockedMDNSServer(z Zone) *mockedMDNSServer        { return &mockedMDNSServer{zone: z} }
+func (ms *mockedMDNSServer) Start(_ *net.Interface) error { return nil }
+func (ms *mockedMDNSServer) Stop() error                  { return nil }
+func (ms *mockedMDNSServer) Zone() Zone                   { return ms.zone }
+
+// a useful mix of NewMockedZone and NewMockedMDNSServer
+func newMockedMDNSServerWithRecord(zr ZoneRecord) *mockedMDNSServer {
+	return newMockedMDNSServer(newMockedZoneWithRecords([]ZoneRecord{zr}))
+}
+func newMockedMDNSServerWithRecords(zrs []ZoneRecord) *mockedMDNSServer {
+	return newMockedMDNSServer(newMockedZoneWithRecords(zrs))
+}
+
+//////////////////////////////////////////////////////////////////
+
+// A mocked mDNS client.
+// This mock asks a group of (potentially mocked) mDNS servers
+type mockedMDNSClient struct {
+	sync.RWMutex
+
+	servers map[ZoneMDNSServer]ZoneMDNSServer
+
+	// Statistics
+	NumLookupsName      int
+	NumLookupsInaddr    int
+	NumInsLookupsName   int
+	NumInsLookupsInaddr int
+}
+
+func newMockedMDNSClient(srvrs []*mockedMDNSServer) *mockedMDNSClient {
+	r := mockedMDNSClient{
+		servers: make(map[ZoneMDNSServer]ZoneMDNSServer),
+	}
+	if srvrs != nil {
+		for _, server := range srvrs {
+			r.servers[server] = server
+		}
+	}
+	return &r
+}
+
+func (mc *mockedMDNSClient) Start(ifi *net.Interface) error { return nil }
+func (mc *mockedMDNSClient) Stop() error                    { return nil }
+func (mc *mockedMDNSClient) Domain() string                 { return DefaultLocalDomain }
+func (mc *mockedMDNSClient) LookupName(name string) ([]ZoneRecord, error) {
+	mc.Lock()
+	defer mc.Unlock()
+
+	// get a random result
+	mc.NumLookupsName += 1
+	for _, server := range mc.servers {
+		r, err := server.Zone().LookupName(name)
+		if err == nil {
+			return r, nil // return the first answer
+		}
+	}
+	return nil, LookupError(name)
+}
+
+func (mc *mockedMDNSClient) LookupInaddr(inaddr string) ([]ZoneRecord, error) {
+	mc.Lock()
+	defer mc.Unlock()
+
+	// get a random result
+	mc.NumLookupsInaddr += 1
+	for _, server := range mc.servers {
+		r, err := server.Zone().LookupInaddr(inaddr)
+		if err == nil {
+			return r, nil // return the first answer
+		}
+	}
+	return nil, LookupError(inaddr)
+}
+
+func (mc *mockedMDNSClient) InsistentLookupName(name string) ([]ZoneRecord, error) {
+	mc.Lock()
+	defer mc.Unlock()
+
+	res := make([]ZoneRecord, 0)
+	mc.NumInsLookupsName += 1
+	for _, server := range mc.servers {
+		r, err := server.Zone().LookupName(name)
+		if err == nil {
+			res = append(res, r...)
+		}
+	}
+	if len(res) == 0 {
+		return nil, LookupError(name)
+	}
+	return res, nil
+}
+
+func (mc *mockedMDNSClient) InsistentLookupInaddr(inaddr string) ([]ZoneRecord, error) {
+	mc.Lock()
+	defer mc.Unlock()
+
+	res := make([]ZoneRecord, 0)
+	mc.NumInsLookupsInaddr += 1
+	for _, server := range mc.servers {
+		r, err := server.Zone().LookupInaddr(inaddr)
+		if err == nil {
+			res = append(res, r...)
+		}
+	}
+	if len(res) == 0 {
+		return nil, LookupError(inaddr)
+	}
+	return res, nil
+}
+
+func (mc *mockedMDNSClient) AddServer(srv ZoneMDNSServer) {
+	mc.Lock()
+	defer mc.Unlock()
+	mc.servers[srv] = srv
+}
+
+func (mc *mockedMDNSClient) RemoveServer(srv ZoneMDNSServer) {
+	mc.Lock()
+	defer mc.Unlock()
+	delete(mc.servers, srv)
+}
+
+//////////////////////////////////////////////////////////////////
+
+type zbmEntry struct {
+	Server *mockedMDNSServer
+	Client *mockedMDNSClient
+	Zone   *zoneDb
+}
+type zoneDbsWithMockedMDns []*zbmEntry
+
+// Creates a group of (real) zone databases linked through mocked mDNS servers and clients
+// This effectively creates a cluster of WeaveDNS peers
+func newZoneDbsWithMockedMDns(num int, config ZoneConfig) zoneDbsWithMockedMDns {
+	res := make(zoneDbsWithMockedMDns, num)
+
+	for i := 0; i < num; i++ {
+		res[i] = new(zbmEntry)
+
+		Debug.Printf("[test] Creating mocked mDNS server #%d", i)
+		res[i].Server = newMockedMDNSServer(nil)
+
+		Debug.Printf("[test] Creating mocked mDNS client #%d", i)
+		res[i].Client = newMockedMDNSClient([]*mockedMDNSServer{})
+
+		cfg := config
+		cfg.MDNSClient = res[i].Client
+		cfg.MDNSServer = res[i].Server
+
+		Debug.Printf("[test] Creating ZoneDb #%d", i)
+		res[i].Zone, _ = NewZoneDb(cfg)
+
+		// link the mDNS server to its zone
+		// ZoneDbs are connected to mDNS in two directions: they use their mDNS
+		// clients for asking other peers about names/IPs, and export their records
+		// through a mDNS server.
+		res[i].Server.zone = res[i].Zone
+	}
+
+	// link all the clients to all the servers
+	for i := 0; i < num; i++ {
+		for j := 0; j < num; j++ {
+			// with this check, the local client will not ask to the local server
+			// this is not what we currently do, but can be convenient in some cases
+			// for finding some bugs...
+			if i != j {
+				Debug.Printf("[test] Linking mocked mDNS client #%d to server #%d", i, j)
+				res[i].Client.AddServer(res[j].Server)
+			}
+		}
+	}
+
+	return res
+}
+
+func (zbs zoneDbsWithMockedMDns) Start() {
+	for _, entry := range zbs {
+		entry.Zone.Start()
+	}
+}
+
+func (zbs zoneDbsWithMockedMDns) Stop() {
+	for _, entry := range zbs {
+		entry.Zone.Stop()
+	}
+}
+
+//////////////////////////////////////////////////////////////////
+
+// A mocked cache where we never find a single thing... ;)
+type mockedCache struct {
+	NumGets     int
+	NumPuts     int
+	NumRemovals int
+}
+
+func newMockedCache() *mockedCache { return &mockedCache{} }
+
+func (c *mockedCache) Get(request *dns.Msg, maxLen int) (reply *dns.Msg, err error) {
+	c.NumGets += 1
+	return nil, nil
+}
+func (c *mockedCache) Put(request *dns.Msg, reply *dns.Msg, ttl int, flags uint8) int {
+	c.NumPuts += 1
+	return 0
+}
+func (c *mockedCache) Remove(question *dns.Question) { c.NumRemovals += 1 }
+func (c *mockedCache) Purge()                        {}
+func (c *mockedCache) Clear()                        {}
+func (c *mockedCache) Len() int                      { return 0 }
+func (c *mockedCache) Capacity() int                 { return DefaultCacheLen }
+
+//////////////////////////////////////////////////////////////////
+
+// A mocked fallback server
+type mockedFallback struct {
+	CliConfig *dns.ClientConfig
+	Addr      string
+	Port      int
+
+	udpSrv *dns.Server
+	tcpSrv *dns.Server
+}
+
+// Create a mocked fallback server
+// You can use the server's `CliConfig` as the `UpstreamCfg` of a `DNSServer`...
+func newMockedFallback(udpH dns.HandlerFunc, tcpH dns.HandlerFunc) (*mockedFallback, error) {
+	udpSrv, fallbackAddr, err := runLocalUDPServer("127.0.0.1:0", udpH)
+	if err != nil {
+		return nil, err
+	}
+	_, fallbackPort, err := net.SplitHostPort(fallbackAddr)
+	if err != nil {
+		return nil, err
+	}
+
+	fallbackPortI, _ := strconv.Atoi(fallbackPort)
+
+	res := mockedFallback{
+		udpSrv: udpSrv,
+		Addr:   fallbackAddr,
+		Port:   fallbackPortI,
+		CliConfig: &dns.ClientConfig{
+			Servers: []string{"127.0.0.1"},
+			Port:    fallbackPort,
+		},
+	}
+
+	if tcpH != nil {
+		fallbackTCPAddr := fmt.Sprintf("127.0.0.1:%s", fallbackPort)
+		res.tcpSrv, _, err = runLocalTCPServer(fallbackTCPAddr, tcpH)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	return &res, nil
+}
+
+// Start the fallback server
+func (mf *mockedFallback) Start() error {
+	return nil
+}
+
+// Stop the fallback server
+func (mf *mockedFallback) Stop() error {
+	if mf.tcpSrv != nil {
+		mf.udpSrv.Shutdown()
+		mf.udpSrv = nil
+	}
+	if mf.tcpSrv != nil {
+		mf.tcpSrv.Shutdown()
+		mf.udpSrv = nil
+	}
+	return nil
+}
+
+// Run a UDP fallback server
+func runLocalUDPServer(laddr string, handler dns.HandlerFunc) (*dns.Server, string, error) {
+	Debug.Printf("[mocked fallback] Starting fallback UDP server at %s", laddr)
+	pc, err := net.ListenPacket("udp", laddr)
+	if err != nil {
+		return nil, "", err
+	}
+	server := &dns.Server{PacketConn: pc, Handler: handler, ReadTimeout: testSocketTimeout * time.Millisecond}
+
+	go func() {
+		server.ActivateAndServe()
+		pc.Close()
+	}()
+
+	Debug.Printf("[mocked fallback] Fallback UDP server listening at %s", pc.LocalAddr())
+	return server, pc.LocalAddr().String(), nil
+}
+
+// Run a TCP fallback server
+func runLocalTCPServer(laddr string, handler dns.HandlerFunc) (*dns.Server, string, error) {
+	Debug.Printf("[mocked fallback] Starting fallback TCP server at %s", laddr)
+	laddrTCP, err := net.ResolveTCPAddr("tcp", laddr)
+	if err != nil {
+		return nil, "", err
+	}
+
+	l, err := net.ListenTCP("tcp", laddrTCP)
+	if err != nil {
+		return nil, "", err
+	}
+	server := &dns.Server{Listener: l, Handler: handler, ReadTimeout: testSocketTimeout * time.Millisecond}
+
+	go func() {
+		server.ActivateAndServe()
+		l.Close()
+	}()
+
+	Debug.Printf("[mocked fallback] Fallback TCP server listening at %s", l.Addr().String())
+	return server, l.Addr().String(), nil
+}
+
+//////////////////////////////////////////////////////////////////
+
+type mockedClock struct {
+	*clock.Mock
+}
+
+func newMockedClock() *mockedClock {
+	return &mockedClock{clock.NewMock()}
+}
+
+func (clk *mockedClock) Forward(secs int) {
+	Debug.Printf(">>>>>>> Moving clock forward %d seconds - Time traveling >>>>>>>", secs)
+	clk.Add(time.Duration(secs) * time.Second)
+	Debug.Printf("<<<<<<< Time travel finished! We are at %s <<<<<<<", clk.Now())
+}
+
+//////////////////////////////////////////////////////////////////
+
+// Perform a DNS query and assert the reply code, number or answers, etc
+func assertExchange(t *testing.T, z string, ty uint16, port int, minAnswers int, maxAnswers int, expErr int) (*dns.Msg, *dns.Msg) {
+	wt.AssertNotEqualInt(t, port, 0, "invalid DNS server port")
+
+	c := &dns.Client{
+		UDPSize: testUDPBufSize,
+	}
+
+	m := new(dns.Msg)
+	m.RecursionDesired = true
+	m.SetQuestion(z, ty)
+	m.SetEdns0(testUDPBufSize, false) // we don't want to play with truncation here...
+
+	lstAddr := fmt.Sprintf("127.0.0.1:%d", port)
+	r, _, err := c.Exchange(m, lstAddr)
+	t.Logf("Response from '%s':\n%+v\n", lstAddr, r)
+	if err != nil {
+		t.Fatalf("Error when querying DNS server at %s: %s", lstAddr, err)
+	}
+	wt.AssertNoErr(t, err)
+	if minAnswers == 0 && maxAnswers == 0 {
+		wt.AssertStatus(t, r.Rcode, expErr, "DNS response code")
+	} else {
+		wt.AssertStatus(t, r.Rcode, dns.RcodeSuccess, "DNS response code")
+	}
+	answers := len(r.Answer)
+	if minAnswers >= 0 && answers < minAnswers {
+		wt.Fatalf(t, "Number of answers >= %d", minAnswers)
+	}
+	if maxAnswers >= 0 && answers > maxAnswers {
+		wt.Fatalf(t, "Number of answers <= %d", maxAnswers)
+	}
+	return m, r
+}
+
+// Assert that we have a response in the cache for a query `q`
+func assertInCache(t *testing.T, cache ZoneCache, q *dns.Msg, desc string) {
+	r, err := cache.Get(q, maxUDPSize)
+	wt.AssertNoErr(t, err)
+	wt.AssertNotNil(t, r, fmt.Sprintf("value in the cache: %s", desc))
+}
+
+// Assert that we have a response in the cache for a query `q`
+func assertNotLocalInCache(t *testing.T, cache ZoneCache, q *dns.Msg, desc string) {
+	r, err := cache.Get(q, maxUDPSize)
+	if !(r == nil && err == errNoLocalReplies) {
+		t.Fatalf("Cache does not return a noLocalReplies error for query %s", q)
+	}
+}
+
+// Assert that we do not have a response in the cache for a query `q`
+func assertNotInCache(t *testing.T, cache ZoneCache, q *dns.Msg, desc string) {
+	r, err := cache.Get(q, maxUDPSize)
+	wt.AssertNoErr(t, err)
+	wt.AssertNil(t, r, fmt.Sprintf("value in the cache: %s\n%s", desc, r))
+}

nameserver/server.go

@@ -3,37 +3,51 @@ package nameserver
 import (
 	"bytes"
 	"fmt"
+	"github.com/benbjohnson/clock"
 	"github.com/miekg/dns"
 	. "github.com/weaveworks/weave/common"
 	"net"
+	"strconv"
 	"sync"
+	"time"
 )
 
 const (
-	DefaultLocalDomain   = "weave.local."     // The default name used for the local domain
-	DefaultServerPort    = 53                 // The default server port
-	DefaultCLICfgFile    = "/etc/resolv.conf" // default "resolv.conf" file to try to load
-	DefaultUDPBuflen     = 4096               // bigger than the default 512
-	DefaultCacheLen      = 8192               // default cache capacity
-	DefaultResolvWorkers = 8                  // default number of resolution workers
-	DefaultTimeout       = 5                  // default timeout for DNS resolutions
+	DefaultServerPort = 53                 // The default server port
+	DefaultCLICfgFile = "/etc/resolv.conf" // default "resolv.conf" file to try to load
+	DefaultUDPBuflen  = 4096               // bigger than the default 512
+	DefaultCacheLen   = 8192               // default cache capacity
+	DefaultTimeout    = 5                  // default timeout for DNS resolutions
+	DefaultMaxAnswers = 1                  // default number of answers provided to users
 )
 
 type DNSServerConfig struct {
+	// The zone
+	Zone Zone
 	// (Optional) client config file for resolving upstream servers
 	UpstreamCfgFile string
 	// (Optional) DNS client config for the fallback server(s)
 	UpstreamCfg *dns.ClientConfig
 	// (Optional) port number (for TCP and UDP)
 	Port int
-	// (Optional) local domain (ie, "weave.local.")
-	LocalDomain string
 	// (Optional) cache size
 	CacheLen int
+	// (Optional) disable the cache
+	CacheDisabled bool
 	// (Optional) timeout for DNS queries
 	Timeout int
 	// (Optional) UDP buffer length
 	UDPBufLen int
+	// (Optional) force a specific cache
+	Cache ZoneCache
+	// (Optional) TTL for negative results in the local domain
+	CacheNegLocalTTL int
+	// (Optional) for a specific clock provider
+	Clock clock.Clock
+	// (Optional) Listening socket read timeout (in milliseconds)
+	ListenReadTimeout int
+	// Maximum number of answers provided to users
+	MaxAnswers int
 }
 
 type dnsProtocol uint8
@@ -54,51 +68,66 @@ func (proto dnsProtocol) String() string {
 }
 
 // get a new dns.Client for a protocol
-func (proto dnsProtocol) GetNewClient(bufsize int) *dns.Client {
+func (proto dnsProtocol) GetNewClient(bufsize int, timeout time.Duration) *dns.Client {
 	switch proto {
 	case protTCP:
-		return &dns.Client{Net: "tcp"}
+		return &dns.Client{Net: "tcp", ReadTimeout: timeout}
 	case protUDP:
-		return &dns.Client{Net: "udp", UDPSize: uint16(bufsize)}
+		return &dns.Client{Net: "udp", ReadTimeout: timeout, UDPSize: uint16(bufsize)}
 	}
 	return nil
 }
 
+// a DNS server
 type DNSServer struct {
-	Zone        Zone
-	Iface       *net.Interface
-	Upstream    *dns.ClientConfig
-	udpSrv      *dns.Server
-	tcpSrv      *dns.Server
-	mdnsCli     *MDNSClient
-	mdnsSrv     *MDNSServer
-	cache       *Cache
-	timeout     int
-	udpBuf      int
-	listenersWg *sync.WaitGroup
-
+	Zone       Zone
+	Upstream   *dns.ClientConfig
 	Domain     string // the local domain
 	ListenAddr string // the address the server is listening at
+
+	udpSrv        *dns.Server
+	tcpSrv        *dns.Server
+	pc            net.PacketConn
+	lst           net.Listener
+	cache         ZoneCache
+	cacheDisabled bool
+	maxAnswers    int
+	negLocalTTL   int
+	timeout       time.Duration
+	readTimeout   time.Duration
+	udpBuf        int
+	listenersWg   *sync.WaitGroup
+	clock         clock.Clock
 }
 
 // Creates a new DNS server
-func NewDNSServer(config DNSServerConfig, zone Zone, iface *net.Interface) (s *DNSServer, err error) {
+func NewDNSServer(config DNSServerConfig) (s *DNSServer, err error) {
 	s = &DNSServer{
-		Zone:        zone,
-		Iface:       iface,
-		listenersWg: new(sync.WaitGroup),
-
+		Zone:       config.Zone,
 		Domain:     DefaultLocalDomain,
-		ListenAddr: fmt.Sprintf(":%d", DefaultServerPort),
+		ListenAddr: fmt.Sprintf(":%d", config.Port),
+
+		listenersWg:   new(sync.WaitGroup),
+		timeout:       time.Duration(config.Timeout) * time.Millisecond,
+		readTimeout:   time.Duration(config.ListenReadTimeout) * time.Millisecond,
+		cacheDisabled: false,
+		maxAnswers:    DefaultMaxAnswers,
+		negLocalTTL:   negLocalTTL,
+		clock:         config.Clock,
+	}
+
+	// check some basic parameters are valid
+	if s.Zone == nil {
+		return nil, fmt.Errorf("No valid Zone provided in server initialization")
+	}
+	if len(s.Domain) == 0 {
+		return nil, fmt.Errorf("No valid Domain provided in server initialization")
+	}
+	if s.clock == nil {
+		s.clock = clock.New()
 	}
 
 	// fill empty parameters with defaults...
-	if config.Port != 0 {
-		s.ListenAddr = fmt.Sprintf(":%d", config.Port)
-	}
-	if len(config.LocalDomain) > 0 {
-		s.Domain = config.LocalDomain
-	}
 	if config.UpstreamCfg != nil {
 		s.Upstream = config.UpstreamCfg
 	} else {
@@ -111,207 +140,226 @@ func NewDNSServer(config DNSServerConfig, zone Zone, iface *net.Interface) (s *D
 		}
 	}
 	if config.Timeout > 0 {
-		s.timeout = config.Timeout
+		s.timeout = time.Duration(config.Timeout) * time.Millisecond
 	}
 	if config.UDPBufLen > 0 {
 		s.udpBuf = config.UDPBufLen
 	}
-	s.mdnsCli, err = NewMDNSClient()
-	if err != nil {
-		return
+	if config.MaxAnswers > 0 {
+		s.maxAnswers = config.MaxAnswers
 	}
-	s.mdnsSrv, err = NewMDNSServer(s.Zone)
-	if err != nil {
-		return
+	if config.CacheNegLocalTTL > 0 {
+		s.negLocalTTL = config.CacheNegLocalTTL
 	}
-	cacheLen := DefaultCacheLen
-	if config.CacheLen > 0 {
-		cacheLen = config.CacheLen
+	if config.CacheDisabled {
+		s.cacheDisabled = true
 	}
-	Debug.Printf("[dns] Initializing cache: %d entries", config.CacheLen)
-	s.cache, err = NewCache(cacheLen, nil)
-	if err != nil {
-		return
+	if !s.cacheDisabled {
+		if config.Cache != nil {
+			s.cache = config.Cache
+		} else {
+			cacheLen := DefaultCacheLen
+			if config.CacheLen > 0 {
+				cacheLen = config.CacheLen
+			}
+			if s.cache, err = NewCache(cacheLen, s.clock); err != nil {
+				return
+			}
+		}
 	}
 
-	// create two DNS request multiplexerers, depending on the protocol used by clients
-	// (we use the same protocol for asking upstream servers)
-	mux := func(proto dnsProtocol) *dns.ServeMux {
-		m := dns.NewServeMux()
-		m.HandleFunc(s.Domain, s.queryHandler(proto))
-		m.HandleFunc(RDNSDomain, s.rdnsHandler(proto))
-		m.HandleFunc(".", s.notUsHandler(proto))
-		return m
-	}
-
-	s.udpSrv = &dns.Server{Addr: s.ListenAddr, Net: "udp", Handler: mux(protUDP)}
-	s.tcpSrv = &dns.Server{Addr: s.ListenAddr, Net: "tcp", Handler: mux(protTCP)}
-
 	return
 }
 
 // Start the DNS server
 func (s *DNSServer) Start() error {
-	Info.Printf("Using mDNS on %v", s.Iface)
-	err := s.mdnsCli.Start(s.Iface)
-	CheckFatal(err)
-	err = s.mdnsSrv.Start(s.Iface)
-	CheckFatal(err)
+	Info.Printf("[dns] Upstream server(s): %+v", s.Upstream)
+	if s.cacheDisabled {
+		Info.Printf("[dns] Cache: disabled")
+	} else {
+		Info.Printf("[dns] Cache: %d entries", s.cache.Capacity())
+	}
+
+	pc, err := net.ListenPacket("udp", s.ListenAddr)
+	if err != nil {
+		return err
+	}
+	s.pc = pc
+
+	_, port, err := net.SplitHostPort(pc.LocalAddr().String())
+	if err != nil {
+		return err
+	}
+	s.ListenAddr = fmt.Sprintf(":%s", port)
+	s.udpSrv = &dns.Server{PacketConn: s.pc, Handler: s.createMux(protUDP), ReadTimeout: s.readTimeout}
+
+	// Bind the TCP socket at the same port, aborting otherwise
+	l, err := net.Listen("tcp", s.ListenAddr)
+	if err != nil {
+		s.Stop()
+		return err
+	}
+	s.lst = l
+	s.tcpSrv = &dns.Server{Listener: l, Handler: s.createMux(protTCP), ReadTimeout: s.readTimeout}
 
 	s.listenersWg.Add(2)
 
 	go func() {
 		defer s.listenersWg.Done()
 
-		Debug.Printf("Listening for DNS on %s (UDP)", s.ListenAddr)
-		err = s.udpSrv.ListenAndServe()
+		Info.Printf("[dns] Listening for DNS on %s (UDP)", s.ListenAddr)
+		err := s.udpSrv.ActivateAndServe()
 		CheckFatal(err)
-		Debug.Printf("DNS UDP server exiting...")
+		Debug.Printf("[dns] DNS UDP server exiting...")
 	}()
 
 	go func() {
 		defer s.listenersWg.Done()
 
-		Debug.Printf("Listening for DNS on %s (TCP)", s.ListenAddr)
-		err = s.tcpSrv.ListenAndServe()
+		Info.Printf("[dns] Listening for DNS on %s (TCP)", s.ListenAddr)
+		err := s.tcpSrv.ActivateAndServe()
 		CheckFatal(err)
-		Debug.Printf("DNS TCP server exiting...")
+		Debug.Printf("[dns] DNS TCP server exiting...")
 	}()
 
 	// Waiting for all goroutines to finish (otherwise they die as main routine dies)
 	s.listenersWg.Wait()
 
-	Info.Printf("WeaveDNS server exiting...")
+	Info.Printf("[dns] Server exiting...")
 	return nil
 }
 
 // Return status string
 func (s *DNSServer) Status() string {
 	var buf bytes.Buffer
-	fmt.Fprintln(&buf, "Local domain", s.Domain)
 	fmt.Fprintln(&buf, "Listen address", s.ListenAddr)
-	fmt.Fprintln(&buf, "mDNS interface", s.Iface)
 	fmt.Fprintln(&buf, "Fallback DNS config", s.Upstream)
-	fmt.Fprintf(&buf, "Zone database:\n%s", s.Zone)
 	return buf.String()
 }
 
 // Perform a graceful shutdown
 func (s *DNSServer) Stop() error {
 	// Stop the listeners/handlers
-	if err := s.tcpSrv.Shutdown(); err != nil {
-		return err
+	if s.tcpSrv != nil {
+		if err := s.tcpSrv.Shutdown(); err != nil {
+			return err
+		}
+		s.lst.Close()
+		s.tcpSrv = nil
 	}
-	if err := s.udpSrv.Shutdown(); err != nil {
-		return err
+	if s.udpSrv != nil {
+		if err := s.udpSrv.Shutdown(); err != nil {
+			return err
+		}
+		s.pc.Close()
+		s.udpSrv = nil
 	}
 	s.listenersWg.Wait()
-
-	// shutdown the mDNS server
-	s.mdnsSrv.Stop()
-
 	return nil
 }
 
-func (s *DNSServer) queryHandler(proto dnsProtocol) dns.HandlerFunc {
-	zoneLookup := func(lookup ZoneLookup, q *dns.Question, r *dns.Msg) (*dns.Msg, []ZoneRecord, error) {
-		ips, err := lookup.LookupName(q.Name)
-		if err != nil {
-			return nil, nil, err
-		}
-		return makeAddressReply(r, q, ips), ips, nil
-	}
-
-	fallback := func(w dns.ResponseWriter, r *dns.Msg) {
+// Create a multiplexer for requests
+// We must create two DNS request multiplexers, depending on the protocol used by
+// clients (as we use the same protocol for asking upstream servers)
+func (s *DNSServer) createMux(proto dnsProtocol) *dns.ServeMux {
+	failFallback := func(w dns.ResponseWriter, r *dns.Msg) {
 		w.WriteMsg(makeDNSFailResponse(r))
 	}
-
-	return s.commonQueryHandler(proto, "Query", dns.TypeA, zoneLookup, fallback)
-}
-
-func (s *DNSServer) rdnsHandler(proto dnsProtocol) dns.HandlerFunc {
-	zoneLookup := func(lookup ZoneLookup, q *dns.Question, r *dns.Msg) (*dns.Msg, []ZoneRecord, error) {
-		names, err := lookup.LookupInaddr(q.Name)
-		if err != nil {
-			return nil, nil, err
-		}
-		return makePTRReply(r, q, names), names, nil
-	}
-
 	notUsHandler := s.notUsHandler(proto)
-	fallback := func(w dns.ResponseWriter, r *dns.Msg) {
+	notUsFallback := func(w dns.ResponseWriter, r *dns.Msg) {
 		Info.Printf("[dns msgid %d] -> sending to fallback server", r.MsgHdr.Id)
 		notUsHandler(w, r)
 	}
 
-	return s.commonQueryHandler(proto, "Reverse query", dns.TypePTR, zoneLookup, fallback)
+	// create the multiplexer
+	m := dns.NewServeMux()
+	m.HandleFunc(s.Zone.Domain(), s.localHandler(proto, "Query", dns.TypeA,
+		s.Zone.DomainLookupName, makeAddressReply, s.Zone.ObserveName, failFallback))
+	m.HandleFunc(RDNSDomain, s.localHandler(proto, "Reverse query", dns.TypePTR,
+		s.Zone.DomainLookupInaddr, makePTRReply, s.Zone.ObserveInaddr, notUsFallback))
+	m.HandleFunc(".", s.notUsHandler(proto))
+	return m
 }
 
-func (s *DNSServer) commonQueryHandler(proto dnsProtocol, kind string, qtype uint16,
-	zoneLookup func(ZoneLookup, *dns.Question, *dns.Msg) (*dns.Msg, []ZoneRecord, error), fallback dns.HandlerFunc) dns.HandlerFunc {
+// Process a request for a name/address in the '.weave.local.' zone
+func (s *DNSServer) localHandler(proto dnsProtocol, kind string, qtype uint16,
+	lookup func(inaddr string) ([]ZoneRecord, error),
+	msgBuilder DNSResponseBuilder, observer ZoneObserverFunc, fallback dns.HandlerFunc) dns.HandlerFunc {
 
 	return func(w dns.ResponseWriter, r *dns.Msg) {
 		q := r.Question[0]
 		Debug.Printf("[dns msgid %d] %s: %+v", r.MsgHdr.Id, kind, q)
 		maxLen := getMaxReplyLen(r, proto)
-		lookups := []ZoneLookup{s.Zone, s.mdnsCli}
 
-		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)
-				lookups = []ZoneLookup{s.Zone}
-			} else {
-				Debug.Printf("[dns msgid %d] Error from cache: %s", r.MsgHdr.Id, err)
-				w.WriteMsg(makeDNSFailResponse(r))
-				return
+		// cache a response if the cache is enabled, and observe the name/IP
+		maybeCache := func(m *dns.Msg, ttl int, flags uint8) {
+			if !s.cacheDisabled {
+				s.cache.Put(r, m, ttl, flags)
+				// any change in the Zone database for this IP/name will lead to this
+				// cache entry being removed...
+				// TODO: this closure results in unnecessary `Remove`s and some wasted
+				// mem... but we can live with that.
+				observer(q.Name, func() { s.cache.Remove(&q) })
 			}
 		}
-		if reply != nil {
-			Debug.Printf("[dns msgid %d] Returning reply from cache: %s/%d answers",
-				r.MsgHdr.Id, dns.RcodeToString[reply.MsgHdr.Rcode], len(reply.Answer))
-			w.WriteMsg(reply)
-			return
+
+		if !s.cacheDisabled {
+			reply, err := s.cache.Get(r, maxLen)
+			if err != nil {
+				if err == errNoLocalReplies {
+					Debug.Printf("[dns msgid %d] Cached no-local-replies", r.MsgHdr.Id)
+					fallback(w, r)
+				} else {
+					Debug.Printf("[dns msgid %d] Error from cache: %s", r.MsgHdr.Id, err)
+					w.WriteMsg(makeDNSFailResponse(r))
+				}
+				return
+			}
+			if reply != nil {
+				reply.Answer = pruneAnswers(shuffleAnswers(reply.Answer), s.maxAnswers)
+				Debug.Printf("[dns msgid %d] Returning reply from cache: %s/%d answers",
+					r.MsgHdr.Id, dns.RcodeToString[reply.MsgHdr.Rcode], len(reply.Answer))
+				w.WriteMsg(reply)
+				return
+			}
 		}
 
 		// catch unsupported queries
 		if q.Qtype != qtype {
 			Debug.Printf("[dns msgid %d] Unsupported query type %s", r.MsgHdr.Id, dns.TypeToString[q.Qtype])
 			m := makeDNSFailResponse(r)
-			s.cache.Put(r, m, negLocalTTL, 0)
+			maybeCache(m, s.negLocalTTL, 0)
 			w.WriteMsg(m)
 			return
 		}
 
-		for _, lookup := range lookups {
-			if m, answers, err := zoneLookup(lookup, &q, r); err == nil {
-				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)
-				w.WriteMsg(m)
-				return
-			}
+		if answers, err := lookup(q.Name); err != nil {
+			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)
+			maybeCache(nil, s.negLocalTTL, CacheNoLocalReplies)
+			fallback(w, r)
+		} else {
+			m := msgBuilder(r, &q, answers)
+			m.Authoritative = true
+			maybeCache(m, nullTTL, 0)
+			m.Answer = pruneAnswers(shuffleAnswers(m.Answer), s.maxAnswers)
+			Debug.Printf("[dns msgid %d] Sending response: %s/%d answers [code:%s]",
+				m.MsgHdr.Id, dns.TypeToString[q.Qtype], len(m.Answer), dns.RcodeToString[m.Rcode])
+			w.WriteMsg(m)
 		}
-
-		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)
-
-		fallback(w, r)
 	}
 }
 
 // When we receive a request for a name outside of our '.weave.local.'
 // domain, ask the configured DNS server as a fallback.
 func (s *DNSServer) notUsHandler(proto dnsProtocol) dns.HandlerFunc {
-	dnsClient := proto.GetNewClient(DefaultUDPBuflen)
+	dnsClient := proto.GetNewClient(DefaultUDPBuflen, s.timeout)
 
 	return func(w dns.ResponseWriter, r *dns.Msg) {
-		maxLen := getMaxReplyLen(r, proto)
 		q := r.Question[0]
 
-		// create a request where we announce our max payload size
+		// announce our max payload size as the max payload our client supports
+		maxLen := getMaxReplyLen(r, proto)
 		rcopy := r
 		rcopy.SetEdns0(uint16(maxLen), false)
 
@@ -337,3 +385,12 @@ func (s *DNSServer) notUsHandler(proto dnsProtocol) dns.HandlerFunc {
 		w.WriteMsg(makeDNSFailResponse(r))
 	}
 }
+
+// Get the listen port
+func (s *DNSServer) GetPort() (int, error) {
+	_, portS, err := net.SplitHostPort(s.ListenAddr)
+	if err != nil {
+		return 0, err
+	}
+	return strconv.Atoi(portS)
+}

nameserver/server_cache_test.go

@@ -0,0 +1,99 @@
+package nameserver
+
+import (
+	"fmt"
+	"github.com/miekg/dns"
+	. "github.com/weaveworks/weave/common"
+	wt "github.com/weaveworks/weave/testing"
+	"net"
+	"testing"
+	"time"
+)
+
+// Check if the names updates lead to cache invalidations
+func TestServerCacheRefresh(t *testing.T) {
+	const (
+		containerID     = "somecontainer"
+		testName1       = "first.weave.local."
+		testName2       = "second.weave.local."
+		refreshInterval = int(localTTL) / 3
+	)
+
+	InitDefaultLogging(testing.Verbose())
+	Info.Println("TestServerCacheRefresh starting")
+	clk := newMockedClock()
+
+	Debug.Printf("Creating 2 zone databases")
+	zoneConfig := ZoneConfig{
+		RefreshInterval: refreshInterval,
+		Clock:           clk,
+	}
+	dbs := newZoneDbsWithMockedMDns(2, zoneConfig)
+	dbs.Start()
+	defer dbs.Stop()
+
+	Debug.Printf("Creating a cache")
+	cache, err := NewCache(1024, clk)
+	wt.AssertNoErr(t, err)
+
+	Debug.Printf("Creating a real DNS server for the first zone database and with the cache")
+	srv, err := NewDNSServer(DNSServerConfig{
+		Zone:              dbs[0].Zone,
+		Cache:             cache,
+		Clock:             clk,
+		ListenReadTimeout: testSocketTimeout,
+		MaxAnswers:        4,
+	})
+	wt.AssertNoErr(t, err)
+	go srv.Start()
+	defer srv.Stop()
+	time.Sleep(100 * time.Millisecond) // Allow sever goroutine to start
+
+	testPort, err := srv.GetPort()
+	wt.AssertNoErr(t, err)
+	wt.AssertNotEqualInt(t, testPort, 0, "listen port")
+
+	Debug.Printf("Adding an IPs to %s", testName1)
+	dbs[1].Zone.AddRecord(containerID, testName1, net.ParseIP("10.2.2.1"))
+
+	// Zone database #2 at this point:
+	//   first.weave.local  = 10.2.2.1
+
+	// testName1 and testName2 should have no IPs yet
+	qName1, _ := assertExchange(t, testName1, dns.TypeA, testPort, 1, 1, 0)
+	qName2, _ := assertExchange(t, testName2, dns.TypeA, testPort, 0, 0, dns.RcodeNameError)
+	assertInCache(t, cache, qName1, "after asking for first name")
+	assertNotLocalInCache(t, cache, qName2, "after asking for second name")
+
+	clk.Forward(refreshInterval / 2)
+
+	Debug.Printf("Adding an IP to %s and to %s", testName1, testName2)
+	dbs[1].Zone.AddRecord(containerID, testName1, net.ParseIP("10.2.2.2"))
+	dbs[1].Zone.AddRecord(containerID, testName2, net.ParseIP("10.9.9.2"))
+
+	// Zone database #2 at this point:
+	//   first.weave.local    = 10.2.2.1 10.2.2.2
+	//   second.weave.local   = 10.9.9.2
+	clk.Forward(refreshInterval/2 + 2)
+
+	// at this point, the testName1 should have been refreshed
+	// so it should have two IPs, and the cache entry should have been invalidated
+	assertNotInCache(t, cache, qName1, fmt.Sprintf("after asking for %s", testName1))
+	assertNotLocalInCache(t, cache, qName2, fmt.Sprintf("after asking for %s", testName2))
+
+	qName1, _ = assertExchange(t, testName1, dns.TypeA, testPort, 2, 2, 0)
+	qName2, _ = assertExchange(t, testName2, dns.TypeA, testPort, 0, 0, dns.RcodeNameError)
+	assertInCache(t, cache, qName1, fmt.Sprintf("after asking for %s", testName1))
+	assertNotLocalInCache(t, cache, qName2, "after asking for a unknown name")
+
+	// delete the IPs, and some time passes by so the cache should be purged...
+	dbs[1].Zone.DeleteRecord(containerID, net.ParseIP("10.2.2.1"))
+	dbs[1].Zone.DeleteRecord(containerID, net.ParseIP("10.2.2.2"))
+	clk.Forward(refreshInterval + 1)
+
+	qName1, _ = assertExchange(t, testName1, dns.TypeA, testPort, 0, 0, dns.RcodeNameError)
+	qName2, _ = assertExchange(t, testName2, dns.TypeA, testPort, 0, 0, dns.RcodeNameError)
+	assertNotLocalInCache(t, cache, qName1, "after asking for a unknown name")
+	assertNotLocalInCache(t, cache, qName2, "after asking for a unknown name")
+
+}

nameserver/server_test.go

@@ -18,10 +18,6 @@ const (
 	testUDPBufSize   = 16384
 )
 
-var (
-	testPort = 17625
-)
-
 func setupForTest(t *testing.T) {
 	// fail early if we cannot find a default multicast interface
 	multicast, err := LinkLocalMulticastListener(nil)
@@ -29,23 +25,29 @@ func setupForTest(t *testing.T) {
 		t.Fatalf("Unable to create multicast listener: %s. No default multicast interface?", err)
 	}
 	multicast.Close()
-
-	testPort++
 }
 
 func TestUDPDNSServer(t *testing.T) {
 	setupForTest(t)
+
 	const (
-		containerID     = "foobar"
 		successTestName = "test1.weave.local."
-		failTestName    = "test2.weave.local."
+		failTestName    = "fail.weave.local."
 		nonLocalName    = "weave.works."
 		testAddr1       = "10.2.2.1"
+		containerID     = "somecontainer"
 	)
 	testCIDR1 := testAddr1 + "/24"
 
-	InitDefaultLogging(true)
-	var zone = NewZoneDb(DefaultLocalDomain)
+	InitDefaultLogging(testing.Verbose())
+	Info.Println("TestUDPDNSServer starting")
+
+	zone, err := NewZoneDb(ZoneConfig{})
+	wt.AssertNoErr(t, err)
+	err = zone.Start()
+	wt.AssertNoErr(t, err)
+	defer zone.Stop()
+
 	ip, _, _ := net.ParseCIDR(testCIDR1)
 	zone.AddRecord(containerID, successTestName, ip)
 
@@ -73,15 +75,17 @@ func TestUDPDNSServer(t *testing.T) {
 	}
 
 	// Run another DNS server for fallback
-	s, fallbackAddr, err := runLocalUDPServer(t, "127.0.0.1:0", fallbackHandler)
+	fallback, err := newMockedFallback(fallbackHandler, nil)
 	wt.AssertNoErr(t, err)
-	defer s.Shutdown()
+	fallback.Start()
+	defer fallback.Stop()
 
-	_, fallbackPort, err := net.SplitHostPort(fallbackAddr)
-	wt.AssertNoErr(t, err)
-
-	config := &dns.ClientConfig{Servers: []string{"127.0.0.1"}, Port: fallbackPort}
-	srv, err := NewDNSServer(DNSServerConfig{UpstreamCfg: config, Port: testPort}, zone, nil)
+	srv, err := NewDNSServer(DNSServerConfig{
+		Zone:              zone,
+		UpstreamCfg:       fallback.CliConfig,
+		CacheDisabled:     true,
+		ListenReadTimeout: testSocketTimeout,
+	})
 	wt.AssertNoErr(t, err)
 	defer srv.Stop()
 	go srv.Start()
@@ -89,29 +93,33 @@ func TestUDPDNSServer(t *testing.T) {
 
 	var r *dns.Msg
 
-	r = assertExchange(t, successTestName, dns.TypeA, 1, 1, 0)
+	testPort, err := srv.GetPort()
+	wt.AssertNoErr(t, err)
+	wt.AssertNotEqualInt(t, testPort, 0, "invalid listen port")
+
+	_, r = assertExchange(t, successTestName, dns.TypeA, testPort, 1, 1, 0)
 	wt.AssertType(t, r.Answer[0], (*dns.A)(nil), "DNS record")
 	wt.AssertEqualString(t, r.Answer[0].(*dns.A).A.String(), testAddr1, "IP address")
 
-	assertExchange(t, failTestName, dns.TypeA, 0, 0, dns.RcodeNameError)
+	assertExchange(t, failTestName, dns.TypeA, testPort, 0, 0, dns.RcodeNameError)
 
-	r = assertExchange(t, testRDNSsuccess, dns.TypePTR, 1, 1, 0)
+	_, r = assertExchange(t, testRDNSsuccess, dns.TypePTR, testPort, 1, 1, 0)
 	wt.AssertType(t, r.Answer[0], (*dns.PTR)(nil), "DNS record")
 	wt.AssertEqualString(t, r.Answer[0].(*dns.PTR).Ptr, successTestName, "IP address")
 
-	assertExchange(t, testRDNSfail, dns.TypePTR, 0, 0, dns.RcodeNameError)
+	assertExchange(t, testRDNSfail, dns.TypePTR, testPort, 0, 0, dns.RcodeNameError)
 
-	// This should fail because we don't have information about MX records
-	assertExchange(t, successTestName, dns.TypeMX, 0, 0, dns.RcodeNameError)
+	// This should fail because we don't handle MX records
+	assertExchange(t, successTestName, dns.TypeMX, testPort, 0, 0, dns.RcodeNameError)
 
 	// This non-local query for an MX record should succeed by being
 	// passed on to the fallback server
-	assertExchange(t, nonLocalName, dns.TypeMX, 1, -1, 0)
+	assertExchange(t, nonLocalName, dns.TypeMX, testPort, 1, -1, 0)
 
 	// Now ask a query that we expect to return a lot of data.
-	assertExchange(t, nonLocalName, dns.TypeANY, 5, -1, 0)
+	assertExchange(t, nonLocalName, dns.TypeANY, testPort, 5, -1, 0)
 
-	assertExchange(t, testRDNSnonlocal, dns.TypePTR, 1, -1, 0)
+	assertExchange(t, testRDNSnonlocal, dns.TypePTR, testPort, 1, -1, 0)
 
 	// Not testing MDNS functionality of server here (yet), since it
 	// needs two servers, each listening on its own address
@@ -119,21 +127,28 @@ func TestUDPDNSServer(t *testing.T) {
 
 func TestTCPDNSServer(t *testing.T) {
 	setupForTest(t)
+
 	const (
 		numAnswers   = 512
 		nonLocalName = "weave.works."
 	)
-	dnsAddr := fmt.Sprintf("localhost:%d", testPort)
 
-	InitDefaultLogging(true)
-	var zone = NewZoneDb(DefaultLocalDomain)
+	InitDefaultLogging(testing.Verbose())
+	Info.Println("TestTCPDNSServer starting")
+
+	zone, err := NewZoneDb(ZoneConfig{})
+	wt.AssertNoErr(t, err)
+	err = zone.Start()
+	wt.AssertNoErr(t, err)
+	defer zone.Stop()
 
 	// generate a list of `numAnswers` IP addresses
 	var addrs []ZoneRecord
 	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, 0})
+		ip := net.IPv4(bs[0], bs[1], bs[2], bs[3])
+		addrs = append(addrs, ZoneRecord(Record{"", ip, 0, 0, 0}))
 	}
 
 	// handler for the fallback server: it will just return a very long response
@@ -160,27 +175,28 @@ func TestTCPDNSServer(t *testing.T) {
 	}
 
 	t.Logf("Running a DNS fallback server with UDP")
-	us, fallbackUDPAddr, err := runLocalUDPServer(t, "127.0.0.1:0", fallbackUDPHandler)
+	fallback, err := newMockedFallback(fallbackUDPHandler, fallbackTCPHandler)
 	wt.AssertNoErr(t, err)
-	defer us.Shutdown()
+	fallback.Start()
+	defer fallback.Stop()
 
-	_, fallbackPort, err := net.SplitHostPort(fallbackUDPAddr)
-	wt.AssertNoErr(t, err)
-
-	t.Logf("Starting another fallback server, with TCP, on the same port as the UDP server")
-	fallbackTCPAddr := fmt.Sprintf("127.0.0.1:%s", fallbackPort)
-	ts, fallbackTCPAddr, err := runLocalTCPServer(t, fallbackTCPAddr, fallbackTCPHandler)
-	wt.AssertNoErr(t, err)
-	defer ts.Shutdown()
-
-	t.Logf("Creating a WeaveDNS server instance, falling back to 127.0.0.1:%s", fallbackPort)
-	config := &dns.ClientConfig{Servers: []string{"127.0.0.1"}, Port: fallbackPort}
-	srv, err := NewDNSServer(DNSServerConfig{UpstreamCfg: config, Port: testPort}, zone, nil)
+	t.Logf("Creating a WeaveDNS server instance, falling back to 127.0.0.1:%s", fallback.Port)
+	srv, err := NewDNSServer(DNSServerConfig{
+		Zone:              zone,
+		UpstreamCfg:       fallback.CliConfig,
+		CacheDisabled:     true,
+		ListenReadTimeout: testSocketTimeout,
+	})
 	wt.AssertNoErr(t, err)
 	defer srv.Stop()
 	go srv.Start()
 	time.Sleep(100 * time.Millisecond) // Allow sever goroutine to start
 
+	testPort, err := srv.GetPort()
+	wt.AssertNoErr(t, err)
+	wt.AssertNotEqualInt(t, testPort, 0, "listen port")
+	dnsAddr := fmt.Sprintf("127.0.0.1:%d", testPort)
+
 	t.Logf("Creating a UDP and a TCP client")
 	uc := new(dns.Client)
 	uc.UDPSize = minUDPSize
@@ -192,8 +208,8 @@ func TestTCPDNSServer(t *testing.T) {
 	m.RecursionDesired = true
 	m.SetQuestion(nonLocalName, dns.TypeA)
 
-	t.Logf("Checking the fallback server at %s returns a truncated response with UDP", fallbackUDPAddr)
-	r, _, err := uc.Exchange(m, fallbackUDPAddr)
+	t.Logf("Checking the fallback server at %s returns a truncated response with UDP", fallback.Addr)
+	r, _, err := uc.Exchange(m, fallback.Addr)
 	t.Logf("Got response from fallback server (UDP) with %d answers", len(r.Answer))
 	t.Logf("Response:\n%+v\n", r)
 	wt.AssertNoErr(t, err)
@@ -229,73 +245,3 @@ func TestTCPDNSServer(t *testing.T) {
 	wt.AssertFalse(t, r.MsgHdr.Truncated, "DNS truncated response flag")
 	wt.AssertEqualInt(t, len(r.Answer), numAnswers, "number of answers (UDP-long)")
 }
-
-//////////////////////////////////////////////////////////////////////////////
-
-// perform a DNS query and assert the reply code, number or answers, etc
-func assertExchange(t *testing.T, z string, ty uint16, minAnswers int, maxAnswers int, expErr int) *dns.Msg {
-	c := new(dns.Client)
-	c.UDPSize = testUDPBufSize
-	m := new(dns.Msg)
-	m.RecursionDesired = true
-	m.SetQuestion(z, ty)
-	m.SetEdns0(testUDPBufSize, false) // we don't want to play with truncation here...
-
-	r, _, err := c.Exchange(m, fmt.Sprintf("127.0.0.1:%d", testPort))
-	t.Logf("Response:\n%+v\n", r)
-	wt.AssertNoErr(t, err)
-	if minAnswers == 0 && maxAnswers == 0 {
-		wt.AssertStatus(t, r.Rcode, expErr, "DNS response code")
-	} else {
-		wt.AssertStatus(t, r.Rcode, dns.RcodeSuccess, "DNS response code")
-	}
-	answers := len(r.Answer)
-	if minAnswers >= 0 && answers < minAnswers {
-		wt.Fatalf(t, "Number of answers >= %d", minAnswers)
-	}
-	if maxAnswers >= 0 && answers > maxAnswers {
-		wt.Fatalf(t, "Number of answers <= %d", maxAnswers)
-	}
-	return r
-}
-
-// run a UDP fallback server
-func runLocalUDPServer(t *testing.T, laddr string, handler dns.HandlerFunc) (*dns.Server, string, error) {
-	t.Logf("Starting fallback UDP server at %s", laddr)
-	pc, err := net.ListenPacket("udp", laddr)
-	if err != nil {
-		return nil, "", err
-	}
-	server := &dns.Server{PacketConn: pc, Handler: handler, ReadTimeout: 100 * time.Millisecond}
-
-	go func() {
-		server.ActivateAndServe()
-		pc.Close()
-	}()
-
-	t.Logf("Fallback UDP server listening at %s", pc.LocalAddr())
-	return server, pc.LocalAddr().String(), nil
-}
-
-// run a TCP fallback server
-func runLocalTCPServer(t *testing.T, laddr string, handler dns.HandlerFunc) (*dns.Server, string, error) {
-	t.Logf("Starting fallback TCP server at %s", laddr)
-	laddrTCP, err := net.ResolveTCPAddr("tcp", laddr)
-	if err != nil {
-		return nil, "", err
-	}
-
-	l, err := net.ListenTCP("tcp", laddrTCP)
-	if err != nil {
-		return nil, "", err
-	}
-	server := &dns.Server{Listener: l, Handler: handler, ReadTimeout: 100 * time.Millisecond}
-
-	go func() {
-		server.ActivateAndServe()
-		l.Close()
-	}()
-
-	t.Logf("Fallback TCP server listening at %s", l.Addr().String())
-	return server, l.Addr().String(), nil
-}

nameserver/zone.go

@@ -3,41 +3,31 @@ package nameserver
 import (
 	"bytes"
 	"fmt"
+	"github.com/benbjohnson/clock"
 	"github.com/miekg/dns"
 	. "github.com/weaveworks/weave/common"
 	"net"
 	"sync"
+	"time"
 )
 
 const (
 	RDNSDomain = "in-addr.arpa."
 )
 
+const (
+	DefaultLocalDomain     = "weave.local." // The default name used for the local domain
+	DefaultRefreshInterval = int(localTTL)  // Period for background updates with mDNS
+	DefaultRelevantTime    = 60             // When to forget info about remote info if nobody asks...
+	DefaultNumUpdaters     = 4              // Default number of background updaters
+
+	defaultRefreshMailbox = 1000           // Number of on-the-fly background queries
+	defaultRemoteIdent    = "weave:remote" // Ident used for info obtained from mDNS
+)
+
 // +1 to also exclude a dot
 var rdnsDomainLen = len(RDNSDomain) + 1
 
-type Zone interface {
-	AddRecord(ident string, name string, ip net.IP) error
-	DeleteRecord(ident string, ip net.IP) error
-	DeleteRecordsFor(ident string) error
-	Domain() string
-	ZoneLookup
-}
-
-type dbRecord struct {
-	Ident string
-	Name  string
-	IP    net.IP
-}
-
-// Very simple data structure for now, with linear searching.
-// TODO: make more sophisticated to improve performance.
-type ZoneDb struct {
-	mx     sync.RWMutex
-	recs   []dbRecord
-	domain string
-}
-
 type LookupError string
 
 func (ops LookupError) Error() string {
@@ -51,103 +41,730 @@ func (dup DuplicateError) Error() string {
 	return "Tried to add a duplicate entry"
 }
 
-func NewZoneDb(domain string) *ZoneDb {
-	return &ZoneDb{
-		domain: domain,
+type Zone interface {
+	ZoneObservable
+	ZoneLookup
+	// The domain where we operate (eg, "weave.local.")
+	Domain() string
+	// Add a record in the local database
+	AddRecord(ident string, name string, ip net.IP) error
+	// Delete a record in the local database
+	DeleteRecord(ident string, ip net.IP) error
+	// Delete all records for an ident in the local database
+	DeleteRecordsFor(ident string) error
+	// Lookup for a name in the whole domain
+	DomainLookupName(name string) ([]ZoneRecord, error)
+	// Lookup for an address in the whole domain
+	DomainLookupInaddr(inaddr string) ([]ZoneRecord, error)
+	// Return a status string
+	Status() string
+}
+
+///////////////////////////////////////////////////////////////////
+//
+// Zone database overview:
+//
+//           idents            names             entry
+//         *-------*          *------*         *------*
+// zone -> | ident |  --*-->  | name |  --*->  | IPv4 |
+//         *-------*    |     *------*    |    *------*
+//         | ident |    *-->  | name |    *->  | IPv4 |
+//         *-------*          *------*         *------*
+//            ...               ...
+//
+// The zone database keeps an ident per container, but also a special
+// ident for names/IPs obtained from remote peers.
+//
+// Each name can have multiple IPs (so far, we only consider IPv4s), and
+// each IP is stored in a `recordEntry` in the database. Entries store
+// some additional information like priority and weight of the IP (for
+// example, for future use with SRV records)
+//
+
+// An entry in the zone database
+type recordEntry struct {
+	Record
+
+	observers []ZoneRecordObserver // the observers for this record
+	insTime   time.Time            // time when this record was inserted
+}
+
+func newRecordEntry(zr ZoneRecord, now time.Time) *recordEntry {
+	return &recordEntry{
+		Record: Record{
+			name:     zr.Name(),
+			ip:       zr.IP(),
+			priority: zr.Priority(),
+			weight:   zr.Weight(),
+			ttl:      zr.TTL(),
+		},
+		insTime:   now,
+		observers: make([]ZoneRecordObserver, 0),
 	}
 }
 
-func (zone *ZoneDb) Domain() string {
-	return zone.domain
+func (i *recordEntry) addIPObserver(zro ZoneRecordObserver) {
+	i.observers = append(i.observers, zro)
 }
 
-func (zone *ZoneDb) indexOf(match func(dbRecord) bool) int {
-	for i, r := range zone.recs {
-		if match(r) {
-			return i
+// Notify all the IP observers and then remove them all
+// IP address observers are notified when
+// - the IP is removed
+// - the name is removed
+func (re *recordEntry) notifyIPObservers() {
+	numObservers := len(re.observers)
+	if numObservers > 0 {
+		Debug.Printf("[zonedb] Notifying %d observers of '%s'", numObservers, re.ip)
+		for _, observer := range re.observers {
+			observer()
 		}
+		re.observers = make([]ZoneRecordObserver, 0)
 	}
-	return -1
 }
 
-func (zone *ZoneDb) String() string {
-	zone.mx.RLock()
-	defer zone.mx.RUnlock()
+func (re *recordEntry) TTL() int {
+	// if we never set the TTL (eg, when using AddRecord()), return the standard value...
+	if re.ttl == 0 {
+		return int(localTTL)
+	}
+	return re.ttl
+}
+
+// Check if the info in this record is still valid according to the TTL
+func (re *recordEntry) hasExpired(now time.Time) bool {
+	return re.ttl > 0 && now.Sub(re.insTime).Seconds() > float64(re.ttl)
+}
+
+func (re *recordEntry) String() string {
 	var buf bytes.Buffer
-	for _, r := range zone.recs {
-		fmt.Fprintf(&buf, "%.12s %s %v\n", r.Ident, r.IP, r.Name)
+	fmt.Fprintf(&buf, "%s", re.Record)
+	lobs := len(re.observers)
+	if lobs > 0 {
+		fmt.Fprintf(&buf, "/OBS:%d", lobs)
 	}
 	return buf.String()
 }
 
-func (zone *ZoneDb) LookupName(name string) ([]ZoneRecord, error) {
+///////////////////////////////////////////////////////////////////
+
+// a name, with group of records by IPv4
+type name struct {
+	name            string
+	ipv4            map[IPv4]*recordEntry // all the IPv4 records for this name
+	observers       []ZoneRecordObserver  // the observers for this name
+	lastRefreshTime time.Time             // last time this name was updated by broadcasting a query
+	lastAccessTime  time.Time             // last time a Lookup* used info from this name
+}
+
+func newName(n string) *name {
+	return &name{
+		name:      n,
+		ipv4:      make(map[IPv4]*recordEntry),
+		observers: make([]ZoneRecordObserver, 0),
+	}
+}
+func (n *name) len() int             { return len(n.ipv4) }
+func (n *name) empty() bool          { return n.len() == 0 }
+func (n *name) hasIP(ip net.IP) bool { return n.hasIPv4(ipToIPv4(ip)) }
+func (n *name) hasIPv4(ip IPv4) bool { _, found := n.ipv4[ip]; return found }
+
+// Get all the entries for this name
+func (n *name) getAllEntries() (res []*recordEntry) {
+	for _, entry := range n.ipv4 {
+		res = append(res, entry)
+	}
+	return
+}
+
+// Get the entry for an IP in this name
+func (n *name) getEntryForIP(ip IPv4) (res *recordEntry) {
+	if entry, found := n.ipv4[ip]; found {
+		res = entry
+	}
+	return
+}
+
+// Add a new IPv4 to this name
+func (n *name) addIP(zr ZoneRecord, now time.Time) (*recordEntry, error) {
+	ne := newRecordEntry(zr, now)
+	ipv4 := ipToIPv4(zr.IP())
+	n.ipv4[ipv4] = ne
+	n.notifyNameObservers()
+	return ne, nil
+}
+
+// Delete an IP for this name
+func (n *name) deleteIP(ip IPv4) bool {
+	if ipRecord, found := n.ipv4[ip]; found {
+		ipRecord.notifyIPObservers()
+		n.notifyNameObservers()
+		delete(n.ipv4, ip)
+		return true
+	}
+	return false
+}
+
+// Update the list of IPs for this name, adding new ones and removing old IPs...
+func (n *name) updateIPs(records []ZoneRecord, now time.Time) (added int, removed int) {
+	ipsMap := make(map[IPv4]bool)
+
+	// add the new IPs
+	for _, record := range records {
+		ipv4 := ipToIPv4(record.IP())
+		ipsMap[ipv4] = true
+		if !n.hasIPv4(ipv4) {
+			n.addIP(record, now)
+			added++
+		}
+	}
+
+	// remove the old IPs
+	for ipv4, ipRecord := range n.ipv4 {
+		if _, isNew := ipsMap[ipv4]; !isNew {
+			ipRecord.notifyIPObservers() // we must notify the observers before removing the IP record...
+			delete(n.ipv4, ipv4)
+			removed++
+		}
+	}
+
+	// notify observers and update times
+	if added > 0 || removed > 0 {
+		n.lastRefreshTime = now
+		n.notifyNameObservers()
+	}
+	return
+}
+
+// Add a observer for this name
+func (n *name) addNameObserver(observer ZoneRecordObserver) {
+	n.observers = append(n.observers, observer)
+}
+
+// Notify and flush all the observers
+// Name observers are notified when
+// - an IP is added
+// - an IP is removed
+// - the name is removed
+func (n *name) notifyNameObservers() {
+	numObservers := len(n.observers)
+	if numObservers > 0 {
+		Debug.Printf("[zonedb] Notifying %d observers of '%s'", numObservers, n.name)
+		for _, observer := range n.observers {
+			observer()
+		}
+		n.observers = make([]ZoneRecordObserver, 0)
+	}
+}
+
+// Convert a set of IP records to a comma-separated string
+func (n *name) String() string {
+	var buf bytes.Buffer
+	for _, ip := range n.ipv4 {
+		fmt.Fprintf(&buf, "%s, ", ip)
+	}
+	l := buf.Len()
+	if l > 2 {
+		buf.Truncate(l - 2)
+	}
+	return buf.String()
+}
+
+///////////////////////////////////////////////////////////////////
+
+// all the names in a ident
+type namesSet struct {
+	names map[string]*name
+}
+
+func newNamesSet() *namesSet     { return &namesSet{names: make(map[string]*name)} }
+func (ns *namesSet) len() int    { return len(ns.names) }
+func (ns *namesSet) empty() bool { return ns.len() == 0 }
+func (ns *namesSet) hasName(name string) bool {
+	if n, found := ns.names[name]; !found {
+		return false
+	} else {
+		// the name must have some valid IPs in order to be considered as "existent"
+		return len(n.ipv4) > 0
+	}
+}
+
+// Get the zone entries for a name
+// If the name does not exist and `create` is true, a new name will be created
+func (ns *namesSet) getName(name string, create bool) (n *name) {
+	if n, found := ns.names[name]; !found && create {
+		newName := newName(name)
+		ns.names[name] = newName
+		return newName
+	} else {
+		return n
+	}
+}
+
+// Get all the entries for a name
+func (ns *namesSet) getEntriesForName(name string) (res []*recordEntry) {
+	n := ns.getName(name, false)
+	if n != nil {
+		res = n.getAllEntries()
+	}
+	return
+}
+
+// Get all the entries for an IP, for all names
+func (ns *namesSet) getEntriesForIP(ip IPv4) (res []*recordEntry) {
+	for _, name := range ns.names {
+		entry := name.getEntryForIP(ip)
+		if entry != nil {
+			res = append(res, entry)
+		}
+	}
+	return
+}
+
+// Add a new IPv4 to a name
+func (ns *namesSet) addIPToName(zr ZoneRecord, now time.Time) (*recordEntry, error) {
+	n := ns.getName(zr.Name(), true)
+	ipv4 := ipToIPv4(zr.IP())
+	if n.hasIPv4(ipv4) {
+		return n.getEntryForIP(ipv4), DuplicateError{}
+	}
+	ns.touchName(zr.Name(), now)
+	return n.addIP(zr, now)
+}
+
+// Delete a name in the names set
+func (ns *namesSet) deleteName(n string) error {
+	if name, found := ns.names[n]; found {
+		for _, ipRecord := range name.getAllEntries() {
+			ipRecord.notifyIPObservers()
+		}
+		name.notifyNameObservers()
+		delete(ns.names, n)
+		return nil
+	} else {
+		return LookupError(fmt.Sprintf("%+v", name))
+	}
+}
+
+// Delete an IPv4 from all names
+func (ns *namesSet) deleteIP(ip IPv4) error {
+	cnt := 0
+	for nkey, n := range ns.names {
+		if n.deleteIP(ip) {
+			cnt++
+		}
+		if n.empty() {
+			delete(ns.names, nkey)
+		}
+	}
+
+	// we must return an error if no records have been found and deleted
+	if cnt == 0 {
+		return LookupError(fmt.Sprintf("%+v", ip))
+	}
+	return nil
+}
+
+// Get the name query time
+func (ns *namesSet) getNameLastAccess(n string) time.Time {
+	if name, found := ns.names[n]; found {
+		return name.lastAccessTime
+	}
+	return time.Time{}
+}
+
+// Touch the last access time for a name
+// The access time is saved only for locally-introduced records (otherwise it could
+// be lost when names are irrelevant...)
+func (ns *namesSet) touchName(name string, now time.Time) {
+	Debug.Printf("[zonedb] Touching name %s", name)
+	n := ns.getName(name, false)
+	if n != nil {
+		n.lastAccessTime = now
+	}
+}
+
+func (ns *namesSet) String() string {
+	var buf bytes.Buffer
+	for _, name := range ns.names {
+		fmt.Fprintf(&buf, "%s, ", name)
+	}
+	l := buf.Len()
+	if l > 2 {
+		buf.Truncate(l - 2)
+	}
+	return buf.String()
+}
+
+///////////////////////////////////////////////////////////////////
+
+// Names sets, by ident name
+type identRecordsSet map[string]*namesSet
+
+func (irs identRecordsSet) String() string {
+	var buf bytes.Buffer
+	for ident, nameset := range irs {
+		fmt.Fprintf(&buf, "%.12s: %s\n", ident, nameset)
+	}
+	return buf.String()
+}
+
+///////////////////////////////////////////////////////////////////
+
+// Invariant on updateRequests: they are in ascending order of time.
+type refreshRequest struct {
+	name string
+	time time.Time
+}
+
+type ZoneConfig struct {
+	// The local domain
+	Domain string
+	// The interface where mDNS operates
+	Iface *net.Interface
+	// Refresh interval for local names in the zone database (disabled by default) (in seconds)
+	RefreshInterval int
+	// Number of background updaters for names
+	RefreshWorkers int
+	// Forget about remote info if nobody asks for it in this time (in seconds)
+	RelevantTime int
+	// Max pending refresh requests
+	MaxRefreshRequests int
+	// Force a specific mDNS client
+	MDNSClient ZoneMDNSClient
+	// Force a specific mDNS server
+	MDNSServer ZoneMDNSServer
+	// For a specific clock provider
+	Clock clock.Clock
+}
+
+// Zone database
+// The zone database contains the locally known information about the zone.
+// The zone database uses some other mechanisms (eg, mDNS) for finding unknown
+// information or for keeping current information up to date.
+type zoneDb struct {
+	mx     sync.RWMutex
+	idents identRecordsSet
+
+	mdnsCli       ZoneMDNSClient
+	mdnsSrv       ZoneMDNSServer
+	iface         *net.Interface
+	domain        string        // the local domain
+	relevantLimit time.Duration // if no one asks for something in this time, it is not relevant a anymore...
+	clock         clock.Clock
+
+	refreshChan      chan refreshRequest // channel for enqueing requests for updating names
+	refreshSchedChan chan refreshRequest // channel for enqueing requests for updating names
+	refreshCloseChan chan bool
+	refreshWg        sync.WaitGroup
+	refreshInterval  time.Duration
+	refreshWorkers   int
+}
+
+// Create a new zone database
+func NewZoneDb(config ZoneConfig) (zone *zoneDb, err error) {
+	maxRefreshRequests := defaultRefreshMailbox
+	if config.MaxRefreshRequests > 0 {
+		maxRefreshRequests = config.MaxRefreshRequests
+	}
+
+	zone = &zoneDb{
+		domain:           config.Domain,
+		idents:           make(identRecordsSet),
+		mdnsCli:          config.MDNSClient,
+		mdnsSrv:          config.MDNSServer,
+		iface:            config.Iface,
+		clock:            config.Clock,
+		relevantLimit:    time.Duration(DefaultRelevantTime) * time.Second,
+		refreshChan:      make(chan refreshRequest, maxRefreshRequests),
+		refreshSchedChan: make(chan refreshRequest, maxRefreshRequests),
+		refreshCloseChan: make(chan bool),
+		refreshWorkers:   DefaultNumUpdaters,
+	}
+
+	// fix the default configuration parameters
+	if zone.clock == nil {
+		zone.clock = clock.New()
+	}
+	if len(zone.domain) == 0 {
+		zone.domain = DefaultLocalDomain
+	}
+	if config.RefreshInterval > 0 {
+		zone.refreshInterval = time.Duration(config.RefreshInterval) * time.Second
+	}
+	if config.RelevantTime > 0 {
+		zone.relevantLimit = time.Duration(config.RelevantTime) * time.Second
+	}
+	if config.RefreshWorkers > 0 {
+		zone.refreshWorkers = config.RefreshWorkers
+	}
+
+	// Create the mDNS client and server
+	if zone.mdnsCli == nil {
+		if zone.mdnsCli, err = NewMDNSClient(); err != nil {
+			return
+		}
+	}
+	if zone.mdnsSrv == nil {
+		if zone.mdnsSrv, err = NewMDNSServer(zone); err != nil {
+			return
+		}
+	}
+
+	return
+}
+
+// Start the zone database
+func (zone *zoneDb) Start() (err error) {
+	if zone.iface != nil {
+		Info.Printf("[zonedb] Using mDNS on %+v", zone.iface)
+	} else {
+		Info.Printf("[zonedb] Using mDNS on all interfaces")
+	}
+
+	if err = zone.mdnsCli.Start(zone.iface); err != nil {
+		return
+	}
+	if err = zone.mdnsSrv.Start(zone.iface); err != nil {
+		return
+	}
+
+	// Start some name refreshers...
+	if zone.refreshInterval > 0 {
+		Info.Printf("[zonedb] Starting %d background name updaters", zone.refreshWorkers)
+		for i := 0; i < zone.refreshWorkers; i++ {
+			zone.refreshWg.Add(1)
+			go zone.updater(i)
+		}
+
+		zone.refreshWg.Add(1)
+		go zone.periodicUpdater()
+	} else {
+		// TODO: the refresher also garbage collects irrelevant remote info... we should have an alternative mechanism
+		//       when we don't use refreshers.
+		Warning.Printf("[zonedb] Remote info will not be garbage collected")
+	}
+
+	return
+}
+
+// Perform a graceful shutdown of the zone database
+func (zone *zoneDb) Stop() error {
+	if zone.refreshInterval > 0 {
+		Debug.Printf("[zonedb] Closing background updaters...")
+		close(zone.refreshCloseChan)
+		zone.refreshWg.Wait()
+	}
+
+	Debug.Printf("[zonedb] Exiting mDNS client and server...")
+	zone.mdnsCli.Stop()
+	zone.mdnsSrv.Stop()
+	return nil
+}
+
+// Obtain the domain where this database keeps information for
+func (zone *zoneDb) Domain() string {
+	return zone.domain
+}
+
+// Return the status string
+func (zone *zoneDb) Status() string {
+	var buf bytes.Buffer
+	fmt.Fprintln(&buf, "Local domain", zone.domain)
+	fmt.Fprintln(&buf, "Interface", zone.iface)
+	fmt.Fprintf(&buf, "Zone database:\n%s", zone)
+	return buf.String()
+}
+
+// Add a record with `name` pointing to `ip`
+func (zone *zoneDb) AddRecord(ident string, name string, ip net.IP) (err error) {
+	zone.mx.Lock()
+	defer zone.mx.Unlock()
+	Debug.Printf("[zonedb] Adding record: '%s'/'%s'[%s]", ident, name, ip)
+	record := Record{dns.Fqdn(name), ip, 0, 0, 0}
+	_, err = zone.getNamesSet(ident).addIPToName(record, zone.clock.Now())
+	return
+}
+
+// Delete all records for an ident
+func (zone *zoneDb) DeleteRecordsFor(ident string) (err error) {
+	zone.mx.Lock()
+	defer zone.mx.Unlock()
+	Debug.Printf("[zonedb] Deleting records for ident '%s'", ident)
+	zone.deleteIdent(ident)
+	return
+}
+
+// Delete an IP in a ident
+func (zone *zoneDb) DeleteRecord(ident string, ip net.IP) (err error) {
+	zone.mx.Lock()
+	defer zone.mx.Unlock()
+
+	Debug.Printf("[zonedb] Deleting record '%s'/[%s]", ident, ip)
+	if _, found := zone.idents[ident]; found {
+		err = zone.idents[ident].deleteIP(ipToIPv4(ip))
+		if zone.idents[ident].empty() {
+			zone.deleteIdent(ident)
+		}
+	} else {
+		err = LookupError(ident)
+	}
+	return
+}
+
+// Observe a name.
+// The name must have at least one valid IP address.
+// Name observers are notified when
+// - an IP is added
+// - an IP is removed
+// - the name is removed
+// The observer will be invoked on any change/removal that affects this name.
+// Each observer will be invoked at least once (but possibly more). After that, they will be removed.
+// The observer should not try to lock the ZoneDB (you will get a deadlock)
+func (zone *zoneDb) ObserveName(name string, observer ZoneRecordObserver) (err error) {
+	zone.mx.Lock()
+	defer zone.mx.Unlock()
+
+	cnt := 0
+	name = dns.Fqdn(name)
+	for _, nameset := range zone.idents {
+		if n := nameset.getName(name, false); n != nil {
+			n.addNameObserver(observer)
+			cnt += 1
+		}
+	}
+	if cnt == 0 {
+		err = LookupError(fmt.Sprintf("%s", name))
+	}
+	return
+}
+
+// Observe a IP address.
+// The IP address must be exit in the database.
+// IP address observers are notified when
+// - the IP is removed
+// - the name is removed
+// The observer will be invoked on any change/removal that affects this IP.
+// Each observer will be invoked at least once (but possibly more). After that, they will be removed.
+// The observer should not try to lock the ZoneDB (you will get a deadlock)
+func (zone *zoneDb) ObserveInaddr(inaddr string, observer ZoneRecordObserver) (err error) {
+	zone.mx.Lock()
+	defer zone.mx.Unlock()
+
+	cnt := 0
+	revIP, err := raddrToIPv4(inaddr)
+	if err != nil {
+		return newParseError("lookup address", inaddr)
+	}
+	for _, nameset := range zone.idents {
+		for _, entry := range nameset.getEntriesForIP(revIP) {
+			entry.addIPObserver(observer)
+			cnt += 1
+		}
+	}
+	if cnt == 0 {
+		err = LookupError(fmt.Sprintf("%+v/%+v", inaddr, revIP))
+	}
+	return
+}
+
+// Get the string representation of a zone
+func (zone *zoneDb) String() string {
 	zone.mx.RLock()
 	defer zone.mx.RUnlock()
-	for _, r := range zone.recs {
-		if r.Name == name {
-			return []ZoneRecord{Record{r.Name, r.IP, 0, 0, 0}}, nil
-		}
-	}
-	return nil, LookupError(name)
+	return zone.idents.String()
 }
 
-func (zone *ZoneDb) LookupInaddr(inaddr string) ([]ZoneRecord, error) {
-	if revIP := net.ParseIP(inaddr[:len(inaddr)-rdnsDomainLen]); revIP != nil {
-		revIP4 := revIP.To4()
-		ip := []byte{revIP4[3], revIP4[2], revIP4[1], revIP4[0]}
-		Debug.Printf("[zonedb] Looking for address: %+v", ip)
-		zone.mx.RLock()
-		defer zone.mx.RUnlock()
-		for _, r := range zone.recs {
-			if r.IP.Equal(ip) {
-				return []ZoneRecord{Record{r.Name, r.IP, 0, 0, 0}}, nil
-			}
-		}
-		return nil, LookupError(inaddr)
-	}
-	Warning.Printf("[zonedb] Asked to reverse lookup %s", inaddr)
-	return nil, LookupError(inaddr)
-}
-
-func (zone *ZoneDb) AddRecord(ident string, name string, ip net.IP) error {
-	zone.mx.Lock()
-	defer zone.mx.Unlock()
-	fqdn := dns.Fqdn(name)
-	pred := func(r dbRecord) bool { return r.Name == fqdn && r.IP.Equal(ip) && r.Ident == ident }
-	if index := zone.indexOf(pred); index != -1 {
-		return DuplicateError{}
-	}
-	zone.recs = append(zone.recs, dbRecord{ident, fqdn, ip})
-	return nil
-}
-
-func (zone *ZoneDb) DeleteRecord(ident string, ip net.IP) error {
-	zone.mx.Lock()
-	defer zone.mx.Unlock()
-	pred := func(r dbRecord) bool { return r.Ident == ident && r.IP.Equal(ip) }
-	if index := zone.indexOf(pred); index != -1 {
-		zone.recs = append(zone.recs[:index], zone.recs[index+1:]...)
-		return nil
-	}
-	return LookupError(ident)
-}
-
-func (zone *ZoneDb) DeleteRecordsFor(ident string) error {
-	zone.mx.Lock()
-	defer zone.mx.Unlock()
-	w := 0 // write index
-
-	for _, r := range zone.recs {
-		if r.Ident != ident {
-			zone.recs[w] = r
-			w++
-		}
-	}
-	zone.recs = zone.recs[:w]
-	return nil
-}
-
-func (zone *ZoneDb) ContainerDied(ident string) error {
+// Notify that a container has died
+func (zone *zoneDb) ContainerDied(ident string) error {
 	Info.Printf("[zonedb] Container %s down. Removing records", ident)
 	return zone.DeleteRecordsFor(ident)
 }
+
+// Return true if a remote name is still relevant
+func (zone *zoneDb) IsNameRelevant(n string) bool {
+	zone.mx.Lock()
+	defer zone.mx.Unlock()
+
+	lastAccess := time.Time{}
+	for _, nameset := range zone.idents {
+		if t := nameset.getNameLastAccess(n); !t.IsZero() {
+			if lastAccess.IsZero() || t.After(lastAccess) {
+				lastAccess = t
+			}
+		}
+	}
+
+	return !lastAccess.IsZero() && zone.clock.Now().Sub(lastAccess) < zone.relevantLimit
+}
+
+// Returns true if the info we have for a remote name has expired. Local names are never expired
+// Returns false if we do not have info for this name from remote peers.
+func (zone *zoneDb) IsNameExpired(n string) bool {
+	zone.mx.Lock()
+	defer zone.mx.Unlock()
+
+	now := zone.clock.Now()
+	remoteNamesSet := zone.getNamesSet(defaultRemoteIdent)
+	entries := remoteNamesSet.getEntriesForName(n)
+	if entries != nil {
+		for _, entry := range entries {
+			if entry.hasExpired(now) {
+				return true
+			}
+		}
+	}
+	return false
+}
+
+func (zone *zoneDb) HasNameLocalInfo(n string) bool {
+	zone.mx.Lock()
+	defer zone.mx.Unlock()
+
+	remoteNamesSet := zone.getNamesSet(defaultRemoteIdent)
+	for _, nameset := range zone.idents {
+		if nameset != remoteNamesSet {
+			if nameset.hasName(n) {
+				return true
+			}
+		}
+	}
+	return false
+}
+
+// Return true if the we have obtained some information for a name from remote peers
+// We can have both local (eg, introduced through the HTTP API) and remote (eg, mDNS)
+// information for a name. This method only checks if there is remote information...
+func (zone *zoneDb) HasNameRemoteInfo(n string) bool {
+	zone.mx.Lock()
+	defer zone.mx.Unlock()
+
+	remoteNamesSet := zone.getNamesSet(defaultRemoteIdent)
+	if remoteNamesSet != nil {
+		return remoteNamesSet.hasName(n)
+	}
+	return false
+}
+
+// Get or create a names set
+func (zone *zoneDb) getNamesSet(ident string) *namesSet {
+	ns, found := zone.idents[ident]
+	if !found {
+		ns = newNamesSet()
+		zone.idents[ident] = ns
+	}
+	return ns
+}
+
+// Delete a ident
+func (zone *zoneDb) deleteIdent(ident string) {
+	Info.Printf("[zonedb] Removing everything for ident '%s'", ident)
+	if ns, found := zone.idents[ident]; found {
+		for _, n := range ns.names {
+			for _, entry := range n.getAllEntries() {
+				entry.notifyIPObservers()
+			}
+			n.notifyNameObservers()
+		}
+		delete(zone.idents, ident)
+	}
+}

nameserver/zone_lookup.go

@@ -0,0 +1,342 @@
+package nameserver
+
+import (
+	"github.com/miekg/dns"
+	. "github.com/weaveworks/weave/common"
+	"time"
+)
+
+type uniqZoneRecordKey struct {
+	name string
+	ipv4 IPv4
+}
+
+// A group of ZoneRecords where there are no duplicates (according to the name & IPv4)
+type uniqZoneRecords map[uniqZoneRecordKey]ZoneRecord
+
+func newUniqZoneRecords() uniqZoneRecords {
+	return make(uniqZoneRecords, 0)
+}
+
+// Add a new ZoneRecord to the group
+func (uzr *uniqZoneRecords) add(zr ZoneRecord) {
+	key := uniqZoneRecordKey{zr.Name(), ipToIPv4(zr.IP())}
+	(*uzr)[key] = zr
+}
+
+// Return the group as an slice
+func (uzr *uniqZoneRecords) toSlice() []ZoneRecord {
+	res := make([]ZoneRecord, len(*uzr))
+	i := 0
+	for _, r := range *uzr {
+		res[i] = r
+		i++
+	}
+	return res
+}
+
+//////////////////////////////////////////////////////////////////////////////
+
+// Lookup in the database for locally-introduced information
+func (zone *zoneDb) lookup(target string, lfun func(ns *namesSet) []*recordEntry) (res []ZoneRecord, err error) {
+	uniq := newUniqZoneRecords()
+	for identName, nameset := range zone.idents {
+		if identName != defaultRemoteIdent {
+			for _, ze := range lfun(nameset) {
+				uniq.add(ze)
+			}
+		}
+	}
+	if len(uniq) == 0 {
+		return nil, LookupError(target)
+	} else {
+		return uniq.toSlice(), nil
+	}
+}
+
+// Perform a lookup for a name in the zone
+// The name can be resolved locally with the local database
+func (zone *zoneDb) LookupName(name string) (res []ZoneRecord, err error) {
+	zone.mx.RLock()
+	defer zone.mx.RUnlock()
+
+	// note: LookupName() is usually called from the mDNS server, so we do not touch the name
+	name = dns.Fqdn(name)
+	Debug.Printf("[zonedb] Looking for name '%s' in local database", name)
+	return zone.lookup(name, func(ns *namesSet) []*recordEntry { return ns.getEntriesForName(name) })
+}
+
+// Perform a lookup for a IP address in the zone
+// The address can be resolved locally with the local database
+func (zone *zoneDb) LookupInaddr(inaddr string) (res []ZoneRecord, err error) {
+	zone.mx.RLock()
+	defer zone.mx.RUnlock()
+
+	// note: LookupInaddr() is usually called from the mDNS server, so we do not touch the name
+	revIPv4, err := raddrToIPv4(inaddr)
+	if err != nil {
+		return nil, newParseError("lookup address", inaddr)
+	}
+	Debug.Printf("[zonedb] Looking for address in local database: '%s' (%s)", revIPv4, inaddr)
+	return zone.lookup(inaddr, func(ns *namesSet) []*recordEntry { return ns.getEntriesForIP(revIPv4) })
+}
+
+// Perform a domain lookup with mDNS
+func (zone *zoneDb) domainLookup(target string, lfun ZoneLookupFunc) (res []ZoneRecord, err error) {
+	// no local results have been obtained in the local database: try with a mDNS query
+	Debug.Printf("[zonedb] '%s' not in local database... trying with mDNS", target)
+	lanswers, err := lfun(target)
+	if err != nil {
+		Debug.Printf("[zonedb] mDNS lookup error for '%s': %s", target, err)
+		return nil, err
+	}
+
+	// if the request has been successful, save the IP in the local database and return the corresponding ZoneRecord
+	// (we do not get the remote ident in the mDNS reply, so we save it in a "remote" ident)
+	Debug.Printf("[zonedb] adding '%s' (obtained with mDNS) to '%s'", lanswers, target)
+	res = make([]ZoneRecord, len(lanswers))
+	zone.mx.Lock()
+	now := zone.clock.Now()
+	uniq := newUniqZoneRecords()
+	for _, answer := range lanswers {
+		r, err := zone.getNamesSet(defaultRemoteIdent).addIPToName(answer, now)
+		if err != nil {
+			zone.mx.Unlock()
+			Warning.Printf("[zonedb] '%s' insertion for %s failed: %s", answer, target, err)
+			return nil, err
+		}
+		uniq.add(r)
+	}
+	zone.mx.Unlock()
+
+	return uniq.toSlice(), nil
+}
+
+// Perform a lookup for a name in the zone
+// The name can be resolved locally with the local database or with some other resolution method (eg, a mDNS query)
+func (zone *zoneDb) DomainLookupName(name string) (res []ZoneRecord, err error) {
+	name = dns.Fqdn(name)
+	Debug.Printf("[zonedb] Looking for name '%s' in local(&remote) database", name)
+
+	zone.mx.RLock()
+	now := zone.clock.Now()
+	uniq := newUniqZoneRecords()
+	for identName, nameset := range zone.idents {
+		for _, ze := range nameset.getEntriesForName(name) {
+			// filter the entries with expired TTL
+			// locally introduced entries are never expired: they always have TTL=0
+			if ze.hasExpired(now) {
+				Debug.Printf("[zonedb] '%s': expired entry '%s' ignored: removing", name, ze)
+				nameset.deleteName(name)
+			} else {
+				uniq.add(ze)
+			}
+		}
+		if identName != defaultRemoteIdent {
+			nameset.touchName(name, now)
+		}
+	}
+	zone.mx.RUnlock()
+
+	if len(uniq) > 0 {
+		Debug.Printf("[zonedb] '%s' resolved in local database", name)
+		res = uniq.toSlice()
+	} else {
+		res, err = zone.domainLookup(name, zone.mdnsCli.LookupName)
+	}
+
+	if len(res) > 0 {
+		zone.startUpdatingName(name)
+		return res, nil
+	}
+	return nil, LookupError(name)
+}
+
+// Perform a lookup for a IP address in the zone
+// The address can be resolved either with the local database or
+// with some other resolution method (eg, a mDNS query)
+func (zone *zoneDb) DomainLookupInaddr(inaddr string) (res []ZoneRecord, err error) {
+	revIPv4, err := raddrToIPv4(inaddr)
+	if err != nil {
+		return nil, newParseError("lookup address", inaddr)
+	}
+
+	Debug.Printf("[zonedb] Looking for address in local(&remote) database: '%s' (%s)", revIPv4, inaddr)
+
+	zone.mx.RLock()
+	now := zone.clock.Now()
+	uniq := newUniqZoneRecords()
+	for identName, nameset := range zone.idents {
+		for _, ze := range nameset.getEntriesForIP(revIPv4) {
+			// filter the entries with expired TTL
+			// locally introduced entries are never expired: they always have TTL=0
+			if ze.hasExpired(now) {
+				Debug.Printf("[zonedb] '%s': expired entry '%s' ignored: removing", revIPv4, ze)
+				nameset.deleteIP(revIPv4)
+			} else {
+				uniq.add(ze)
+				if identName != defaultRemoteIdent {
+					nameset.touchName(ze.Name(), now)
+				}
+			}
+		}
+	}
+	zone.mx.RUnlock()
+
+	if len(uniq) > 0 {
+		Debug.Printf("[zonedb] '%s' resolved in local database", inaddr)
+		res = uniq.toSlice()
+	} else {
+		res, err = zone.domainLookup(inaddr, zone.mdnsCli.LookupInaddr)
+	}
+
+	if len(res) > 0 {
+		// note: even for reverse addresses, we perform the background updates in the name, not in the IP
+		//       this simplifies the process and produces basically the same results...
+		// note: we do not spend time trying to update names that did not return an initial response...
+		for _, r := range res {
+			zone.startUpdatingName(r.Name())
+		}
+
+		return res, nil
+	}
+	return nil, LookupError(inaddr)
+}
+
+//////////////////////////////////////////////////////////////////////////////
+
+// Names updates try to find all the IPs for a given name with a mDNS query
+//
+// There are two types of names updates:
+//
+// - immediate updates.
+//   After a `DomainLookup*()` for a name not in the database we will return the
+//   first IP we can get with mDNS from other peers. Waiting for more responses would
+//   mean more latency in the response to the client, so we send only one answer BUT
+//   we also trigger an immediate update request for that name in order to get all
+//   the other IPs we didn't wait for...
+//
+// - periodic updates
+//   once we have obtained the first group of IPs for a name, we schedule a periodic
+//   refresh for that name, so we keep the list of IPs for that name up to date.
+//
+// These names updates are repeated until either
+//
+//  a) there is no interest in the name, determined by a global 'relevant time'
+//     and the last time some local client asked about the name,
+//     or
+//  b) no peers answer one of our refresh requests (because the name has probably
+//     disappeared from the network)
+//
+
+// TODO: for the sake of simplicity, we implement this mechanism with two channels: one for immediate
+//       updates and another one for scheduled requests. We could use a heap of "next-time"s and a timer,
+//       but that would require timers cancellations/updates/etc on insertions/removals/etc, and
+//       it is probably not worth the trouble as we always use the same refresh period.
+//       It could be useful if we move to a solution where we set update times from the responses TTLs,
+//       but we currently use a fixed TTL (30secs), the same as the refresh period...
+//       Anyway, maybe we will move to a gossip-based solution instead of doing this polling...
+
+// Check if we must start updating a name and, in that case, trigger a immediate update
+func (zone *zoneDb) startUpdatingName(name string) {
+	if zone.refreshInterval > 0 {
+		zone.mx.Lock()
+		defer zone.mx.Unlock()
+
+		// check if we should enqueue a refresh request for this name
+		n := zone.getNamesSet(defaultRemoteIdent).getName(name, true)
+		if n.lastRefreshTime.IsZero() {
+			now := zone.clock.Now()
+			n.lastRefreshTime = now
+
+			Debug.Printf("[zonedb] Creating new immediate refresh request for '%s'", name)
+			zone.refreshChan <- refreshRequest{name: name, time: now}
+		}
+	}
+}
+
+// A worker for updating the list of IPs we have for a name
+func (zone *zoneDb) updater(num int) {
+	defer zone.refreshWg.Done()
+
+	Debug.Printf("[zonedb] Starting background updater #%d...", num)
+	for {
+		select {
+		case <-zone.refreshCloseChan:
+			Debug.Printf("[zonedb] Background updater #%d: interrupted while waiting for requests: exiting", num)
+			return
+
+		case request := <-zone.refreshChan:
+			// if nobody has asked for this name for long time, just forget about it...
+			// this will eventually garbage collect the `refreshChan` and all remote info in absence of activity
+			if !zone.IsNameRelevant(request.name) || zone.IsNameExpired(request.name) {
+				Debug.Printf("[zonedb] '%s' seem to be irrelevant now: removing any remote information", request.name)
+				zone.mx.Lock()
+				zone.getNamesSet(defaultRemoteIdent).deleteName(request.name)
+				zone.mx.Unlock()
+				continue
+			}
+
+			// perform the refresh for this name
+			name := dns.Fqdn(request.name)
+			Debug.Printf("[zonedb] Refreshing name '%s' with mDNS...", name)
+			res, _ := zone.mdnsCli.InsistentLookupName(request.name)
+			if res != nil && len(res) > 0 {
+				numIps := len(res)
+				zone.mx.Lock()
+				now := zone.clock.Now()
+				added, removed := zone.getNamesSet(defaultRemoteIdent).getName(name, true).updateIPs(res, now)
+				zone.mx.Unlock()
+				Debug.Printf("[zonedb] Obtained %d IPs for name '%s' with mDNS: %d added, %d removed",
+					numIps, name, added, removed)
+
+				// once the name has been updated, we insert the request (back) in the periodic requests channel
+				now = zone.clock.Now()
+				request.time = now.Add(zone.refreshInterval)
+				Debug.Printf("[zonedb] Rescheduling update for '%s' in %.2f secs",
+					request.name, zone.refreshInterval.Seconds())
+				zone.refreshSchedChan <- request
+			} else {
+				Debug.Printf("[zonedb] nobody knows about '%s'... removing", name)
+				zone.mx.Lock()
+				zone.getNamesSet(defaultRemoteIdent).deleteName(request.name)
+				zone.mx.Unlock()
+			}
+		}
+	}
+}
+
+// The periodic updater
+// Consume requests from the `refreshSchedChan`, where requests with increasing scheduling time are enqueued
+// for refreshing names...
+func (zone *zoneDb) periodicUpdater() {
+	defer zone.refreshWg.Done()
+	for {
+		select {
+		case <-zone.refreshCloseChan:
+			Debug.Printf("[zonedb] Periodic updater: interrupted while waiting for requests: exiting")
+			return
+
+		case request := <-zone.refreshSchedChan:
+			now := zone.clock.Now()
+			// we can sleep until the update time has arrived, as requests are sorted by scheduled time (new request
+			// in this channel will always be scheduled later than the last item in the channel), so we can
+			// safely suspend this goroutine until then...
+			if request.time.After(now) {
+				ddiff := time.Duration(request.time.Sub(now).Nanoseconds())
+				timer := zone.clock.Timer(ddiff)
+				Debug.Printf("[zonedb] Periodic updater: new request for %s: sleeping for %.2f secs...",
+					request.name, ddiff.Seconds())
+				select {
+				case <-zone.refreshCloseChan:
+					Debug.Printf("[zonedb] Periodic updater: interrupted while sleeping: exiting")
+					return
+				case <-timer.C:
+				}
+			}
+
+			// once the time has arrived, we insert the request in the immediate refresh channel
+			zone.refreshChan <- request
+		}
+	}
+}

nameserver/zone_lookup_test.go

@@ -0,0 +1,108 @@
+package nameserver
+
+import (
+	. "github.com/weaveworks/weave/common"
+	wt "github.com/weaveworks/weave/testing"
+	"net"
+	"testing"
+	"time"
+)
+
+// Check that the refreshing mechanism works as expected
+func TestZoneRefresh(t *testing.T) {
+	const (
+		refreshInterval = 5
+		relevantTime    = 30
+	)
+	var (
+		name  = "mysql.weave.local."
+		addr1 = "10.2.8.4"
+		addr2 = "10.2.8.5"
+		addr3 = "10.2.8.6"
+		addr4 = "10.2.8.7"
+	)
+
+	InitDefaultLogging(testing.Verbose())
+	Info.Println("TestZoneRefresh starting")
+
+	clk := newMockedClock()
+
+	// Create multiple zone databases, linked through mocked mDNS connections
+	zc := ZoneConfig{
+		RefreshInterval: refreshInterval,
+		RelevantTime:    relevantTime,
+		Clock:           clk,
+	}
+	dbs := newZoneDbsWithMockedMDns(2, zc)
+	dbs.Start()
+	defer dbs.Stop()
+
+	time.Sleep(100 * time.Millisecond) // allow for server to get going
+
+	Debug.Printf("Adding '%s' to Db #1", name)
+	dbs[0].Zone.AddRecord("someident", name, net.ParseIP(addr1))
+
+	Debug.Printf("Checking that the name %s is relevant (as it has been locally inserted) and not remote", name)
+	wt.AssertTrue(t, dbs[0].Zone.IsNameRelevant(name), "name relevant")
+	wt.AssertTrue(t, dbs[0].Zone.HasNameLocalInfo(name), "local name info")
+	wt.AssertFalse(t, dbs[0].Zone.HasNameRemoteInfo(name), "remote name info")
+
+	Debug.Printf("Asking for '%s' to Db #1: should get 1 IP from the local database...", name)
+	res, err := dbs[0].Zone.DomainLookupName(name)
+	wt.AssertNoErr(t, err)
+	Debug.Printf("Got: %s", res)
+	t.Logf("Db #1 after the lookup:\n%s", dbs[0].Zone)
+	wt.AssertEqualInt(t, len(res), 1, "lookup result")
+
+	clk.Forward(refreshInterval / 2)
+	Debug.Printf("A couple of seconds later, we should still have one IP for that name")
+	res, err = dbs[0].Zone.DomainLookupName(name)
+	wt.AssertNoErr(t, err)
+	wt.AssertEqualInt(t, len(res), 1, "lookup result")
+
+	Debug.Printf("And then we add 2 IPs for that name at ZoneDb 2")
+	clk.Forward(1)
+	Debug.Printf("Adding 2 IPs to '%s' in Db #2", name)
+	dbs[1].Zone.AddRecord("someident", name, net.ParseIP(addr2))
+	dbs[1].Zone.AddRecord("someident", name, net.ParseIP(addr3))
+
+	Debug.Printf("Perform a lookup, to ensure the name will be updated in the background...")
+	dbs[0].Zone.DomainLookupName(name)
+
+	Debug.Printf("Wait for a while, until a refresh is performed...")
+	clk.Forward(refreshInterval + 1)
+
+	Debug.Printf("A refresh should have been scheduled now: we should have 3 IPs:")
+	Debug.Printf("the first (local) IP and the others obtained from zone2 with a mDNS query")
+	Debug.Printf("Asking for '%s' again... we should have 3 IPs now", name)
+	res, err = dbs[0].Zone.DomainLookupName(name)
+	Debug.Printf("Got: %s", res)
+	t.Logf("Db #1 after the second lookup:\n%s", dbs[0].Zone)
+	wt.AssertEqualInt(t, len(res), 3, "lookup result length")
+
+	Debug.Printf("We will not ask for `name` for a while, so it will become irrelevant and will be removed...")
+	clk.Forward(relevantTime + refreshInterval + 1)
+
+	// the name should be irrelevant now, and all remote info should have been
+	// removed from the zone database
+	Debug.Printf("Name '%s' should not be in the remote database in ZoneDb 1", name)
+	wt.AssertFalse(t, dbs[0].Zone.IsNameRelevant(name), "name still relevant after some inactivity time")
+	wt.AssertTrue(t, dbs[0].Zone.HasNameLocalInfo(name), "local name info")
+	wt.AssertFalse(t, dbs[0].Zone.HasNameRemoteInfo(name), "remote name info")
+
+	Debug.Printf("There is no remote info about this name at zone 1: a new IP appears remotely meanwhile...")
+	clk.Forward(1)
+	Debug.Printf("Adding '%s' to Db #2", name)
+	dbs[1].Zone.AddRecord("someident", name, net.ParseIP(addr4))
+
+	Debug.Printf("When we ask about this name again, we get 4 IPs (1 local, 3 remote)")
+	clk.Forward(1)
+	Debug.Printf("Asking for '%s' again... the first lookup will return only the local results", name)
+	res, err = dbs[0].Zone.DomainLookupName(name)
+	wt.AssertEqualInt(t, len(res), 1, "lookup result length")
+	Debug.Printf("... but a second lookup should return all the results in the network")
+	clk.Forward(1)
+	res, err = dbs[0].Zone.DomainLookupName(name)
+	Debug.Printf("Got: %s", res)
+	wt.AssertEqualInt(t, len(res), 4, "lookup result length") // TODO: this fails 1% of the runs... !!??
+}

nameserver/zone_test.go

@@ -1,6 +1,8 @@
 package nameserver
 
 import (
+	"fmt"
+	. "github.com/weaveworks/weave/common"
 	wt "github.com/weaveworks/weave/testing"
 	"net"
 	"testing"
@@ -8,60 +10,118 @@ import (
 
 func TestZone(t *testing.T) {
 	var (
-		containerID      = "deadbeef"
-		otherContainerID = "cowjuice"
-		successTestName  = "test1.weave."
-		testAddr1        = "10.2.2.1/24"
+		container1    = "deadbeef"
+		container2    = "cowjuice"
+		name1         = "test1.weave."
+		name1Addr1    = "10.9.2.1/24"
+		name1Addr2    = "10.9.2.2/24"
+		revName1Addr1 = "1.2.9.10.in-addr.arpa."
 	)
 
-	var zone = NewZoneDb(DefaultLocalDomain)
+	InitDefaultLogging(testing.Verbose())
 
-	ip, _, _ := net.ParseCIDR(testAddr1)
-	err := zone.AddRecord(containerID, successTestName, ip)
+	zone, err := NewZoneDb(ZoneConfig{})
+	wt.AssertNoErr(t, err)
+	err = zone.Start()
+	wt.AssertNoErr(t, err)
+	defer zone.Stop()
+
+	ip1, _, _ := net.ParseCIDR(name1Addr1)
+	t.Logf("Adding '%s'/%s to '%s'", name1, ip1, container1)
+	err = zone.AddRecord(container1, name1, ip1)
 	wt.AssertNoErr(t, err)
 
 	// Add a few more records to make the job harder
-	err = zone.AddRecord("abcdef0123", "adummy.weave.", net.ParseIP("10.2.0.1"))
+	t.Logf("Adding 'adummy.weave.' to 'abcdef0123'")
+	err = zone.AddRecord("abcdef0123", "adummy.weave.", net.ParseIP("10.9.0.1"))
 	wt.AssertNoErr(t, err)
-	err = zone.AddRecord("0123abcdef", "zdummy.weave.", net.ParseIP("10.2.0.2"))
+	t.Logf("Adding 'zdummy.weave.' to '0123abcdef'")
+	err = zone.AddRecord("0123abcdef", "zdummy.weave.", net.ParseIP("10.9.0.2"))
+	wt.AssertNoErr(t, err)
+	t.Logf("Zone database:\n%s", zone)
+
+	t.Logf("Checking if we can find the name '%s'", name1)
+	foundIPs, err := zone.LookupName(name1)
 	wt.AssertNoErr(t, err)
 
-	// Check that the address is now there.
-	foundIP, err := zone.LookupName(successTestName)
-	wt.AssertNoErr(t, err)
-
-	if !foundIP[0].IP().Equal(ip) {
-		t.Fatal("Unexpected result for", successTestName, foundIP)
+	if !foundIPs[0].IP().Equal(ip1) {
+		t.Fatal("Unexpected result for", name1, foundIPs)
 	}
 
-	// See if we can find the address by IP.
-	foundName, err := zone.LookupInaddr("1.2.2.10.in-addr.arpa.")
+	t.Logf("Checking if we cannot find some silly name like 'something.wrong'")
+	foundIPs, err = zone.LookupName("something.wrong")
+	wt.AssertErrorType(t, err, (*LookupError)(nil), fmt.Sprintf("unknown name: %+v", foundIPs))
+
+	ip2, _, _ := net.ParseCIDR(name1Addr2)
+	t.Logf("Adding a second IP for '%s'/%s to '%s'", name1, ip2, container1)
+	err = zone.AddRecord(container1, name1, ip2)
 	wt.AssertNoErr(t, err)
 
-	if foundName[0].Name() != successTestName {
-		t.Fatal("Unexpected result for", ip, foundName)
+	t.Logf("Checking if we can find both the old IP and the new IP for '%s'", name1)
+	foundIPs, err = zone.LookupName(name1)
+	wt.AssertNoErr(t, err)
+	if !(foundIPs[0].IP().Equal(ip1) || foundIPs[1].IP().Equal(ip1)) {
+		t.Fatal("Unexpected result for", name1, foundIPs)
+	}
+	if !(foundIPs[0].IP().Equal(ip2) || foundIPs[1].IP().Equal(ip2)) {
+		t.Fatal("Unexpected result for", name1, foundIPs)
 	}
 
-	err = zone.AddRecord(containerID, successTestName, ip)
+	t.Logf("Checking if we can find the address by IP '1.2.9.10.in-addr.arpa.'")
+	foundNames, err := zone.LookupInaddr("1.2.9.10.in-addr.arpa.")
+	wt.AssertNoErr(t, err)
+
+	if foundNames[0].Name() != name1 {
+		t.Fatal("Unexpected result for", ip1, foundNames)
+	}
+
+	t.Logf("Checking we can not find an unknown address '30.20.10.1.in-addr.arpa.'")
+	foundNames, err = zone.LookupInaddr("30.20.10.1.in-addr.arpa.")
+	wt.AssertErrorType(t, err, (*LookupError)(nil), fmt.Sprintf("unknown IP: %+v", foundNames))
+
+	t.Logf("Checking if adding again '%s'/%s in %s results in an error", name1, ip1, container1)
+	err = zone.AddRecord(container1, name1, ip1)
 	wt.AssertErrorType(t, err, (*DuplicateError)(nil), "duplicate add")
 
-	err = zone.AddRecord(otherContainerID, successTestName, ip)
-	// Delete the record for the original container
-	err = zone.DeleteRecord(containerID, ip)
+	t.Logf("Adding '%s'/%s in %s too", name1, ip1, container2)
+	err = zone.AddRecord(container2, name1, ip1)
 	wt.AssertNoErr(t, err)
 
-	_, err = zone.LookupName(successTestName)
+	name1Removed := 0
+	err = zone.ObserveName(name1, func() { t.Logf("Observer #1 for '%s' notified.", name1); name1Removed++ })
+	wt.AssertNoErr(t, err)
+	err = zone.ObserveName(name1, func() { t.Logf("Observer #2 for '%s' notified.", name1); name1Removed++ })
+	wt.AssertNoErr(t, err)
+	err = zone.ObserveInaddr(revName1Addr1, func() { t.Logf("Observer #1 for '%s' notified.", revName1Addr1); name1Removed++ })
 	wt.AssertNoErr(t, err)
 
-	err = zone.DeleteRecord(otherContainerID, ip)
+	t.Logf("Zone database:\n%s", zone)
+	t.Logf("Deleting the %s in %s", ip1, container1)
+	err = zone.DeleteRecord(container1, ip1)
+	wt.AssertNoErr(t, err)
+	t.Logf("Zone database:\n%s", zone)
+
+	t.Logf("Checking %s's observers have been notified on removal", name1)
+	if name1Removed < 3 {
+		t.Logf("Zone database:\n%s", zone)
+		t.Fatalf("Unexpected number (%d) of calls to observers", name1Removed)
+	}
+
+	t.Logf("Checking %s can be found", name1)
+	_, err = zone.LookupName(name1)
 	wt.AssertNoErr(t, err)
 
-	// Check that the address is not there now.
-	_, err = zone.LookupName(successTestName)
+	t.Logf("Checking %s is not found after removing %s it in %s and %s in %s",
+		name1, ip1, container2, ip2, container1)
+	err = zone.DeleteRecord(container1, ip2)
+	err = zone.DeleteRecord(container2, ip1)
+	wt.AssertNoErr(t, err)
+	t.Logf("Zone database:\n%s", zone)
+	_, err = zone.LookupName(name1)
 	wt.AssertErrorType(t, err, (*LookupError)(nil), "after deleting record")
 
-	// Delete a record that isn't there
-	err = zone.DeleteRecord(containerID, net.ParseIP("0.0.0.0"))
+	t.Logf("Checking if removing an unknown record results in an error")
+	err = zone.DeleteRecord(container1, net.ParseIP("0.0.0.0"))
 	wt.AssertErrorType(t, err, (*LookupError)(nil), "when deleting record that doesn't exist")
 }
 
@@ -72,14 +132,22 @@ func TestDeleteFor(t *testing.T) {
 		addr1 = "10.2.2.3/24"
 		addr2 = "10.2.7.8/24"
 	)
-	zone := NewZoneDb(DefaultLocalDomain)
+
+	InitDefaultLogging(testing.Verbose())
+
+	zone, err := NewZoneDb(ZoneConfig{})
+	wt.AssertNoErr(t, err)
+	err = zone.Start()
+	wt.AssertNoErr(t, err)
+	defer zone.Stop()
+
 	for _, addr := range []string{addr1, addr2} {
 		ip, _, _ := net.ParseCIDR(addr)
 		err := zone.AddRecord(id, name, ip)
 		wt.AssertNoErr(t, err)
 	}
 
-	_, err := zone.LookupName(name)
+	_, err = zone.LookupName(name)
 	wt.AssertNoErr(t, err)
 
 	err = zone.DeleteRecordsFor(id)

site/weavedns.md

@@ -44,10 +44,9 @@ $ docker attach $shell1
 ```
 
 Note that it is permissible to register multiple containers with the
-same name. weaveDNS picks one address to return when asked for the
-name. Since weaveDNS removes any container that dies, this is a simple
-way to implement redundancy.  In the current implementation it does
-not attempt to do load-balancing.
+same name: weaveDNS picks one address at random on each request. Since
+weaveDNS removes any container that dies, this is a simple way to
+implement redundancy.
 
 Each weaveDNS container started with `launch-dns` needs to be given
 its own, unique, IP address, in a subnet that is a) common to all
@@ -103,6 +102,9 @@ The inverse operation can be carried out using the `dns-remove` command:
 $ weave dns-remove 10.2.1.27 $shell2
 ```
 
+When queried about a name with multiple IPs, weaveDNS returns a random
+result from the set of IPs available.
+
 ## <a name="hot-swapping"></a>Hot-swapping service containers
 
 If you would like to deploy a new version of a service, keep the old

test/210_dns_multicast_test.sh

@@ -3,7 +3,9 @@
 . ./config.sh
 
 C1=10.2.3.78
-C2=10.2.3.34
+C2_1=10.2.3.34
+C2_2=10.2.3.35
+C2_3=10.2.3.36
 UNIVERSE=10.2.4.0/24
 NAME2=seetwo.weave.local
 NAME4=seefour.weave.local
@@ -13,16 +15,19 @@ start_suite "Resolve names across hosts (with and without IPAM)"
 weave_on $HOST1 launch -iprange $UNIVERSE
 weave_on $HOST2 launch -iprange $UNIVERSE $HOST1
 
-weave_on $HOST1 launch-dns 10.2.254.1/24
-weave_on $HOST2 launch-dns 10.2.254.2/24
+weave_on $HOST1 launch-dns 10.2.254.1/24 --no-cache
+weave_on $HOST2 launch-dns 10.2.254.2/24 --no-cache
 
-start_container          $HOST2 $C2/24 --name=c2 -h $NAME2
-start_container          $HOST2        --name=c4 -h $NAME4
-start_container_with_dns $HOST1 $C1/24 --name=c1
-start_container_with_dns $HOST1        --name=c3
+start_container          $HOST2 $C2_1/24 --name=c2 -h $NAME2
+start_container          $HOST2          --name=c4 -h $NAME4
+start_container_with_dns $HOST1 $C1/24   --name=c1
+start_container_with_dns $HOST1          --name=c3
 C4=$(container_ip $HOST2 c4)
 
-assert_dns_record $HOST1 c1 $NAME2 $C2
+tell_dns PUT $HOST2 c2 $NAME $C2_2 $NAME2
+tell_dns PUT $HOST2 c2 $NAME $C2_3 $NAME2
+
+assert_dns_record $HOST1 c1 $NAME2 $C2_1 $C2_2 $C2_3
 assert_dns_record $HOST1 c3 $NAME4 $C4
 
 assert_raises "exec_on $HOST1 c1 getent hosts 8.8.8.8 | grep google"

test/config.sh

@@ -99,6 +99,25 @@ exec_on() {
     docker -H tcp://$host:$DOCKER_PORT exec $container "$@"
 }
 
+# get a container IP: <host> <container>
+get_container_ip() {
+    docker -H tcp://$1:2375 inspect --format '{{ .NetworkSettings.IPAddress }}' $2
+}
+
+tell_dns() {
+    method=$1
+    host=$2
+    container=$3
+    name=$4
+    ip=$5
+    
+    dns=$(get_container_ip $host weavedns)
+    [[ -z "$DEBUG" ]] || greyly echo "$method'ing $name=$ip at $host/$dns"
+    run_on $host  \
+        curl --connect-timeout 3 -s -X $method \
+        "http://$dns:6785/name/$container/$ip" -d fqdn=$name
+}
+
 start_container() {
     host=$1
     shift 1
@@ -117,8 +136,20 @@ container_ip() {
 
 # assert_dns_record <host> <container> <name> <ip>
 assert_dns_record() {
-    assert "exec_on $1 $2 getent hosts $3 | tr -s ' '" "$4 $3"
-    assert "exec_on $1 $2 getent hosts $4 | tr -s ' '" "$4 $3"
+    host=$1
+    container=$2
+    name=$3
+    shift 3
+    exp_ips_regex=$(echo $@ | sed -r 's/ /\\\|/g')
+
+    [[ -z "$DEBUG" ]] || greyly echo "Checking wether $name exists at $host"
+    got_ip=$(exec_on $host $container getent hosts $name | tr -s ' ')
+    assert "echo \"$got_ip\" | grep -q \"$exp_ips_regex\" && echo found" "found"
+
+    [[ -z "$DEBUG" ]] || greyly echo "Checking wether the IPs \"$ips\" exists at $host"
+    for ip ; do
+        assert "exec_on $host $container getent hosts $ip | tr -s ' '" "$ip $name"
+    done
 }
 
 start_suite() {

weavedns/main.go

@@ -15,19 +15,24 @@ var version = "(unreleased version)"
 
 func main() {
 	var (
-		justVersion bool
-		ifaceName   string
-		apiPath     string
-		domain      string
-		dnsPort     int
-		httpPort    int
-		wait        int
-		timeout     int
-		udpbuf      int
-		cacheLen    int
-		watch       bool
-		debug       bool
-		err         error
+		justVersion     bool
+		ifaceName       string
+		apiPath         string
+		domain          string
+		dnsPort         int
+		httpPort        int
+		wait            int
+		timeout         int
+		udpbuf          int
+		refreshInterval int
+		refreshWorkers  int
+		relevantTime    int
+		maxAnswers      int
+		cacheLen        int
+		cacheDisabled   bool
+		watch           bool
+		debug           bool
+		err             error
 	)
 
 	flag.BoolVar(&justVersion, "version", false, "print version and exit")
@@ -37,11 +42,18 @@ func main() {
 	flag.IntVar(&wait, "wait", 0, "number of seconds to wait for interface to be created and come up")
 	flag.IntVar(&dnsPort, "dnsport", weavedns.DefaultServerPort, "port to listen to DNS requests")
 	flag.IntVar(&httpPort, "httpport", 6785, "port to listen to HTTP requests")
-	flag.IntVar(&timeout, "timeout", weavedns.DefaultTimeout, "timeout for resolutions")
-	flag.IntVar(&udpbuf, "udpbuf", weavedns.DefaultUDPBuflen, "UDP buffer length")
 	flag.IntVar(&cacheLen, "cache", weavedns.DefaultCacheLen, "cache length")
 	flag.BoolVar(&watch, "watch", true, "watch the docker socket for container events")
 	flag.BoolVar(&debug, "debug", false, "output debugging info to stderr")
+	// advanced options
+	flag.IntVar(&refreshInterval, "refresh", weavedns.DefaultRefreshInterval, "refresh interval (in secs) for local names (0=disable)")
+	flag.IntVar(&refreshWorkers, "refresh-workers", weavedns.DefaultNumUpdaters, "default number of background updaters")
+	flag.IntVar(&maxAnswers, "max-answers", weavedns.DefaultMaxAnswers, "maximum number of answers returned to clients (0=unlimited)")
+	flag.IntVar(&relevantTime, "relevant", weavedns.DefaultRelevantTime, "life time for info in the absence of queries (in secs)")
+	flag.IntVar(&udpbuf, "udpbuf", weavedns.DefaultUDPBuflen, "UDP buffer length")
+	flag.IntVar(&timeout, "timeout", weavedns.DefaultTimeout, "timeout for resolutions (in millisecs)")
+	flag.BoolVar(&cacheDisabled, "no-cache", false, "disable the cache")
+
 	flag.Parse()
 
 	if justVersion {
@@ -50,16 +62,7 @@ func main() {
 	}
 
 	InitDefaultLogging(debug)
-	Info.Printf("WeaveDNS version %s\n", version) // first thing in log: the version
-
-	var zone = weavedns.NewZoneDb(domain)
-
-	if watch {
-		err := updater.Start(apiPath, zone)
-		if err != nil {
-			Error.Fatal("Unable to start watcher", err)
-		}
-	}
+	Info.Printf("[main] WeaveDNS version %s\n", version) // first thing in log: the version
 
 	var iface *net.Interface
 	if ifaceName != "" {
@@ -73,19 +76,44 @@ func main() {
 		}
 	}
 
-	srvConfig := weavedns.DNSServerConfig{
-		Port:        dnsPort,
-		CacheLen:    cacheLen,
-		LocalDomain: domain,
-		Timeout:     timeout,
-		UDPBufLen:   udpbuf,
+	zoneConfig := weavedns.ZoneConfig{
+		Domain:          domain,
+		Iface:           iface,
+		RefreshInterval: refreshInterval,
+		RefreshWorkers:  refreshWorkers,
+		RelevantTime:    relevantTime,
+	}
+	zone, err := weavedns.NewZoneDb(zoneConfig)
+	if err != nil {
+		Error.Fatal("[main] Unable to initialize the Zone database", err)
+	}
+	err = zone.Start()
+	if err != nil {
+		Error.Fatal("[main] Unable to start the Zone database", err)
+	}
+	defer zone.Stop()
+
+	if watch {
+		err := updater.Start(apiPath, zone)
+		if err != nil {
+			Error.Fatal("[main] Unable to start watcher", err)
+		}
 	}
 
-	srv, err := weavedns.NewDNSServer(srvConfig, zone, iface)
+	srvConfig := weavedns.DNSServerConfig{
+		Zone:          zone,
+		Port:          dnsPort,
+		CacheLen:      cacheLen,
+		MaxAnswers:    maxAnswers,
+		Timeout:       timeout,
+		UDPBufLen:     udpbuf,
+		CacheDisabled: cacheDisabled,
+	}
+
+	srv, err := weavedns.NewDNSServer(srvConfig)
 	if err != nil {
-		Error.Fatal("Failed to initialize the WeaveDNS server", err)
+		Error.Fatal("[main] Failed to initialize the WeaveDNS server", err)
 	}
-	Info.Println("Upstream", srv.Upstream)
 
 	go SignalHandlerLoop(srv)
 	go weavedns.ListenHTTP(version, srv, domain, zone, httpPort)