Merge pull request #559 from inercia/weave-338

Support for multiple responses in mDNS
This commit is contained in:
Adam Harrison committed 2015-04-24 12:01:43 +01:00
commit 443b11646d
17 files changed
+506 -216

No files matched your search

+111
View File
@@ -0,0 +1,111 @@
package nameserver
import (
"bytes"
"fmt"
"net"
)
// A zone record
// Note: we will try to keep all the zone records information is concentrated here. Maybe not all queries will
// use things like priorities or weights, but we do not want to create a hierarchy for dealing with all
// possible queries...
type ZoneRecord interface {
Name() string // The name for this IP
IP() net.IP // The IP (v4) address
Priority() int // The priority
Weight() int // The weight
}
type ZoneLookup interface {
// Lookup for a name
LookupName(name string) ([]ZoneRecord, error)
// Lookup for an address
LookupInaddr(inaddr string) ([]ZoneRecord, error)
}
type ZoneObserver 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 ZoneRecordObserver func()
/////////////////////////////////////////////////////////////
// A basic record struct (satisfying the ZoneRecord interface)
type Record struct {
name string
ip net.IP
priority int
weight int
}
func (r Record) Name() string { return r.name }
func (r Record) IP() net.IP { return r.ip }
func (r Record) Priority() int { return r.priority }
func (r Record) Weight() int { return r.weight }
func (i Record) String() string {
var buf bytes.Buffer
if len(i.Name()) > 0 {
fmt.Fprintf(&buf, "%s", i.Name())
}
if !i.IP().IsUnspecified() {
fmt.Fprintf(&buf, "[%s]", i.IP())
}
if i.Priority() > 0 {
fmt.Fprintf(&buf, "/P:%d", i.Priority())
}
if i.Weight() > 0 {
fmt.Fprintf(&buf, "/W:%d", i.Weight())
}
return buf.String()
}
/////////////////////////////////////////////////////////////
func newParseError(reason string, addr string) *net.ParseError {
return &net.ParseError{Type: reason, Text: addr}
}
// simple IPv4 type that can be used as a key in a map (in contrast with net.IP), used for sets of IPs, etc...
type IPv4 [4]byte
func (ip IPv4) toNetIP() net.IP { return net.IPv4(ip[0], ip[1], ip[2], ip[3]) }
func (ip IPv4) String() string { return ip.toNetIP().String() }
func ipToIPv4(nip net.IP) IPv4 { ip := nip.To4(); return IPv4([4]byte{ip[0], ip[1], ip[2], ip[3]}) }
// Parse an address (eg, "10.13.12.11") and return the corresponding IPv4 (eg, "10.13.12.11")
func addrToIPv4(addr string) (IPv4, error) {
ip := net.ParseIP(addr)
if ip == nil {
return IPv4{}, newParseError("IP address", addr)
}
return ipToIPv4(ip), nil
}
// Parse a reverse address (eg, "11.12.13.10.in-addr.arpa.") and return the corresponding IPv4 (eg, "10.13.12.11")
func raddrToIPv4(addr string) (IPv4, error) {
l := len(addr)
if l < rdnsDomainLen+1 {
return IPv4{}, newParseError("too short reverse IP address", addr)
}
suffixLen := l - rdnsDomainLen
suffix := addr[suffixLen+1:]
if suffix != RDNSDomain {
return IPv4{}, newParseError("suffix of reverse IP address", addr)
}
ipStr := addr[:suffixLen]
revIP := net.ParseIP(ipStr)
if revIP == nil {
return IPv4{}, newParseError("reverse IP address", addr)
}
revIP4 := revIP.To4()
return IPv4([4]byte{revIP4[3], revIP4[2], revIP4[1], revIP4[0]}), nil
}
+30
View File
@@ -0,0 +1,30 @@
package nameserver
import (
. "github.com/weaveworks/weave/common"
wt "github.com/weaveworks/weave/testing"
"net"
"testing"
)
func TestAddrs(t *testing.T) {
InitDefaultLogging(testing.Verbose())
ip, err := addrToIPv4("10.13.12.11")
wt.AssertNoErr(t, err)
wt.AssertTrue(t, net.ParseIP("10.13.12.11").Equal(ip.toNetIP()), "IP")
ip, err = raddrToIPv4("11.12.13.10.in-addr.arpa.")
wt.AssertNoErr(t, err)
wt.AssertTrue(t, net.ParseIP("10.13.12.11").Equal(ip.toNetIP()), "IP")
// some malformed addresses
ip, err = addrToIPv4("10.13.12")
wt.AssertTrue(t, err != nil, "when parsing malformed address")
ip, err = addrToIPv4("10.13.AA.12")
wt.AssertTrue(t, err != nil, "when parsing malformed address")
ip, err = raddrToIPv4("11.12.13.10.in-axxx.arpa.")
wt.AssertTrue(t, err != nil, "when parsing malformed address")
ip, err = raddrToIPv4("11.12.13.10.in-addr")
wt.AssertTrue(t, err != nil, "when parsing malformed address")
}
+3 -2
View File
@@ -247,6 +247,7 @@ func (c *Cache) Purge(now time.Time) {
}
// Add adds a reply to the cache.
// When `ttl` is equal to `nullTTL`, the cache entry will be valid until the closest TTL in the `reply`
func (c *Cache) Put(request *dns.Msg, reply *dns.Msg, ttl int, flags uint8, now time.Time) int {
c.lock.Lock()
defer c.lock.Unlock()
@@ -274,8 +275,7 @@ func (c *Cache) Put(request *dns.Msg, reply *dns.Msg, ttl int, flags uint8, now
}
// Look up for a question's reply from the cache.
// If no reply is stored in the cache, it returns a `nil` reply and no error. The caller can then `Wait()`
// for another goroutine `Put`ing a reply in the cache.
// If no reply is stored in the cache, it returns a `nil` reply and no error.
func (c *Cache) Get(request *dns.Msg, maxLen int, now time.Time) (reply *dns.Msg, err error) {
c.lock.Lock()
defer c.lock.Unlock()
@@ -307,6 +307,7 @@ func (c *Cache) Remove(question *dns.Question) {
key := cacheKey(*question)
if entry, found := c.entries[key]; found {
Debug.Printf("[cache] removing %s-response for '%s'", dns.TypeToString[question.Qtype], question.Name)
if entry.index > 0 {
heap.Remove(&c.entriesH, entry.index)
}
+15 -8
View File
@@ -23,15 +23,17 @@ func TestCacheLength(t *testing.T) {
t.Logf("Inserting 256 questions in the cache at '%s', with TTL from 0 to 255", insTime)
for i := 0; i < cacheLen*2; i++ {
name := fmt.Sprintf("name%d", i)
questionMsg := new(dns.Msg)
questionMsg.SetQuestion(fmt.Sprintf("name%d", i), dns.TypeA)
questionMsg.SetQuestion(name, dns.TypeA)
questionMsg.RecursionDesired = true
question := &questionMsg.Question[0]
ip := net.ParseIP(fmt.Sprintf("10.0.1.%d", i))
ips := []net.IP{ip}
reply := makeAddressReply(questionMsg, question, ips)
records := []ZoneRecord{Record{name, ip, 0, 0}}
reply := makeAddressReply(questionMsg, question, records)
reply.Answer[0].Header().Ttl = uint32(i)
l.Put(questionMsg, reply, 0, 0, insTime)
@@ -81,7 +83,8 @@ func TestCacheEntries(t *testing.T) {
}
t.Logf("Inserting the reply")
reply1 := makeAddressReply(questionMsg, question, []net.IP{net.ParseIP("10.0.1.1")})
records := []ZoneRecord{Record{"some.name", net.ParseIP("10.0.1.1"), 0, 0}}
reply1 := makeAddressReply(questionMsg, question, records)
l.Put(questionMsg, reply1, nullTTL, 0, time.Now())
timeGet1 := time.Now()
@@ -113,7 +116,8 @@ func TestCacheEntries(t *testing.T) {
}
t.Logf("Checking that an Remove() results in Get() returning nothing")
replyTemp := makeAddressReply(questionMsg, question, []net.IP{net.ParseIP("10.0.9.9")})
records = []ZoneRecord{Record{"some.name", net.ParseIP("10.0.9.9"), 0, 0}}
replyTemp := makeAddressReply(questionMsg, question, records)
l.Put(questionMsg, replyTemp, nullTTL, 0, time.Now())
lenBefore := l.Len()
l.Remove(question)
@@ -127,10 +131,12 @@ func TestCacheEntries(t *testing.T) {
t.Logf("Inserting a two replies for the same query")
timePut2 := time.Now()
reply2 := makeAddressReply(questionMsg, question, []net.IP{net.ParseIP("10.0.1.2")})
records = []ZoneRecord{Record{"some.name", net.ParseIP("10.0.1.2"), 0, 0}}
reply2 := makeAddressReply(questionMsg, question, records)
l.Put(questionMsg, reply2, nullTTL, 0, timePut2)
timePut3 := timePut2.Add(time.Duration(1) * time.Second)
reply3 := makeAddressReply(questionMsg, question, []net.IP{net.ParseIP("10.0.1.3")})
records = []ZoneRecord{Record{"some.name", net.ParseIP("10.0.1.3"), 0, 0}}
reply3 := makeAddressReply(questionMsg, question, records)
l.Put(questionMsg, reply3, nullTTL, 0, timePut3)
t.Logf("Checking we get the last one...")
@@ -171,7 +177,8 @@ func TestCacheEntries(t *testing.T) {
wt.AssertNil(t, resp, "reponse from Get() yet")
t.Logf("Checking that an Remove() between Get() and Put() does not break things")
replyTemp2 := makeAddressReply(questionMsg2, question2, []net.IP{net.ParseIP("10.0.9.9")})
records = []ZoneRecord{Record{"some.name", net.ParseIP("10.0.9.9"), 0, 0}}
replyTemp2 := makeAddressReply(questionMsg2, question2, records)
l.Remove(question2)
l.Put(questionMsg2, replyTemp2, nullTTL, 0, time.Now())
resp, err = l.Get(questionMsg2, minUDPSize, time.Now())
+10 -8
View File
@@ -2,7 +2,6 @@ package nameserver
import (
"github.com/miekg/dns"
"net"
)
const (
@@ -34,20 +33,23 @@ func makeTruncatedReply(r *dns.Msg) *dns.Msg {
return reply
}
func makeAddressReply(r *dns.Msg, q *dns.Question, addrs []net.IP) *dns.Msg {
func makeAddressReply(r *dns.Msg, q *dns.Question, addrs []ZoneRecord) *dns.Msg {
answers := make([]dns.RR, len(addrs))
header := makeHeader(r, q)
count := 0
for _, addr := range addrs {
ip := addr.IP()
ip4 := ip.To4()
switch q.Qtype {
case dns.TypeA:
if ip4 := addr.To4(); ip4 != nil {
answers[count] = &dns.A{Hdr: *header, A: addr}
if ip4 != nil {
answers[count] = &dns.A{Hdr: *header, A: ip}
count++
}
case dns.TypeAAAA:
if ip4 := addr.To4(); ip4 == nil {
answers[count] = &dns.AAAA{Hdr: *header, AAAA: addr}
if ip4 == nil {
answers[count] = &dns.AAAA{Hdr: *header, AAAA: ip}
count++
}
}
@@ -55,11 +57,11 @@ func makeAddressReply(r *dns.Msg, q *dns.Question, addrs []net.IP) *dns.Msg {
return makeReply(r, answers[:count])
}
func makePTRReply(r *dns.Msg, q *dns.Question, names []string) *dns.Msg {
func makePTRReply(r *dns.Msg, q *dns.Question, names []ZoneRecord) *dns.Msg {
answers := make([]dns.RR, len(names))
header := makeHeader(r, q)
for i, name := range names {
answers[i] = &dns.PTR{Hdr: *header, Ptr: name}
answers[i] = &dns.PTR{Hdr: *header, Ptr: name.Name()}
}
return makeReply(r, answers)
}
+9 -6
View File
@@ -27,7 +27,8 @@ func TestHttp(t *testing.T) {
dockerIP = "9.8.7.6"
)
var zone = new(ZoneDb)
zone := NewZoneDb(DefaultLocalDomain)
port := rand.Intn(10000) + 32768
fmt.Println("Http test on port", port)
go ListenHTTP("", nil, testDomain, zone, port)
@@ -43,11 +44,11 @@ func TestHttp(t *testing.T) {
wt.AssertStatus(t, resp.StatusCode, http.StatusOK, "http response")
// Check that the address is now there.
ip, _, _ := net.ParseCIDR(testAddr1)
foundIP, err := zone.LookupName(successTestName)
wt.AssertNoErr(t, err)
ip, _, _ := net.ParseCIDR(testAddr1)
if !foundIP.Equal(ip) {
t.Fatal("Unexpected result for", successTestName, foundIP)
if !foundIP[0].IP().Equal(ip) {
t.Fatalf("Unexpected result for %s: received %s, expected %s", successTestName, foundIP, ip)
}
// Adding exactly the same address should be OK
@@ -70,7 +71,8 @@ func TestHttp(t *testing.T) {
wt.AssertStatus(t, resp.StatusCode, http.StatusOK, "http response")
// Check that the address is still resolvable.
_, err = zone.LookupName(successTestName)
x, err := zone.LookupName(successTestName)
t.Logf("Got %s", x)
wt.AssertNoErr(t, err)
// Delete the address record mentioning the other container
@@ -79,7 +81,8 @@ func TestHttp(t *testing.T) {
wt.AssertStatus(t, resp.StatusCode, http.StatusOK, "http response")
// Check that the address is gone
_, err = zone.LookupName(successTestName)
x, err = zone.LookupName(successTestName)
t.Logf("Got %s", x)
wt.AssertErrorType(t, err, (*LookupError)(nil), "fully-removed address")
// Would like to shut down the http server at the end of this test
+67 -26
View File
@@ -27,14 +27,33 @@ var (
)
type Response struct {
Name string
Addr net.IP
Err error
name string
addr net.IP
err error
}
func (r Response) Name() string { return r.name }
func (r Response) IP() net.IP { return r.addr }
func (r Response) Priority() int { return 0 }
func (r Response) Weight() int { return 0 }
func (r Response) Equal(r2 *Response) bool {
if r.name != r2.name {
return false
}
if !r.addr.Equal(r2.addr) {
return false
}
if r.err != r2.err {
return false
}
return true
}
type responseInfo struct {
timeout time.Time // if no answer by this time, give up
ch chan<- *Response
timeout time.Time // if no answer by this time, give up
insistent bool // insistent queries are not removed on the first reply
ch chan<- *Response
}
// Represents one query that we have sent for one name.
@@ -55,6 +74,7 @@ type MDNSClient struct {
addr *net.UDPAddr
inflight map[string]*inflightQuery
actionChan chan<- MDNSAction
running bool
}
type mDNSQueryInfo struct {
@@ -64,17 +84,21 @@ type mDNSQueryInfo struct {
}
func NewMDNSClient() (*MDNSClient, error) {
if conn, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.IPv4zero, Port: 0}); err != nil {
return nil, err
} else {
return &MDNSClient{
conn: conn,
addr: ipv4Addr,
inflight: make(map[string]*inflightQuery)}, nil
}
return &MDNSClient{
addr: ipv4Addr,
running: false,
inflight: make(map[string]*inflightQuery)}, nil
}
func (c *MDNSClient) Start(ifi *net.Interface) error {
func (c *MDNSClient) Start(ifi *net.Interface) (err error) {
if c.running {
return nil
}
if c.conn, err = net.ListenUDP("udp4", &net.UDPAddr{IP: net.IPv4zero, Port: 0}); err != nil {
return err
}
multicast, err := LinkLocalMulticastListener(ifi)
if err != nil {
return err
@@ -93,7 +117,13 @@ func (c *MDNSClient) Start(ifi *net.Interface) error {
actionChan := make(chan MDNSAction, MailboxSize)
c.actionChan = actionChan
go c.actorLoop(actionChan)
return nil
}
func (c *MDNSClient) Stop() error {
if c.running {
c.actionChan <- nil
}
return nil
}
@@ -106,7 +136,7 @@ func LinkLocalMulticastListener(ifi *net.Interface) (net.PacketConn, error) {
type MDNSAction func()
// Async
func (c *MDNSClient) SendQuery(name string, querytype uint16, responseCh chan<- *Response) {
func (c *MDNSClient) SendQuery(name string, querytype uint16, insistent bool, responseCh chan<- *Response) {
c.actionChan <- func() {
query, found := c.inflight[name]
if !found {
@@ -115,7 +145,7 @@ func (c *MDNSClient) SendQuery(name string, querytype uint16, responseCh chan<-
m.RecursionDesired = false
buf, err := m.Pack()
if err != nil {
responseCh <- &Response{Err: err}
responseCh <- &Response{err: err}
close(responseCh)
return
}
@@ -124,15 +154,16 @@ func (c *MDNSClient) SendQuery(name string, querytype uint16, responseCh chan<-
id: m.Id,
}
if _, err = c.conn.WriteTo(buf, c.addr); err != nil {
responseCh <- &Response{Err: err}
responseCh <- &Response{err: err}
close(responseCh)
return
}
c.inflight[name] = query
}
info := &responseInfo{
ch: responseCh,
timeout: time.Now().Add(mDNSTimeout),
ch: responseCh,
timeout: time.Now().Add(mDNSTimeout),
insistent: insistent,
}
// Invariant on responseInfos: they are in ascending order of
// timeout. Since we use a fixed interval from Now(), this
@@ -151,20 +182,30 @@ func (c *MDNSClient) ResponseCallback(r *dns.Msg) {
switch rr := answer.(type) {
case *dns.A:
name = rr.Hdr.Name
res = &Response{Addr: rr.A}
res = &Response{addr: rr.A}
case *dns.PTR:
name = rr.Hdr.Name
res = &Response{Name: rr.Ptr}
res = &Response{name: rr.Ptr}
default:
return
}
if query, found := c.inflight[name]; found {
newResponseInfos := make([]*responseInfo, 0)
for _, resp := range query.responseInfos {
resp.ch <- res
close(resp.ch)
// insistent queries are not removed on the first reply, but on the timeout
if resp.insistent {
newResponseInfos = append(newResponseInfos, resp)
} else {
close(resp.ch)
}
}
if len(newResponseInfos) == 0 {
delete(c.inflight, name)
} else {
query.responseInfos = newResponseInfos
}
delete(c.inflight, name)
} else {
// We've received a response that didn't match a query
// Do we want to cache it?
@@ -207,13 +248,13 @@ func (c *MDNSClient) checkInFlightQueries() time.Duration {
func (c *MDNSClient) actorLoop(actionChan <-chan MDNSAction) {
timer := time.NewTimer(MaxDuration)
run := func() { timer.Reset(c.checkInFlightQueries()) }
terminate := false
for !terminate {
c.running = true
for c.running {
select {
case action := <-actionChan:
if action == nil {
c.listener.Shutdown()
terminate = true
c.running = false
} else {
action()
run()
+14 -14
View File
@@ -108,9 +108,9 @@ func (c *testContext) checkResponse(t *testing.T, channelOk bool, resp *Response
c.channel = nil
return
}
wt.AssertNoErr(t, resp.Err)
log.Printf("Got address response %s addr %s", resp.Name, resp.Addr)
c.receivedAddr = resp.Addr
wt.AssertNoErr(t, resp.err)
log.Printf("Got address response %s addr %s", resp.Name, resp.IP())
c.receivedAddr = resp.IP()
c.receivedCount++
}
@@ -123,7 +123,7 @@ func TestSimpleQuery(t *testing.T) {
context := newTestContext()
// First, a test we expect to succeed
mdnsClient.SendQuery(successTestName, dns.TypeA, context.channel)
mdnsClient.SendQuery(successTestName, dns.TypeA, false, context.channel)
for resp := range context.channel {
context.checkResponse(t, true, resp)
}
@@ -134,7 +134,7 @@ func TestSimpleQuery(t *testing.T) {
// Now, a test we expect to time out with no responses
context = newTestContext()
mdnsClient.SendQuery("test2.weave.", dns.TypeA, context.channel)
mdnsClient.SendQuery("test2.weave.", dns.TypeA, false, context.channel)
for resp := range context.channel {
context.checkResponse(t, true, resp)
}
@@ -153,8 +153,8 @@ func TestParallelQuery(t *testing.T) {
context1 := newTestContext()
context2 := newTestContext()
go mdnsClient.SendQuery(successTestName, dns.TypeA, context1.channel)
go mdnsClient.SendQuery(successTestName, dns.TypeA, context2.channel)
go mdnsClient.SendQuery(successTestName, dns.TypeA, false, context1.channel)
go mdnsClient.SendQuery(successTestName, dns.TypeA, false, context2.channel)
timeout := time.After(2 * time.Second)
outerloop:
for context1.channel != nil || context2.channel != nil {
@@ -178,18 +178,18 @@ func TestAsLookup(t *testing.T) {
defer mdnsClient.Shutdown()
defer server.Shutdown()
ip, err := mdnsClient.LookupName(successTestName)
ips, err := mdnsClient.LookupName(successTestName)
wt.AssertNoErr(t, err)
if !testAddr.Equal(ip) {
t.Fatalf("Returned address incorrect %s", ip)
if !testAddr.Equal(ips[0].IP()) {
t.Fatalf("Returned address incorrect %s", ips)
}
ip, err = mdnsClient.LookupName("foo.example.com.")
ips, err = mdnsClient.LookupName("foo.example.com.")
wt.AssertErrorType(t, err, (*LookupError)(nil), "unknown hostname")
name, err := mdnsClient.LookupInaddr(testInAddr)
names, err := mdnsClient.LookupInaddr(testInAddr)
wt.AssertNoErr(t, err)
if !(successTestName == name) {
t.Fatalf("Expected name %s, got %s", successTestName, name)
if !(successTestName == names[0].Name()) {
t.Fatalf("Expected name %s, got %s", successTestName, names)
}
}
+54 -22
View File
@@ -3,37 +3,69 @@ package nameserver
import (
"github.com/miekg/dns"
. "github.com/weaveworks/weave/common"
"net"
)
func mdnsLookup(client *MDNSClient, name string, qtype uint16) (*Response, error) {
func mdnsLookup(client *MDNSClient, name string, qtype uint16, insistent bool) ([]ZoneRecord, error) {
responses := make([]*Response, 0)
channel := make(chan *Response)
client.SendQuery(name, qtype, channel)
client.SendQuery(name, qtype, insistent, channel)
for resp := range channel {
if err := resp.Err; err != nil {
Debug.Printf("[mdns] Error for query type %s name %s: %s",
dns.TypeToString[qtype], name, err)
if err := resp.err; err != nil {
Debug.Printf("[mdns] Error for query type %s name %s: %s", dns.TypeToString[qtype], name, err)
return nil, err
}
Debug.Printf("[mdns] Got response name for query type %s name %s: name %s, addr %s",
dns.TypeToString[qtype], name, resp.Name, resp.Addr)
return resp, nil
Debug.Printf("[mdns] Got response name for %s-query about name '%s': name '%s', addr '%s'",
dns.TypeToString[qtype], name, resp.name, resp.addr)
if !insistent {
return []ZoneRecord{resp}, nil // non-background queries return the first response...
}
// add the response, avoiding duplicates
// TODO: replace this linear search by something better... (not a big deal as we don't expect long responses)
duplicate := false
for _, r := range responses {
if r.Equal(resp) {
duplicate = true
break
}
}
if !duplicate {
responses = append(responses, resp)
}
}
return nil, LookupError(name)
if len(responses) == 0 {
return nil, LookupError(name)
}
// convert []Response to []ZoneRecord
res := make([]ZoneRecord, len(responses))
for i, rr := range responses {
res[i] = ZoneRecord(rr)
}
return res, nil
}
func (client *MDNSClient) LookupName(name string) (net.IP, error) {
if r, e := mdnsLookup(client, name, dns.TypeA); r != nil {
return r.Addr, nil
} else {
return nil, e
}
// Perform a lookup for a name, returning either an IP address or an error
func (client *MDNSClient) LookupName(name string) ([]ZoneRecord, error) {
return mdnsLookup(client, name, dns.TypeA, false)
}
func (client *MDNSClient) LookupInaddr(inaddr string) (string, error) {
if r, e := mdnsLookup(client, inaddr, dns.TypePTR); r != nil {
return r.Name, nil
} else {
return "", e
}
// Perform a lookup for an IP address, returning either a name or an error
func (client *MDNSClient) LookupInaddr(inaddr string) ([]ZoneRecord, error) {
return mdnsLookup(client, inaddr, dns.TypePTR, false)
}
// Perform lookup for a name, returning either a list of IP addresses or an error
// Insistent queries will wait up to `mDNSTimeout` milliseconds for responses
func (client *MDNSClient) InsistentLookupName(name string) ([]ZoneRecord, error) {
return mdnsLookup(client, name, dns.TypeA, true)
}
// Perform an insistent lookup for an IP address, returning either a list of names or an error
// Insistent queries will wait up to `mDNSTimeout` milliseconds for responses
func (client *MDNSClient) InsistentLookupInaddr(inaddr string) ([]ZoneRecord, error) {
return mdnsLookup(client, inaddr, dns.TypePTR, true)
}
+40 -21
View File
@@ -11,15 +11,13 @@ type MDNSServer struct {
sendconn *net.UDPConn
srv *dns.Server
zone Zone
running bool
}
// Create a new mDNS server
// Nothing will be done (including port bindings) until you `Start()` the server
func NewMDNSServer(zone Zone) (*MDNSServer, error) {
// This is a bit of a kludge - per the RFC we should send responses from 5353, but that doesn't seem to work
sendconn, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.IPv4zero, Port: 0})
if err != nil {
return nil, err
}
return &MDNSServer{sendconn: sendconn, zone: zone}, nil
return &MDNSServer{zone: zone}, nil
}
// Return true if testaddr is a UDP address with IP matching my local i/f
@@ -36,7 +34,19 @@ func (s *MDNSServer) addrIsLocal(testaddr net.Addr) bool {
return false
}
func (s *MDNSServer) Start(ifi *net.Interface, localDomain string) error {
// Start the mDNS server
func (s *MDNSServer) Start(ifi *net.Interface) (err error) {
// skip double initialization
if s.running {
return nil
}
// This is a bit of a kludge - per the RFC we should send responses from 5353, but that doesn't seem to work
s.sendconn, err = net.ListenUDP("udp4", &net.UDPAddr{IP: net.IPv4zero, Port: 0})
if err != nil {
return err
}
conn, err := LinkLocalMulticastListener(ifi)
if err != nil {
return err
@@ -52,23 +62,23 @@ func (s *MDNSServer) Start(ifi *net.Interface, localDomain string) error {
}
handleLocal := s.makeHandler(dns.TypeA,
func(zone Lookup, r *dns.Msg, q *dns.Question) *dns.Msg {
if ip, err := zone.LookupName(q.Name); err == nil {
return makeAddressReply(r, q, []net.IP{ip})
func(zone ZoneLookup, r *dns.Msg, q *dns.Question) *dns.Msg {
if ips, err := zone.LookupName(q.Name); err == nil {
return makeAddressReply(r, q, ips)
}
return nil
})
handleReverse := s.makeHandler(dns.TypePTR,
func(zone Lookup, r *dns.Msg, q *dns.Question) *dns.Msg {
if name, err := zone.LookupInaddr(q.Name); err == nil {
return makePTRReply(r, q, []string{name})
func(zone ZoneLookup, r *dns.Msg, q *dns.Question) *dns.Msg {
if names, err := zone.LookupInaddr(q.Name); err == nil {
return makePTRReply(r, q, names)
}
return nil
})
mux := dns.NewServeMux()
mux.HandleFunc(localDomain, handleLocal)
mux.HandleFunc(s.zone.Domain(), handleLocal)
mux.HandleFunc(RDNSDomain, handleReverse)
s.srv = &dns.Server{
@@ -77,17 +87,24 @@ func (s *MDNSServer) Start(ifi *net.Interface, localDomain string) error {
Handler: mux,
}
go s.srv.ActivateAndServe()
s.running = true
return err
}
// Stop the mDNS server
func (s *MDNSServer) Stop() error {
s.running = false
return s.srv.Shutdown()
}
type LookupFunc func(Lookup, *dns.Msg, *dns.Question) *dns.Msg
func (s *MDNSServer) Zone() Zone {
return s.zone
}
type LookupFunc func(ZoneLookup, *dns.Msg, *dns.Question) *dns.Msg
func (s *MDNSServer) makeHandler(qtype uint16, lookup LookupFunc) dns.HandlerFunc {
return func(_ dns.ResponseWriter, r *dns.Msg) {
return func(rw dns.ResponseWriter, r *dns.Msg) {
// Handle only questions, ignore answers. We might also ignore
// questions that arise locally (i.e., that come from an IP we
// think is local), but in the interest of avoiding
@@ -95,18 +112,20 @@ func (s *MDNSServer) makeHandler(qtype uint16, lookup LookupFunc) dns.HandlerFun
// assumption that the client wouldn't ask if it already knew
// the answer, and if it does ask, it'll be happy to get an
// answer.
// TODO: use any Answer in the query for adding records in the Zone database...
if len(r.Answer) == 0 && len(r.Question) > 0 {
q := &r.Question[0]
if q.Qtype == qtype {
Debug.Printf("[mdns msgid %d] Trying to answer to mDNS query '%s'", r.MsgHdr.Id, q.Name)
if m := lookup(s.zone, r, q); m != nil {
Debug.Printf("[mdns msgid %d] Found local answer to mDNS query %s",
r.MsgHdr.Id, q.Name)
Debug.Printf("[mdns msgid %d] - found local answer to mDNS query '%s'", r.MsgHdr.Id, q.Name)
if err := s.sendResponse(m); err != nil {
Warning.Printf("[mdns msgid %d] Error writing to %v",
r.MsgHdr.Id, s.sendconn)
Warning.Printf("[mdns msgid %d] - error writing to %v", r.MsgHdr.Id, s.sendconn)
} else {
Debug.Printf("[mdns msgid %d] - response sent: %d answers", r.MsgHdr.Id, len(m.Answer))
}
} else {
Debug.Printf("[mdns msgid %d] No local answer for mDNS query %s",
Debug.Printf("[mdns msgid %d] - no local answer for mDNS query '%s'",
r.MsgHdr.Id, q.Name)
}
}
+53 -70
View File
@@ -2,82 +2,49 @@ package nameserver
import (
"github.com/miekg/dns"
"github.com/weaveworks/weave/common"
. "github.com/weaveworks/weave/common"
wt "github.com/weaveworks/weave/testing"
"log"
"net"
"testing"
"time"
)
var (
containerID = "deadbeef"
testName = "test.weave.local."
testAddr1 = "10.2.2.1/24"
testInAddr1 = "1.2.2.10.in-addr.arpa."
)
func sendQuery(name string, querytype uint16) error {
m := new(dns.Msg)
m.SetQuestion(name, querytype)
m.RecursionDesired = false
buf, err := m.Pack()
if err != nil {
return err
}
conn, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.IPv4zero, Port: 0})
if err != nil {
return err
}
_, err = conn.WriteTo(buf, ipv4Addr)
return err
}
func TestServerSimpleQuery(t *testing.T) {
// The ff can be handy for debugging (obvs)
common.InitDefaultLogging(true)
InitDefaultLogging(testing.Verbose())
log.Println("TestServerSimpleQuery starting")
var zone = new(ZoneDb)
ip, _, _ := net.ParseCIDR(testAddr1)
zone.AddRecord(containerID, testName, ip)
var (
testRecord1 = Record{"test.weave.local.", net.ParseIP("10.20.20.10"), 0, 0}
testInAddr1 = "10.20.20.10.in-addr.arpa."
)
mzone := NewMockedZone(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
var recvChan chan interface{}
receivedCount := 0
reset := func() {
receivedAddr = nil
receivedName = ""
receivedCount = 0
recvChan = make(chan interface{})
}
wait := func() {
select {
case <-recvChan:
return
case <-time.After(100 * time.Millisecond):
return
}
}
// Implement a minimal listener for responses
multicast, err := LinkLocalMulticastListener(nil)
if err != nil {
t.Fatalf("Unable to run test server: %s. No default multicast interface?", err)
}
wt.AssertNoErr(t, err)
handleMDNS := func(w dns.ResponseWriter, r *dns.Msg) {
// Only handle responses here
if len(r.Answer) > 0 {
t.Logf("Received %d answer(s)", len(r.Answer))
for _, answer := range r.Answer {
switch rr := answer.(type) {
case *dns.A:
t.Logf("... A:\n%+v", rr)
receivedAddr = rr.A
receivedCount++
case *dns.PTR:
t.Logf("... PTR:\n%+v", rr)
receivedName = rr.Ptr
receivedCount++
}
@@ -86,6 +53,33 @@ func TestServerSimpleQuery(t *testing.T) {
}
}
sendQuery := func(name string, querytype uint16) {
receivedAddr = nil
receivedName = ""
receivedCount = 0
recvChan = make(chan interface{})
m := new(dns.Msg)
m.SetQuestion(name, querytype)
m.RecursionDesired = false
buf, err := m.Pack()
wt.AssertNoErr(t, err)
conn, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.IPv4zero, Port: 0})
wt.AssertNoErr(t, err)
Debug.Printf("Sending UDP packet to %s", ipv4Addr)
_, err = conn.WriteTo(buf, ipv4Addr)
wt.AssertNoErr(t, err)
Debug.Printf("Waiting for response")
select {
case <-recvChan:
return
case <-time.After(100 * time.Millisecond):
Debug.Printf("Timeout while waiting for response")
return
}
}
listener := &dns.Server{
Unsafe: true,
PacketConn: multicast,
@@ -94,39 +88,28 @@ func TestServerSimpleQuery(t *testing.T) {
go listener.ActivateAndServe()
defer listener.Shutdown()
mdnsServer, err := NewMDNSServer(zone)
wt.AssertNoErr(t, err)
err = mdnsServer.Start(nil, DefaultLocalDomain)
wt.AssertNoErr(t, err)
time.Sleep(100 * time.Millisecond) // Allow for server to get going
reset()
sendQuery(testName, dns.TypeA)
wait()
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, testName)
t.Fatalf("Unexpected result count %d for %s", receivedCount, testRecord1.Name())
}
if !receivedAddr.Equal(ip) {
t.Fatalf("Unexpected result %s for %s", receivedAddr, testName)
if !receivedAddr.Equal(testRecord1.IP()) {
t.Fatalf("Unexpected result %s for %s", receivedAddr, testRecord1.Name())
}
reset()
Debug.Printf("Query: testfail.weave. dns.TypeA")
sendQuery("testfail.weave.", dns.TypeA)
wait()
if receivedCount != 0 {
t.Fatalf("Unexpected result count %d for testfail.weave", receivedCount)
}
reset()
Debug.Printf("Query: %s dns.TypePTR", testInAddr1)
sendQuery(testInAddr1, dns.TypePTR)
wait()
if receivedCount != 1 {
t.Fatalf("Expected an answer to %s, got %d answers", testInAddr1, receivedCount)
} else if !(testName == receivedName) {
t.Fatalf("Expected answer %s to query for %s, got %s", testName, testInAddr1, receivedName)
} else if !(testRecord1.Name() == receivedName) {
t.Fatalf("Expected answer %s to query for %s, got %s", testRecord1.Name(), testInAddr1, receivedName)
}
}
+53
View File
@@ -0,0 +1,53 @@
package nameserver
import (
. "github.com/weaveworks/weave/common"
"net"
)
// Warn about some methods that some day should be implemented...
func notImplWarn() { Warning.Printf("Mocked method. Not implemented.") }
// A mocked Zone that always returns the same record
type mockedZone struct {
record ZoneRecord
// Statistics
NumLookupsName int
NumLookupsInaddr int
}
func NewMockedZone(zr ZoneRecord) *mockedZone { return &mockedZone{record: zr} }
func (mz mockedZone) Domain() string { return DefaultLocalDomain }
func (mz mockedZone) LookupName(name string) ([]ZoneRecord, error) {
Debug.Printf("[mocked zone]: LookupName: returning record %s", mz.record)
mz.NumLookupsName += 1
return []ZoneRecord{mz.record}, nil
}
func (mz mockedZone) LookupInaddr(inaddr string) ([]ZoneRecord, error) {
Debug.Printf("[mocked zone]: LookupInaddr: returning record %s", mz.record)
mz.NumLookupsInaddr += 1
return []ZoneRecord{mz.record}, nil
}
// the following methods are not currently needed...
func (mz mockedZone) Status() string { notImplWarn(); return "nothing" }
func (mz mockedZone) AddRecord(ident string, name string, ip net.IP) error { notImplWarn(); return nil }
func (mz mockedZone) DeleteRecord(ident string, ip net.IP) error { notImplWarn(); return nil }
func (mz mockedZone) DeleteRecordsFor(ident string) error { notImplWarn(); return nil }
func (mz mockedZone) DomainLookupName(name string) ([]ZoneRecord, error) {
notImplWarn()
return nil, nil
}
func (mz mockedZone) DomainLookupInaddr(inaddr string) ([]ZoneRecord, error) {
notImplWarn()
return nil, nil
}
func (mz mockedZone) ObserveName(name string, observer ZoneRecordObserver) error {
notImplWarn()
return nil
}
func (mz mockedZone) ObserveInaddr(inaddr string, observer ZoneRecordObserver) error {
notImplWarn()
return nil
}
+11 -11
View File
@@ -156,7 +156,7 @@ 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, s.Domain)
err = s.mdnsSrv.Start(s.Iface)
CheckFatal(err)
s.listenersWg.Add(2)
@@ -219,14 +219,14 @@ func (s *DNSServer) queryHandler(proto dnsProtocol) dns.HandlerFunc {
now := time.Now()
q := r.Question[0]
maxLen := getMaxReplyLen(r, proto)
lookups := []Lookup{s.Zone, s.mdnsCli}
lookups := []ZoneLookup{s.Zone, s.mdnsCli}
Debug.Printf("Query: %+v", q)
reply, err := s.cache.Get(r, maxLen, time.Now())
if err != nil {
if err == errNoLocalReplies {
Debug.Printf("[dns msgid %d] Cached 'no local replies' - skipping local lookup", r.MsgHdr.Id)
lookups = []Lookup{s.Zone}
lookups = []ZoneLookup{s.Zone}
} else {
Debug.Printf("[dns msgid %d] Error from cache: %s", r.MsgHdr.Id, err)
w.WriteMsg(makeDNSFailResponse(r))
@@ -250,12 +250,12 @@ func (s *DNSServer) queryHandler(proto dnsProtocol) dns.HandlerFunc {
}
for _, lookup := range lookups {
if ip, err := lookup.LookupName(q.Name); err == nil {
m := makeAddressReply(r, &q, []net.IP{ip})
if ips, err := lookup.LookupName(q.Name); err == nil {
m := makeAddressReply(r, &q, ips)
m.Authoritative = true
Debug.Printf("[dns msgid %d] Caching response for %s-query for \"%s\": %s [code:%s]",
m.MsgHdr.Id, dns.TypeToString[q.Qtype], q.Name, ip, dns.RcodeToString[m.Rcode])
m.MsgHdr.Id, dns.TypeToString[q.Qtype], q.Name, ips, dns.RcodeToString[m.Rcode])
s.cache.Put(r, m, nullTTL, 0, now)
w.WriteMsg(m)
return
@@ -275,7 +275,7 @@ func (s *DNSServer) rdnsHandler(proto dnsProtocol) dns.HandlerFunc {
return func(w dns.ResponseWriter, r *dns.Msg) {
now := time.Now()
q := r.Question[0]
lookups := []Lookup{s.Zone, s.mdnsCli}
lookups := []ZoneLookup{s.Zone, s.mdnsCli}
maxLen := getMaxReplyLen(r, proto)
Debug.Printf("Reverse query: %+v", q)
@@ -283,7 +283,7 @@ func (s *DNSServer) rdnsHandler(proto dnsProtocol) dns.HandlerFunc {
if err != nil {
if err == errNoLocalReplies {
Debug.Printf("[dns msgid %d] Cached 'no local replies' - skipping local lookup", r.MsgHdr.Id)
lookups = []Lookup{s.Zone}
lookups = []ZoneLookup{s.Zone}
} else {
Debug.Printf("[dns msgid %d] Error from cache: %s", r.MsgHdr.Id, err)
w.WriteMsg(makeDNSFailResponse(r))
@@ -308,12 +308,12 @@ func (s *DNSServer) rdnsHandler(proto dnsProtocol) dns.HandlerFunc {
}
for _, lookup := range lookups {
if name, err := lookup.LookupInaddr(q.Name); err == nil {
m := makePTRReply(r, &q, []string{name})
if names, err := lookup.LookupInaddr(q.Name); err == nil {
m := makePTRReply(r, &q, names)
m.Authoritative = true
Debug.Printf("[dns msgid %d] Caching response for %s-query for \"%s\": %s [code:%s]",
m.MsgHdr.Id, dns.TypeToString[q.Qtype], q.Name, name, dns.RcodeToString[m.Rcode])
m.MsgHdr.Id, dns.TypeToString[q.Qtype], q.Name, names, dns.RcodeToString[m.Rcode])
s.cache.Put(r, m, nullTTL, 0, now)
w.WriteMsg(m)
return
+5 -4
View File
@@ -36,6 +36,7 @@ func setupForTest(t *testing.T) {
func TestUDPDNSServer(t *testing.T) {
setupForTest(t)
const (
containerID = "foobar"
successTestName = "test1.weave.local."
failTestName = "test2.weave.local."
nonLocalName = "weave.works."
@@ -44,7 +45,7 @@ func TestUDPDNSServer(t *testing.T) {
testCIDR1 := testAddr1 + "/24"
InitDefaultLogging(true)
var zone = new(ZoneDb)
var zone = NewZoneDb(DefaultLocalDomain)
ip, _, _ := net.ParseCIDR(testCIDR1)
zone.AddRecord(containerID, successTestName, ip)
@@ -125,14 +126,14 @@ func TestTCPDNSServer(t *testing.T) {
dnsAddr := fmt.Sprintf("localhost:%d", testPort)
InitDefaultLogging(true)
var zone = new(ZoneDb)
var zone = NewZoneDb(DefaultLocalDomain)
// generate a list of `numAnswers` IP addresses
var addrs []net.IP
var addrs []ZoneRecord
bs := make([]byte, 4)
for i := 0; i < numAnswers; i++ {
binary.LittleEndian.PutUint32(bs, uint32(i))
addrs = append(addrs, net.IPv4(bs[0], bs[1], bs[2], bs[3]))
addrs = append(addrs, Record{"", net.IPv4(bs[0], bs[1], bs[2], bs[3]), 0, 0})
}
// handler for the fallback server: it will just return a very long response
+26 -19
View File
@@ -16,19 +16,15 @@ const (
// +1 to also exclude a dot
var rdnsDomainLen = len(RDNSDomain) + 1
type Lookup interface {
LookupName(name string) (net.IP, error)
LookupInaddr(inaddr string) (string, error)
}
type Zone interface {
AddRecord(ident string, name string, ip net.IP) error
DeleteRecord(ident string, ip net.IP) error
DeleteRecordsFor(ident string) error
Lookup
Domain() string
ZoneLookup
}
type Record struct {
type dbRecord struct {
Ident string
Name string
IP net.IP
@@ -37,8 +33,9 @@ type Record struct {
// Very simple data structure for now, with linear searching.
// TODO: make more sophisticated to improve performance.
type ZoneDb struct {
mx sync.RWMutex
recs []Record
mx sync.RWMutex
recs []dbRecord
domain string
}
type LookupError string
@@ -54,7 +51,17 @@ func (dup DuplicateError) Error() string {
return "Tried to add a duplicate entry"
}
func (zone *ZoneDb) indexOf(match func(Record) bool) int {
func NewZoneDb(domain string) *ZoneDb {
return &ZoneDb{
domain: domain,
}
}
func (zone *ZoneDb) Domain() string {
return zone.domain
}
func (zone *ZoneDb) indexOf(match func(dbRecord) bool) int {
for i, r := range zone.recs {
if match(r) {
return i
@@ -73,18 +80,18 @@ func (zone *ZoneDb) String() string {
return buf.String()
}
func (zone *ZoneDb) LookupName(name string) (net.IP, error) {
func (zone *ZoneDb) LookupName(name string) ([]ZoneRecord, error) {
zone.mx.RLock()
defer zone.mx.RUnlock()
for _, r := range zone.recs {
if r.Name == name {
return r.IP, nil
return []ZoneRecord{Record{r.Name, r.IP, 0, 0}}, nil
}
}
return nil, LookupError(name)
}
func (zone *ZoneDb) LookupInaddr(inaddr string) (string, error) {
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]}
@@ -93,31 +100,31 @@ func (zone *ZoneDb) LookupInaddr(inaddr string) (string, error) {
defer zone.mx.RUnlock()
for _, r := range zone.recs {
if r.IP.Equal(ip) {
return r.Name, nil
return []ZoneRecord{Record{r.Name, r.IP, 0, 0}}, nil
}
}
return "", LookupError(inaddr)
return nil, LookupError(inaddr)
}
Warning.Printf("[zonedb] Asked to reverse lookup %s", inaddr)
return "", LookupError(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 Record) bool { return r.Name == fqdn && r.IP.Equal(ip) && r.Ident == ident }
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, Record{ident, fqdn, ip})
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 Record) bool { return r.Ident == ident && r.IP.Equal(ip) }
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
+4 -4
View File
@@ -14,7 +14,7 @@ func TestZone(t *testing.T) {
testAddr1 = "10.2.2.1/24"
)
var zone = new(ZoneDb)
var zone = NewZoneDb(DefaultLocalDomain)
ip, _, _ := net.ParseCIDR(testAddr1)
err := zone.AddRecord(containerID, successTestName, ip)
@@ -30,7 +30,7 @@ func TestZone(t *testing.T) {
foundIP, err := zone.LookupName(successTestName)
wt.AssertNoErr(t, err)
if !foundIP.Equal(ip) {
if !foundIP[0].IP().Equal(ip) {
t.Fatal("Unexpected result for", successTestName, foundIP)
}
@@ -38,7 +38,7 @@ func TestZone(t *testing.T) {
foundName, err := zone.LookupInaddr("1.2.2.10.in-addr.arpa.")
wt.AssertNoErr(t, err)
if foundName != successTestName {
if foundName[0].Name() != successTestName {
t.Fatal("Unexpected result for", ip, foundName)
}
@@ -72,7 +72,7 @@ func TestDeleteFor(t *testing.T) {
addr1 = "10.2.2.3/24"
addr2 = "10.2.7.8/24"
)
zone := new(ZoneDb)
zone := NewZoneDb(DefaultLocalDomain)
for _, addr := range []string{addr1, addr2} {
ip, _, _ := net.ParseCIDR(addr)
err := zone.AddRecord(id, name, ip)
+1 -1
View File
@@ -55,7 +55,7 @@ func main() {
InitDefaultLogging(debug)
Info.Printf("WeaveDNS version %s\n", version) // first thing in log: the version
var zone = new(weavedns.ZoneDb)
var zone = weavedns.NewZoneDb(domain)
if watch {
err := updater.Start(apiPath, zone)