diff --git a/nameserver/addrs.go b/nameserver/addrs.go
index 0bdb0dca..bb4ac27c 100644
--- a/nameserver/addrs.go
+++ b/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()
/////////////////////////////////////////////////////////////
diff --git a/nameserver/cache.go b/nameserver/cache.go
index 644814f0..7a5cc0c1 100644
--- a/nameserver/cache.go
+++ b/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
}
diff --git a/nameserver/dns.go b/nameserver/dns.go
index 904a9bf2..3f919993 100644
--- a/nameserver/dns.go
+++ b/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
+}
diff --git a/nameserver/dns_test.go b/nameserver/dns_test.go
new file mode 100644
index 00000000..58e53784
--- /dev/null
+++ b/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")
+}
diff --git a/nameserver/http_test.go b/nameserver/http_test.go
index 5979d683..f3cee7a9 100644
--- a/nameserver/http_test.go
+++ b/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)
diff --git a/nameserver/mdns.go b/nameserver/mdns.go
new file mode 100644
index 00000000..e476fb4f
--- /dev/null
+++ b/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)
+}
diff --git a/nameserver/mdns_test.go b/nameserver/mdns_test.go
new file mode 100644
index 00000000..dc0fe11f
--- /dev/null
+++ b/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)
+ }
+}
diff --git a/nameserver/mocks_test.go b/nameserver/mocks_test.go
index 5f3ba870..ca3c941d 100644
--- a/nameserver/mocks_test.go
+++ b/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))
+}
diff --git a/nameserver/server.go b/nameserver/server.go
index 3077c8f4..79a4e802 100644
--- a/nameserver/server.go
+++ b/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)
+}
diff --git a/nameserver/server_cache_test.go b/nameserver/server_cache_test.go
new file mode 100644
index 00000000..cd16537a
--- /dev/null
+++ b/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")
+
+}
diff --git a/nameserver/server_test.go b/nameserver/server_test.go
index 7e893c5b..8f1897d1 100644
--- a/nameserver/server_test.go
+++ b/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
-}
diff --git a/nameserver/zone.go b/nameserver/zone.go
index ca23dab1..c797779c 100644
--- a/nameserver/zone.go
+++ b/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)
+ }
+}
diff --git a/nameserver/zone_lookup.go b/nameserver/zone_lookup.go
new file mode 100644
index 00000000..b38971d3
--- /dev/null
+++ b/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
+ }
+ }
+}
diff --git a/nameserver/zone_lookup_test.go b/nameserver/zone_lookup_test.go
new file mode 100644
index 00000000..6c09ab6b
--- /dev/null
+++ b/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... !!??
+}
diff --git a/nameserver/zone_test.go b/nameserver/zone_test.go
index a807464f..968b4b7d 100644
--- a/nameserver/zone_test.go
+++ b/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)
diff --git a/site/weavedns.md b/site/weavedns.md
index db736eac..d13009b7 100644
--- a/site/weavedns.md
+++ b/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.
+
## Hot-swapping service containers
If you would like to deploy a new version of a service, keep the old
diff --git a/test/210_dns_multicast_test.sh b/test/210_dns_multicast_test.sh
index 097ef027..ca959c1a 100755
--- a/test/210_dns_multicast_test.sh
+++ b/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"
diff --git a/test/config.sh b/test/config.sh
index 8e74c5ab..c8bdcf06 100644
--- a/test/config.sh
+++ b/test/config.sh
@@ -99,6 +99,25 @@ exec_on() {
docker -H tcp://$host:$DOCKER_PORT exec $container "$@"
}
+# get a container IP:
+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
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() {
diff --git a/weavedns/main.go b/weavedns/main.go
index 6a888896..db38cbcd 100644
--- a/weavedns/main.go
+++ b/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)