diff --git a/.gitignore b/.gitignore new file mode 100644 index 00000000..a23bb285 --- /dev/null +++ b/.gitignore @@ -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 diff --git a/nameserver/dns.go b/nameserver/dns.go new file mode 100644 index 00000000..bd0466e0 --- /dev/null +++ b/nameserver/dns.go @@ -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) +} diff --git a/nameserver/http.go b/nameserver/http.go new file mode 100644 index 00000000..b25b9204 --- /dev/null +++ b/nameserver/http.go @@ -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// +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) + } +} diff --git a/nameserver/http_test.go b/nameserver/http_test.go new file mode 100644 index 00000000..9e41dbd8 --- /dev/null +++ b/nameserver/http_test.go @@ -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 +} diff --git a/nameserver/if_utils.go b/nameserver/if_utils.go new file mode 100644 index 00000000..2159b125 --- /dev/null +++ b/nameserver/if_utils.go @@ -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 +} diff --git a/nameserver/logging.go b/nameserver/logging.go new file mode 100644 index 00000000..03c1529b --- /dev/null +++ b/nameserver/logging.go @@ -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) +} diff --git a/nameserver/mdns_client.go b/nameserver/mdns_client.go new file mode 100644 index 00000000..9bf9dda2 --- /dev/null +++ b/nameserver/mdns_client.go @@ -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? + } + } + } +} diff --git a/nameserver/mdns_client_test.go b/nameserver/mdns_client_test.go new file mode 100644 index 00000000..2e0206a7 --- /dev/null +++ b/nameserver/mdns_client_test.go @@ -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) + } +} diff --git a/nameserver/mdns_server.go b/nameserver/mdns_server.go new file mode 100644 index 00000000..7cc25fae --- /dev/null +++ b/nameserver/mdns_server.go @@ -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 +} diff --git a/nameserver/mdns_server_test.go b/nameserver/mdns_server_test.go new file mode 100644 index 00000000..51fd5f14 --- /dev/null +++ b/nameserver/mdns_server_test.go @@ -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) + } +} diff --git a/nameserver/server.go b/nameserver/server.go new file mode 100644 index 00000000..bd492a57 --- /dev/null +++ b/nameserver/server.go @@ -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 +} diff --git a/nameserver/testing_utils.go b/nameserver/testing_utils.go new file mode 100644 index 00000000..0a4835ae --- /dev/null +++ b/nameserver/testing_utils.go @@ -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) + } +} diff --git a/nameserver/updater.go b/nameserver/updater.go new file mode 100644 index 00000000..9d73090b --- /dev/null +++ b/nameserver/updater.go @@ -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 +} diff --git a/nameserver/zone.go b/nameserver/zone.go new file mode 100644 index 00000000..97c88cd5 --- /dev/null +++ b/nameserver/zone.go @@ -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 +} diff --git a/nameserver/zone_test.go b/nameserver/zone_test.go new file mode 100644 index 00000000..3ada988c --- /dev/null +++ b/nameserver/zone_test.go @@ -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") +} diff --git a/weavedns/Dockerfile b/weavedns/Dockerfile new file mode 100644 index 00000000..cb93e00c --- /dev/null +++ b/weavedns/Dockerfile @@ -0,0 +1,5 @@ +FROM scratch +# EXPOSE 53/udp +WORKDIR /home/weave +ADD ./weavedns /home/weave/ +ENTRYPOINT ["/home/weave/weavedns", "-wait", "5", "-iface", "ethwe"] diff --git a/weavedns/Makefile b/weavedns/Makefile new file mode 100644 index 00000000..237ac97e --- /dev/null +++ b/weavedns/Makefile @@ -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 diff --git a/weavedns/README.md b/weavedns/README.md new file mode 100644 index 00000000..45e05318 --- /dev/null +++ b/weavedns/README.md @@ -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. diff --git a/weavedns/design.md b/weavedns/design.md new file mode 100644 index 00000000..67c44214 --- /dev/null +++ b/weavedns/design.md @@ -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//` + +Put a record for an IP, bound to a host-scoped identifier (e.g., a +container ID), in the DNS database. + +`DELETE /name//` + +Remove a specific record for an IP and host-scoped identifier. + +`DELETE /name/` + +Remove all records for the host-scoped identifier. + +`GET /name/` + +List all the records in the database. + +`GET /name/` + +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. diff --git a/weavedns/main.go b/weavedns/main.go new file mode 100644 index 00000000..79c8a51a --- /dev/null +++ b/weavedns/main.go @@ -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) + } +}