ADD ignoreemptyservice option for kubernetes plugin (#1813)

* ADD: ignoreemptyservice option for kubernetes plugin

* Modify documentation and rename option to add space

* UPD: Add unit tests

* UPD: gofmt

* Add unit test for ignore emptyservice

* gofmt

* xfr tests failed

* Rename emptyservice to empty_service

plugin/kubernetes/README.md

@@ -101,6 +101,9 @@ kubernetes [ZONES...] {
   the query. If **[ZONES...]** is omitted, then fallthrough happens for all zones for which the plugin
   is authoritative. If specific zones are listed (for example `in-addr.arpa` and `ip6.arpa`), then only
   queries for those zones will be subject to fallthrough.
+* `ignore empty_service` return NXDOMAIN for services without any ready endpoint addresses (e.g. ready pods). 
+  This allows the querying pod to continue searching for the service in the search path. 
+  The search path could, for example, include another kubernetes cluster.
 
 ## Health
 

plugin/kubernetes/controller.go

@@ -79,6 +79,7 @@ type dnsControlOpts struct {
 	initPodCache       bool
 	initEndpointsCache bool
 	resyncPeriod       time.Duration
+	ignoreEmptyService bool
 	// Label handling.
 	labelSelector *meta.LabelSelector
 	selector      labels.Selector

plugin/kubernetes/handler_ignore_emptyservice_test.go

+package kubernetes
+
+import (
+	"context"
+	"github.com/coredns/coredns/plugin/pkg/dnstest"
+	"github.com/coredns/coredns/plugin/test"
+	"testing"
+
+	"github.com/miekg/dns"
+)
+
+var dnsEmptyServiceTestCases = []test.Case{
+	// A Service
+	{
+		Qname: "svcempty.testns.svc.cluster.local.", Qtype: dns.TypeA,
+		Rcode: dns.RcodeNameError,
+		Ns: []dns.RR{
+			test.SOA("cluster.local.	300	IN	SOA	ns.dns.cluster.local. hostmaster.cluster.local. 1499347823 7200 1800 86400 60"),
+		},
+	},
+}
+
+func TestServeDNSEmptyService(t *testing.T) {
+
+	k := New([]string{"cluster.local."})
+	k.APIConn = &APIConnServeTest{}
+	k.opts.ignoreEmptyService = true
+	k.Next = test.NextHandler(dns.RcodeSuccess, nil)
+	ctx := context.TODO()
+
+	for i, tc := range dnsEmptyServiceTestCases {
+		r := tc.Msg()
+
+		w := dnstest.NewRecorder(&test.ResponseWriter{})
+
+		_, err := k.ServeDNS(ctx, w, r)
+		if err != tc.Error {
+			t.Errorf("Test %d expected no error, got %v", i, err)
+			return
+		}
+		if tc.Error != nil {
+			continue
+		}
+
+		resp := w.Msg
+		if resp == nil {
+			t.Fatalf("Test %d, got nil message and no error for %q", i, r.Question[0].Name)
+		}
+
+		// Before sorting, make sure that CNAMES do not appear after their target records
+		test.CNAMEOrder(t, resp)
+
+		test.SortAndCheck(t, resp, tc)
+	}
+}

plugin/kubernetes/handler_test.go

@@ -22,6 +22,13 @@ var dnsTestCases = []test.Case{
 			test.A("svc1.testns.svc.cluster.local.	5	IN	A	10.0.0.1"),
 		},
 	},
+	{
+		Qname: "svcempty.testns.svc.cluster.local.", Qtype: dns.TypeA,
+		Rcode: dns.RcodeSuccess,
+		Answer: []dns.RR{
+			test.A("svcempty.testns.svc.cluster.local.	5	IN	A	10.0.0.1"),
+		},
+	},
 	// A Service (wildcard)
 	{
 		Qname: "svc1.*.svc.cluster.local.", Qtype: dns.TypeA,
@@ -36,6 +43,12 @@ var dnsTestCases = []test.Case{
 		Answer: []dns.RR{test.SRV("svc1.testns.svc.cluster.local.	5	IN	SRV	0 100 80 svc1.testns.svc.cluster.local.")},
 		Extra: []dns.RR{test.A("svc1.testns.svc.cluster.local.  5       IN      A       10.0.0.1")},
 	},
+	{
+		Qname: "svcempty.testns.svc.cluster.local.", Qtype: dns.TypeSRV,
+		Rcode: dns.RcodeSuccess,
+		Answer: []dns.RR{test.SRV("svcempty.testns.svc.cluster.local.	5	IN	SRV	0 100 80 svcempty.testns.svc.cluster.local.")},
+		Extra: []dns.RR{test.A("svcempty.testns.svc.cluster.local.  5       IN      A       10.0.0.1")},
+	},
 	{
 		Qname: "svc6.testns.svc.cluster.local.", Qtype: dns.TypeSRV,
 		Rcode: dns.RcodeSuccess,
@@ -49,6 +62,12 @@ var dnsTestCases = []test.Case{
 		Answer: []dns.RR{test.SRV("svc1.*.svc.cluster.local.	5	IN	SRV	0 100 80 svc1.testns.svc.cluster.local.")},
 		Extra: []dns.RR{test.A("svc1.testns.svc.cluster.local.  5       IN      A       10.0.0.1")},
 	},
+	{
+		Qname: "svcempty.*.svc.cluster.local.", Qtype: dns.TypeSRV,
+		Rcode: dns.RcodeSuccess,
+		Answer: []dns.RR{test.SRV("svcempty.*.svc.cluster.local.	5	IN	SRV	0 100 80 svcempty.testns.svc.cluster.local.")},
+		Extra: []dns.RR{test.A("svcempty.testns.svc.cluster.local.  5       IN      A       10.0.0.1")},
+	},
 	// SRV Service (wildcards)
 	{
 		Qname: "*.any.svc1.*.svc.cluster.local.", Qtype: dns.TypeSRV,
@@ -83,6 +102,16 @@ var dnsTestCases = []test.Case{
 			test.A("svc1.testns.svc.cluster.local.	5	IN	A	10.0.0.1"),
 		},
 	},
+	{
+		Qname: "_http._tcp.svcempty.testns.svc.cluster.local.", Qtype: dns.TypeSRV,
+		Rcode: dns.RcodeSuccess,
+		Answer: []dns.RR{
+			test.SRV("_http._tcp.svcempty.testns.svc.cluster.local.	5	IN	SRV	0 100 80 svcempty.testns.svc.cluster.local."),
+		},
+		Extra: []dns.RR{
+			test.A("svcempty.testns.svc.cluster.local.	5	IN	A	10.0.0.1"),
+		},
+	},
 	// A Service (Headless)
 	{
 		Qname: "hdls1.testns.svc.cluster.local.", Qtype: dns.TypeA,
@@ -332,6 +361,21 @@ var svcIndex = map[string][]*api.Service{
 			}},
 		},
 	}},
+	"svcempty.testns": {{
+		ObjectMeta: meta.ObjectMeta{
+			Name:      "svcempty",
+			Namespace: "testns",
+		},
+		Spec: api.ServiceSpec{
+			Type:      api.ServiceTypeClusterIP,
+			ClusterIP: "10.0.0.1",
+			Ports: []api.ServicePort{{
+				Name:     "http",
+				Protocol: "tcp",
+				Port:     80,
+			}},
+		},
+	}},
 	"svc6.testns": {{
 		ObjectMeta: meta.ObjectMeta{
 			Name:      "svc6",
@@ -410,6 +454,24 @@ var epsIndex = map[string][]*api.Endpoints{
 			Namespace: "testns",
 		},
 	}},
+	"svcempty.testns": {{
+		Subsets: []api.EndpointSubset{
+			{
+				Addresses: nil,
+				Ports: []api.EndpointPort{
+					{
+						Port:     80,
+						Protocol: "tcp",
+						Name:     "http",
+					},
+				},
+			},
+		},
+		ObjectMeta: meta.ObjectMeta{
+			Name:      "svcempty",
+			Namespace: "testns",
+		},
+	}},
 	"hdls1.testns": {{
 		Subsets: []api.EndpointSubset{
 			{

plugin/kubernetes/kubernetes.go

@@ -89,7 +89,6 @@ var (
 
 // Services implements the ServiceBackend interface.
 func (k *Kubernetes) Services(state request.Request, exact bool, opt plugin.Options) (svcs []msg.Service, err error) {
-
 	// We're looking again at types, which we've already done in ServeDNS, but there are some types k8s just can't answer.
 	switch state.QType() {
 
@@ -240,7 +239,6 @@ func (k *Kubernetes) getClientConfig() (*rest.Config, error) {
 
 // InitKubeCache initializes a new Kubernetes cache.
 func (k *Kubernetes) InitKubeCache() (err error) {
-
 	config, err := k.getClientConfig()
 	if err != nil {
 		return err
@@ -398,7 +396,6 @@ func (k *Kubernetes) findServices(r recordRequest, zone string) (services []msg.
 	}
 
 	for _, svc := range serviceList {
-
 		if !(match(r.namespace, svc.Namespace) && match(r.service, svc.Name)) {
 			continue
 		}
@@ -409,6 +406,20 @@ func (k *Kubernetes) findServices(r recordRequest, zone string) (services []msg.
 			continue
 		}
 
+		if k.opts.ignoreEmptyService && svc.Spec.ClusterIP != api.ClusterIPNone {
+			// serve NXDOMAIN if no endpoint is able to answer
+			podsCount := 0
+			for _, ep := range endpointsListFunc() {
+				for _, eps := range ep.Subsets {
+					podsCount = podsCount + len(eps.Addresses)
+				}
+			}
+
+			if podsCount == 0 {
+				continue
+			}
+		}
+
 		// Endpoint query or headless service
 		if svc.Spec.ClusterIP == api.ClusterIPNone || r.endpoint != "" {
 			if endpointsList == nil {

plugin/kubernetes/setup.go

@@ -115,6 +115,7 @@ func ParseStanza(c *caddy.Controller) (*Kubernetes, error) {
 
 	opts := dnsControlOpts{
 		initEndpointsCache: true,
+		ignoreEmptyService: false,
 		resyncPeriod:       defaultResyncPeriod,
 	}
 	k8s.opts = opts
@@ -249,10 +250,22 @@ func ParseStanza(c *caddy.Controller) (*Kubernetes, error) {
 				return nil, c.ArgErr()
 			}
 			k8s.opts.initEndpointsCache = false
+		case "ignore":
+			args := c.RemainingArgs()
+			if len(args) > 0 {
+				ignore := args[0]
+				if ignore == "empty_service" {
+					k8s.opts.ignoreEmptyService = true
+					continue
+				} else {
+					return nil, fmt.Errorf("unable to parse ignore value: '%v'", ignore)
+				}
+			}
 		default:
 			return nil, c.Errf("unknown property '%s'", c.Val())
 		}
 	}
+
 	return k8s, nil
 }
 

plugin/kubernetes/setup_test.go

@@ -615,3 +615,73 @@ func TestKubernetesParseNoEndpoints(t *testing.T) {
 		}
 	}
 }
+
+func TestKubernetesParseIgnoreEmptyService(t *testing.T) {
+	tests := []struct {
+		input                 string // Corefile data as string
+		shouldErr             bool   // true if test case is exected to produce an error.
+		expectedErrContent    string // substring from the expected error. Empty for positive cases.
+		expectedEndpointsInit bool
+	}{
+		// valid
+		{
+			`kubernetes coredns.local {
+	ignore empty_service
+}`,
+			false,
+			"",
+			true,
+		},
+		// invalid
+		{
+			`kubernetes coredns.local {
+	ignore ixnay on the endpointsay
+}`,
+			true,
+			"unable to parse ignore value",
+			false,
+		},
+		{
+			`kubernetes coredns.local {
+	ignore empty_service ixnay on the endpointsay
+}`,
+			false,
+			"",
+			true,
+		},
+		// not set
+		{
+			`kubernetes coredns.local {
+}`,
+			false,
+			"",
+			false,
+		},
+	}
+
+	for i, test := range tests {
+		c := caddy.NewTestController("dns", test.input)
+		k8sController, err := kubernetesParse(c)
+
+		if test.shouldErr && err == nil {
+			t.Errorf("Test %d: Expected error, but did not find error for input '%s'. Error was: '%v'", i, test.input, err)
+		}
+
+		if err != nil {
+			if !test.shouldErr {
+				t.Errorf("Test %d: Expected no error but found one for input %s. Error was: %v", i, test.input, err)
+				continue
+			}
+
+			if !strings.Contains(err.Error(), test.expectedErrContent) {
+				t.Errorf("Test %d: Expected error to contain: %v, found error: %v, input: %s", i, test.expectedErrContent, err, test.input)
+			}
+			continue
+		}
+
+		foundIgnoreEmptyService := k8sController.opts.ignoreEmptyService
+		if foundIgnoreEmptyService != test.expectedEndpointsInit {
+			t.Errorf("Test %d: Expected kubernetes controller to be initialized with ignore empty_service '%v'. Instead found ignore empty_service watch '%v' for input '%s'", i, test.expectedEndpointsInit, foundIgnoreEmptyService, test.input)
+		}
+	}
+}