Adding label selector support to Corefile (#208)

* Adding parsing for label selector to Corefile

* Updating comment typo in k8sCorefile

* Adding implementation of label support to filter exposed objects

* Updating TODO list

conf/k8sCorefile

@@ -11,6 +11,13 @@
         template {service}.{namespace}.{zone}
         # Only expose the k8s namespace "demo"
         namespaces demo
+        # Only expose the records for kubernetes objects
+        # that matches this label selector. The label
+        # selector syntax is described in the kubernetes
+        # API documentation: http://kubernetes.io/docs/user-guide/labels/
+        # Example selector below only exposes objects tagged as
+        # "application=nginx" in the staging or qa environments.
+        #labels environment in (staging, qa),application=nginx
     }
     # Perform DNS response caching for the coredns.local zone
     # Cache timeout is provided by the integer in seconds

core/setup/kubernetes.go

@@ -10,6 +10,7 @@ import (
 	"github.com/miekg/coredns/middleware"
 	"github.com/miekg/coredns/middleware/kubernetes"
 	"github.com/miekg/coredns/middleware/kubernetes/nametemplate"
+	unversionedapi "k8s.io/kubernetes/pkg/api/unversioned"
 )
 
 const (
@@ -109,6 +110,20 @@ func kubernetesParse(c *Controller) (kubernetes.Kubernetes, error) {
 						log.Printf("[debug] 'resyncperiod' keyword provided without any duration value.")
 						return kubernetes.Kubernetes{}, c.ArgErr()
 					}
+				case "labels":
+					args := c.RemainingArgs()
+					if len(args) != 0 {
+						labelSelectorString := strings.Join(args, " ")
+						k8s.LabelSelector, err = unversionedapi.ParseToLabelSelector(labelSelectorString)
+						if err != nil {
+							err = errors.New(fmt.Sprintf("Unable to parse label selector. Value provided was '%v'. Error was: %v", labelSelectorString, err))
+							log.Printf("[ERROR] %v", err)
+							return kubernetes.Kubernetes{}, err
+						}
+					} else {
+						log.Printf("[debug] 'labels' keyword provided without any selector value.")
+						return kubernetes.Kubernetes{}, c.ArgErr()
+					}
 				}
 			}
 			return k8s, nil

core/setup/kubernetes_test.go

