plugin/kubernetes: Add upstream @self and loop count (#1484)

* add upstream @self and loop count

* 1st round of feedback

* allow argless upstream

* update test

* readmes

* feedback

core/dnsserver/server.go

@@ -95,7 +95,7 @@ func NewServer(addr string, group []*Config) (*Server, error) {
 func (s *Server) Serve(l net.Listener) error {
 	s.m.Lock()
 	s.server[tcp] = &dns.Server{Listener: l, Net: "tcp", Handler: dns.HandlerFunc(func(w dns.ResponseWriter, r *dns.Msg) {
-		ctx := context.Background()
+		ctx := context.WithValue(context.Background(), Key{}, s)
 		s.ServeDNS(ctx, w, r)
 	})}
 	s.m.Unlock()
@@ -108,7 +108,7 @@ func (s *Server) Serve(l net.Listener) error {
 func (s *Server) ServePacket(p net.PacketConn) error {
 	s.m.Lock()
 	s.server[udp] = &dns.Server{PacketConn: p, Net: "udp", Handler: dns.HandlerFunc(func(w dns.ResponseWriter, r *dns.Msg) {
-		ctx := context.Background()
+		ctx := context.WithValue(context.Background(), Key{}, s)
 		s.ServeDNS(ctx, w, r)
 	})}
 	s.m.Unlock()
@@ -207,6 +207,12 @@ func (s *Server) ServeDNS(ctx context.Context, w dns.ResponseWriter, r *dns.Msg)
 		return
 	}
 
+	ctx, err := incrementDepthAndCheck(ctx)
+	if err != nil {
+		DefaultErrorFunc(w, r, dns.RcodeServerFailure)
+		return
+	}
+
 	q := r.Question[0].Name
 	b := make([]byte, len(q))
 	var off int
@@ -329,11 +335,36 @@ func DefaultErrorFunc(w dns.ResponseWriter, r *dns.Msg, rc int) {
 	w.WriteMsg(answer)
 }
 
+// incrementDepthAndCheck increments the loop counter in the context, and returns an error if
+// the counter exceeds the max number of re-entries
+func incrementDepthAndCheck(ctx context.Context) (context.Context, error) {
+	// Loop counter for self directed lookups
+	loop := ctx.Value(loopKey{})
+	if loop == nil {
+		ctx = context.WithValue(ctx, loopKey{}, 0)
+		return ctx, nil
+	}
+
+	iloop := loop.(int) + 1
+	if iloop > maxreentries {
+		return ctx, fmt.Errorf("too deep")
+	}
+	ctx = context.WithValue(ctx, loopKey{}, iloop)
+	return ctx, nil
+}
+
 const (
-	tcp = 0
+	tcp          = 0
-	udp = 1
+	udp          = 1
+	maxreentries = 10
 )
 
+// Key is the context key for the current server
+type Key struct{}
+
+// loopKey is the context key for counting self loops
+type loopKey struct{}
+
 // enableChaos is a map with plugin names for which we should open CH class queries as
 // we block these by default.
 var enableChaos = map[string]bool{

core/dnsserver/server_test.go

@@ -48,6 +48,21 @@ func TestNewServer(t *testing.T) {
 	}
 }
 
+func TestIncrementDepthAndCheck(t *testing.T) {
+	ctx := context.Background()
+	var err error
+	for i := 0; i <= maxreentries; i++ {
+		ctx, err = incrementDepthAndCheck(ctx)
+		if err != nil {
+			t.Errorf("Expected no error for depthCheck (i=%v), got %s", i, err)
+		}
+	}
+	_, err = incrementDepthAndCheck(ctx)
+	if err == nil {
+		t.Errorf("Expected error for depthCheck (i=%v)", maxreentries+1)
+	}
+}
+
 func BenchmarkCoreServeDNS(b *testing.B) {
 	s, err := NewServer("127.0.0.1:53", []*Config{testConfig("dns", testPlugin{})})
 	if err != nil {

plugin/backend.go

@@ -3,9 +3,10 @@ package plugin
 import (
 	"github.com/coredns/coredns/plugin/etcd/msg"
 	"github.com/coredns/coredns/request"
-	"golang.org/x/net/context"
 
 	"github.com/miekg/dns"
+
+	"golang.org/x/net/context"
 )
 
 // ServiceBackend defines a (dynamic) backend that returns a slice of service definitions.

plugin/etcd/README.md

@@ -32,7 +32,7 @@ etcd [ZONES...] {
     fallthrough [ZONES...]
     path PATH
     endpoint ENDPOINT...
-    upstream ADDRESS...
+    upstream [ADDRESS...]
     tls CERT KEY CACERT
 }
 ~~~
@@ -47,8 +47,9 @@ etcd [ZONES...] {
 * **ENDPOINT** the etcd endpoints. Defaults to "http://localhost:2379".
 * `upstream` upstream resolvers to be used resolve external names found in etcd (think CNAMEs)
   pointing to external names. If you want CoreDNS to act as a proxy for clients, you'll need to add
-  the proxy plugin. **ADDRESS** can be an IP address, and IP:port or a string pointing to a file
+  the proxy plugin. If no **ADDRESS** is given, CoreDNS will resolve CNAMEs against itself.
-  that is structured as /etc/resolv.conf.
+  **ADDRESS** can be an IP address, and IP:port or a string pointing to a file that is structured
+  as /etc/resolv.conf.
 * `tls` followed by:
 
     * no arguments, if the server certificate is signed by a system-installed CA and no client cert is needed

plugin/etcd/etcd.go

@@ -13,6 +13,7 @@ import (
 	"github.com/coredns/coredns/plugin/proxy"
 	"github.com/coredns/coredns/request"
 
+	"github.com/coredns/coredns/plugin/pkg/upstream"
 	etcdc "github.com/coreos/etcd/client"
 	"github.com/miekg/dns"
 	"golang.org/x/net/context"
@@ -24,7 +25,7 @@ type Etcd struct {
 	Fall       fall.F
 	Zones      []string
 	PathPrefix string
-	Proxy      proxy.Proxy // Proxy for looking up names during the resolution process
+	Upstream   upstream.Upstream // Proxy for looking up names during the resolution process
 	Client     etcdc.KeysAPI
 	Ctx        context.Context
 	Stubmap    *map[string]proxy.Proxy // list of proxies for stub resolving.
@@ -50,7 +51,7 @@ func (e *Etcd) Reverse(state request.Request, exact bool, opt plugin.Options) (s
 
 // Lookup implements the ServiceBackend interface.
 func (e *Etcd) Lookup(state request.Request, name string, typ uint16) (*dns.Msg, error) {
-	return e.Proxy.Lookup(state, name, typ)
+	return e.Upstream.Lookup(state, name, typ)
 }
 
 // IsNameError implements the ServiceBackend interface.

plugin/etcd/handler.go

@@ -12,7 +12,7 @@ import (
 // ServeDNS implements the plugin.Handler interface.
 func (e *Etcd) ServeDNS(ctx context.Context, w dns.ResponseWriter, r *dns.Msg) (int, error) {
 	opt := plugin.Options{}
-	state := request.Request{W: w, Req: r}
+	state := request.Request{W: w, Req: r, Context: ctx}
 
 	name := state.Name()
 

plugin/etcd/lookup_test.go

@@ -10,6 +10,7 @@ import (
 	"github.com/coredns/coredns/plugin/etcd/msg"
 	"github.com/coredns/coredns/plugin/pkg/dnstest"
 	"github.com/coredns/coredns/plugin/pkg/tls"
+	"github.com/coredns/coredns/plugin/pkg/upstream"
 	"github.com/coredns/coredns/plugin/proxy"
 	"github.com/coredns/coredns/plugin/test"
 
@@ -227,8 +228,9 @@ func newEtcdPlugin() *Etcd {
 	tlsc, _ := tls.NewTLSConfigFromArgs()
 	client, _ := newEtcdClient(endpoints, tlsc)
 
+	p := proxy.NewLookup([]string{"8.8.8.8:53"})
 	return &Etcd{
-		Proxy:      proxy.NewLookup([]string{"8.8.8.8:53"}),
+		Upstream:   upstream.Upstream{Forward: &p},
 		PathPrefix: "skydns",
 		Ctx:        context.Background(),
 		Zones:      []string{"skydns.test.", "skydns_extra.test.", "in-addr.arpa."},

plugin/etcd/setup.go

@@ -5,8 +5,8 @@ import (
 
 	"github.com/coredns/coredns/core/dnsserver"
 	"github.com/coredns/coredns/plugin"
-	"github.com/coredns/coredns/plugin/pkg/dnsutil"
 	mwtls "github.com/coredns/coredns/plugin/pkg/tls"
+	"github.com/coredns/coredns/plugin/pkg/upstream"
 	"github.com/coredns/coredns/plugin/proxy"
 
 	etcdc "github.com/coreos/etcd/client"
@@ -90,13 +90,13 @@ func etcdParse(c *caddy.Controller) (*Etcd, bool, error) {
 				case "upstream":
 					args := c.RemainingArgs()
 					if len(args) == 0 {
-						return &Etcd{}, false, c.ArgErr()
+						return nil, false, c.ArgErr()
 					}
-					ups, err := dnsutil.ParseHostPortOrFile(args...)
+					u, err := upstream.NewUpstream(args)
 					if err != nil {
-						return &Etcd{}, false, err
+						return nil, false, err
 					}
-					etc.Proxy = proxy.NewLookup(ups)
+					etc.Upstream = u
 				case "tls": // cert key cacertfile
 					args := c.RemainingArgs()
 					tlsConfig, err = mwtls.NewTLSConfigFromArgs(args...)

plugin/kubernetes/README.md

@@ -13,8 +13,8 @@ CoreDNS running the kubernetes plugin can be used as a replacement of kube-dns i
 cluster.  See the [deployment](https://github.com/coredns/deployment) repository for details on [how
 to deploy CoreDNS in Kubernetes](https://github.com/coredns/deployment/tree/master/kubernetes).
 
-[stubDomains](http://blog.kubernetes.io/2017/04/configuring-private-dns-zones-upstream-nameservers-kubernetes.html)
+[stubDomains and upstreamNameservers](http://blog.kubernetes.io/2017/04/configuring-private-dns-zones-upstream-nameservers-kubernetes.html)
-are implemented via the *proxy* plugin.
+are implemented via the *proxy* plugin and kubernetes *upstream*. See example below.
 
 ## Syntax
 
@@ -36,7 +36,7 @@ kubernetes [ZONES...] {
     labels EXPRESSION
     pods POD-MODE
     endpoint_pod_names
-    upstream ADDRESS...
+    upstream [ADDRESS...]
     ttl TTL
     fallthrough [ZONES...]
 }
@@ -80,8 +80,9 @@ kubernetes [ZONES...] {
    follows: Use the hostname of the endpoint, or if hostname is not set, use the
    pod name of the pod targeted by the endpoint. If there is no pod targeted by
    the endpoint, use the dashed IP address form.
-* `upstream` **ADDRESS [ADDRESS...]** defines the upstream resolvers used for resolving services
+* `upstream` [**ADDRESS**...] defines the upstream resolvers used for resolving services
-  that point to external hosts (External Services).  **ADDRESS** can be an IP, an IP:port, or a path
+  that point to external hosts (aka External Services aka CNAMEs).  If no **ADDRESS** is given, CoreDNS
+  will resolve External Services against itself. **ADDRESS** can be an IP, an IP:port, or a path
   to a file structured like resolv.conf.
 * `ttl` allows you to set a custom TTL for responses. The default (and allowed minimum) is to use
   5 seconds, the maximum is capped at 3600 seconds.
@@ -129,24 +130,33 @@ kubernetes cluster.local {
 }
 ~~~
 
-Here we use the *proxy* plugin to implement stubDomains that forwards `example.org` and
+
-`example.com` to another nameserver.
+## stubDomains and upstreamNameservers
+
+Here we use the *proxy* plugin to implement a stubDomain that forwards `example.local` to the nameserver `10.100.0.10:53`.
+The *upstream* option in kubernetes means that ExternalName services (CNAMEs) will be resolved using the respective proxy.
+Also configured is an upstreamNameserver `8.8.8.8:53` that will be used for resolving names that do not fall in `cluster.local`
+or `example.local`.
 
 ~~~ txt
-cluster.local {
+.:53 {
-    kubernetes {
+    kubernetes cluster.local {
-        endpoint https://k8s-endpoint:8443
+        upstream
-        tls cert key cacert
     }
-}
+    proxy example.local 10.100.0.10:53
-example.org {
-    proxy . 8.8.8.8:53
-}
-example.com {
     proxy . 8.8.8.8:53
 }
 ~~~
 
+The configuration above represents the following Kube-DNS stubDomains and upstreamNameservers configuration.
+
+~~~ txt
+  stubDomains: |
+    {“example.local”: [“10.100.0.10:53”]}
+  upstreamNameservers: |
+    [“8.8.8.8:53”]
+~~~
+
 ## AutoPath
 
 The *kubernetes* plugin can be used in conjunction with the *autopath* plugin.  Using this

plugin/kubernetes/handler.go

@@ -11,7 +11,8 @@ import (
 
 // ServeDNS implements the plugin.Handler interface.
 func (k Kubernetes) ServeDNS(ctx context.Context, w dns.ResponseWriter, r *dns.Msg) (int, error) {
-	state := request.Request{W: w, Req: r}
+	opt := plugin.Options{}
+	state := request.Request{W: w, Req: r, Context: ctx}
 
 	m := new(dns.Msg)
 	m.SetReply(r)
@@ -32,24 +33,24 @@ func (k Kubernetes) ServeDNS(ctx context.Context, w dns.ResponseWriter, r *dns.M
 
 	switch state.QType() {
 	case dns.TypeA:
-		records, err = plugin.A(&k, zone, state, nil, plugin.Options{})
+		records, err = plugin.A(&k, zone, state, nil, opt)
 	case dns.TypeAAAA:
-		records, err = plugin.AAAA(&k, zone, state, nil, plugin.Options{})
+		records, err = plugin.AAAA(&k, zone, state, nil, opt)
 	case dns.TypeTXT:
-		records, err = plugin.TXT(&k, zone, state, plugin.Options{})
+		records, err = plugin.TXT(&k, zone, state, opt)
 	case dns.TypeCNAME:
-		records, err = plugin.CNAME(&k, zone, state, plugin.Options{})
+		records, err = plugin.CNAME(&k, zone, state, opt)
 	case dns.TypePTR:
-		records, err = plugin.PTR(&k, zone, state, plugin.Options{})
+		records, err = plugin.PTR(&k, zone, state, opt)
 	case dns.TypeMX:
-		records, extra, err = plugin.MX(&k, zone, state, plugin.Options{})
+		records, extra, err = plugin.MX(&k, zone, state, opt)
 	case dns.TypeSRV:
-		records, extra, err = plugin.SRV(&k, zone, state, plugin.Options{})
+		records, extra, err = plugin.SRV(&k, zone, state, opt)
 	case dns.TypeSOA:
-		records, err = plugin.SOA(&k, zone, state, plugin.Options{})
+		records, err = plugin.SOA(&k, zone, state, opt)
 	case dns.TypeNS:
 		if state.Name() == zone {
-			records, extra, err = plugin.NS(&k, zone, state, plugin.Options{})
+			records, extra, err = plugin.NS(&k, zone, state, opt)
 			break
 		}
 		fallthrough
@@ -57,21 +58,21 @@ func (k Kubernetes) ServeDNS(ctx context.Context, w dns.ResponseWriter, r *dns.M
 		k.Transfer(ctx, state)
 	default:
 		// Do a fake A lookup, so we can distinguish between NODATA and NXDOMAIN
-		_, err = plugin.A(&k, zone, state, nil, plugin.Options{})
+		_, err = plugin.A(&k, zone, state, nil, opt)
 	}
 
 	if k.IsNameError(err) {
 		if k.Fall.Through(state.Name()) {
 			return plugin.NextOrFailure(k.Name(), k.Next, ctx, w, r)
 		}
-		return plugin.BackendError(&k, zone, dns.RcodeNameError, state, nil /* err */, plugin.Options{})
+		return plugin.BackendError(&k, zone, dns.RcodeNameError, state, nil /* err */, opt)
 	}
 	if err != nil {
 		return dns.RcodeServerFailure, err
 	}
 
 	if len(records) == 0 {
-		return plugin.BackendError(&k, zone, dns.RcodeSuccess, state, nil, plugin.Options{})
+		return plugin.BackendError(&k, zone, dns.RcodeSuccess, state, nil, opt)
 	}
 
 	m.Answer = append(m.Answer, records...)

plugin/kubernetes/kubernetes.go

@@ -14,7 +14,7 @@ import (
 	"github.com/coredns/coredns/plugin/pkg/dnsutil"
 	"github.com/coredns/coredns/plugin/pkg/fall"
 	"github.com/coredns/coredns/plugin/pkg/healthcheck"
-	"github.com/coredns/coredns/plugin/proxy"
+	"github.com/coredns/coredns/plugin/pkg/upstream"
 	"github.com/coredns/coredns/request"
 
 	"github.com/miekg/dns"
@@ -31,7 +31,7 @@ import (
 type Kubernetes struct {
 	Next             plugin.Handler
 	Zones            []string
-	Proxy            proxy.Proxy // Proxy for looking up names during the resolution process
+	Upstream         upstream.Upstream
 	APIServerList    []string
 	APIProxy         *apiProxy
 	APICertAuth      string
@@ -59,7 +59,6 @@ func New(zones []string) *Kubernetes {
 	k.Namespaces = make(map[string]bool)
 	k.interfaceAddrsFunc = func() net.IP { return net.ParseIP("127.0.0.1") }
 	k.podMode = podModeDisabled
-	k.Proxy = proxy.Proxy{}
 	k.ttl = defaultTTL
 
 	return k
@@ -146,7 +145,7 @@ func (k *Kubernetes) primaryZone() string { return k.Zones[k.primaryZoneIndex] }
 
 // Lookup implements the ServiceBackend interface.
 func (k *Kubernetes) Lookup(state request.Request, name string, typ uint16) (*dns.Msg, error) {
-	return k.Proxy.Lookup(state, name, typ)
+	return k.Upstream.Lookup(state, name, typ)
 }
 
 // IsNameError implements the ServiceBackend interface.

plugin/kubernetes/setup.go

@@ -9,10 +9,9 @@ import (
 
 	"github.com/coredns/coredns/core/dnsserver"
 	"github.com/coredns/coredns/plugin"
-	"github.com/coredns/coredns/plugin/pkg/dnsutil"
 	"github.com/coredns/coredns/plugin/pkg/parse"
-	"github.com/coredns/coredns/plugin/proxy"
 
+	"github.com/coredns/coredns/plugin/pkg/upstream"
 	"github.com/mholt/caddy"
 	"github.com/miekg/dns"
 	meta "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -195,14 +194,11 @@ func ParseStanza(c *caddy.Controller) (*Kubernetes, error) {
 			k8s.Fall.SetZonesFromArgs(c.RemainingArgs())
 		case "upstream":
 			args := c.RemainingArgs()
-			if len(args) == 0 {
+			u, err := upstream.NewUpstream(args)
-				return nil, c.ArgErr()
-			}
-			ups, err := dnsutil.ParseHostPortOrFile(args...)
 			if err != nil {
 				return nil, err
 			}
-			k8s.Proxy = proxy.NewLookup(ups)
+			k8s.Upstream = u
 		case "ttl":
 			args := c.RemainingArgs()
 			if len(args) == 0 {

plugin/kubernetes/setup_test.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/coredns/coredns/plugin/pkg/fall"
 
+	"github.com/coredns/coredns/plugin/proxy"
 	"github.com/mholt/caddy"
 	meta "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
@@ -463,7 +464,10 @@ kubernetes cluster.local`,
 			t.Errorf("Test %d: Expected kubernetes controller to be initialized with fallthrough '%v'. Instead found fallthrough '%v' for input '%s'", i, test.expectedFallthrough, k8sController.Fall, test.input)
 		}
 		// upstream
-		foundUpstreams := k8sController.Proxy.Upstreams
+		var foundUpstreams *[]proxy.Upstream
+		if k8sController.Upstream.Forward != nil {
+			foundUpstreams = k8sController.Upstream.Forward.Upstreams
+		}
 		if test.expectedUpstreams == nil {
 			if foundUpstreams != nil {
 				t.Errorf("Test %d: Expected kubernetes controller to not be initialized with upstreams for input '%s'", i, test.input)

plugin/pkg/upstream/upstream.go

@@ -0,0 +1,55 @@
+// Package upstream abstracts a upstream lookups so that plugins
+// can handle them in an unified way.
+package upstream
+
+import (
+	"github.com/miekg/dns"
+
+	"github.com/coredns/coredns/core/dnsserver"
+	"github.com/coredns/coredns/plugin/pkg/dnsutil"
+	"github.com/coredns/coredns/plugin/pkg/nonwriter"
+	"github.com/coredns/coredns/plugin/proxy"
+	"github.com/coredns/coredns/request"
+)
+
+// Upstream is used to resolve CNAME targets
+type Upstream struct {
+	self    bool
+	Forward *proxy.Proxy
+}
+
+// NewUpstream creates a new Upstream for given destination(s)
+func NewUpstream(dests []string) (Upstream, error) {
+	u := Upstream{}
+	if len(dests) == 0 {
+		u.self = true
+		return u, nil
+	}
+	u.self = false
+	ups, err := dnsutil.ParseHostPortOrFile(dests...)
+	if err != nil {
+		return u, err
+	}
+	p := proxy.NewLookup(ups)
+	u.Forward = &p
+	return u, nil
+}
+
+// Lookup routes lookups to Self or Forward
+func (u Upstream) Lookup(state request.Request, name string, typ uint16) (*dns.Msg, error) {
+	if u.self {
+		// lookup via self
+		req := new(dns.Msg)
+		req.SetQuestion(name, typ)
+		state.SizeAndDo(req)
+		nw := nonwriter.New(state.W)
+		state2 := request.Request{W: nw, Req: req}
+		server := state.Context.Value(dnsserver.Key{}).(*dnsserver.Server)
+		server.ServeDNS(state.Context, state2.W, req)
+		return nw.Msg, nil
+	}
+	if u.Forward != nil {
+		return u.Forward.Lookup(state, name, typ)
+	}
+	return &dns.Msg{}, nil
+}

request/request.go

@@ -9,6 +9,7 @@ import (
 	"github.com/coredns/coredns/plugin/pkg/edns"
 
 	"github.com/miekg/dns"
+	"golang.org/x/net/context"
 )
 
 // Request contains some connection state and is useful in plugin.
@@ -19,6 +20,8 @@ type Request struct {
 	// Optional lowercased zone of this query.
 	Zone string
 
+	Context context.Context
+
 	// Cache size after first call to Size or Do.
 	size int
 	do   int // 0: not, 1: true: 2: false