Merge pull request #45 from miekg/etcd-stub

Etcd stub

core/setup/etcd.go

@@ -21,10 +21,16 @@ const defaultEndpoint = "http://127.0.0.1:2379"
 
 // Etcd sets up the etcd middleware.
 func Etcd(c *Controller) (middleware.Middleware, error) {
-	etcd, err := etcdParse(c)
+	etcd, stubzones, err := etcdParse(c)
 	if err != nil {
 		return nil, err
 	}
+	if stubzones {
+		c.Startup = append(c.Startup, func() error {
+			etcd.UpdateStubZones()
+			return nil
+		})
+	}
 
 	return func(next middleware.Handler) middleware.Handler {
 		etcd.Next = next
@@ -32,31 +38,113 @@ func Etcd(c *Controller) (middleware.Middleware, error) {
 	}, nil
 }
 
-func etcdParse(c *Controller) (etcd.Etcd, error) {
+func etcdParse(c *Controller) (etcd.Etcd, bool, error) {
+	stub := make(map[string]proxy.Proxy)
 	etc := etcd.Etcd{
-		// make stuff configurable
 		Proxy:      proxy.New([]string{"8.8.8.8:53"}),
 		PathPrefix: "skydns",
 		Ctx:        context.Background(),
 		Inflight:   &singleflight.Group{},
+		Stubmap:    &stub,
 	}
+	var (
+		client        etcdc.KeysAPI
+		tlsCertFile   = ""
+		tlsKeyFile    = ""
+		tlsCAcertFile = ""
+		endpoints     = []string{defaultEndpoint}
+		stubzones     = false
+	)
 	for c.Next() {
 		if c.Val() == "etcd" {
-			// etcd [origin...]
-			client, err := newEtcdClient([]string{defaultEndpoint}, "", "", "")
-			if err != nil {
-				return etcd.Etcd{}, err
-			}
 			etc.Client = client
 			etc.Zones = c.RemainingArgs()
 			if len(etc.Zones) == 0 {
 				etc.Zones = c.ServerBlockHosts
 			}
 			middleware.Zones(etc.Zones).FullyQualify()
-			return etc, nil
+			if c.NextBlock() {
+				// TODO(miek): 2 switches?
+				switch c.Val() {
+				case "stubzones":
+					stubzones = true
+				case "path":
+					if !c.NextArg() {
+						return etcd.Etcd{}, false, c.ArgErr()
+					}
+					etc.PathPrefix = c.Val()
+				case "endpoint":
+					args := c.RemainingArgs()
+					if len(args) == 0 {
+						return etcd.Etcd{}, false, c.ArgErr()
+					}
+					endpoints = args
+				case "upstream":
+					args := c.RemainingArgs()
+					if len(args) == 0 {
+						return etcd.Etcd{}, false, c.ArgErr()
+					}
+					for i := 0; i < len(args); i++ {
+						h, p, e := net.SplitHostPort(args[i])
+						if e != nil && p == "" {
+							args[i] = h + ":53"
+						}
+					}
+					endpoints = args
+					etc.Proxy = proxy.New(args)
+				case "tls": // cert key cacertfile
+					args := c.RemainingArgs()
+					if len(args) != 3 {
+						return etcd.Etcd{}, false, c.ArgErr()
+					}
+					tlsCertFile, tlsKeyFile, tlsCAcertFile = args[0], args[1], args[2]
+				}
+				for c.Next() {
+					switch c.Val() {
+					case "stubzones":
+						stubzones = true
+					case "path":
+						if !c.NextArg() {
+							return etcd.Etcd{}, false, c.ArgErr()
+						}
+						etc.PathPrefix = c.Val()
+					case "endpoint":
+						args := c.RemainingArgs()
+						if len(args) == 0 {
+							return etcd.Etcd{}, false, c.ArgErr()
+						}
+						endpoints = args
+					case "upstream":
+						args := c.RemainingArgs()
+						if len(args) == 0 {
+							return etcd.Etcd{}, false, c.ArgErr()
+						}
+						for i := 0; i < len(args); i++ {
+							h, p, e := net.SplitHostPort(args[i])
+							if e != nil && p == "" {
+								args[i] = h + ":53"
+							}
+						}
+						endpoints = args
+						etc.Proxy = proxy.New(args)
+					case "tls": // cert key cacertfile
+						args := c.RemainingArgs()
+						if len(args) != 3 {
+							return etcd.Etcd{}, false, c.ArgErr()
+						}
+						tlsCertFile, tlsKeyFile, tlsCAcertFile = args[0], args[1], args[2]
+					}
+				}
+			}
+			client, err := newEtcdClient(endpoints, tlsCertFile, tlsKeyFile, tlsCAcertFile)
+			if err != nil {
+				return etcd.Etcd{}, false, err
+			}
+			etc.Client = client
+			return etc, stubzones, nil
 		}
 	}
-	return etcd.Etcd{}, nil
+	return etcd.Etcd{}, false, nil
 }
 
 func newEtcdClient(endpoints []string, tlsCert, tlsKey, tlsCACert string) (etcdc.KeysAPI, error) {

middleware/etcd/etcd.go

@@ -18,11 +18,12 @@ import (
 type Etcd struct {
 	Next       middleware.Handler
 	Zones      []string
+	PathPrefix string
 	Proxy      proxy.Proxy // Proxy for looking up names during the resolution process
 	Client     etcdc.KeysAPI
 	Ctx        context.Context
 	Inflight   *singleflight.Group
-	PathPrefix string
+	Stubmap    *map[string]proxy.Proxy // List of proxies for stub resolving.
 }
 
 // Records looks up records in etcd. If exact is true, it will lookup just

middleware/etcd/etcd.md

@@ -13,7 +13,7 @@ other servers in the network.
 etcd [zones...]
 ~~~
 
-* `zones` zones it should be authoritative for.
+* `zones` zones etcd should be authoritative for.
 
 The will default to `/skydns` as the path and the local etcd proxy (http://127.0.0.1:2379).
 If no zones are specified the block's zone will be used as the zone.
@@ -21,15 +21,19 @@ If no zones are specified the block's zone will be used as the zone.
 If you want to `round robin` A and AAAA responses look at the `loadbalance` middleware.
 
 ~~~
-etcd {
+etcd [zones...] {
+    stubzones
     path /skydns
     endpoint endpoint...
-    stubzones
+    upstream address...
+    tls cert key cacert
 }
 ~~~
 
-* `path` /skydns
+* `stubzones` enable the stub zones feature.
-* `endpoint` endpoints...
+* `path` the path inside etcd, defaults to "/skydns".
-* `stubzones`
+* `endpoint` the etcd endpoints, default to "http://localhost:2397".
+* `upstream` upstream resolvers to be used resolve external names found in etcd.
+* `tls` followed the cert, key and the CA's cert filenames.
 
 ## Examples

middleware/etcd/handler.go

@@ -1,6 +1,8 @@
 package etcd
 
 import (
+	"strings"
+
 	"github.com/miekg/coredns/middleware"
 
 	"github.com/miekg/dns"
@@ -9,6 +11,20 @@ import (
 
 func (e Etcd) ServeDNS(ctx context.Context, w dns.ResponseWriter, r *dns.Msg) (int, error) {
 	state := middleware.State{W: w, Req: r}
+
+	// We need to check stubzones first, because we may get a request for a zone we
+	// are not auth. for *but* do have a stubzone forward for. If we do the stubzone
+	// handler will handle the request.
+	name := state.Name()
+	if len(*e.Stubmap) > 0 {
+		for zone, _ := range *e.Stubmap {
+			if strings.HasSuffix(name, zone) {
+				stub := Stub{Etcd: e, Zone: zone}
+				return stub.ServeDNS(ctx, w, r)
+			}
+		}
+	}
+
 	zone := middleware.Zones(e.Zones).Matches(state.Name())
 	if zone == "" {
 		return e.Next.ServeDNS(ctx, w, r)

middleware/etcd/stub.go

@@ -4,104 +4,68 @@ import (
 	"net"
 	"strconv"
 	"strings"
+	"time"
+
+	"github.com/miekg/coredns/middleware/proxy"
 
 	"github.com/miekg/dns"
 )
 
-// hasStubEdns0 checks if the message is carrying our special
+func (e Etcd) UpdateStubZones() {
-// edns0 zero option.
+	go func() {
-func hasStubEdns0(m *dns.Msg) bool {
+		for {
-	option := m.IsEdns0()
+			e.updateStubZones()
-	if option == nil {
+			time.Sleep(15 * time.Second)
-		return false
-	}
-	for _, o := range option.Option {
-		if o.Option() == ednsStubCode && len(o.(*dns.EDNS0_LOCAL).Data) == 1 &&
-			o.(*dns.EDNS0_LOCAL).Data[0] == 1 {
-			return true
 		}
-	}
+	}()
-	return false
-}
-
-// addStubEdns0 adds our special option to the message's OPT record.
-func addStubEdns0(m *dns.Msg) *dns.Msg {
-	option := m.IsEdns0()
-	// Add a custom EDNS0 option to the packet, so we can detect loops
-	// when 2 stubs are forwarding to each other.
-	if option != nil {
-		option.Option = append(option.Option, &dns.EDNS0_LOCAL{ednsStubCode, []byte{1}})
-	} else {
-		m.Extra = append(m.Extra, ednsStub)
-	}
-	return m
 }
 
 // Look in .../dns/stub/<zone>/xx for msg.Services. Loop through them
 // extract <zone> and add them as forwarders (ip:port-combos) for
 // the stub zones. Only numeric (i.e. IP address) hosts are used.
-// TODO(miek): makes this Startup Function.
+func (e Etcd) updateStubZones() {
-func (e Etcd) UpdateStubZones(zone string) error {
+	stubmap := make(map[string]proxy.Proxy)
-	stubmap := make(map[string][]string)
+	for _, zone := range e.Zones {
-
+		services, err := e.Records(stubDomain+"."+zone, false)
-	services, err := e.Records("stub.dns."+zone, false)
+		if err != nil {
-	if err != nil {
-		return err
-	}
-	for _, serv := range services {
-		if serv.Port == 0 {
-			serv.Port = 53
-		}
-		ip := net.ParseIP(serv.Host)
-		if ip == nil {
-			//logf("stub zone non-address %s seen for: %s", serv.Key, serv.Host)
 			continue
 		}
 
-		domain := e.Domain(serv.Key)
+		// track the nameservers on a per domain basis, but allow a list on the domain.
-		labels := dns.SplitDomainName(domain)
+		nameservers := map[string][]string{}
 
-		// If the remaining name equals any of the zones we have, we ignore it.
+		for _, serv := range services {
-		for _, z := range e.Zones {
+			if serv.Port == 0 {
-			// Chop of left most label, because that is used as the nameserver place holder
+				serv.Port = 53
-			// and drop the right most labels that belong to zone.
+			}
-			domain = dns.Fqdn(strings.Join(labels[1:len(labels)-dns.CountLabel(z)], "."))
+			ip := net.ParseIP(serv.Host)
-			if domain == z {
+			if ip == nil {
 				continue
 			}
-			stubmap[domain] = append(stubmap[domain], net.JoinHostPort(serv.Host, strconv.Itoa(serv.Port)))
+
+			domain := e.Domain(serv.Key)
+			labels := dns.SplitDomainName(domain)
+			// nameserver need to be tracked by domain and *then* added
+
+			// If the remaining name equals any of the zones we have, we ignore it.
+			for _, z := range e.Zones {
+				// Chop of left most label, because that is used as the nameserver place holder
+				// and drop the right most labels that belong to zone.
+				domain = dns.Fqdn(strings.Join(labels[1:len(labels)-dns.CountLabel(z)], "."))
+				if domain == z {
+					continue
+				}
+				nameservers[domain] = append(nameservers[domain], net.JoinHostPort(serv.Host, strconv.Itoa(serv.Port)))
+			}
+		}
+		for domain, nss := range nameservers {
+			stubmap[domain] = proxy.New(nss)
 		}
 	}
 
-	// TODO(miek): add to etcd structure and startup with a StartFunction
+	// atomic swap (at least that's what we hope it is)
-	//	e.stub = &stubmap
+	if len(stubmap) > 0 {
-	// stubmap contains proxy is best way forward... I think.
+		e.Stubmap = &stubmap
-	// TODO(miek): setup a proxy that forward to these
-	// StubProxy type?
-	return nil
-}
-
-func ServeDNSStubForward(w dns.ResponseWriter, req *dns.Msg) *dns.Msg {
-	if !hasStubEdns0(req) {
-		return nil
 	}
-	req = addStubEdns0(req)
+	return
-	// proxy woxy
-	return nil
 }
-
-// ednsStub is the EDNS0 record we add to stub queries. Queries which have this record are
-// not forwarded again.
-var ednsStub = func() *dns.OPT {
-	o := new(dns.OPT)
-	o.Hdr.Name = "."
-	o.Hdr.Rrtype = dns.TypeOPT
-
-	e := new(dns.EDNS0_LOCAL)
-	e.Code = ednsStubCode
-	e.Data = []byte{1}
-	o.Option = append(o.Option, e)
-	return o
-}()
-
-const ednsStubCode = dns.EDNS0LOCALSTART + 10

middleware/etcd/stub_handler.go

@@ -0,0 +1,78 @@
+package etcd
+
+import (
+	"github.com/miekg/coredns/middleware"
+	"github.com/miekg/dns"
+	"golang.org/x/net/context"
+)
+
+// Stub wraps an Etcd. We have this type so that it can have a ServeDNS method.
+type Stub struct {
+	Etcd
+	Zone string // for what zone (and thus what nameservers are we called)
+}
+
+func (s Stub) ServeDNS(ctx context.Context, w dns.ResponseWriter, req *dns.Msg) (int, error) {
+	if hasStubEdns0(req) {
+		// TODO(miek): actual error here
+		return dns.RcodeServerFailure, nil
+	}
+	req = addStubEdns0(req)
+	proxy, ok := (*s.Etcd.Stubmap)[s.Zone]
+	if !ok { // somebody made a mistake..
+		return dns.RcodeServerFailure, nil
+	}
+	state := middleware.State{W: w, Req: req}
+
+	m1, e1 := proxy.Forward(state)
+	if e1 != nil {
+		return dns.RcodeServerFailure, e1
+	}
+	m1.RecursionAvailable, m1.Compress = true, true
+	state.W.WriteMsg(m1)
+	return dns.RcodeSuccess, nil
+}
+
+// hasStubEdns0 checks if the message is carrying our special edns0 zero option.
+func hasStubEdns0(m *dns.Msg) bool {
+	option := m.IsEdns0()
+	if option == nil {
+		return false
+	}
+	for _, o := range option.Option {
+		if o.Option() == ednsStubCode && len(o.(*dns.EDNS0_LOCAL).Data) == 1 &&
+			o.(*dns.EDNS0_LOCAL).Data[0] == 1 {
+			return true
+		}
+	}
+	return false
+}
+
+// addStubEdns0 adds our special option to the message's OPT record.
+func addStubEdns0(m *dns.Msg) *dns.Msg {
+	option := m.IsEdns0()
+	// Add a custom EDNS0 option to the packet, so we can detect loops when 2 stubs are forwarding to each other.
+	if option != nil {
+		option.Option = append(option.Option, &dns.EDNS0_LOCAL{ednsStubCode, []byte{1}})
+	} else {
+		m.Extra = append(m.Extra, ednsStub)
+	}
+	return m
+}
+
+const (
+	ednsStubCode = dns.EDNS0LOCALSTART + 10
+	stubDomain   = "stub.dns"
+)
+
+var ednsStub = func() *dns.OPT {
+	o := new(dns.OPT)
+	o.Hdr.Name = "."
+	o.Hdr.Rrtype = dns.TypeOPT
+
+	e := new(dns.EDNS0_LOCAL)
+	e.Code = ednsStubCode
+	e.Data = []byte{1}
+	o.Option = append(o.Option, e)
+	return o
+}()

middleware/etcd/stub_test.go

@@ -0,0 +1,145 @@
+// +build etcd
+
+package etcd
+
+import "testing"
+
+func TestStubLookup(t *testing.T) {
+	// e.updateStubZones()
+}
+
+/*
+func TestDNSStubForward(t *testing.T) {
+	s := newTestServer(t, false)
+	defer s.Stop()
+
+	c := new(dns.Client)
+	m := new(dns.Msg)
+
+	stubEx := &msg.Service{
+		// IP address of a.iana-servers.net.
+		Host: "199.43.132.53", Key: "a.example.com.stub.dns.skydns.test.",
+	}
+	stubBroken := &msg.Service{
+		Host: "127.0.0.1", Port: 5454, Key: "b.example.org.stub.dns.skydns.test.",
+	}
+	stubLoop := &msg.Service{
+		Host: "127.0.0.1", Port: Port, Key: "b.example.net.stub.dns.skydns.test.",
+	}
+	addService(t, s, stubEx.Key, 0, stubEx)
+	defer delService(t, s, stubEx.Key)
+	addService(t, s, stubBroken.Key, 0, stubBroken)
+	defer delService(t, s, stubBroken.Key)
+	addService(t, s, stubLoop.Key, 0, stubLoop)
+	defer delService(t, s, stubLoop.Key)
+
+	s.UpdateStubZones()
+
+	m.SetQuestion("www.example.com.", dns.TypeA)
+	resp, _, err := c.Exchange(m, "127.0.0.1:"+StrPort)
+	if err != nil {
+		// try twice
+		resp, _, err = c.Exchange(m, "127.0.0.1:"+StrPort)
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+	if len(resp.Answer) == 0 || resp.Rcode != dns.RcodeSuccess {
+		t.Fatal("answer expected to have A records or rcode not equal to RcodeSuccess")
+	}
+	// The main diff. here is that we expect the AA bit to be set, because we directly
+	// queried the authoritative servers.
+	if resp.Authoritative != true {
+		t.Fatal("answer expected to have AA bit set")
+	}
+
+	// This should fail.
+	m.SetQuestion("www.example.org.", dns.TypeA)
+	resp, _, err = c.Exchange(m, "127.0.0.1:"+StrPort)
+	if len(resp.Answer) != 0 || resp.Rcode != dns.RcodeServerFailure {
+		t.Fatal("answer expected to fail for example.org")
+	}
+
+	// This should really fail with a timeout.
+	m.SetQuestion("www.example.net.", dns.TypeA)
+	resp, _, err = c.Exchange(m, "127.0.0.1:"+StrPort)
+	if err == nil {
+		t.Fatal("answer expected to fail for example.net")
+	} else {
+		t.Logf("succesfully failing %s", err)
+	}
+
+	// Packet with EDNS0
+	m.SetEdns0(4096, true)
+	resp, _, err = c.Exchange(m, "127.0.0.1:"+StrPort)
+	if err == nil {
+		t.Fatal("answer expected to fail for example.net")
+	} else {
+		t.Logf("succesfully failing %s", err)
+	}
+
+	// Now start another SkyDNS instance on a different port,
+	// add a stubservice for it and check if the forwarding is
+	// actually working.
+	oldStrPort := StrPort
+
+	s1 := newTestServer(t, false)
+	defer s1.Stop()
+	s1.config.Domain = "skydns.com."
+
+	// Add forwarding IP for internal.skydns.com. Use Port to point to server s.
+	stubForward := &msg.Service{
+		Host: "127.0.0.1", Port: Port, Key: "b.internal.skydns.com.stub.dns.skydns.test.",
+	}
+	addService(t, s, stubForward.Key, 0, stubForward)
+	defer delService(t, s, stubForward.Key)
+	s.UpdateStubZones()
+
+	// Add an answer for this in our "new" server.
+	stubReply := &msg.Service{
+		Host: "127.1.1.1", Key: "www.internal.skydns.com.",
+	}
+	addService(t, s1, stubReply.Key, 0, stubReply)
+	defer delService(t, s1, stubReply.Key)
+
+	m = new(dns.Msg)
+	m.SetQuestion("www.internal.skydns.com.", dns.TypeA)
+	resp, _, err = c.Exchange(m, "127.0.0.1:"+oldStrPort)
+	if err != nil {
+		t.Fatalf("failed to forward %s", err)
+	}
+	if resp.Answer[0].(*dns.A).A.String() != "127.1.1.1" {
+		t.Fatalf("failed to get correct reply")
+	}
+
+	// Adding an in baliwick internal domain forward.
+	s2 := newTestServer(t, false)
+	defer s2.Stop()
+	s2.config.Domain = "internal.skydns.net."
+
+	// Add forwarding IP for internal.skydns.net. Use Port to point to server s.
+	stubForward1 := &msg.Service{
+		Host: "127.0.0.1", Port: Port, Key: "b.internal.skydns.net.stub.dns.skydns.test.",
+	}
+	addService(t, s, stubForward1.Key, 0, stubForward1)
+	defer delService(t, s, stubForward1.Key)
+	s.UpdateStubZones()
+
+	// Add an answer for this in our "new" server.
+	stubReply1 := &msg.Service{
+		Host: "127.10.10.10", Key: "www.internal.skydns.net.",
+	}
+	addService(t, s2, stubReply1.Key, 0, stubReply1)
+	defer delService(t, s2, stubReply1.Key)
+
+	m = new(dns.Msg)
+	m.SetQuestion("www.internal.skydns.net.", dns.TypeA)
+	resp, _, err = c.Exchange(m, "127.0.0.1:"+oldStrPort)
+	if err != nil {
+		t.Fatalf("failed to forward %s", err)
+	}
+	if resp.Answer[0].(*dns.A).A.String() != "127.10.10.10" {
+		t.Fatalf("failed to get correct reply")
+	}
+}
+*/

middleware/proxy/lookup.go

@@ -52,6 +52,9 @@ func New(hosts []string) Proxy {
 	return p
 }
 
+// Lookup will use name and tpe to forge a new message and will send that upstream. It will
+// set any EDNS0 options correctly so that downstream will be able to process the reply.
+// Lookup is not suitable for forwarding request. So Forward for that.
 func (p Proxy) Lookup(state middleware.State, name string, tpe uint16) (*dns.Msg, error) {
 	req := new(dns.Msg)
 	req.SetQuestion(name, tpe)
@@ -62,13 +65,17 @@ func (p Proxy) Lookup(state middleware.State, name string, tpe uint16) (*dns.Msg
 	return p.lookup(state, req)
 }
 
+func (p Proxy) Forward(state middleware.State) (*dns.Msg, error) {
+	return p.lookup(state, state.Req)
+}
+
 func (p Proxy) lookup(state middleware.State, r *dns.Msg) (*dns.Msg, error) {
 	var (
 		reply *dns.Msg
 		err   error
 	)
 	for _, upstream := range p.Upstreams {
-		// allowed bla bla bla TODO(miek): fix full proxy spec from caddy
+		// allowed bla bla bla TODO(miek): fix full proxy spec from caddy?
 		start := time.Now()
 
 		// Since Select() should give us "up" hosts, keep retrying