Merge branch 'arrange_for_merge' of ../weavedns into integrate_weavedns

This pulls the weavedns repo commits into the weave repo, for merging
into the main code base.
This commit is contained in:
Michael Bridgen committed 2014-10-29 12:25:20 +00:00
commit 190e708045
20 files changed
+1585

No files matched your search

+30
View File
@@ -0,0 +1,30 @@
# Compiled Object files, Static and Dynamic libs (Shared Objects)
*.o
*.a
*.so
# Folders
_obj
_test
# Architecture specific extensions/prefixes
*.[568vq]
[568vq].out
*.cgo1.go
*.cgo2.c
_cgo_defun.c
_cgo_gotypes.go
_cgo_export.*
_testmain.go
*.exe
*.test
*.prof
# Emacs backup files
*~
# Project specific
weavedns/weavedns
+55
View File
@@ -0,0 +1,55 @@
package nameserver
import (
"github.com/miekg/dns"
"net"
)
const (
localTTL uint32 = 300 // somewhat arbitrary; we don't expect anyone
// downstream to cache results
)
func makeHeader(r *dns.Msg, q *dns.Question) *dns.RR_Header {
return &dns.RR_Header{
Name: q.Name, Rrtype: q.Qtype,
Class: dns.ClassINET, Ttl: localTTL}
}
func makeReply(r *dns.Msg, as []dns.RR) *dns.Msg {
m := new(dns.Msg)
m.SetReply(r)
m.RecursionAvailable = true
m.Answer = as
return m
}
func makeAReply(r *dns.Msg, q *dns.Question, addrs []net.IP) *dns.Msg {
answers := make([]dns.RR, len(addrs))
header := makeHeader(r, q)
count := 0
for _, addr := range addrs {
switch q.Qtype {
case dns.TypeA:
if ip4 := addr.To4(); ip4 != nil {
answers[count] = &dns.A{*header, addr}
count++
}
case dns.TypeAAAA:
if ip4 := addr.To4(); ip4 == nil {
answers[count] = &dns.AAAA{*header, addr}
count++
}
}
}
return makeReply(r, answers[:count])
}
func makePTRReply(r *dns.Msg, q *dns.Question, names []string) *dns.Msg {
answers := make([]dns.RR, len(names))
header := makeHeader(r, q)
for i, name := range names {
answers[i] = &dns.PTR{*header, name}
}
return makeReply(r, answers)
}
+107
View File
@@ -0,0 +1,107 @@
package nameserver
import (
"errors"
"fmt"
"github.com/miekg/dns"
"log"
"net"
"net/http"
"strings"
)
// Parse a URL of the form PUT /name/<identifier>/<ip-address>
func parseUrl(url string) (identifier string, ipaddr string, err error) {
parts := strings.Split(url, "/")
if len(parts) != 4 {
return "", "", errors.New("Unable to parse url: " + url)
}
return parts[2], parts[3], nil
}
func httpErrorAndLog(level *log.Logger, w http.ResponseWriter, msg string,
status int, logmsg string, logargs ...interface{}) {
http.Error(w, msg, status)
level.Printf(logmsg, logargs...)
}
func ListenHttp(domain string, db Zone, port int) {
http.HandleFunc("/name/", func(w http.ResponseWriter, r *http.Request) {
reqError := func(msg string, logmsg string, logargs ...interface{}) {
httpErrorAndLog(Warning, w, msg, http.StatusBadRequest,
logmsg, logargs...)
}
switch r.Method {
case "PUT":
ident, weaveIPstr, err := parseUrl(r.URL.Path)
name := r.FormValue("fqdn")
localIPstr := r.FormValue("local_ip")
if ident == "" || weaveIPstr == "" || name == "" || localIPstr == "" {
reqError("Invalid request", "Invalid request: %s, %s", r.URL, r.Form)
return
}
localIP := net.ParseIP(localIPstr)
if localIP == nil {
reqError("Invalid IP in request", "Invalid IP in request: %s", localIPstr)
return
}
weaveIP := net.ParseIP(weaveIPstr)
if weaveIP == nil {
reqError("Invalid weave IP", "Invalid weave IP in request: %s", weaveIPstr)
return
}
if dns.IsSubDomain(domain, name) {
Info.Printf("Adding %s (%s) -> %s", name, localIPstr, weaveIPstr)
if err = db.AddRecord(ident, name, localIP, weaveIP); err != nil {
if dup, ok := err.(DuplicateError); !ok {
httpErrorAndLog(
Error, w, "Internal error", http.StatusInternalServerError,
"Unexpected error from DB: %s", err)
return
} else if dup.Ident != ident {
http.Error(w, err.Error(), http.StatusConflict)
return
} // else we are golden
}
} else {
Info.Printf("Ignoring name %s, not in %s", name, domain)
}
case "DELETE":
ident, weaveIPstr, err := parseUrl(r.URL.Path)
if ident == "" || weaveIPstr == "" {
reqError("Invalid Request", "Invalid request: %s, %s", r.URL, r.Form)
return
}
weaveIP := net.ParseIP(weaveIPstr)
if weaveIP == nil {
reqError("Invalid IP in request", "Invalid IP in request: %s", weaveIPstr)
return
}
Info.Printf("Deleting %s (%s)", ident, weaveIPstr)
if err = db.DeleteRecord(ident, weaveIP); err != nil {
if _, ok := err.(LookupError); !ok {
httpErrorAndLog(
Error, w, "Internal error", http.StatusInternalServerError,
"Unexpected error from DB: %s", err)
return
}
}
default:
msg := "Unexpected http method: " + r.Method
reqError(msg, msg)
return
}
})
address := fmt.Sprintf(":%d", port)
if err := http.ListenAndServe(address, nil); err != nil {
Error.Fatal("Unable to create http listener: ", err)
}
}
+82
View File
@@ -0,0 +1,82 @@
package nameserver
import (
"fmt"
"math/rand"
"net"
"net/http"
"net/url"
"strings"
"testing"
"time"
)
func genForm(method string, url string, data url.Values) (resp *http.Response, err error) {
req, err := http.NewRequest(method, url, strings.NewReader(data.Encode()))
req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
return http.DefaultClient.Do(req)
}
func TestHttp(t *testing.T) {
var (
containerID = "deadbeef"
testDomain = "weave.local."
successTestName = "test1." + testDomain
testAddr1 = "10.0.2.1/24"
dockerIP = "9.8.7.6"
)
var zone = new(ZoneDb)
port := rand.Intn(10000) + 32768
fmt.Println("Http test on port", port)
go ListenHttp(testDomain, zone, port)
time.Sleep(100 * time.Millisecond) // Allow for http server to get going
// Ask the http server to add our test address into the database
addrParts := strings.Split(testAddr1, "/")
addrUrl := fmt.Sprintf("http://localhost:%d/name/%s/%s", port, containerID, addrParts[0])
resp, err := genForm("PUT", addrUrl,
url.Values{"fqdn": {successTestName}, "local_ip": {dockerIP}, "routing_prefix": {addrParts[1]}})
assertNoErr(t, err)
assertStatus(t, resp.StatusCode, http.StatusOK, "http response")
// Check that the address is now there.
ip, err := zone.MatchLocal(successTestName)
assertNoErr(t, err)
weaveIP, _, _ := net.ParseCIDR(testAddr1)
if !ip.Equal(weaveIP) {
t.Fatal("Unexpected result for", successTestName, ip)
}
// Adding exactly the same address should be OK
resp, err = genForm("PUT", addrUrl,
url.Values{"fqdn": {successTestName}, "local_ip": {dockerIP}, "routing_prefix": {addrParts[1]}})
assertNoErr(t, err)
assertStatus(t, resp.StatusCode, http.StatusOK, "http success response for duplicate add")
// Now try adding the same address again with a different ident - should fail
otherUrl := fmt.Sprintf("http://localhost:%d/name/%s/%s", port, "other", addrParts[0])
resp, err = genForm("PUT", otherUrl,
url.Values{"fqdn": {successTestName}, "local_ip": {dockerIP}, "routing_prefix": {addrParts[1]}})
assertNoErr(t, err)
assertStatus(t, resp.StatusCode, http.StatusConflict, "http response")
// Delete the address
resp, err = genForm("DELETE", addrUrl, nil)
assertNoErr(t, err)
assertStatus(t, resp.StatusCode, http.StatusOK, "http response")
// Check that the address is not there now.
_, err = zone.MatchLocal(successTestName)
assertErrorType(t, err, (*LookupError)(nil), "nonexistent lookup")
// Delete the address again, it should accept this
resp, err = genForm("DELETE", addrUrl, nil)
assertNoErr(t, err)
assertStatus(t, resp.StatusCode, http.StatusOK, "http response")
// Would like to shut down the http server at the end of this test
// but it's complicated.
// See https://groups.google.com/forum/#!topic/golang-nuts/vLHWa5sHnCE
}
+32
View File
@@ -0,0 +1,32 @@
package nameserver
import (
"fmt"
"net"
"time"
)
func EnsureInterface(ifaceName string, wait int) (iface *net.Interface, err error) {
if iface, err = findInterface(ifaceName); err == nil || wait == 0 {
return
}
Info.Println("Waiting for interface", ifaceName, "to come up")
for ; err != nil && wait > 0; wait -= 1 {
time.Sleep(1 * time.Second)
iface, err = findInterface(ifaceName)
}
if err == nil {
Info.Println("Interface", ifaceName, "is up")
}
return
}
func findInterface(ifaceName string) (iface *net.Interface, err error) {
if iface, err = net.InterfaceByName(ifaceName); err != nil {
return iface, fmt.Errorf("Unable to find interface %s", ifaceName)
}
if 0 == (net.FlagUp & iface.Flags) {
return iface, fmt.Errorf("Interface %s is not up", ifaceName)
}
return
}
+37
View File
@@ -0,0 +1,37 @@
package nameserver
import (
"io"
"io/ioutil"
"log"
"os"
)
const (
standardLogFlags = log.Ldate | log.Ltime | log.Lmicroseconds
)
// Largely taken from
// http://www.goinggo.net/2013/11/using-log-package-in-go.html
var (
Debug *log.Logger
Info *log.Logger
Warning *log.Logger
Error *log.Logger
)
func InitLogging(debugHandle io.Writer,
infoHandle io.Writer,
warningHandle io.Writer,
errorHandle io.Writer) {
Debug = log.New(debugHandle, "DEBUG: ", standardLogFlags)
Info = log.New(infoHandle, "INFO: ", standardLogFlags)
Warning = log.New(warningHandle, "WARNING: ", standardLogFlags)
Error = log.New(errorHandle, "ERROR: ", standardLogFlags)
}
func init() {
InitLogging(ioutil.Discard, os.Stdout, os.Stdout, os.Stderr)
}
+251
View File
@@ -0,0 +1,251 @@
package nameserver
import (
"github.com/miekg/dns"
"math"
"net"
"time"
)
// Portions of this code taken from github.com/armon/mdns
const (
ipv4mdns = "224.0.0.251" // link-local multicast address
mdnsPort = 5353 // mDNS assigned port
// We wait this long to hear a response from other mDNS servers on
// the network.
mDNSTimeout = 500 * time.Millisecond
MaxDuration = time.Duration(math.MaxInt64)
MailboxSize = 16
)
var (
ipv4Addr = &net.UDPAddr{
IP: net.ParseIP(ipv4mdns),
Port: mdnsPort,
}
)
type ResponseA struct {
Name string
Addr net.IP
Err error
}
type responseInfo struct {
timeout time.Time // if no answer by this time, give up
ch chan<- *ResponseA
}
// Represents one query that we have sent for one name.
// If we, internally, get several requests for the same name while we have
// a query in flight, then we don't want to send more queries out.
// Invariant on responseInfos: they are in ascending order of timeout.
type inflightQuery struct {
name string
id uint16 // the DNS message ID
responseInfos []*responseInfo
}
type MDNSClient struct {
server *dns.Server
conn *net.UDPConn
addr *net.UDPAddr
inflight map[string]*inflightQuery
queryChan chan<- *MDNSInteraction
}
type mDNSQueryInfo struct {
name string
querytype uint16
responseCh chan<- *ResponseA
}
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
}
}
func (c *MDNSClient) Start(ifi *net.Interface) error {
multicast, err := LinkLocalMulticastListener(ifi)
if err != nil {
return err
}
handleMDNS := func(w dns.ResponseWriter, r *dns.Msg) {
// Don't want to handle queries here, so filter anything out that isn't a response
if len(r.Answer) > 0 {
c.ResponseCallback(r)
}
}
c.server = &dns.Server{Listener: nil, PacketConn: multicast, Handler: dns.HandlerFunc(handleMDNS)}
go c.server.ActivateAndServe()
queryChan := make(chan *MDNSInteraction, MailboxSize)
c.queryChan = queryChan
go c.queryLoop(queryChan)
return nil
}
func LinkLocalMulticastListener(ifi *net.Interface) (net.PacketConn, error) {
return net.ListenMulticastUDP("udp", ifi, ipv4Addr)
}
// ACTOR client API
const (
CSendQuery = iota
CShutdown = iota
CMessageReceived = iota
)
type MDNSInteraction struct {
code int
resultChan chan<- interface{}
payload interface{}
}
// Async
func (c *MDNSClient) Shutdown() {
c.queryChan <- &MDNSInteraction{code: CShutdown}
}
// Async
func (c *MDNSClient) SendQuery(name string, querytype uint16, responseCh chan<- *ResponseA) {
c.queryChan <- &MDNSInteraction{
code: CSendQuery,
payload: mDNSQueryInfo{name, querytype, responseCh},
}
}
// Async - called from dns library multiplexer
func (c *MDNSClient) ResponseCallback(r *dns.Msg) {
c.queryChan <- &MDNSInteraction{code: CMessageReceived, payload: r}
}
// ACTOR server
// Check all in-flight queries, close all that have already timed out,
// and return the duration until the next timeout
func (c *MDNSClient) checkInFlightQueries() time.Duration {
now := time.Now()
after := MaxDuration
for name, query := range c.inflight {
// Invariant on responseInfos: they are in ascending order of timeout.
numClosed := 0
for _, item := range query.responseInfos {
duration := item.timeout.Sub(now)
if duration <= 0 { // timed out
close(item.ch)
numClosed++
} else {
if duration < after {
after = duration
}
break // don't need to look at any more for this query
}
}
// Remove timed-out items from the slice
query.responseInfos = query.responseInfos[numClosed:]
if len(query.responseInfos) == 0 {
delete(c.inflight, name)
}
}
return after
}
func (c *MDNSClient) queryLoop(queryChan <-chan *MDNSInteraction) {
timer := time.NewTimer(MaxDuration)
run := func() {
timer.Reset(c.checkInFlightQueries())
}
terminate := false
for !terminate {
select {
case query, ok := <-queryChan:
if !ok {
break
}
switch query.code {
case CShutdown:
c.server.Shutdown()
terminate = true
case CSendQuery:
c.handleSendQuery(query.payload.(mDNSQueryInfo))
run()
case CMessageReceived:
c.handleResponse(query.payload.(*dns.Msg))
run()
}
case <-timer.C:
run()
}
}
// Close all response channels at termination
for _, query := range c.inflight {
for _, item := range query.responseInfos {
close(item.ch)
}
}
}
func (c *MDNSClient) handleSendQuery(q mDNSQueryInfo) {
query, found := c.inflight[q.name]
if !found {
m := new(dns.Msg)
m.SetQuestion(q.name, q.querytype)
m.RecursionDesired = false
buf, err := m.Pack()
if err != nil {
q.responseCh <- &ResponseA{Err: err}
close(q.responseCh)
return
}
query = &inflightQuery{
name: q.name,
id: m.Id,
}
if _, err = c.conn.WriteTo(buf, c.addr); err != nil {
q.responseCh <- &ResponseA{Err: err}
close(q.responseCh)
return
}
c.inflight[q.name] = query
}
info := &responseInfo{
ch: q.responseCh,
timeout: time.Now().Add(mDNSTimeout),
}
// Invariant on responseInfos: they are in ascending order of timeout.
// Since we use a fixed interval from Now(), this must be after all existing timeouts.
query.responseInfos = append(query.responseInfos, info)
}
func (c *MDNSClient) handleResponse(r *dns.Msg) {
for _, answer := range r.Answer {
switch rr := answer.(type) {
case *dns.A:
name := rr.Hdr.Name
if query, found := c.inflight[name]; found {
for _, resp := range query.responseInfos {
resp.ch <- &ResponseA{Name: rr.Hdr.Name, Addr: rr.A}
close(resp.ch)
}
delete(c.inflight, name)
} else {
// We've received a response that didn't match a query
// Do we want to cache it?
}
}
}
}
+153
View File
@@ -0,0 +1,153 @@
package nameserver
import (
"github.com/miekg/dns"
"log"
"net"
"testing"
"time"
)
var (
successTestName = "test1.weave."
failTestName = "test2.weave."
testAddr = net.ParseIP("9.8.7.6")
)
func minimalServer(w dns.ResponseWriter, req *dns.Msg) {
//log.Println("minimalServer received:", req)
if len(req.Answer) > 0 {
return // Only interested in questions.
}
if len(req.Question) != 1 {
return // We only handle single-question messages
}
if req.Question[0].Name != successTestName {
return // This is not the DNS record you are looking for
}
m := new(dns.Msg)
m.SetReply(req)
hdr := dns.RR_Header{Name: m.Question[0].Name, Rrtype: dns.TypeA,
Class: dns.ClassINET, Ttl: 3600}
a := &dns.A{hdr, testAddr}
m.Answer = append(m.Answer, a)
buf, err := m.Pack()
checkFatal(err)
if buf == nil {
log.Fatal("Nil buffer")
}
//log.Println("minimalServer sending:", buf)
// 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.DialUDP("udp", nil, ipv4Addr)
checkFatal(err)
_, err = sendconn.Write(buf)
sendconn.Close()
checkFatal(err)
}
func RunLocalMulticastServer() (*dns.Server, error) {
multicast, err := net.ListenMulticastUDP("udp", nil, ipv4Addr)
if err != nil {
return nil, err
}
server := &dns.Server{Listener: nil, PacketConn: multicast, Handler: dns.HandlerFunc(minimalServer)}
go server.ActivateAndServe()
return server, nil
}
func setup(t *testing.T) (*MDNSClient, *dns.Server, error) {
mdnsClient, err := NewMDNSClient()
assertNoErr(t, err)
err = mdnsClient.Start(nil)
assertNoErr(t, err)
server, err := RunLocalMulticastServer()
if err != nil {
t.Fatalf("Unable to run test server: %s", err)
}
return mdnsClient, server, err
}
type testContext struct {
receivedAddr net.IP
receivedCount int
channel chan *ResponseA
}
func newTestContext() *testContext {
return &testContext{channel: make(chan *ResponseA, 4)}
}
func (c *testContext) checkResponse(t *testing.T, channelOk bool, resp *ResponseA) {
if !channelOk {
c.channel = nil
return
}
assertNoErr(t, resp.Err)
log.Printf("Got address response %s addr %s", resp.Name, resp.Addr)
c.receivedAddr = resp.Addr
c.receivedCount++
}
func TestSimpleQuery(t *testing.T) {
log.Println("TestSimpleQuery starting")
mdnsClient, server, _ := setup(t)
defer mdnsClient.Shutdown()
defer server.Shutdown()
context := newTestContext()
// First, a test we expect to succeed
mdnsClient.SendQuery(successTestName, dns.TypeA, context.channel)
for resp := range context.channel {
context.checkResponse(t, true, resp)
}
if !context.receivedAddr.Equal(testAddr) || context.receivedCount != 1 {
t.Fatal("Unexpected result for", successTestName, context.receivedAddr, context.receivedCount)
}
// Now, a test we expect to time out with no responses
context = newTestContext()
mdnsClient.SendQuery("test2.weave.", dns.TypeA, context.channel)
for resp := range context.channel {
context.checkResponse(t, true, resp)
}
if context.receivedCount > 0 {
t.Fatal("Unexpected result for test2.weave", context.receivedAddr)
}
}
func TestParallelQuery(t *testing.T) {
log.Println("TestParallelQuery starting")
mdnsClient, server, _ := setup(t)
defer mdnsClient.Shutdown()
defer server.Shutdown()
context1 := newTestContext()
context2 := newTestContext()
go mdnsClient.SendQuery(successTestName, dns.TypeA, context1.channel)
go mdnsClient.SendQuery(successTestName, dns.TypeA, context2.channel)
timeout := time.After(2 * time.Second)
outerloop:
for context1.channel != nil || context2.channel != nil {
select {
case resp, ok := <-context1.channel:
context1.checkResponse(t, ok, resp)
case resp, ok := <-context2.channel:
context2.checkResponse(t, ok, resp)
case <-timeout:
break outerloop
}
}
if !context1.receivedAddr.Equal(testAddr) || !context2.receivedAddr.Equal(testAddr) || context1.receivedCount != 1 || context2.receivedCount != 1 {
t.Fatal("Unexpected result for", successTestName, context1.receivedAddr, context2.receivedAddr, context1.receivedCount, context2.receivedCount)
}
}
+79
View File
@@ -0,0 +1,79 @@
package nameserver
import (
"github.com/miekg/dns"
"net"
)
type MDNSServer struct {
localAddrs []net.Addr
sendconn *net.UDPConn
zone Zone
}
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 true if testaddr is a UDP address with IP matching my local i/f
func (s *MDNSServer) addrIsLocal(testaddr net.Addr) bool {
if udpaddr, ok := testaddr.(*net.UDPAddr); ok {
for _, localaddr := range s.localAddrs {
if ipnetlocal, ok := localaddr.(*net.IPNet); ok {
if ipnetlocal.IP.Equal(udpaddr.IP) {
return true
}
}
}
}
return false
}
func (s *MDNSServer) Start(ifi *net.Interface) error {
conn, err := LinkLocalMulticastListener(ifi)
if err != nil {
return err
}
if ifi == nil {
s.localAddrs, err = net.InterfaceAddrs()
} else {
s.localAddrs, err = ifi.Addrs()
}
if err != nil {
return err
}
handleMDNS := func(w dns.ResponseWriter, r *dns.Msg) {
// Ignore answers to other questions
if len(r.Answer) == 0 && len(r.Question) > 0 {
q := r.Question[0]
if ip, err := s.zone.MatchLocal(q.Name); err == nil {
m := makeAReply(r, &q, []net.IP{ip})
if err = s.sendResponse(m); err != nil {
Warning.Printf("Error writing to %s", w)
}
} else if s.addrIsLocal(w.RemoteAddr()) {
// ignore this - it's our own query received via multicast
} else {
Debug.Printf("Failed MDNS lookup for %s", q.Name)
}
}
}
go dns.ActivateAndServe(nil, conn, dns.HandlerFunc(handleMDNS))
return err
}
func (s *MDNSServer) sendResponse(m *dns.Msg) error {
buf, err := m.Pack()
if err != nil {
return err
}
_, err = s.sendconn.WriteTo(buf, ipv4Addr)
return err
}
+88
View File
@@ -0,0 +1,88 @@
package nameserver
import (
"github.com/miekg/dns"
"log"
"net"
"testing"
"time"
)
var (
containerID = "deadbeef"
testAddr1 = "10.0.2.1/24"
)
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) {
log.Println("TestServerSimpleQuery starting")
var zone = new(ZoneDb)
dockerIP := net.ParseIP("9.8.7.6")
weaveIP, _, _ := net.ParseCIDR(testAddr1)
zone.AddRecord(containerID, "test.weave.", dockerIP, weaveIP)
mdnsServer, err := NewMDNSServer(zone)
assertNoErr(t, err)
err = mdnsServer.Start(nil)
assertNoErr(t, err)
var receivedAddr net.IP
receivedCount := 0
// Implement a minimal listener for responses
multicast, err := LinkLocalMulticastListener(nil)
assertNoErr(t, err)
handleMDNS := func(w dns.ResponseWriter, r *dns.Msg) {
// Only handle responses here
if len(r.Answer) > 0 {
for _, answer := range r.Answer {
switch rr := answer.(type) {
case *dns.A:
receivedAddr = rr.A
receivedCount++
}
}
}
}
server := &dns.Server{Listener: nil, PacketConn: multicast, Handler: dns.HandlerFunc(handleMDNS)}
go server.ActivateAndServe()
defer server.Shutdown()
time.Sleep(100 * time.Millisecond) // Allow for server to get going
sendQuery("test.weave.", dns.TypeA)
time.Sleep(time.Second)
if receivedCount != 1 {
t.Fatal("Unexpected result count for test.weave", receivedCount)
}
if !receivedAddr.Equal(weaveIP) {
t.Fatal("Unexpected result for test.weave", receivedAddr)
}
receivedCount = 0
sendQuery("testfail.weave.", dns.TypeA)
if receivedCount != 0 {
t.Fatal("Unexpected result count for testfail.weave", receivedCount)
}
}
+152
View File
@@ -0,0 +1,152 @@
package nameserver
import (
"fmt"
"github.com/miekg/dns"
"net"
)
const (
LOCAL_DOMAIN = "weave.local."
RDNS_DOMAIN = "in-addr.arpa."
)
// +1 to also exclude a dot
var rdnsDomainLen = len(RDNS_DOMAIN) + 1
func checkFatal(e error) {
if e != nil {
Error.Fatal(e)
}
}
func checkWarn(e error) {
if e != nil {
Warning.Println(e)
}
}
func makeDNSFailResponse(r *dns.Msg) *dns.Msg {
m := new(dns.Msg)
m.SetReply(r)
m.RecursionAvailable = true
m.Rcode = dns.RcodeNameError
return m
}
func queryHandler(zone Zone, mdnsClient *MDNSClient) dns.HandlerFunc {
return func(w dns.ResponseWriter, r *dns.Msg) {
q := r.Question[0]
Debug.Printf("Local query: %+v", q)
if q.Qtype == dns.TypeA {
if ip, err := zone.MatchLocal(q.Name); err == nil {
m := makeAReply(r, &q, []net.IP{ip})
w.WriteMsg(m)
} else {
Debug.Printf("Failed lookup for %s; sending mDNS query", q.Name)
// We don't know the answer; see if someone else does
channel := make(chan *ResponseA)
replies := make([]net.IP, 0)
go func() {
for resp := range channel {
Debug.Printf("Got address response %s to query %s addr %s", resp.Name, q.Name, resp.Addr)
replies = append(replies, resp.Addr)
}
var responseMsg *dns.Msg
if len(replies) > 0 {
responseMsg = makeAReply(r, &q, replies)
} else {
responseMsg = makeDNSFailResponse(r)
}
w.WriteMsg(responseMsg)
}()
mdnsClient.SendQuery(q.Name, dns.TypeA, channel)
}
} else {
Warning.Printf("Local query not handled: %+v", q)
m := makeDNSFailResponse(r)
w.WriteMsg(m)
}
return
}
}
func rdnsHandler(zone Zone, mdnsClient *MDNSClient) dns.HandlerFunc {
return func(w dns.ResponseWriter, r *dns.Msg) {
q := r.Question[0]
Debug.Printf("Local rdns query: %+v", q)
if q.Qtype == dns.TypePTR {
if ip := net.ParseIP(q.Name[:len(q.Name)-rdnsDomainLen]); ip != nil {
ip4 := ip.To4()
revIP := []byte{ip4[3], ip4[2], ip4[1], ip4[0]}
Debug.Printf("Looking for address: %+v", revIP)
if name, err := zone.MatchLocalIP(revIP); err == nil {
Debug.Printf("Found name: %s", name)
m := makePTRReply(r, &q, []string{name})
w.WriteMsg(m)
} else {
Debug.Printf("Failed lookup for %s; sending mDNS query", q.Name)
// We don't know the answer; see if someone else does
// TODO
}
}
}
}
}
/* When we receive a request for a name outside of our '.weave' domain, call
the underlying lookup mechanism and return the answer(s) it gives.
Unfortunately, this means that TTLs from a real DNS server are lost - FIXME.
*/
func notUsHandler() dns.HandlerFunc {
return func(w dns.ResponseWriter, r *dns.Msg) {
q := r.Question[0]
Debug.Printf("Non-local query: %+v", q)
var responseMsg *dns.Msg
if q.Qtype == dns.TypeA {
if addrs, err := net.LookupIP(q.Name); err == nil {
responseMsg = makeAReply(r, &q, addrs)
} else {
responseMsg = makeDNSFailResponse(r)
Debug.Print("Failed fallback lookup ", err)
}
} else {
Warning.Printf("Non-local query not handled: %+v", q)
responseMsg = makeDNSFailResponse(r)
}
w.WriteMsg(responseMsg)
}
}
func StartServer(zone Zone, iface *net.Interface, dnsPort int, httpPort int, wait int) error {
go ListenHttp(LOCAL_DOMAIN, zone, httpPort)
mdnsClient, err := NewMDNSClient()
checkFatal(err)
ifaceName := "default interface"
if iface != nil {
ifaceName = iface.Name
}
Info.Printf("Using mDNS on %s", ifaceName)
err = mdnsClient.Start(iface)
checkFatal(err)
LocalServeMux := dns.NewServeMux()
LocalServeMux.HandleFunc(LOCAL_DOMAIN, queryHandler(zone, mdnsClient))
LocalServeMux.HandleFunc(RDNS_DOMAIN, rdnsHandler(zone, mdnsClient))
LocalServeMux.HandleFunc(".", notUsHandler())
mdnsServer, err := NewMDNSServer(zone)
checkFatal(err)
err = mdnsServer.Start(iface)
checkFatal(err)
address := fmt.Sprintf(":%d", dnsPort)
Info.Printf("Listening for DNS on %s", address)
err = dns.ListenAndServe(address, "udp", LocalServeMux)
checkFatal(err)
return nil
}
+32
View File
@@ -0,0 +1,32 @@
package nameserver
import (
"reflect"
"testing"
)
func assertNoErr(t *testing.T, err error) {
if err != nil {
t.Fatal(err)
}
}
func assertStatus(t *testing.T, got int, wanted int, desc string) {
if got != wanted {
t.Fatalf("Expected %s %d but got %d", desc, wanted, got)
}
}
func assertErrorInterface(t *testing.T, got interface{}, wanted interface{}, desc string) {
gotT, wantedT := reflect.TypeOf(got), reflect.TypeOf(wanted).Elem()
if !gotT.Implements(wantedT) {
t.Fatalf("Expected %s but got %s (%s)", wantedT.String(), gotT.String(), desc)
}
}
func assertErrorType(t *testing.T, got interface{}, wanted interface{}, desc string) {
gotT, wantedT := reflect.TypeOf(got), reflect.TypeOf(wanted).Elem()
if gotT != wantedT {
t.Fatalf("Expected %s but got %s (%s)", wantedT.String(), gotT.String(), desc)
}
}
+34
View File
@@ -0,0 +1,34 @@
package nameserver
import (
"github.com/fsouza/go-dockerclient"
)
func StartUpdater(apiPath string, zone Zone) error {
client, err := docker.NewClient(apiPath)
if err != nil {
Error.Fatalf("Unable to connect to Docker API on %s: %s", apiPath, err)
}
events := make(chan *docker.APIEvents)
client.AddEventListener(events)
Info.Printf("Using Docker API on %s", apiPath)
go func() {
for event := range events {
handleEvent(zone, event, client)
}
}()
return nil
}
func handleEvent(zone Zone, event *docker.APIEvents, client *docker.Client) error {
switch event.Status {
case "die":
id := event.ID
Info.Printf("Container %s down. Removing records", id)
zone.DeleteRecordsFor(id)
}
return nil
}
+115
View File
@@ -0,0 +1,115 @@
package nameserver
import (
"github.com/miekg/dns"
"net"
"sync"
)
type Zone interface {
AddRecord(ident string, name string, localIP net.IP, weaveIP net.IP) error
DeleteRecord(ident string, weaveIp net.IP) error
DeleteRecordsFor(ident string) error
MatchLocal(name string) (net.IP, error)
MatchLocalIP(ip net.IP) (string, error)
}
type Record struct {
Ident string
Name string
LocalIP net.IP
WeaveIP 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 []Record
}
type LookupError string
func (ops LookupError) Error() string {
return "Unable to find " + string(ops)
}
type DuplicateError struct {
Name string
WeaveIP net.IP
Ident string
}
func (err DuplicateError) Error() string {
return "Duplicate " + err.Name + "," + err.WeaveIP.String() + " for identity " + err.Ident
}
func (zone *ZoneDb) indexOf(match func(Record) bool) int {
for i, r := range zone.recs {
if match(r) {
return i
}
}
return -1
}
func (zone *ZoneDb) MatchLocal(name string) (net.IP, error) {
zone.mx.RLock()
defer zone.mx.RUnlock()
for _, r := range zone.recs {
if r.Name == name {
return r.WeaveIP, nil
}
}
return nil, LookupError(name)
}
func (zone *ZoneDb) MatchLocalIP(ip net.IP) (string, error) {
zone.mx.RLock()
defer zone.mx.RUnlock()
for _, r := range zone.recs {
if r.WeaveIP.Equal(ip) {
return r.Name, nil
}
}
return "", LookupError(ip.String())
}
func (zone *ZoneDb) AddRecord(ident string, name string, localIP net.IP, weaveIP net.IP) error {
zone.mx.Lock()
defer zone.mx.Unlock()
fqdn := dns.Fqdn(name)
if index := zone.indexOf(
func(r Record) bool { return r.Name == fqdn && r.WeaveIP.Equal(weaveIP) }); index != -1 {
return DuplicateError{fqdn, weaveIP, zone.recs[index].Ident}
}
zone.recs = append(zone.recs, Record{ident, fqdn, localIP, weaveIP})
return nil
}
func (zone *ZoneDb) DeleteRecord(ident string, weaveIP net.IP) error {
zone.mx.Lock()
defer zone.mx.Unlock()
if index := zone.indexOf(
func(r Record) bool { return r.Ident == ident && r.WeaveIP.Equal(weaveIP) }); index == -1 {
return LookupError(ident)
} else {
zone.recs = append(zone.recs[:index], zone.recs[index+1:]...)
}
return nil
}
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
}
+84
View File
@@ -0,0 +1,84 @@
package nameserver
import (
"net"
"testing"
)
func TestZone(t *testing.T) {
var (
containerID = "deadbeef"
successTestName = "test1.weave."
testAddr1 = "10.0.2.1/24"
dockerIP = "9.8.7.6"
)
var zone = new(ZoneDb)
ip := net.ParseIP(dockerIP)
weaveIP, _, _ := net.ParseCIDR(testAddr1)
err := zone.AddRecord(containerID, successTestName, ip, weaveIP)
assertNoErr(t, err)
// Add a few more records to make the job harder
err = zone.AddRecord("abcdef0123", "adummy.weave.", net.ParseIP("1.1.1.1"), net.ParseIP("10.0.0.1"))
assertNoErr(t, err)
err = zone.AddRecord("0123abcdef", "zdummy.weave.", net.ParseIP("2.2.2.2"), net.ParseIP("10.0.0.2"))
assertNoErr(t, err)
// Check that the address is now there.
foundIP, err := zone.MatchLocal(successTestName)
assertNoErr(t, err)
if !foundIP.Equal(weaveIP) {
t.Fatal("Unexpected result for", successTestName, foundIP)
}
// See if we can find the address by IP.
foundName, err := zone.MatchLocalIP(weaveIP)
assertNoErr(t, err)
if foundName != successTestName {
t.Fatal("Unexpected result for", weaveIP, foundName)
}
// Now try to add the same address again
err = zone.AddRecord(containerID, successTestName, ip, weaveIP)
assertErrorType(t, err, (*DuplicateError)(nil), "duplicate add")
// Now delete the record
err = zone.DeleteRecord(containerID, weaveIP)
assertNoErr(t, err)
// Check that the address is not there now.
_, err = zone.MatchLocal(successTestName)
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"))
assertErrorType(t, err, (*LookupError)(nil), "when deleting record that doesn't exist")
}
func TestDeleteFor(t *testing.T) {
var (
id = "foobar"
name = "foo.weave."
addr1 = "10.1.2.3/24"
addr2 = "10.2.7.8/24"
dockerIP = "172.16.0.4"
)
zone := new(ZoneDb)
ip := net.ParseIP(dockerIP)
for _, addr := range []string{addr1, addr2} {
weaveIP, _, _ := net.ParseCIDR(addr)
err := zone.AddRecord(id, name, ip, weaveIP)
assertNoErr(t, err)
}
_, err := zone.MatchLocal(name)
assertNoErr(t, err)
err = zone.DeleteRecordsFor(id)
_, err = zone.MatchLocal(name)
assertErrorType(t, err, (*LookupError)(nil), "after deleting records for ident")
}
+5
View File
@@ -0,0 +1,5 @@
FROM scratch
# EXPOSE 53/udp
WORKDIR /home/weave
ADD ./weavedns /home/weave/
ENTRYPOINT ["/home/weave/weavedns", "-wait", "5", "-iface", "ethwe"]
+19
View File
@@ -0,0 +1,19 @@
.DEFAULT: all
all: server-image
weavedns: ../nameserver/*.go ./main.go
go get -tags netgo && go build -tags netgo -ldflags '-linkmode external -extldflags "-static"' -o $@ -v
.PHONY: server-image
server-image: /var/tmp/weavedns.tar
/var/tmp/weavedns.tar: Dockerfile weavedns
sudo -E docker build -t zettio/weavedns .
sudo -E docker save zettio/weavedns > $@
clean:
-sudo docker rmi zettio/weavedns
rm -f weavedns
rm -f /var/tmp/weavedns.tar
+61
View File
@@ -0,0 +1,61 @@
# Weave DNS server
The Weave DNS server answers name queries in a Weave network. It is
run per-host, to be supplied as the nameserver for containers on that
host. It is then told about hostnames for the local containers. For
other names it will ask the other weave hosts, or fall back to using
the host's configured name server.
## Use
The Weave DNS container should be started after the Weave router --
i.e., after running `weave launch` -- and joined to the Weave network.
In general, it should be given an IP in the same subnet as the
router. To watch for container events it must have the docker socker
mounted.
```bash
$ weave launch 10.0.0.1/16
$ weave run 10.0.0.2/16 --name=weavedns -v /var/run/docker.sock:/var/run/docker.sock zettio/weavedns
```
The docker IP for the container can then be provided as the `--dns` value
for other containers:
```bash
$ dns_ip=$(docker inspect --format='{{ .NetworkSettings.IPAddress }}' weavedns)
$ shell1=$(weave run 10.0.1.2/24 -h shell.weave.local --dns=$dns_ip -t ubuntu /bin/bash)
```
To associate a hostname with an IP, use an HTTP PUT to the DNS server:
```bash
$ shell1_ip=$(docker inspect --format='{{ .NetworkSettings.IPAddress }}' $shell1)
$ curl -X PUT "http://$dns_ip:6785/name/$shell1/10.0.1.2?fqdn=shell1.weave.local&routing_prefix=24&local_ip=$shell1_ip"
```
You should now be able to look up the name:
```bash
$ dig @$dns_ip +short shell1.weave.local
```
By default, the server will also watch docker events and remove
entries for any containers that die. You can tell it not to by adding
`--watch=false` to the container args:
```bash
$ weave run 10.0.0.2/16 --name=weavedns -v /var/run/docker.sock:/var/run/docker.sock zettio/weavedns --watch=false
```
## Present limitations
* The server will not know about restarted containers, unless you
re-PUT the name.
* The server will forget names if it is itself restarted.
* The server may give unreachable IPs as answers, since it
doesn't try to filter by subnet yet.
* We use UDP broadcast to find out about remote names (from Weave DNS
servers on other hosts); this likely won't scale well beyond a
certain point T.B.D., so we'll have to come up with another scheme.
+110
View File
@@ -0,0 +1,110 @@
# Weave DNS (service discovery) design
The model is that each host has a service that is notified of
hostnames and weave addresses for containers on the host. It binds to
the host bridge to answer DNS queries from local containers; anything
it can't answer, it asks with mDNS on the weave network.
The service is divided into two components: the first is the DNS
server, which serves the records it is told about, and otherwise asks
via mDNS on the weave network. The other component is the DNS updater,
specific to Docker, which monitors the running containers so to add
and remove records from the DNS server.
## DNS server API
The DNS server accepts HTTP requests on the following URL (patterns)
and methods:
`PUT /name/<identifier>/<ip-address>`
Put a record for an IP, bound to a host-scoped identifier (e.g., a
container ID), in the DNS database.
`DELETE /name/<identifier>/<ip-address>`
Remove a specific record for an IP and host-scoped identifier.
`DELETE /name/<identifier>`
Remove all records for the host-scoped identifier.
`GET /name/`
List all the records in the database.
`GET /name/<identifier>`
List all the records for a host-scoped identifier.
`GET /status`
Give the server status.
### Record structure
The record structure is *either* form fields (useful for scripting
PUTs), or a JSON object with the same fields. The fields are:
```js
{
"fqdn": string,
"routing_prefix": number,
"local_ip": string
}
```
> Also TTL?
## DNS server behaviour
The DNS server listens for the HTTP requests as above, and maintains a
database of host-local names accordingly.
It also listens on the host-local network for DNS queries (port
53). Anything that is in its database it responds to immediately. For
other requests, it asks using multicast DNS on the weave network.
Meanwhile, the DNS server is listening for mDNS queries, and responds
if it the queried name is in its database.
### Caching mDNS answers
The DNS server will hear the answers to its own and other servers'
mDNS queries sent on the weave network. It can cache these, to respond
to local DNS requests; the cache is consulted before using mDNS.
## DNS updater
The updater component uses the Docker remote API to monitor containers
coming and going, and tells the DNS server to update its records via
its HTTP interface. It does not need to be attached to the weave
network.
The updater starts by subscribing to the events, and getting a list of
the current containers. Any containers given a domain ending with
".weave" are considered for inclusion in the name database.
When it sees a container start or stop, the updater checks the weave
network attachment of the container, and updates the DNS server.
> How does it check the network attachment from within a container?
> Will it need to delay slightly so that `attach` has a chance to run?
> Perhaps it could put containers on a watch list when it's noticed
> them.
## Container startup
> TBD
## Grouping and load balancing
> TBD
## Extensions
* An updater that checks health more specifically, e.g., by testing
that a particular port is being listened to, or that it can get a
200 OK. These could be encoded in the container environment
variables, as per registrator.
+59
View File
@@ -0,0 +1,59 @@
package main
import (
"flag"
weavedns "github.com/zettio/weavedns/nameserver"
"io/ioutil"
"net"
"os"
)
func main() {
var (
ifaceName string
apiPath string
dnsPort int
httpPort int
wait int
watch bool
debug bool
)
flag.StringVar(&ifaceName, "iface", "", "name of interface to use for multicast")
flag.StringVar(&apiPath, "api", "unix:///var/run/docker.sock", "Path to Docker API socket")
flag.IntVar(&wait, "wait", 0, "number of seconds to wait for interface to be created and come up")
flag.IntVar(&dnsPort, "dnsport", 53, "port to listen to dns requests")
flag.IntVar(&httpPort, "httpport", 6785, "port to listen to HTTP requests")
flag.BoolVar(&watch, "watch", true, "watch the docker socket for container events")
flag.BoolVar(&debug, "debug", false, "output debugging info to stderr")
flag.Parse()
debugOut := ioutil.Discard
if debug {
debugOut = os.Stderr
}
weavedns.InitLogging(debugOut, os.Stdout, os.Stdout, os.Stderr)
var zone = new(weavedns.ZoneDb)
if watch {
err := weavedns.StartUpdater(apiPath, zone)
if err != nil {
weavedns.Error.Fatal("Unable to start watcher", err)
}
}
var iface *net.Interface = nil
if ifaceName != "" {
var err error
iface, err = weavedns.EnsureInterface(ifaceName, wait)
if err != nil {
weavedns.Error.Fatal(err)
}
}
err := weavedns.StartServer(zone, iface, dnsPort, httpPort, wait)
if err != nil {
weavedns.Error.Fatal("Failed to start server", err)
}
}