@@ -4,18 +4,21 @@ import (
 	"strings"
 	"testing"
 	"time"
+
+	unversionedapi "k8s.io/kubernetes/pkg/api/unversioned"
 )
 
 func TestKubernetesParse(t *testing.T) {
 	tests := []struct {
-		description          string        // Human-facing description of test case
-		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.
-		expectedZoneCount    int           // expected count of defined zones.
-		expectedNTValid      bool          // NameTemplate to be initialized and valid
-		expectedNSCount      int           // expected count of namespaces.
-		expectedResyncPeriod time.Duration // expected resync period value
+		description           string        // Human-facing description of test case
+		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.
+		expectedZoneCount     int           // expected count of defined zones.
+		expectedNTValid       bool          // NameTemplate to be initialized and valid
+		expectedNSCount       int           // expected count of namespaces.
+		expectedResyncPeriod  time.Duration // expected resync period value
+		expectedLabelSelector string        // expected label selector value
 	}{
 		// positive
 		{
@@ -27,6 +30,7 @@ func TestKubernetesParse(t *testing.T) {
 			true,
 			0,
 			defaultResyncPeriod,
+			"",
 		},
 		{
 			"kubernetes keyword with multiple zones",
@@ -37,6 +41,7 @@ func TestKubernetesParse(t *testing.T) {
 			true,
 			0,
 			defaultResyncPeriod,
+			"",
 		},
 		{
 			"kubernetes keyword with zone and empty braces",
@@ -48,6 +53,7 @@ func TestKubernetesParse(t *testing.T) {
 			true,
 			0,
 			defaultResyncPeriod,
+			"",
 		},
 		{
 			"endpoint keyword with url",
@@ -60,6 +66,7 @@ func TestKubernetesParse(t *testing.T) {
 			true,
 			0,
 			defaultResyncPeriod,
+			"",
 		},
 		{
 			"template keyword with valid template",
@@ -72,6 +79,7 @@ func TestKubernetesParse(t *testing.T) {
 			true,
 			0,
 			defaultResyncPeriod,
+			"",
 		},
 		{
 			"namespaces keyword with one namespace",
@@ -84,6 +92,7 @@ func TestKubernetesParse(t *testing.T) {
 			true,
 			1,
 			defaultResyncPeriod,
+			"",
 		},
 		{
 			"namespaces keyword with multiple namespaces",
@@ -96,6 +105,7 @@ func TestKubernetesParse(t *testing.T) {
 			true,
 			2,
 			defaultResyncPeriod,
+			"",
 		},
 		{
 			"resync period in seconds",
@@ -108,6 +118,7 @@ func TestKubernetesParse(t *testing.T) {
 			true,
 			0,
 			30 * time.Second,
+			"",
 		},
 		{
 			"resync period in minutes",
@@ -120,6 +131,33 @@ func TestKubernetesParse(t *testing.T) {
 			true,
 			0,
 			15 * time.Minute,
+			"",
+		},
+		{
+			"basic label selector",
+			`kubernetes coredns.local {
+    labels environment=prod
+}`,
+			false,
+			"",
+			1,
+			true,
+			0,
+			defaultResyncPeriod,
+			"environment=prod",
+		},
+		{
+			"multi-label selector",
+			`kubernetes coredns.local {
+    labels environment in (production, staging, qa),application=nginx
+}`,
+			false,
+			"",
+			1,
+			true,
+			0,
+			defaultResyncPeriod,
+			"application=nginx,environment in (production,qa,staging)",
 		},
 		{
 			"fully specified valid config",
@@ -128,6 +166,7 @@ func TestKubernetesParse(t *testing.T) {
 	endpoint http://localhost:8080
 	template {service}.{namespace}.{zone}
 	namespaces demo test
+    labels environment in (production, staging, qa),application=nginx
 }`,
 			false,
 			"",
@@ -135,6 +174,7 @@ func TestKubernetesParse(t *testing.T) {
 			true,
 			2,
 			15 * time.Minute,
+			"application=nginx,environment in (production,qa,staging)",
 		},
 		// negative
 		{
@@ -146,6 +186,7 @@ func TestKubernetesParse(t *testing.T) {
 			false,
 			-1,
 			defaultResyncPeriod,
+			"",
 		},
 		{
 			"kubernetes keyword without a zone",
@@ -156,6 +197,7 @@ func TestKubernetesParse(t *testing.T) {
 			true,
 			0,
 			defaultResyncPeriod,
+			"",
 		},
 		{
 			"endpoint keyword without an endpoint value",
@@ -168,6 +210,7 @@ func TestKubernetesParse(t *testing.T) {
 			true,
 			-1,
 			defaultResyncPeriod,
+			"",
 		},
 		{
 			"template keyword without a template value",
@@ -180,6 +223,7 @@ func TestKubernetesParse(t *testing.T) {
 			false,
 			0,
 			defaultResyncPeriod,
+			"",
 		},
 		{
 			"template keyword with an invalid template value",
@@ -192,6 +236,7 @@ func TestKubernetesParse(t *testing.T) {
 			false,
 			0,
 			defaultResyncPeriod,
+			"",
 		},
 		{
 			"namespace keyword without a namespace value",
@@ -204,6 +249,7 @@ func TestKubernetesParse(t *testing.T) {
 			true,
 			-1,
 			defaultResyncPeriod,
+			"",
 		},
 		{
 			"resyncperiod keyword without a duration value",
@@ -216,6 +262,7 @@ func TestKubernetesParse(t *testing.T) {
 			true,
 			0,
 			0 * time.Minute,
+			"",
 		},
 		{
 			"resync period no units",
@@ -228,6 +275,7 @@ func TestKubernetesParse(t *testing.T) {
 			true,
 			0,
 			0 * time.Second,
+			"",
 		},
 		{
 			"resync period invalid",
@@ -240,6 +288,33 @@ func TestKubernetesParse(t *testing.T) {
 			true,
 			0,
 			0 * time.Second,
+			"",
+		},
+		{
+			"labels with no selector value",
+			`kubernetes coredns.local {
+    labels
+}`,
+			true,
+			"Wrong argument count or unexpected line ending after 'labels'",
+			-1,
+			true,
+			0,
+			0 * time.Second,
+			"",
+		},
+		{
+			"labels with invalid selector value",
+			`kubernetes coredns.local {
+    labels environment in (production, qa
+}`,
+			true,
+			"Unable to parse label selector. Value provided was",
+			-1,
+			true,
+			0,
+			0 * time.Second,
+			"",
 		},
 	}
 
@@ -300,7 +375,15 @@ func TestKubernetesParse(t *testing.T) {
 		//    ResyncPeriod
 		foundResyncPeriod := k8sController.ResyncPeriod
 		if foundResyncPeriod != test.expectedResyncPeriod {
-			t.Errorf("Test %d: Expected kubernetes controller to be initialized with resync period '%s'. Instead found period '%s' for input '%s'", test.expectedResyncPeriod, foundResyncPeriod, test.input)
+			t.Errorf("Test %d: Expected kubernetes controller to be initialized with resync period '%s'. Instead found period '%s' for input '%s'", i, test.expectedResyncPeriod, foundResyncPeriod, test.input)
+		}
+
+		//    Labels
+		if k8sController.LabelSelector != nil {
+			foundLabelSelectorString := unversionedapi.FormatLabelSelector(k8sController.LabelSelector)
+			if foundLabelSelectorString != test.expectedLabelSelector {
+				t.Errorf("Test %d: Expected kubernetes controller to be initialized with label selector '%s'. Instead found selector '%s' for input '%s'", i, test.expectedLabelSelector, foundLabelSelectorString, test.input)
+			}
 		}
 	}
 }

middleware/kubernetes/README.md

@@ -44,6 +44,13 @@ This is the default kubernetes setup, with everything specified in full:
         template {service}.{namespace}.{zone}
         # Only expose the k8s namespace "demo"
         namespaces demo
+        # Only expose the records for kubernetes objects
+        # that matches this label selector. The label
+        # selector syntax is described in the kubernetes
+        # API documentation: http://kubernetes.io/docs/user-guide/labels/
+        # Example selector below only exposes objects tagged as
+        # "application=nginx" in the staging or qa environments.
+        #labels environment in (staging, qa),application=nginx
     }
     # Perform DNS response caching for the coredns.local zone
     # Cache timeout is provided by the integer in seconds
@@ -51,10 +58,13 @@ This is the default kubernetes setup, with everything specified in full:
 }
 ~~~
 
-Notes:
+Defaults:
 * If the `namespaces` keyword is omitted, all kubernetes namespaces are exposed.
 * If the `template` keyword is omitted, the default template of "{service}.{namespace}.{zone}" is used.
 * If the `resyncperiod` keyword is omitted, the default resync period is 5 minutes.
+* The `labels` keyword is only used when filtering of results based on kubernetes label selector syntax
+  is required. The label selector syntax is described in the kubernetes API documentation at: 
+  http://kubernetes.io/docs/user-guide/labels/
 
 ### Basic Setup
 
@@ -191,7 +201,7 @@ mynginx.demo.coredns.local. 0   IN  A   10.0.0.10
 
 ## Implementation Notes/Ideas
 
-### Basic Zone Mapping (implemented)
+### Basic Zone Mapping
 The middleware is configured with a "zone" string. For 
 example: "zone = coredns.local".
 
@@ -200,8 +210,8 @@ to: "myservice.mynamespace.coredns.local".
 
 The middleware should publish an A record for that service and a service record.
 
-Initial implementation just performs the above simple mapping. Subsequent 
-revisions should allow different namespaces to be published under different zones.
+If multiple zone names are specified, the records for kubernetes objects are
+exposed in all listed zones.
 
 For example:
 
@@ -262,11 +272,6 @@ return the IP addresses for all services with "nginx" in the service name.
 
 TBD:
 * How does this relate the the k8s load-balancer configuration?
-* Do wildcards search across namespaces? (Yes)
-* Initial implementation assumes that a namespace maps to the first DNS label
-  below the zone managed by the kubernetes middleware. This assumption may
-  need to be revised. (Template scheme for record names removes this assumption.)
-
 
 ## TODO
 * SkyDNS compatibility/equivalency:
@@ -318,19 +323,19 @@ TBD:
 * Additional features:
 	* Reverse IN-ADDR entries for services. (Is there any value in supporting 
 	  reverse lookup records?) (need tests, functionality should work based on @aledbf's code.)
-	* How to support label specification in Corefile to allow use of labels to 
-	  indicate zone? (Is this even useful?) For example, the following
+	* (done) ~~How to support label specification in Corefile to allow use of labels to 
+	  indicate zone? For example, the following
 	  configuration exposes all services labeled for the "staging" environment
 	  and tenant "customerB" in the zone "customerB.stage.local":
 
 			kubernetes customerB.stage.local {
 				# Use url for k8s API endpoint
 				endpoint http://localhost:8080
-				label "environment" : "staging", "tenant" : "customerB"
+				labels environment in (staging),tenant=customerB
 			}
 
 	  Note: label specification/selection is a killer feature for segmenting
-	  test vs staging vs prod environments.
+	  test vs staging vs prod environments.~~ Need label testing.
 	* Implement IP selection and ordering (internal/external). Related to
 	  wildcards and SkyDNS use of CNAMES.
 	* Flatten service and namespace names to valid DNS characters. (service names

middleware/kubernetes/controller.go

@@ -12,6 +12,7 @@ import (
 	"k8s.io/kubernetes/pkg/client/cache"
 	client "k8s.io/kubernetes/pkg/client/unversioned"
 	"k8s.io/kubernetes/pkg/controller/framework"
+    "k8s.io/kubernetes/pkg/labels"
 	"k8s.io/kubernetes/pkg/runtime"
 	"k8s.io/kubernetes/pkg/watch"
 )
@@ -23,6 +24,8 @@ var (
 type dnsController struct {
 	client *client.Client
 
+    selector *labels.Selector
+
 	endpController *framework.Controller
 	svcController  *framework.Controller
 	nsController   *framework.Controller
@@ -40,68 +43,87 @@ type dnsController struct {
 }
 
 // newDNSController creates a controller for coredns
-func newdnsController(kubeClient *client.Client, resyncPeriod time.Duration) *dnsController {
+func newdnsController(kubeClient *client.Client, resyncPeriod time.Duration, lselector *labels.Selector) *dnsController {
 	dns := dnsController{
 		client: kubeClient,
+        selector: lselector,
 		stopCh: make(chan struct{}),
 	}
 
 	dns.endpLister.Store, dns.endpController = framework.NewInformer(
 		&cache.ListWatch{
-			ListFunc:  endpointsListFunc(dns.client, namespace),
-			WatchFunc: endpointsWatchFunc(dns.client, namespace),
+			ListFunc:  endpointsListFunc(dns.client, namespace, dns.selector),
+			WatchFunc: endpointsWatchFunc(dns.client, namespace, dns.selector),
 		},
 		&api.Endpoints{}, resyncPeriod, framework.ResourceEventHandlerFuncs{})
 
 	dns.svcLister.Store, dns.svcController = framework.NewInformer(
 		&cache.ListWatch{
-			ListFunc:  serviceListFunc(dns.client, namespace),
-			WatchFunc: serviceWatchFunc(dns.client, namespace),
+			ListFunc:  serviceListFunc(dns.client, namespace, dns.selector),
+			WatchFunc: serviceWatchFunc(dns.client, namespace, dns.selector),
 		},
 		&api.Service{}, resyncPeriod, framework.ResourceEventHandlerFuncs{})
 
 	dns.nsLister.Store, dns.nsController = framework.NewInformer(
 		&cache.ListWatch{
-			ListFunc:  namespaceListFunc(dns.client),
-			WatchFunc: namespaceWatchFunc(dns.client),
+			ListFunc:  namespaceListFunc(dns.client, dns.selector),
+			WatchFunc: namespaceWatchFunc(dns.client, dns.selector),
 		},
 		&api.Namespace{}, resyncPeriod, framework.ResourceEventHandlerFuncs{})
 
 	return &dns
 }
 
-func serviceListFunc(c *client.Client, ns string) func(api.ListOptions) (runtime.Object, error) {
+func serviceListFunc(c *client.Client, ns string, s *labels.Selector) func(api.ListOptions) (runtime.Object, error) {
 	return func(opts api.ListOptions) (runtime.Object, error) {
+        if s != nil {
+            opts.LabelSelector = *s
+        }
 		return c.Services(ns).List(opts)
 	}
 }
 
-func serviceWatchFunc(c *client.Client, ns string) func(options api.ListOptions) (watch.Interface, error) {
+func serviceWatchFunc(c *client.Client, ns string, s *labels.Selector) func(options api.ListOptions) (watch.Interface, error) {
 	return func(options api.ListOptions) (watch.Interface, error) {
+        if s != nil {
+            options.LabelSelector = *s
+        }
 		return c.Services(ns).Watch(options)
 	}
 }
 
-func endpointsListFunc(c *client.Client, ns string) func(api.ListOptions) (runtime.Object, error) {
+func endpointsListFunc(c *client.Client, ns string, s *labels.Selector) func(api.ListOptions) (runtime.Object, error) {
 	return func(opts api.ListOptions) (runtime.Object, error) {
+        if s != nil {
+            opts.LabelSelector = *s
+        }
 		return c.Endpoints(ns).List(opts)
 	}
 }
 
-func endpointsWatchFunc(c *client.Client, ns string) func(options api.ListOptions) (watch.Interface, error) {
+func endpointsWatchFunc(c *client.Client, ns string, s *labels.Selector) func(options api.ListOptions) (watch.Interface, error) {
 	return func(options api.ListOptions) (watch.Interface, error) {
+        if s != nil {
+            options.LabelSelector = *s
+        }
 		return c.Endpoints(ns).Watch(options)
 	}
 }
 
-func namespaceListFunc(c *client.Client) func(api.ListOptions) (runtime.Object, error) {
+func namespaceListFunc(c *client.Client, s *labels.Selector) func(api.ListOptions) (runtime.Object, error) {
 	return func(opts api.ListOptions) (runtime.Object, error) {
+        if s != nil {
+            opts.LabelSelector = *s
+        }
 		return c.Namespaces().List(opts)
 	}
 }
 
-func namespaceWatchFunc(c *client.Client) func(options api.ListOptions) (watch.Interface, error) {
+func namespaceWatchFunc(c *client.Client, s *labels.Selector) func(options api.ListOptions) (watch.Interface, error) {
 	return func(options api.ListOptions) (watch.Interface, error) {
+        if s != nil {
+            options.LabelSelector = *s
+        }
 		return c.Namespaces().Watch(options)
 	}
 }
@@ -149,7 +171,6 @@ func (dns *dnsController) GetNamespaceList() *api.NamespaceList {
 }
 
 func (dns *dnsController) GetServiceList() *api.ServiceList {
-	log.Printf("[debug] here in GetServiceList")
 	svcList, err := dns.svcLister.List()
 	if err != nil {
 		return &api.ServiceList{}

middleware/kubernetes/kubernetes.go

@@ -15,20 +15,24 @@ import (
 
 	"github.com/miekg/dns"
 	"k8s.io/kubernetes/pkg/api"
-	"k8s.io/kubernetes/pkg/client/unversioned"
+	unversionedapi "k8s.io/kubernetes/pkg/api/unversioned"
+	"k8s.io/kubernetes/pkg/labels"
+	unversionedclient "k8s.io/kubernetes/pkg/client/unversioned"
 	"k8s.io/kubernetes/pkg/client/unversioned/clientcmd"
 	clientcmdapi "k8s.io/kubernetes/pkg/client/unversioned/clientcmd/api"
 )
 
 type Kubernetes struct {
-	Next         middleware.Handler
-	Zones        []string
-	Proxy        proxy.Proxy // Proxy for looking up names during the resolution process
-	APIEndpoint  string
-	APIConn      *dnsController
-	ResyncPeriod time.Duration
-	NameTemplate *nametemplate.NameTemplate
-	Namespaces   []string
+	Next          middleware.Handler
+	Zones         []string
+	Proxy         proxy.Proxy // Proxy for looking up names during the resolution process
+	APIEndpoint   string
+	APIConn       *dnsController
+	ResyncPeriod  time.Duration
+	NameTemplate  *nametemplate.NameTemplate
+	Namespaces    []string
+	LabelSelector *unversionedapi.LabelSelector
+	Selector      *labels.Selector 
 }
 
 func (g *Kubernetes) StartKubeCache() error {
@@ -45,14 +49,26 @@ func (g *Kubernetes) StartKubeCache() error {
 		log.Printf("[debug] error connecting to the client: %v", err)
 		return err
 	}
-	kubeClient, err := unversioned.New(config)
+	kubeClient, err := unversionedclient.New(config)
 
 	if err != nil {
 		log.Printf("[ERROR] Failed to create kubernetes notification controller: %v", err)
 		return err
 	}
+	if g.LabelSelector == nil {
+		log.Printf("[INFO] Kubernetes middleware configured without a label selector. No label-based filtering will be operformed.")
+	} else {
+        var selector labels.Selector
+		selector, err = unversionedapi.LabelSelectorAsSelector(g.LabelSelector)
+        g.Selector = &selector
+        if err != nil {
+            log.Printf("[ERROR] Unable to create Selector for LabelSelector '%s'.Error was: %s", g.LabelSelector, err)
+            return err
+        }
+		log.Printf("[INFO] Kubernetes middleware configured with the label selector '%s'. Only kubernetes objects matching this label selector will be exposed.", unversionedapi.FormatLabelSelector(g.LabelSelector))
+	}
 	log.Printf("[debug] Starting kubernetes middleware with k8s API resync period: %s", g.ResyncPeriod)
-	g.APIConn = newdnsController(kubeClient, g.ResyncPeriod)
+	g.APIConn = newdnsController(kubeClient, g.ResyncPeriod, g.Selector)
 
 	go g.APIConn.Run()